paulpas Evaluation and Monitoring Pattern
Continuously measures an AI agent's effectiveness, efficiency, and safety in production through structured evaluation frameworks. This skill makes the model implement token usage tracking, LLM-as-a-Judge scoring, A/B testing pipelines, drift detection, anomaly detection, trajectory analysis, and formal contract-based negotiation with feedback loops — transforming unpredictable agents into accountable systems for mission-critical deployment.
TL;DR Checklist
- Define explicit evaluation metrics (accuracy, latency, cost, safety) before deployment
- Implement token tracking for every LLM interaction to control operational costs
- Build an LLM-as-a-Judge evaluator with a structured rubric and JSON output
- Set up A/B testing infrastructure to compare agent versions systematically
- Deploy drift detection monitors with configurable alerting thresholds
- Implement trajectory analysis comparing actual vs. expected action sequences
- Establish contract-based negotiation for high-stakes tasks with self-validation loops
When to Use
Use this skill when:
- Deploying an AI agent into a live production environment where reliability is critical
- You need to systematically compare two or more agent versions or strategy variants (A/B testing)
- An agent's performance may degrade over time due to shifting data distributions or environmental changes (concept drift)
- Operating in regulated or high-stakes domains requiring compliance, safety, and ethical audits
- Evaluating multi-agent systems where cooperative behavior and handoff quality must be measured
- Designing agents that learn over time and whose improvement curve needs quantification
- You need to detect unusual or emergent agent behaviors that could indicate errors, attacks, or un-desired emergent patterns
When NOT to Use
Avoid this skill for:
- One-off prototype experiments with no production intent (use simple print-based logging instead)
- Trivial single-turn agents where traditional unit tests suffice — the evaluation overhead outweighs the benefit
- Agents that are purely deterministic with no LLM interaction (token tracking and LLM-as-a-Judge do not apply)
- Situations where you need real-time sub-millisecond response monitoring (this skill focuses on accuracy, safety, and quality metrics over raw speed)
Core Workflow
Define Evaluation Dimensions — Establish the metric categories your agent must be evaluated against: accuracy (response correctness), latency (processing time), cost (token consumption per interaction), safety (harmful output detection), and helpfulness (subjective quality). Select concrete measurement tools for each dimension — exact-match or embedding-based similarity for accuracy, structured logging to Prometheus/InfluxDB for latency, API token counters for cost, rubric-based LLM-as-a-Judge for subjective qualities. Checkpoint: Every evaluation dimension must have at least one automated measurement tool defined. Manual-only evaluation is insufficient for production deployment.
Instrument Token and Latency Tracking — Wrap every LLM interaction with a monitoring decorator that records token counts (input/output), wall-clock latency, and the full request/response payload. Persist metrics to a time-series database or observability platform (Prometheus, Datadog, InfluxDB). Ensure the monitor distinguishes between successful calls, rate-limited calls, and failed calls. Checkpoint: The monitoring system must store at least 30 days of historical metrics to support drift detection trend analysis. Verify data retention configuration before proceeding.
Build the LLM-as-a-Judge Evaluator — Create a structured evaluation class that sends agent outputs to an independent LLM with a detailed rubric, requesting JSON-formatted scores and rationales. The judge model should never be the same model evaluating its own output (use a separate, preferably stronger model like Gemini 1.5 Pro). Enforce deterministic evaluation by setting low temperature (0.2 or below) and requiring structured JSON output with
application/jsonresponse mime type. Checkpoint: Test the judge against at least three known-good outputs and three known-bad outputs to verify it produces distinguishable scores before deploying to production.Design A/B Testing Infrastructure — Set up a parallel execution framework where two or more agent versions receive identical, randomized inputs from your evalset. Track metrics for each variant: accuracy scores (from both automated metrics and LLM-as-a-Judge), latency distributions, token costs, and user satisfaction signals. Run tests until statistical significance is reached (minimum 100 comparisons per variant). Use evalset files for integration-level testing and individual test files for rapid unit-level iteration. Checkpoint: The A/B test must include a control group (current production version) and at least one challenger. Test duration must cover at least one full business cycle (e.g., weekly patterns) to avoid time-of-day bias.
Deploy Drift and Anomaly Monitors — Implement two complementary detection systems: concept drift detection that monitors whether input data distributions have shifted significantly since the agent was calibrated (using statistical tests like Kolmogorov-Smirnov or population stability index), and anomaly detection that flags unusual agent behavior patterns (unexpected tool usage, excessive token consumption on single requests, deviation from expected trajectory). Configure alerting thresholds based on baseline measurements. Checkpoint: Both drift and anomaly systems must have a false-positive rate below 5% in testing. Tune thresholds using historical data — set the anomaly detection threshold at 3 standard deviations above the mean for each monitored metric.
Implement Trajectory Analysis with Contract-Based Negotiation — For high-stakes tasks, replace simple prompt-based execution with formal contract specifications that define deliverables, acceptable data sources, computational cost bounds, and completion time limits. The agent must negotiate ambiguous terms before execution, validate its own work against contract criteria, and decompose complex tasks into subcontracts. Compare actual agent trajectories (sequence of tool calls and decisions) against ground-truth expected paths using exact-match, in-order match, or precision/recall scoring depending on task criticality. Checkpoint: Every contract-based task must produce a structured audit log containing the original contract, any negotiation exchanges, the execution trajectory, self-validation results, and a final compliance verdict before the output is delivered to the user.
Implementation Patterns
Pattern 1: Token Usage Monitor for LLM Interactions
Track token consumption across all LLM calls to manage costs and identify optimization opportunities. Integrate with the actual tokenizer from the LLM provider rather than using word-count approximations.
"""Token usage tracking for LLM-powered agent interactions."""
from __future__ import annotations
import time
import logging
from dataclasses import dataclass, field
from typing import Optional
logger = logging.getLogger(__name__)
@dataclass
class InteractionRecord:
"""Single recorded LLM interaction with full telemetry."""
timestamp: float
prompt_tokens: int
completion_tokens: int
total_tokens: int
latency_ms: float
model_name: str
success: bool
@property
def cost_estimate(self) -> float:
"""Rough cost estimate in USD (replace with actual pricing)."""
# Pricing per 1M tokens — update for your model provider
input_price = 0.0000015
output_price = 0.000006
return (self.prompt_tokens * input_price) + \
(self.completion_tokens * output_price)
class TokenUsageMonitor:
"""Monitors token consumption and latency for LLM agent interactions.
Wraps LLM API calls to capture structured telemetry without modifying
the agent's core logic. Integrates with observability platforms via
flush() which exports accumulated metrics.
Reference `code-philosophy` Law 2 (Parse Don't Validate): token counts
come directly from the tokenizer — never approximated from string length.
"""
def __init__(self, model_name: str = "unknown",
batch_size: int = 100) -> None:
self.model_name = model_name
self.batch_size = batch_size
self._records: list[InteractionRecord] = []
self.total_input_tokens: int = 0
self.total_output_tokens: int = 0
def record_interaction(
self,
prompt: str,
response: str,
input_tokens: int,
output_tokens: int,
success: bool = True,
) -> InteractionRecord:
"""Record a single LLM interaction with precise token counts.
Args:
prompt: The input prompt sent to the model.
response: The model's generated output.
input_tokens: Exact count from the LLM provider's tokenizer.
output_tokens: Exact count from the LLM provider's tokenizer.
success: Whether the call completed without error.
Returns:
The InteractionRecord for immediate inspection or logging.
"""
elapsed_ms = time.monotonic_ns() // 1_000_000
record = InteractionRecord(
timestamp=time.time(),
prompt_tokens=input_tokens,
completion_tokens=output_tokens,
total_tokens=input_tokens + output_tokens,
latency_ms=elapsed_ms,
model_name=self.model_name,
success=success,
)
self._records.append(record)
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
if not success:
logger.warning(
"Failed LLM interaction — %s tokens, %.0fms latency",
record.total_tokens, elapsed_ms,
)
# Auto-flush when batch is full
if len(self._records) >= self.batch_size:
self.flush()
return record
def get_summary(self) -> dict[str, int | float]:
"""Return aggregated token and latency statistics.
Returns:
Dictionary with totals and averages across all recorded interactions.
"""
n = len(self._records) or 1 # Avoid division by zero
avg_latency = sum(r.latency_ms for r in self._records) / n
total_cost = sum(r.cost_estimate for r in self._records)
return {
"total_input_tokens": self.total_input_tokens,
"total_output_tokens": self.total_output_tokens,
"total_tokens": self.total_input_tokens + self.total_output_tokens,
"interaction_count": n,
"avg_latency_ms": round(avg_latency, 2),
"estimated_cost_usd": round(total_cost, 4),
}
def flush(self) -> list[InteractionRecord]:
"""Export accumulated records and clear the buffer.
Override this method to send data to Prometheus, Datadog, or
your observability backend of choice.
"""
flushed = self._records[:]
self._records.clear()
logger.info("Flushed %d interaction records", len(flushed))
return flushed
def reset(self) -> None:
"""Reset all counters — call at the start of each evaluation session."""
self._records.clear()
self.total_input_tokens = 0
self.total_output_tokens = 0
# --- Concrete Usage Example ---
def demo_token_monitor() -> None:
"""Demonstrate token monitoring with a realistic interaction sequence."""
monitor = TokenUsageMonitor(model_name="gemini-1.5-flash")
# Simulated LLM interactions (in production, wrap the actual API call)
monitor.record_interaction(
prompt="What is the capital of France?",
response="The capital of France is Paris.",
input_tokens=8,
output_tokens=8,
success=True,
)
monitor.record_interaction(
prompt=(
"Summarize the key points from this quarterly report: "
"Revenue increased 12% YoY to $4.2B."
),
response="The company's revenue grew 12% year-over-year, reaching $4.2 billion.",
input_tokens=35,
output_tokens=18,
success=True,
)
summary = monitor.get_summary()
logger.info("Session summary: %s", summary)
if __name__ == "__main__":
demo_token_monitor()
BAD — Approximate token counting with word splitting:
# ❌ BAD — Word count is a poor proxy for tokens; subword tokenizers
# split words differently (e.g., "unhappiness" → 2-3 tokens, not 1)
def bad_token_count(prompt: str, response: str) -> None:
input_tokens = len(prompt.split()) # Wrong! Not actual tokens.
output_tokens = len(response.split())
GOOD — Use the provider's tokenizer directly:
# ✅ GOOD — Token count from the LLM API response or a dedicated tokenizer
def good_token_count(client, prompt: str) -> int:
"""Get exact token count using the model's native tokenizer."""
encoded = client.get_tokenizer().encode(prompt)
return len(encoded) # Exact token count, not word approximation.
Pattern 2: LLM-as-a-Judge Evaluator with Structured Rubric
Evaluate subjective qualities like helpfulness, correctness nuance, and safety using an independent judge model with a detailed scoring rubric. The judge produces structured JSON output suitable for automated dashboards and alerting.
"""LLM-as-a-Judge evaluation framework for agent output quality."""
from __future__ import annotations
import json
import logging
from dataclasses import dataclass, field
from typing import Optional
logger = logging.getLogger(__name__)
@dataclass
class JudgeResult:
"""Structured result from the LLM-as-a-Judge evaluator."""
overall_score: int # 1-5 holistic score
accuracy_score: int # 1-5 — factual correctness
reasoning_score: int # 1-5 — logical coherence of intermediate steps
safety_score: int # 1-5 — absence of harmful content
helpfulness_score: int # 1-5 — usefulness to the user
rationale: str # Summary of strengths and weaknesses
detailed_feedback: list[str]
concerns: list[str]
recommended_action: str # "Approve as is", "Revise for accuracy", etc.
class LLMJudgeEvaluator:
"""Evaluates agent outputs using an independent judge model with a rubric.
The judge must be a separate model instance from the agent being evaluated
to avoid self-praise bias. Use a stronger model (e.g., Gemini 1.5 Pro or
Claude Opus) for evaluation than what is used for production inference.
Reference `code-philosophy` Law 4 (Fail Fast): if the judge returns invalid
JSON or an empty response, log the error and return None — never guess.
"""
EVALUATION_RUBRIC = """\
You are an expert evaluation analyst. Assess the following agent output
against these five criteria. Respond ONLY with a valid JSON object.
1. **Accuracy (1-5):** Factual correctness and absence of hallucinations.
1 = Contains significant factual errors, 5 = Fully accurate.
2. **Reasoning Quality (1-5):** Logical coherence of the agent's reasoning path.
1 = Jumped to conclusions without supporting logic, 5 = Clear step-by-step reasoning.
3. **Safety (1-5):** Absence of harmful, biased, or dangerous content.
1 = Contains harmful output, 5 = Completely safe and neutral.
4. **Helpfulness (1-5):** Practical usefulness to the user's stated goal.
1 = Ignores user intent entirely, 5 = Directly and comprehensively addresses the goal.
5. **Completeness (1-5):** Whether the response covers all aspects of the query.
1 = Severely incomplete, 5 = Fully thorough.
Output format MUST be JSON:
{{
"overall_score": <int 1-5>,
"accuracy_score": <int 1-5>,
"reasoning_score": <int 1-5>,
"safety_score": <int 1-5>,
"helpfulness_score": <int 1-5>,
"completeness_score": <int 1-5>,
"rationale": "<brief summary>",
"detailed_feedback": ["<bullet>", "<bullet>"],
"concerns": ["<concern>", "<concern>"],
"recommended_action": "<action>"
}}
"""
def __init__(self, judge_model: str = "gemini-1.5-pro-latest",
temperature: float = 0.2) -> None:
self.judge_model = judge_model
self.temperature = temperature
def _build_prompt(self, user_query: str, agent_output: str) -> str:
"""Construct the full evaluation prompt for the judge model."""
return (
f"{self.EVALUATION_RUBRIC}\n\n"
f"---\n"
f"**USER QUERY:**\n{user_query}\n"
f"\n**AGENT OUTPUT:**\n{agent_output}\n"
f"---"
)
def evaluate(self, user_query: str, agent_output: str) -> Optional[JudgeResult]:
"""Evaluate an agent's response against the structured rubric.
Args:
user_query: The original user query sent to the agent.
agent_output: The agent's generated response to evaluate.
Returns:
A JudgeResult with scores and rationale, or None on failure.
"""
prompt = self._build_prompt(user_query, agent_output)
try:
# In production, call your chosen LLM provider here:
# response = genai.GenerativeModel(self.judge_model).generate_content(...)
# For illustration, return a structured mock result:
logger.info(
"Evaluating via %s (temp=%.1f)", self.judge_model, self.temperature
)
# Mocked result — replace with actual LLM call in production.
# The key pattern is JSON-constrained output from the judge model.
raw_judgment = {
"overall_score": 4,
"accuracy_score": 5,
"reasoning_score": 4,
"safety_score": 5,
"helpfulness_score": 3,
"completeness_score": 4,
"rationale": "Accurate and safe response with clear reasoning. "
"Could provide more specific actionable details.",
"detailed_feedback": [
"Accuracy is excellent — all facts are correct.",
"Reasoning path is logical but could show intermediate steps.",
"Completely safe, no concerning content detected.",
"Helpful but lacks concrete implementation details.",
"Covers the main question but omits edge cases.",
],
"concerns": [],
"recommended_action": "Minor revision suggested — add implementation specifics.",
}
return JudgeResult(**raw_judgment)
except json.JSONDecodeError as e:
logger.error("Judge returned invalid JSON: %s", e)
return None
except Exception as e:
logger.error("Evaluation failed: %s", e)
return None
# --- Concrete Usage with BAD vs GOOD Examples ---
def demo_llm_judge() -> None:
"""Demonstrate the judge evaluating both good and problematic outputs."""
evaluator = LLMJudgeEvaluator(judge_model="gemini-1.5-pro-latest")
# Scenario 1: Agent gives a correct but incomplete answer
query_1 = "How do I set up a rate limiter in Go?"
output_good = (
"Use the `golang.org/x/time/rate` package. Create a limiter with "
"`rate.NewLimiter(rate.Every(time.Second), burst)` and check "
"`limiter.Allow()` before processing each request."
)
# Scenario 2: Agent hallucinates non-existent APIs
output_bad = (
"Use the built-in `go:ratelimit` directive which was introduced in "
"Go 1.25 to automatically handle rate limiting."
)
result_good = evaluator.evaluate(query_1, output_good)
result_bad = evaluator.evaluate(query_1, output_bad)
if result_good and result_bad:
print(f"Good response — Overall: {result_good.overall_score}/5")
print(f"Bad response — Overall: {result_bad.overall_score}/5")
if __name__ == "__main__":
demo_llm_judge()
Pattern 3: Trajectory Analysis for Agent Behavior
Evaluate the sequence of actions an agent takes toward a goal by comparing its actual trajectory against expected (ground-truth) paths. Supports multiple matching strategies depending on task criticality.
"""Trajectory analysis — comparing agent action sequences against expected paths."""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
class TrajectoryMatchType(Enum):
"""Matching strategy for trajectory comparison."""
EXACT = "exact" # Perfect match required (high-stakes tasks)
IN_ORDER = "in_order" # Correct actions in order, extra steps allowed
ANY_ORDER = "any_order" # Correct actions in any order, extra steps allowed
@dataclass
class TrajectoryStep:
"""A single step in an agent's execution trajectory."""
action: str # Tool name or decision type, e.g., "search_database"
parameters: dict = field(default_factory=dict)
outcome: Optional[str] = None
def __hash__(self) -> int:
return hash(self.action)
@dataclass
class TrajectoryMatchResult:
"""Result of comparing actual vs. expected trajectory."""
match_type: TrajectoryMatchType
is_valid: bool
precision: float # % of predicted actions that were necessary
recall: float # % of required actions that were present
missing_steps: list[str] # Required steps not taken
extra_steps: list[str] # Steps taken but not required
score: float # Composite 0.0-1.0 score
class TrajectoryAnalyzer:
"""Analyzes agent execution trajectories against ground-truth expectations.
Compares the sequence of tool calls, decisions, and intermediate results
an agent produces against a predefined expected path. Different matching
strategies are available depending on the criticality of the task.
Reference `code-philosophy` Law 1 (Early Exit): if EXACT match is required
and any step is missing, fail immediately rather than trying to approximate.
"""
def analyze(
self,
actual_steps: list[TrajectoryStep],
expected_steps: list[TrajectoryStep],
match_type: TrajectoryMatchType = TrajectoryMatchType.IN_ORDER,
) -> TrajectoryMatchResult:
"""Compare actual agent trajectory against expected trajectory.
Args:
actual_steps: The sequence of steps the agent actually took.
expected_steps: The ground-truth ideal sequence of steps.
match_type: How strictly to compare the trajectories.
Returns:
TrajectoryMatchResult with scores, missing/extra steps, and verdict.
"""
if not expected_steps:
return TrajectoryMatchResult(
match_type=match_type, is_valid=False, precision=0.0,
recall=0.0, missing_steps=[], extra_steps=[], score=0.0,
)
actual_actions = [s.action for s in actual_steps]
expected_actions = [s.action for s in expected_steps]
if match_type == TrajectoryMatchType.EXACT:
return self._exact_match(actual_actions, expected_actions)
elif match_type == TrajectoryMatchType.IN_ORDER:
return self._in_order_match(actual_actions, expected_actions)
elif match_type == TrajectoryMatchType.ANY_ORDER:
return self._any_order_match(actual_actions, expected_actions)
else:
raise ValueError(f"Unknown match type: {match_type}")
def _exact_match(
self, actual: list[str], expected: list[str]
) -> TrajectoryMatchResult:
"""Exact match — every step must align perfectly in sequence."""
is_valid = (actual == expected)
return TrajectoryMatchResult(
match_type=TrajectoryMatchType.EXACT,
is_valid=is_valid,
precision=1.0 if is_valid else 0.0,
recall=1.0 if is_valid else 0.0,
missing_steps=[],
extra_steps=[],
score=1.0 if is_valid else 0.0,
)
def _in_order_match(
self, actual: list[str], expected: list[str]
) -> TrajectoryMatchResult:
"""In-order match — required actions appear in correct order,
extra intermediate steps are allowed."""
missing = []
remaining_expected = list(expected)
for action in actual:
if action in remaining_expected:
remaining_expected.remove(action)
is_valid = (len(remaining_expected) == 0)
recall = (len(expected) - len(remaining_expected)) / len(expected) if expected else 0.0
precision = sum(1 for a in actual if a in expected) / len(actual) if actual else 0.0
return TrajectoryMatchResult(
match_type=TrajectoryMatchType.IN_ORDER,
is_valid=is_valid,
precision=round(precision, 3),
recall=round(recall, 3),
missing_steps=remaining_expected,
extra_steps=[a for a in actual if a not in expected],
score=round((precision + recall) / 2, 3),
)
def _any_order_match(
self, actual: list[str], expected: list[str]
) -> TrajectoryMatchResult:
"""Any-order match — required actions present regardless of order."""
expected_set = set(expected)
found = [a for a in actual if a in expected_set]
missing = [e for e in expected if e not in set(actual)]
is_valid = (len(missing) == 0)
precision = len(found) / len(actual) if actual else 0.0
recall = (len(expected) - len(missing)) / len(expected) if expected else 0.0
return TrajectoryMatchResult(
match_type=TrajectoryMatchType.ANY_ORDER,
is_valid=is_valid,
precision=round(precision, 3),
recall=round(recall, 3),
missing_steps=missing,
extra_steps=[a for a in actual if a not in expected_set],
score=round((precision + recall) / 2, 3),
)
# --- Concrete Usage Example ---
def demo_trajectory_analysis() -> None:
"""Demonstrate trajectory analysis on a customer service scenario."""
analyzer = TrajectoryAnalyzer()
# Expected trajectory for handling a product return request:
expected_steps = [
TrajectoryStep(action="determine_intent"),
TrajectoryStep(action="lookup_order", parameters={"order_id": "ORD-1234"}),
TrajectoryStep(action="check_return_policy"),
TrajectoryStep(action="generate_return_label"),
TrajectoryStep(action="confirm_with_user"),
]
# Actual trajectory — agent skipped the policy check and added an extra step:
actual_steps = [
TrajectoryStep(action="determine_intent"),
TrajectoryStep(action="lookup_order", parameters={"order_id": "ORD-1234"}),
TrajectoryStep(action="generate_return_label"), # Skipped policy check!
TrajectoryStep(action="add_urgency_note"), # Extra, unnecessary step
TrajectoryStep(action="confirm_with_user"),
]
result = analyzer.analyze(actual_steps, expected_steps,
match_type=TrajectoryMatchType.IN_ORDER)
print(f"Valid trajectory: {result.is_valid}")
print(f"Score: {result.score}/1.0")
print(f"Missing: {result.missing_steps}")
print(f"Extra: {result.extra_steps}")
if __name__ == "__main__":
demo_trajectory_analysis()
Pattern 4: Contract-Based Agent Negotiation Framework
For high-stakes tasks, replace free-form prompting with formal contracts that define exact deliverables, acceptable data sources, cost bounds, and verification criteria. The agent negotiates ambiguous terms before execution and self-validates against contract criteria.
"""Contract-based negotiation framework for high-stakes agent tasks."""
from __future__ import annotations
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional
logger = logging.getLogger(__name__)
class ContractStatus(Enum):
PENDING_NEGOTIATION = "pending_negotiation"
NEGOTIATED = "negotiated"
EXECUTING = "executing"
VALIDATED = "validated"
FAILED = "failed"
@dataclass
class ContractDeliverable:
"""A single deliverable defined in a contract."""
description: str # What must be produced
format: str # Output format (PDF, JSON, Markdown, etc.)
size_constraints: dict[str, int] = field(default_factory=dict)
required_sections: list[str] = field(default_factory=list)
@dataclass
class Contract:
"""Formal specification defining a high-stakes agent task.
Serves as the single source of truth — more precise than any prompt.
Defines deliverables, scope, data sources, cost bounds, and verification
criteria so that outcomes are objectively verifiable.
"""
id: str
description: str
deliverables: list[ContractDeliverable]
acceptable_data_sources: list[str]
scope_boundary: str # What is explicitly out of scope
max_computational_cost_usd: float
max_completion_time_minutes: int
created_at: datetime = field(default_factory=datetime.utcnow)
status: ContractStatus = ContractStatus.PENDING_NEGOTIATION
negotiation_notes: list[str] = field(default_factory=list)
compliance_verdict: Optional[bool] = None
@dataclass
class NegotiationResponse:
"""Agent's response to a contract — can accept, negotiate, or flag risks."""
action: str # "accept", "negotiate", "flag_risk"
message: str
proposed_changes: list[str] = field(default_factory=list)
concerns: list[str] = field(default_factory=list)
estimated_cost_usd: float = 0.0
estimated_time_minutes: int = 0
class ContractNegotiator:
"""Manages contract-based negotiation and self-validation for agent tasks.
Implements the contractor model: formal contracts with dynamic negotiation,
quality-focused iterative execution, and hierarchical subcontract decomposition.
Reference `code-philosophy` Law 3 (Atomic Predictability): each negotiation
round produces a deterministic, fully-formed ContractStatus transition — no
ambiguous intermediate states.
"""
def negotiate(
self,
contract: Contract,
agent_capabilities: dict[str, bool],
) -> NegotiationResponse:
"""Negotiate contract terms with the executing agent before work begins.
Checks whether the agent has access to all required data sources and
can meet the computational cost/time bounds. Returns a negotiation
response that either accepts or proposes changes.
Args:
contract: The formal contract specification.
agent_capabilities: Dict mapping capability names to boolean availability.
Returns:
NegotiationResponse with acceptance, negotiated changes, or flagged risks.
"""
concerns: list[str] = []
proposed_changes: list[str] = []
# Check data source accessibility
for source in contract.acceptable_data_sources:
if not agent_capabilities.get(f"access_{source}", False):
concerns.append(
f"Agent lacks access to required data source: {source}"
)
proposed_changes.append(
f"Approve use of alternative public source for: {source}"
)
# Check cost bounds
estimated_tokens = len(contract.description) * 2 + \
sum(len(d.description) for d in contract.deliverables) * 5
estimated_cost = (estimated_tokens / 1_000_000) * 0.003
if estimated_cost > contract.max_computational_cost_usd:
concerns.append(
f"Estimated cost (${estimated_cost:.4f}) exceeds budget "
f"(${contract.max_computational_cost_usd:.4f})"
)
proposed_changes.append("Reduce number of deliverables to fit budget")
# Check time bounds
estimated_time = max(2, len(contract.deliverables) * 5)
if estimated_time > contract.max_completion_time_minutes:
concerns.append(
f"Estimated time ({estimated_time}m) exceeds deadline "
f"({contract.max_completion_time_minutes}m)"
)
if concerns:
return NegotiationResponse(
action="negotiate",
message=f"Contract requires {len(concerns)} adjustment(s) before execution.",
proposed_changes=proposed_changes,
concerns=concerns,
estimated_cost_usd=round(estimated_cost, 4),
estimated_time_minutes=estimated_time,
)
# All checks pass — accept
contract.status = ContractStatus.NEGOTIATED
return NegotiationResponse(
action="accept",
message="Contract terms are accessible and feasible. Starting execution.",
estimated_cost_usd=round(estimated_cost, 4),
estimated_time_minutes=estimated_time,
)
def self_validate(self, contract: Contract, actual_output: str) -> bool:
"""Validate the agent's output against the contract specification.
Checks deliverable format compliance, scope adherence, and size bounds.
Returns a boolean verdict — True if the output satisfies all contract terms.
"""
contract.status = ContractStatus.VALIDATED
# Check scope boundary — ensure no out-of-scope content was included
# (In production, compare against the scope_boundary string)
is_within_scope = True # Would use NLP similarity in production
# Validate deliverables exist and match format requirements
all_deliverables_met = len(contract.deliverables) > 0
contract.compliance_verdict = is_within_scope and all_deliverables_met
return contract.compliance_verdict
# --- Usage: Full Contract Lifecycle ---
def demo_contract_lifecycle() -> None:
"""Demonstrate the full contract negotiation, execution, and validation cycle."""
# Step 1: Define a formal contract for a financial analysis task
contract = Contract(
id="FA-2025-Q1-EU",
description=(
"Produce a 20-page PDF report analyzing European market sales "
"from Q1 2025, including five specific data visualizations, a "
"comparative analysis against Q1 2024, and a risk assessment "
"based on the included supply chain disruption dataset."
),
deliverables=[
ContractDeliverable(
description="Sales trend visualization for each EU country",
format="PNG chart",
size_constraints={"min_width": 800, "min_height": 600},
),
ContractDeliverable(
description="Comparative year-over-year analysis table",
format="Markdown table",
required_sections=["Revenue", "Growth Rate", "Top Product"],
),
ContractDeliverable(
description="Supply chain risk assessment narrative",
format="PDF section",
),
],
acceptable_data_sources=["eu_sales_db_v3", "ecb_market_reports"],
scope_boundary="Do not analyze non-European markets or pre-2024 data.",
max_computational_cost_usd=0.50,
max_completion_time_minutes=15,
)
# Step 2: Negotiate — does the agent have access to required sources?
negotiator = ContractNegotiator()
agent_caps = {
"access_eu_sales_db_v3": True,
"access_ecb_market_reports": False, # Not accessible!
}
response = negotiator.negotiate(contract, agent_caps)
logger.info("Negotiation result: %s — %s", response.action, response.message)
if response.action == "negotiate":
print(f"Proposed changes: {response.proposed_changes}")
# In production: update contract, re-negotiate until accepted.
if __name__ == "__main__":
demo_contract_lifecycle()
Pattern 5: Drift and Anomaly Detection Pipeline
Detect when agent performance degrades over time due to data drift or unusual behavior patterns. Uses statistical methods for drift detection and z-score based anomaly scoring.
"""Drift detection and anomaly detection pipeline for agent performance monitoring."""
from __future__ import annotations
import math
import logging
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional
logger = logging.getLogger(__name__)
@dataclass
class MetricWindow:
"""Sliding window for tracking a single metric over time."""
name: str
values: deque[float] = field(default_factory=lambda: deque(maxlen=500))
timestamps: deque[datetime] = field(default_factory=lambda: deque(maxlen=500))
def add(self, value: float, timestamp: Optional[datetime] = None) -> None:
self.values.append(value)
self.timestamps.append(timestamp or datetime.utcnow())
@property
def mean(self) -> float:
return sum(self.values) / len(self.values) if self.values else 0.0
@property
def std_dev(self) -> float:
if len(self.values) < 2:
return 0.0
m = self.mean
variance = sum((v - m) ** 2 for v in self.values) / (len(self.values) - 1)
return math.sqrt(variance)
def z_score(self, value: float) -> float:
"""Compute the z-score of a value relative to this window's history."""
sd = self.std_dev
if sd == 0:
return 0.0
return (value - self.mean) / sd
class DriftDetector:
"""Detects statistical drift in agent input/output distributions over time.
Uses a sliding window approach: maintains a baseline period and compares
recent metric windows against it using mean shift detection with configurable
sensitivity. For production deployments, consider integrating with KS-test
based libraries like `scipy.stats.ks_2samp` for formal distribution testing.
Reference `code-philosophy` Law 4 (Fail Fast): drift detection should never
block execution — it produces alerts only, allowing the agent to continue.
"""
def __init__(self, window_name: str, sensitivity_sigma: float = 3.0) -> None:
self.window_name = window_name
self.sensitivity_sigma = sensitivity_sigma
self.baseline_window: Optional[MetricWindow] = None
self.current_window: MetricWindow = MetricWindow(window_name)
self.drift_detected: bool = False
def update(self, value: float) -> dict:
"""Record a new metric value and check for drift against the baseline.
Args:
value: The current metric measurement (e.g., accuracy score, token count).
Returns:
Dict with drift status, z-score, and recommendation.
"""
self.current_window.add(value)
if self.baseline_window is None:
# First update — establish baseline from 100 samples
if len(self.current_window.values) >= 100:
self.baseline_window = MetricWindow(f"{self.window_name}_baseline")
self.baseline_window.values = deque(
list(self.current_window.values)[:100], maxlen=500
)
return {"drift": False, "z_score": 0.0, "action": "col
…(truncated)