Monitoring Production Agents for Safety and Drift: Detecting Misaligned Behavior

Your agent worked perfectly in testing. It handled edge cases gracefully, stayed on task, and never once did anything weird. Then you shipped it to production, and three weeks later a user screenshots it recommending something it absolutely should not have recommended. You have no idea when it started doing that, why, or how many users saw it. This is the problem that agent safety monitoring solves — and most teams don’t implement it until after something goes wrong.

This article is about building the monitoring layer that catches behavioral drift, unsafe outputs, and unexpected capability changes before they become support tickets, PR disasters, or worse. We’ll cover what to measure, how to instrument your agents, and working code you can drop into a production system today.

Why Agent Behavior Drifts (And Why You Won’t Notice Without Monitoring)

Agents drift for reasons that have nothing to do with your code changing. The underlying model gets updated silently. Your prompt templates interact differently with new user inputs you hadn’t anticipated. A retrieval system starts surfacing different documents as your knowledge base grows. A tool your agent calls returns slightly different data structures after an upstream API change.

Any one of these can shift your agent’s behavior in ways that are subtle but consequential. The agent still works — it responds, it completes tasks — it just does them slightly differently than it used to. Output length drifts. Refusal rates change. The tone shifts. Certain topics start getting handled differently. None of this trips an error, none of it shows up in your latency graphs, and your users often don’t complain until the damage is already done.

The monitoring approach that actually works treats your agent like a statistical process, not a deterministic function. You’re not just checking “did it return a response” — you’re tracking the distribution of what those responses look like over time.

What to Actually Measure: The Four Signal Categories

1. Output Structure Signals

These are the easiest to instrument and often the first to show drift. Track: response length (token count distribution), format compliance (does JSON output still parse correctly, do structured responses still match your schema), and section presence (does the agent still include required components like citations, disclaimers, or confidence scores).

A 40% increase in average response length over two weeks isn’t necessarily a problem, but it’s a signal worth investigating. Maybe a model update made it more verbose. Maybe a prompt change had an unintended effect. Either way, you want to know.

2. Semantic Drift Signals

This is where it gets more interesting. You want to detect when the meaning of your agent’s outputs shifts, not just the surface statistics. The practical approach: embed a sample of agent outputs daily and track centroid distance over time. When the embedding centroid drifts significantly from your baseline, flag it for review.

You can also maintain a set of “anchor prompts” — fixed inputs you run against your agent on a schedule — and track how the responses to those evolve. If your customer service agent starts responding to “how do I cancel my subscription?” differently than it did last month, you want to know that before a customer notices.

3. Policy Compliance Signals

This is the safety-critical layer. Define the behaviors your agent should never exhibit — recommending competitors, making specific medical/legal/financial claims, discussing certain topics, using particular language — and build classifiers or rule-based detectors for them. Run every output through this layer before it reaches the user, or at minimum log everything and run async batch checks.

For most teams, a combination works best: cheap regex/keyword rules for the obvious stuff (near-zero latency, very high recall), plus an LLM-based classifier running async for nuanced policy violations.

4. Behavioral Consistency Signals

Does your agent handle the same input consistently? In production, you’ll often see the same or nearly-identical queries hitting your agent. Track the variance in responses to semantically similar inputs. High variance isn’t always wrong, but sudden increases in variance often signal a model update or a prompt regression.

Instrumenting Your Agent: Working Code

Here’s a monitoring wrapper that handles logging, policy checking, and drift detection hooks. This is designed to wrap any LLM call, so it works regardless of whether you’re using Claude, GPT-4, or an open-source model.

import time
import json
import hashlib
import statistics
from datetime import datetime
from typing import Any, Callable, Optional
import anthropic

# Simple in-memory store — replace with your DB (Postgres, Redis, etc.)
_output_log = []
_anchor_baselines = {}

POLICY_KEYWORDS = [
    "i cannot provide medical advice",  # check if agent is being overly cautious
    "competitor_name_here",              # check for competitor mentions
    # add your own policy signals here
]

