The Slow Degradation Problem
Agent systems degrade in ways that are subtle, gradual, and often invisible to traditional monitoring. An agent that achieved 95% task success in week one may be at 87% by week six without any single obvious failure event. The error rate did not spike. Latency did not change. No circuit breakers tripped. But the agent is quietly producing lower-quality outputs that erode user trust, generate more support tickets, and correlate with falling user retention.
The degradation happens for multiple reasons. Model providers update their models continuously. The 'claude-sonnet-4-5' you are calling today may behave subtly differently from the one you called last month, because providers roll out continuous improvements and safety updates. Your data distribution shifts: the kinds of queries users send evolve over time, and the agent was not trained on the distribution you are now serving. Prompt drift occurs as small incremental changes to your system prompt accumulate into a meaningfully different prompt. And tool outputs change. If the APIs your tools call return different data structures or different information, the agent's reasoning may silently degrade.
This covers the full detection and response stack: how to detect semantic drift using embedding-based distribution comparison, how to detect hallucinations using automated verification pipelines, and how to build an alerting system that catches these problems before users do.
Types of Agent Drift and Why They Happen
Semantic Drift: Outputs Shift in Content
Semantic drift occurs when the meaning and content of agent outputs shifts over time even though the inputs remain similar. The most reliable way to detect semantic drift is to embed agent outputs into a vector space and compare the distribution of embeddings from a baseline period against the current period. If the distributions have diverged significantly (measured by KL divergence, Wasserstein distance, or Jensen-Shannon divergence), semantic drift has occurred.
Behavioral Drift: Response Patterns Change
Behavioral drift refers to changes in how the agent responds, response length, clarification rate, refusal rate, tone, rather than what it says. An agent that suddenly starts giving much shorter responses may have had its effective context truncated by an upstream change. An agent with a rising refusal rate may be overly conservative due to a safety update. Behavioral drift is detectable through aggregate statistics on response characteristics rather than semantic analysis.
Factual Drift: Accuracy Declines
Factual drift occurs when the agent's accuracy on verifiable claims declines. This is the most serious type of drift because it directly harms users who act on incorrect information. Factual drift can result from: model updates that changed how the model handles specific domains, retrieval-augmented systems where the knowledge base has become stale, or prompt changes that inadvertently encouraged over-confident responses on uncertain topics.
Detecting Semantic Drift Automatically
The core of automatic semantic drift detection is a comparison of output embedding distributions over time. Embed a sample of agent outputs each day using a consistent embedding model (text-embedding-3-large or equivalent). Store these embeddings with timestamps. Periodically compare the distribution of recent embeddings against a baseline distribution from your reference period (typically the first two weeks after a successful deployment).
import numpy as np
from scipy.stats import entropy
from openai import AsyncOpenAI
from datetime import datetime, timedelta
client = AsyncOpenAI()
async def embed_text(text: str) -> np.ndarray:
response = await client.embeddings.create(
model="text-embedding-3-large",
input=text[:8000], # Truncate to model limit
)
return np.array(response.data[0].embedding)
async def compute_distribution_divergence(
reference_embeddings: list[np.ndarray],
current_embeddings: list[np.ndarray],
n_bins: int = 50,
) -> dict:
ref_array = np.stack(reference_embeddings)
cur_array = np.stack(current_embeddings)
divergences = []
for dim in range(0, ref_array.shape[1], 32): # Sample dimensions
ref_hist, bins = np.histogram(ref_array[:, dim], bins=n_bins, density=True)
cur_hist, _ = np.histogram(cur_array[:, dim], bins=bins, density=True)
# Add small epsilon to avoid log(0)
ref_hist = ref_hist + 1e-10
cur_hist = cur_hist + 1e-10
ref_hist /= ref_hist.sum()
cur_hist /= cur_hist.sum()
kl_div = entropy(cur_hist, ref_hist)
divergences.append(kl_div)
mean_kl = np.mean(divergences)
return {
"mean_kl_divergence": float(mean_kl),
"max_kl_divergence": float(np.max(divergences)),
"drift_detected": mean_kl > 0.1, # Threshold from empirical calibration
"severity": "high" if mean_kl > 0.2 else ("medium" if mean_kl > 0.1 else "low"),
}
async def daily_drift_check(agent_id: str) -> dict:
reference = await load_embeddings(agent_id, days_back=14, reference=True)
current = await load_embeddings(agent_id, days_back=1, reference=False)
if len(current) < 30: # Insufficient sample size
return {"skipped": True, "reason": "insufficient_samples"}
result = await compute_distribution_divergence(reference, current)
result["agent_id"] = agent_id
result["checked_at"] = datetime.utcnow().isoformat()
if result["drift_detected"]:
await send_drift_alert(result)
return resultBehavioral Metrics for Drift Detection
Beyond semantic embedding comparison, track a set of behavioral metrics daily and alert when they deviate significantly from their baseline. These metrics are cheap to compute from existing logs and catch different types of drift than embedding comparison.
from dataclasses import dataclass
import statistics
@dataclass
class BehavioralMetrics:
mean_response_length: float # Characters per response
refusal_rate: float # Fraction of responses that decline the request
clarification_rate: float # Fraction that ask clarifying questions
list_usage_rate: float # Fraction that use bullet lists
mean_confidence_language: float # Hedging words per response (uncertainty signal)
tool_call_rate: float # Fraction of responses that invoke tools
mean_tool_calls_per_request: float
def compute_behavioral_metrics(responses: list[str]) -> BehavioralMetrics:
REFUSAL_MARKERS = ["i cannot", "i'm unable", "i don't have", "i can't assist"]
CLARIFICATION_MARKERS = ["could you clarify", "could you tell me more", "what do you mean"]
HEDGING_WORDS = ["however", "although", "note that", "it's worth", "be aware"]
lengths = [len(r) for r in responses]
return BehavioralMetrics(
mean_response_length=statistics.mean(lengths),
refusal_rate=sum(any(m in r.lower() for m in REFUSAL_MARKERS) for r in responses) / len(responses),
clarification_rate=sum(any(m in r.lower() for m in CLARIFICATION_MARKERS) for r in responses) / len(responses),
list_usage_rate=sum("•" in r or "- " in r for r in responses) / len(responses),
mean_confidence_language=sum(sum(w in r.lower() for w in HEDGING_WORDS) for r in responses) / len(responses),
tool_call_rate=0.0,
mean_tool_calls_per_request=0.0,
)
def check_behavioral_drift(
baseline: BehavioralMetrics,
current: BehavioralMetrics,
threshold: float = 0.3, # 30% deviation triggers alert
) -> list[str]:
alerts = []
fields = ["mean_response_length", "refusal_rate", "clarification_rate",
"tool_call_rate", "mean_confidence_language"]
for field in fields:
base_val = getattr(baseline, field)
curr_val = getattr(current, field)
if base_val == 0:
continue
deviation = abs(curr_val - base_val) / base_val
if deviation > threshold:
direction = "increased" if curr_val > base_val else "decreased"
alerts.append(f"{field} {direction} by {deviation:.0%} (baseline: {base_val:.2f}, current: {curr_val:.2f})")
return alertsCategories of Hallucination in Agents
Hallucination in AI agents takes several distinct forms, each requiring different detection strategies. Understanding which type of hallucination your use case is most vulnerable to helps you prioritize your detection investment.
- Factual hallucination: The agent asserts false facts with confidence. A research agent that fabricates citations, statistics, or historical events. Detectable by cross-referencing claims against known-good sources.
- Contextual hallucination: The agent generates information that is plausible in isolation but inconsistent with the provided context. For example, summarizing a document and including facts not present in the document. Detectable by comparing the output against the input context.
- Self-referential hallucination: The agent makes false claims about its own capabilities, training data, or previous statements in the conversation. Detectable by tracking factual claims the agent makes about itself.
- Tool result hallucination: After a tool call returns data, the agent misrepresents or fabricates additional details beyond what the tool actually returned. Detectable by comparing tool outputs against how the agent describes them.
- Confidence miscalibration: The agent expresses high confidence about uncertain information, or expresses uncertainty about well-established facts. This is the hardest to detect automatically and requires calibration measurement over time.
Automated Hallucination Detection Pipeline
A production hallucination detection pipeline runs asynchronously after each agent response is generated. It does not block the response to the user. The detection latency (often 2 to 5 seconds per response) is incompatible with real-time serving. Instead, it flags responses for review, powers aggregate hallucination rate metrics, and triggers alerts when rates exceed thresholds.
from anthropic import AsyncAnthropic
import re
client = AsyncAnthropic()
async def check_contextual_hallucination(
context: str, # The source document or retrieved knowledge
response: str, # The agent's response
model: str = "claude-haiku-4-5", # Use cheap model for detection
) -> dict:
detection_prompt = f"""You are a factual accuracy checker. Compare the RESPONSE against the CONTEXT.
CONTEXT (ground truth):
{context[:4000]}
RESPONSE:
{response}
Check for claims in the response that:
1. Are not supported by the context
2. Contradict the context
3. Go beyond what the context states
Respond with JSON:
{{
"unsupported_claims": ["list of specific claims not in context"],
"contradicted_claims": ["list of claims contradicting context"],
"hallucination_score": 0.0 to 1.0,
"rationale": "brief explanation"
}}"""
detection_response = await client.messages.create(
model=model,
system="You are a precise factual accuracy evaluator. Output only valid JSON.",
messages=[{"role": "user", "content": detection_prompt}],
max_tokens=1024,
)
try:
result = json.loads(detection_response.content[0].text)
return result
except json.JSONDecodeError:
return {"error": "detection_failed", "hallucination_score": None}
async def run_hallucination_pipeline(
response: str,
context: str,
request_id: str,
) -> dict:
result = await check_contextual_hallucination(context, response)
score = result.get("hallucination_score", 0)
action = "serve" if score < 0.3 else ("flag" if score < 0.7 else "block")
await store_hallucination_result(request_id, result)
await update_hallucination_rate_metric(score)
if score >= 0.7:
await send_hallucination_alert(request_id, result)
return {"action": action, "score": score, **result}LLM-as-Judge for Quality Scoring
Using a language model to evaluate the outputs of another language model is called 'LLM-as-judge'. It enables automated quality scoring at scale without human evaluators, though it requires careful prompt design to avoid the judge sharing biases or blindspots with the model being evaluated. For best results, use a different model family as judge than the one generating responses.
JUDGE_PROMPT_TEMPLATE = """You are an objective quality evaluator for AI assistant responses.
Evaluate the RESPONSE against the QUERY on the following criteria.
Score each criterion 1-5 where 5 is excellent.
QUERY: {query}
RESPONSE: {response}
Criteria:
1. Relevance: Does the response address what was actually asked?
2. Accuracy: Are factual claims correct and well-supported?
3. Completeness: Does the response cover the key aspects of the question?
4. Clarity: Is the response clear, well-structured, and easy to understand?
5. Appropriate confidence: Does the response express appropriate uncertainty where relevant?
Respond with valid JSON:
{{
"relevance": 1-5,
"accuracy": 1-5,
"completeness": 1-5,
"clarity": 1-5,
"appropriate_confidence": 1-5,
"overall_score": (average of above),
"flags": ["list of specific quality issues, or empty list"],
"rationale": "one sentence summary"
}}"""
async def judge_response(
query: str,
response: str,
judge_model: str = "gpt-4o-mini", # Different family than Claude responses
) -> dict:
prompt = JUDGE_PROMPT_TEMPLATE.format(query=query, response=response)
result = await openai_client.chat.completions.create(
model=judge_model,
messages=[
{"role": "system", "content": "You are an objective evaluator. Output only valid JSON."},
{"role": "user", "content": prompt},
],
max_tokens=512,
temperature=0, # Deterministic for consistency
)
return json.loads(result.choices[0].message.content)Building the Alerting Pipeline
Drift and hallucination alerts differ from standard infrastructure alerts in an important way: they are based on aggregate statistics computed over time windows, not on individual event thresholds. This means they need a batch evaluation pipeline that runs on a schedule (hourly, daily) rather than real-time event streaming.
from dataclasses import dataclass
from datetime import datetime
@dataclass
class DriftAlert:
agent_id: str
alert_type: str # semantic_drift, behavioral_drift, hallucination_rate
severity: str # low, medium, high, critical
metric_name: str
baseline_value: float
current_value: float
deviation_pct: float
detected_at: datetime
recommended_action: str
ALERT_THRESHOLDS = {
"hallucination_rate": {"warning": 0.05, "critical": 0.15},
"semantic_drift_kl": {"warning": 0.10, "critical": 0.25},
"refusal_rate_delta": {"warning": 0.15, "critical": 0.40},
"quality_score_drop": {"warning": 0.10, "critical": 0.20},
}
async def evaluate_and_alert(agent_id: str):
metrics = await compute_current_metrics(agent_id)
baseline = await load_baseline_metrics(agent_id)
alerts = []
for metric, thresholds in ALERT_THRESHOLDS.items():
current = metrics.get(metric, 0)
base = baseline.get(metric, 0)
if base == 0:
continue
deviation = abs(current - base) / base
if deviation >= thresholds["critical"]:
severity = "critical"
elif deviation >= thresholds["warning"]:
severity = "warning"
else:
continue
alerts.append(DriftAlert(
agent_id=agent_id,
alert_type=metric,
severity=severity,
metric_name=metric,
baseline_value=base,
current_value=current,
deviation_pct=deviation,
detected_at=datetime.utcnow(),
recommended_action=get_recommended_action(metric, severity),
))
for alert in alerts:
await dispatch_alert(alert)
return alertsMaintaining Reference Baselines
The reference baseline is the distribution of outputs your agent produced during a known-good period, typically the first two to three weeks after a successful deployment, after initial user feedback confirmed acceptable quality. The baseline must be updated deliberately: you should not automatically roll the baseline forward, because that would mask gradual drift. Instead, update the baseline only after a deliberate evaluation that confirms the new behavior is acceptable.
- Baseline snapshot frequency: Compute and store baseline snapshots weekly. Retain at least 8 weeks of snapshots to enable retrospective analysis.
- Baseline update triggers: Update the baseline after a major model upgrade, a major prompt version change, or a deliberate behavior modification, not automatically.
- Multi-period comparison: Compare against the most recent baseline AND the original deployment baseline to distinguish new drift from accumulated drift.
- Stratified baselines: For agents handling multiple task types, maintain separate baselines per task type. A shift in query distribution would look like drift in an aggregate baseline but is not actually degradation.
Response Playbook When Drift Is Detected
When a drift or hallucination alert fires, the on-call team needs a clear playbook. The severity of the alert determines the urgency and the set of response options.
Low Severity: Monitor and Investigate
For low-severity drift (10 to 30% deviation from baseline), the immediate response is increased monitoring. Reduce the alert evaluation window from daily to hourly, enable full content logging for a sample of requests, and schedule a human evaluation session within 48 hours. Do not immediately roll back or modify the agent; instead confirm whether the drift is an artifact of shifting query distribution or actual quality degradation.
High Severity: Activate Safeguards
For high-severity drift (more than 30% deviation) or a hallucination rate above the critical threshold, activate safeguards immediately: enable shadow mode for a portion of traffic (run the old and new versions in parallel and compare outputs), add output validation that rejects responses below a quality threshold, or roll back to the previous agent configuration. Do not wait for a human evaluation. The customer impact at this severity level is already significant.
Continuous Evaluation Architecture
The gold standard for production agent quality is a continuous evaluation pipeline that runs on a daily cadence against a curated test set. Unlike drift detection (which compares distributions), continuous evaluation gives you absolute quality scores on known test cases. It can detect regressions even when the test distribution stays fixed.
import asyncio
from datetime import datetime
@dataclass
class EvalResult:
test_case_id: str
input_query: str
agent_response: str
judge_scores: dict
overall_score: float
regression_from_baseline: float
async def run_continuous_eval(
agent_fn,
test_suite: list[dict],
baseline_scores: dict[str, float],
judge_model: str = "gpt-4o-mini",
) -> dict:
results = []
semaphore = asyncio.Semaphore(5) # Limit concurrent evals
async def eval_case(case: dict) -> EvalResult:
async with semaphore:
response = await agent_fn(case["query"])
scores = await judge_response(case["query"], response, judge_model)
return EvalResult(
test_case_id=case["id"],
input_query=case["query"],
agent_response=response,
judge_scores=scores,
overall_score=scores["overall_score"],
regression_from_baseline=scores["overall_score"] - baseline_scores.get(case["id"], 3.5),
)
results = await asyncio.gather(*[eval_case(c) for c in test_suite])
mean_score = sum(r.overall_score for r in results) / len(results)
regressions = [r for r in results if r.regression_from_baseline < -0.5]
report = {
"evaluated_at": datetime.utcnow().isoformat(),
"mean_score": mean_score,
"pass_rate": sum(r.overall_score >= 3.5 for r in results) / len(results),
"regressions": len(regressions),
"worst_regressions": sorted(regressions, key=lambda x: x.regression_from_baseline)[:5],
}
if len(regressions) > len(test_suite) * 0.1: # More than 10% regressions
await send_regression_alert(report)
return reportFrequently Asked Questions
Build secure agentic systems with AgentixForce
We help companies design, build, and secure production-grade AI agent systems. From architecture reviews to full implementation, our team brings deep expertise to every engagement.