Continuously monitors agent performance in production through token tracking, LLM-as-a-Judge evaluation, A/B testing for improvements, drift detection, anomaly detection, and trajectory analysis with structured feedback loops.
Scanned 9/4/2026
Install to Claude Code
Skills Directorynpx -y skills add paulpas/agent-skill-router --skill evaluation-monitoring --agent claude-codeInstalls into .claude/skills of the current project.
Are you the author of Evaluation Monitoring?
Add the live security badge to your README — it updates automatically with every re-scan.
[](https://www.skillsdirectory.com/skills/paulpas-evaluation-monitoring)More formats (shields.io, HTML) on the badges page.
---
name: evaluation-monitoring
description: Continuously monitors agent performance in production through token tracking, LLM-as-a-Judge evaluation, A/B testing for improvements, drift detection, anomaly detection, and trajectory analysis with structured feedback loops.
license: MIT
compatibility: opencode
metadata:
version: "1.0.0"
domain: agent
role: implementation
scope: implementation
output-format: code
triggers: evaluation, LLM as judge, A/B testing agents, drift detection, anomaly detection, trajectory analysis, how do i monitor agent performance, token tracking
related-skills: reflection-loop, goal-setting-monitoring, rag-patterns
archetypes:
- tactical
anti_triggers:
- brainstorming
- vague ideation
- single-agent monolith
response_profile:
verbosity: medium
directive_strength: high
abstraction_level: operational
---
# 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
1. **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.
2. **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.
3. **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/json` response 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.
4. **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.
5. **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.
6. **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.
```python
"""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:**
```python
# ❌ 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:**
```python
# ✅ 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.
```python
"""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.
```python
"""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.
```python
"""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.
```python
"""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": "collecting_baseline"}
z = self.current_window.z_score(value)
is_drift = abs(z) > self.sensitivity_sigma
if is_drift and not self.drift_detected:
self.drift_detected = True
logger.warning(
"DRIFT DETECTED in %s: z-score=%.2f (threshold=%dσ)",
self.window_name, z, int(self.sensitivity_sigma),
)
return {
"drift": is_drift,
"z_score": round(z, 3),
"current_mean": round(self.current_window.mean, 4),
"baseline_mean": round(self.baseline_window.mean, 4),
"action": "investigate_and_retrain" if is_drift else "normal",
}
def reset_baseline(self) -> None:
"""Reset the baseline after retraining or corrective action."""
self.baseline_window = None
self.drift_detected = False
class AnomalyDetector:
"""Detects anomalous agent behavior using z-score based outlier detection.
Monitors multiple metrics concurrently and flags combinations that suggest
unusual behavior (e.g., high latency + low accuracy simultaneously).
"""
def __init__(self, window_size: int = 200) -> None:
self.accuracy_window = MetricWindow("accuracy")
self.latency_window = MetricWindow("latency_ms")
self.token_window = MetricWindow("total_tokens")
self.window_size = window_size
def record(self, accuracy: float, latency_ms: float, tokens: int) -> dict:
"""Record a single agent interaction and check for anomalies.
Args:
accuracy: Response accuracy score (0.0-1.0).
latency_ms: Processing time in milliseconds.
tokens: Total token count for the interaction.
Returns:
Dict with anomaly flags and per-metric z-scores.
"""
self.accuracy_window.add(accuracy)
self.latency_window.add(latency_ms)
self.token_window.add(float(tokens))
anomalies = {
"low_accuracy": self._flag_anomaly(self.accuracy_window, accuracy, direction="below"),
"high_latency": self._flag_anomaly(self.latency_window, latency_ms, direction="above"),
"token_spike": self._flag_anomaly(self.token_window, float(tokens), direction="above"),
}
is_anomalous = any(anomalies.values())
anomaly_count = sum(anomalies.values())
return {
"is_anomaly": is_anomalous,
"anomaly_type": self._classify_anomaly(anomalies),
"flags": anomalies,
"severity": "high" if anomaly_count >= 2 else "low",
}
def _flag_anomaly(
self, window: MetricWindow, value: float, direction: str = "above"
) -> bool:
"""Flag a single metric as anomalous based on z-score threshold."""
z = abs(window.z_score(value))
threshold = 3.0 if len(window.values) >= 50 else 4.0
return (z > threshold and ((direction == "above" and value > window.mean) or
(direction == "below" and value < window.mean)))
def _classify_anomaly(self, flags: dict) -> str:
"""Classify the type of anomaly based on which metrics are flagged."""
if flags["low_accuracy"] and flags["high_latency"]:
return "degraded_performance"
elif flags["token_spike"] and flags["high_latency"]:
return "resource_exhaustion_risk"
elif flags["low_accuracy"]:
return "quality_degradation"
elif flags["token_spike"]:
return "unusual_token_consumption"
else:
return "minor_anomaly"
# --- Usage: Full Monitoring Pipeline ---
def demo_monitoring_pipeline() -> None:
"""Demonstrate the combined drift + anomaly detection pipeline."""
drift_detector = DriftDetector("accuracy_score", sensitivity_sigma=2.5)
# Simulate 150 interactions with a gradual accuracy decline (drift)
base_accuracy = 0.95
for i in range(150):
# Accuracy gradually degrades from 0.95 to ~0.78
current_accuracy = base_accuracy - (i * 0.0011)
drift_result = drift_detector.update(current_accuracy)
if drift_result["drift"]:
print(f" Drift detected at sample {i}: "
f"accuracy={current_accuracy:.3f}, z={drift_result['z_score']}")
# Anomaly detection on individual interactions
anomaly_detector = AnomalyDetector()
# Normal interaction
normal = anomaly_detector.record(accuracy=0.92, latency_ms=1200, tokens=250)
print(f"Normal: {normal['is_anomaly']}, severity={normal['severity']}")
# Degraded — low accuracy + high latency
degraded = anomaly_detector.record(accuracy=0.45, latency_ms=8500, tokens=310)
print(f"Degraded: {degraded['is_anomaly']}, "
f"type={degraded['anomaly_type']}, severity={degraded['severity']}")
if __name__ == "__main__":
demo_monitoring_pipeline()
```
---
## Constraints
### MUST DO
1. **Define explicit evaluation dimensions before deployment** — Never deploy an agent without first specifying which metrics (accuracy, latency, cost, safety) will be tracked and how each is measured. Reference `code-philosophy` Law 2 (Parse Don't Validate): use actual tokenizer output and provider metrics, never string-length approximations for token counts.
2. **Use an independent judge model** — The LLM-as-a-Judge evaluator must never be the same model being evaluated. Use a stronger model (e.g., Gemini 1.5 Pro or Claude Opus) to avoid self-praise bias and ensure evaluation quality exceeds production quality.
3. **Require structured JSON output from judges** — Every LLM-as-a-Judge prompt must enforce `response_mime_type="application/json"` with a clearly defined schema. This enables automated dashboarding, alerting, and regression testing without brittle text parsing.
4. **Run A/B tests with statistical significance** — Minimum 100 comparisons per variant, running for at least one full business cycle to avoid time-of-day or day-of-week bias. Always include a control group (current production version).
5. **Set drift detection thresholds from historical baselines** — Do not use arbitrary threshold values. Establish baseline metrics over the first 100+ production interactions and tune z-score sensitivity based on observed variance. Default to 3σ for stable systems, 2σ for early-stage deployments.
6. **Compare trajectories using task-appropriate matching** — High-stakes tasks (financial advice, medical triage) require `EXACT` match type with zero tolerance. Flexible creative tasks can use `IN_ORDER` or `ANY_ORDER`. Never apply the same strictness universally — calibrate to risk level.
7. **Implement contract negotiation before high-stakes execution** — For any task exceeding defined cost/time thresholds, run the contractor negotiation flow before execution begins. This catches missing data sources and scope ambiguities that would otherwise cause costly failures.
8. **Store all evaluation telemetry for at least 30 days** — Drift detection requires historical trend analysis. Ensure your observability backend (Prometheus, Datadog, etc.) is configured with adequate retention to support this analysis.
### MUST NOT DO
1. **Never skip A/B testing before promoting a new agent version to production** — Rolling out changes without controlled comparison risks silently degrading user experience and performance across all users.
2. **Never evaluate an agent with the same model it uses for inference** — Self-evaluation produces inflated scores that mask real quality issues. Always use a separate, preferably stronger judge model.
3. **Never rely solely on automated evaluation for subjective qualities like helpfulness or safety** — LLM-as-a-Judge is a complement to, not replacement for, periodic human review. Schedule quarterly manual audits of flagged evaluations.
4. **Never deploy drift detection without a defined response procedure** — A detection alert without an action plan (investigate → retrain → verify) is noise. Every monitored metric must have an owned escalation path.
5. **Never use exact-match trajectory comparison for probabilistic agent tasks** — Agents operate probabilistically; demanding perfect step-for-step alignment against a ground-truth path will flag legitimate variations as failures, inflating false-positive rates above 80%.
6. **Never allow contract costs to exceed pre-defined computational bounds without explicit negotiation** — This directly violates the contractor model's cost control principle and leads to unbounded spending on complex tasks.
---
## Output Template
When applying this skill to design or audit an agent evaluation and monitoring system, produce:
1. **Evaluation Metrics Specification** — Complete list of metrics to track (accuracy, latency, cost, safety), each with measurement method, target threshold, alerting configuration, and data retention period
2. **LLM-as-a-Judge Rubric Document** — The full scoring rubric used for the judge model evaluation, including all criteria with 1-5 scoring definitions, expected JSON output schema, and judge model selection rationale
3. **A/B Test Design Document** — Control and challenger definitions, test inputs (from evalset), comparison metrics, minimum sample size calculation, statistical significance method, test duration plan, and rollback criteria
4. **Drift and Anomaly Configuration** — Baseline metric values from the first N interactions, configured z-score thresholds per metric, alert routing rules (who gets notified for each anomaly type), and response procedures for each detected condition
5. **Trajectory Analysis Results** — Comparison of actual agent execution paths against expected ground-truth paths, including match type used, precision/recall scores, missing/extra steps identified, and trend over time
6. **Contract Compliance Audit Log** — For each contract-based task: the original contract terms, negotiation exchanges, self-validation results, final compliance verdict, and any deviations from specifications with root cause analysis
---
## Related Skills
| Skill | Purpose |
|---|---|
| `reflection-loop` | Agent self-reflection patterns for iterative improvement — feeds back evaluation results to refine agent behavior |
| `goal-setting-monitoring` | Goal tracking and progress monitoring — complementary to performance evaluation in live systems |
| `rag-patterns` | RAG-specific evaluation metrics (faithfulness, relevance) — extends LLM-as-a-Judge with retrieval-focused rubrics |
---
## Live References
> Authoritative documentation for evaluation frameworks and agent monitoring.
- [Google ADK Evaluation](https://google.github.io/adk-docs/evaluate/) — Programmatic evaluation using pytest integration
- [ADK Web UI](https://github.com/google/adk-web) — Interactive session creation and eval dataset generation
- [Survey: Evaluation of LLM-based Agents](https://arxiv.org/abs/2503.16416) — Comprehensive survey of agent evaluation methodologies
- [Agent-as-a-Judge](https://arxiv.org/abs/2410.10934) — Evaluate agents with agents using a judge model framework
- [Agent Companion Paper](https://www.kaggle.com/whitepaper-agent-companion) — Contract-based contractor model for deterministic agent execution
---
*This skill implements the Evaluation and Monitoring design pattern from Chapter 19 of the Agentic Systems reference text. It transforms unpredictable, probabilistic agents into accountable systems through structured evaluation frameworks, formal contracts, and continuous monitoring.*
Is this your skill, or is something wrong with this listing? Request removal or report an issue. Author removals are honored within 72 hours.
No comments yet. Be the first to comment!