def compute_response_stats(text: str) -> dict:
    words = text.split()
    sentences = text.split('.')
    return {
        "char_count": len(text),
        "word_count": len(words),
        "sentence_count": len([s for s in sentences if s.strip()]),
        "avg_word_length": statistics.mean([len(w) for w in words]) if words else 0,
    }

def check_policy_flags(text: str) -> list[str]:
    """Returns list of triggered policy flags."""
    text_lower = text.lower()
    return [kw for kw in POLICY_KEYWORDS if kw.lower() in text_lower]

def log_agent_output(
    session_id: str,
    prompt_hash: str,
    response: str,
    model: str,
    latency_ms: float,
    metadata: Optional[dict] = None
):
    stats = compute_response_stats(response)
    flags = check_policy_flags(response)
    
    record = {
        "timestamp": datetime.utcnow().isoformat(),
        "session_id": session_id,
        "prompt_hash": prompt_hash,
        "model": model,
        "latency_ms": latency_ms,
        "response_preview": response[:200],  # don't log full PII-containing responses
        "stats": stats,
        "policy_flags": flags,
        "metadata": metadata or {}
    }
    
    _output_log.append(record)
    
    # Alert on policy violations immediately
    if flags:
        trigger_alert(record, alert_type="policy_violation")
    
    return record

def trigger_alert(record: dict, alert_type: str):
    """Send to your alerting system — Slack, PagerDuty, email, whatever."""
    print(f"[ALERT] {alert_type.upper()} at {record['timestamp']}")
    print(f"  Session: {record['session_id']}")
    print(f"  Flags: {record.get('policy_flags', [])}")
    # Replace with: requests.post(SLACK_WEBHOOK, json={"text": ...})

def monitored_agent_call(
    client: anthropic.Anthropic,
    system_prompt: str,
    user_message: str,
    session_id: str,
    model: str = "claude-3-haiku-20240307",
    **kwargs
) -> str:
    prompt_hash = hashlib.md5(
        (system_prompt + user_message).encode()
    ).hexdigest()[:8]
    
    start = time.time()
    
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": user_message}],
        **kwargs
    )
    
    latency_ms = (time.time() - start) * 1000
    output_text = response.content[0].text
    
    log_agent_output(
        session_id=session_id,
        prompt_hash=prompt_hash,
        response=output_text,
        model=model,
        latency_ms=latency_ms,
        metadata={"input_tokens": response.usage.input_tokens,
                  "output_tokens": response.usage.output_tokens}
    )
    
    return output_text

# Usage
client = anthropic.Anthropic()
response = monitored_agent_call(
    client=client,
    system_prompt="You are a helpful customer service agent for AcmeCorp.",
    user_message="How do I reset my password?",
    session_id="user_123_session_456"
)

This runs at roughly zero added latency for the synchronous policy check. The async embedding drift detection — where the real cost sits — runs in a separate worker. At Haiku pricing (~$0.00025 per 1K input tokens), even if you’re running 10,000 agent calls per day, the monitoring overhead is negligible compared to the operational risk you’re mitigating.

Drift Detection With Anchor Prompts

The anchor prompt system is the most practical drift detection approach I’ve found for production agents. Define 10-20 representative inputs that cover your agent’s main use cases and edge cases. Run them against your agent on a daily or weekly schedule. Store the outputs. Compare them.

import json
from openai import OpenAI  # using OpenAI for embeddings — cheap and consistent

def embed_text(text: str, openai_client: OpenAI) -> list[float]:
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",  # $0.00002 per 1K tokens
        input=text[:8000]  # truncate for safety
    )
    return response.data[0].embedding

def cosine_similarity(a: list[float], b: list[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    norm_a = sum(x ** 2 for x in a) ** 0.5
    norm_b = sum(x ** 2 for x in b) ** 0.5
    return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0

ANCHOR_PROMPTS = [
    "What is your refund policy?",
    "I want to cancel my account",
    "Can you help me with a billing dispute?",
    # ... your representative queries
]

DRIFT_THRESHOLD = 0.85  # flag if similarity drops below this

def run_anchor_check(
    anthropic_client: anthropic.Anthropic,
    openai_client: OpenAI,
    system_prompt: str,
    baseline: Optional[dict] = None
) -> dict:
    results = {}
    
    for prompt in ANCHOR_PROMPTS:
        response = anthropic_client.messages.create(
            model="claude-3-haiku-20240307",
            max_tokens=512,
            system=system_prompt,
            messages=[{"role": "user", "content": prompt}]
        )
        output = response.content[0].text
        embedding = embed_text(output, openai_client)
        
        result = {
            "output": output,
            "embedding": embedding,
            "timestamp": datetime.utcnow().isoformat()
        }
        
        # Compare to baseline if we have one
        if baseline and prompt in baseline:
            similarity = cosine_similarity(
                embedding, 
                baseline[prompt]["embedding"]
            )
            result["baseline_similarity"] = similarity
            
            if similarity < DRIFT_THRESHOLD:
                print(f"[DRIFT DETECTED] Prompt: '{prompt[:50]}...'")
                print(f"  Similarity to baseline: {similarity:.3f}")
                # trigger_alert(...)
        
        results[prompt] = result
    
    return results

Running 20 anchor prompts daily costs roughly $0.003 in Haiku API calls plus negligible embedding costs. That’s about $1/month for meaningful behavioral drift detection.

Building the Alerting Layer That Doesn’t Cry Wolf

Bad alerting is worse than no alerting because it trains your team to ignore alerts. Here’s what works in production:

• Immediate alerts (synchronous, high confidence): Hard policy violations caught by keyword/regex. If the agent says something it should never say, you need to know now.
• Daily digest alerts: Statistical drift — response length changes, anchor prompt similarity scores, token usage anomalies. Batch these into a single daily summary so they don’t spam you.
• Weekly trend reports: Slower-moving signals like refusal rate trends, user satisfaction correlation with agent behavior changes. These require more data to be meaningful.

For the statistical alerts, use rolling z-scores rather than fixed thresholds. A response length of 400 words might be normal for your agent — or it might be 3 standard deviations above its usual output. Fixed thresholds require constant tuning; z-scores adapt to your agent’s actual behavior baseline automatically.

What Breaks in Production (And How to Handle It)

The biggest failure mode I’ve seen: logging everything but never looking at it. You build the monitoring system, it dutifully records every output, and then nobody has a process for actually reviewing the data. Fix this by making the monitoring output actionable by default — not a dashboard you have to remember to check, but a daily Slack message or email that surfaces the top 3 anomalies from the previous 24 hours.

The second failure mode: monitoring the wrong thing. Teams often focus on latency and error rates (because those metrics already exist in their infra tooling) while ignoring semantic drift entirely. A response that takes 200ms and returns HTTP 200 can still be completely wrong.

The third: not having a baseline. If you didn’t record what “normal” looks like when your agent was behaving correctly, you have nothing to compare drift against. Start logging immediately, even before you build detection logic — the historical data is what makes detection possible.

When to Use This Approach vs. Simpler Options

If you’re running a low-stakes agent with low traffic — say, an internal tool with 50 users — a simpler approach works fine: manually review a random sample of outputs weekly, keep a changelog of every prompt change, and set up keyword alerts for obvious violations. The full monitoring stack described here is overkill for that scenario.

Where agent safety monitoring at this level pays for itself: customer-facing agents handling sensitive topics (finance, health, legal-adjacent), agents with tool use that can take real-world actions (sending emails, modifying data, making purchases), and any agent operating at scale where you genuinely can’t manually review outputs.

For solo founders: start with the logging wrapper and policy keyword checks. That alone catches 80% of real production incidents and takes an afternoon to implement. Add drift detection when you have enough volume to make it statistically meaningful (rough threshold: 1,000+ agent calls per day).

For teams with a dedicated ML engineer: add embedding-based drift detection, build out the anchor prompt system, and integrate with your existing observability stack (Datadog, Grafana, whatever you already have). The code here is the starting point, not the final implementation.

The bottom line on agent safety monitoring: the cost of building it is an afternoon; the cost of not building it is finding out three weeks after the fact that your agent has been doing something it shouldn’t. That’s not a difficult tradeoff.