Skip to content
BSWEN
Blog

What Is Agent Behavior Drift and How to Detect It?

Cowrie
Cowrie
Dev @ Bswen

My agent tests were passing. Every single one. The outputs looked correct, the assertions were green, and CI was happy. Yet users kept complaining about degraded experiences.

I checked the logs. No errors. I ran the tests again. All passed. I was confused.

Then I realized: my agent was returning the right answers, but taking a completely different path to get there. Same destination, different journey. And that journey mattered.

This is agent behavior drift, and it’s a silent killer in production LLM systems.

The Problem: When “Correct” Isn’t Enough

I had a math agent that I thought was working perfectly:

def test_agent():
    result = agent.run("Calculate 25% of 400")
    assert result.answer == 100  # Passes!

The test passed every time. But here’s what I didn’t see:

  • The agent was skipping the validation step
  • It was using the wrong tool first, then correcting
  • It was calling tools in a different order than before

The output was correct. The behavior had drifted. And my tests couldn’t catch it.

Why this happens:

  • LLM outputs are probabilistic even at temperature=0
  • Model updates change underlying behavior patterns
  • Context window variations affect tool selection
  • Prompt modifications have cascading effects

I learned this the hard way: most production issues with agents aren’t “quality” problems. They’re “behavior drift” problems.

What Behavior Drift Looks Like

Let me show you what I mean. Here’s what a good trajectory looks like:

Query: "Calculate 25% of 400"
Step 1: parse_number("400") -> 400
Step 2: parse_percentage("25%") -> 0.25
Step 3: multiply(400, 0.25) -> 100
Step 4: validate_result(100) -> True
Final: 100

And here’s a drifted trajectory:

Query: "Calculate 25% of 400"
Step 1: multiply(400, 0.25) -> 100  # Skipped parsing!
Step 2: validate_result(100) -> True
Final: 100

Same answer. But the agent skipped parsing steps. What happens when the input is “twenty-five percent”? The drifted agent fails.

My Failed Attempts at Detection

I tried a few things before finding the right approach:

Attempt 1: Add more assertions

def test_agent_detailed():
    result = agent.run("Calculate 25% of 400")
    assert result.answer == 100
    assert result.steps_taken == 4  # Too brittle
    assert "parse_number" in result.tools_used  # Order doesn't matter here

This was fragile. Step counts changed legitimately when I improved the agent. I was constantly updating tests.

Attempt 2: Mock everything

def test_agent_mocks():
    with mock.patch("agent.parse_number") as mock_parse:
        result = agent.run("Calculate 25% of 400")
        assert mock_parse.called  # Doesn't check order

Still couldn’t detect order changes. And mocking LLM calls is a nightmare.

Attempt 3: Record and compare tool calls

This was the breakthrough. I needed to treat agent behavior like snapshot tests. Record the trajectory when it’s working, save it as a baseline, diff after every change.

The Solution: Snapshot Testing for Agent Trajectories

I built a system to capture and compare agent execution paths. Here’s how it works.

Step 1: Capture Complete Trajectories

First, I needed to record everything the agent does:

capture_trajectory.py
import json
from datetime import datetime
from typing import TypedDict, List, Dict, Any


class ToolCall(TypedDict):
    tool_name: str
    parameters: Dict[str, Any]
    timestamp: str
    result_summary: str


class AgentTrajectory(TypedDict):
    query: str
    tool_calls: List[ToolCall]
    final_answer: str
    total_steps: int
    execution_time_ms: float


def capture_trajectory(agent, query: str) -> AgentTrajectory:
    """Capture complete execution path for snapshot testing."""
    start_time = datetime.utcnow()
    tool_calls = []


    # Instrument agent to record every tool call
    original_execute = agent.execute_tool


    def traced_execute(tool_name: str, params: dict):
        result = original_execute(tool_name, params)
        tool_calls.append({
            "tool_name": tool_name,
            "parameters": params,
            "timestamp": datetime.utcnow().isoformat(),
            "result_summary": summarize(result)
        })
        return result


    agent.execute_tool = traced_execute
    result = agent.run(query)


    return AgentTrajectory(
        query=query,
        tool_calls=tool_calls,
        final_answer=result.answer,
        total_steps=len(tool_calls),
        execution_time_ms=(datetime.utcnow() - start_time).total_seconds() * 1000
    )

Now I can record exactly what happened during execution.

Step 2: Normalize Tool Calls Across Providers

I ran into a problem: OpenAI, Anthropic, and Google all structure tool calls differently. I needed a unified format.

normalizer.py
from abc import ABC, abstractmethod
from typing import Dict, Any, List
import json


class NormalizedToolCall:
    """Provider-agnostic tool call representation."""


    def __init__(self, tool_name: str, parameters: Dict[str, Any]):
        self.tool_name = tool_name
        self.parameters = parameters


    def __eq__(self, other):
        """Compare tool calls semantically, not structurally."""
        return (
            self.tool_name == other.tool_name and
            self.parameters == other.parameters
        )


    def __hash__(self):
        return hash((self.tool_name, tuple(sorted(self.parameters.items()))))


class ToolCallNormalizer:
    """Normalize tool calls across LLM providers."""


    @staticmethod
    def normalize_openai(tool_call: dict) -> NormalizedToolCall:
        """OpenAI format: {'function': {'name': '...', 'arguments': '{...}'}}"""
        return NormalizedToolCall(
            tool_name=tool_call["function"]["name"],
            parameters=json.loads(tool_call["function"]["arguments"])
        )


    @staticmethod
    def normalize_anthropic(tool_call: dict) -> NormalizedToolCall:
        """Anthropic format: {'name': '...', 'input': {...}}"""
        return NormalizedToolCall(
            tool_name=tool_call["name"],
            parameters=tool_call["input"]
        )


    @staticmethod
    def normalize_google(tool_call: dict) -> NormalizedToolCall:
        """Google format: {'functionCall': {'name': '...', 'args': {...}}}"""
        return NormalizedToolCall(
            tool_name=tool_call["functionCall"]["name"],
            parameters=tool_call["functionCall"]["args"]
        )


    @classmethod
    def normalize(cls, tool_call: dict, provider: str) -> NormalizedToolCall:
        """Normalize any provider's tool call format."""
        normalizers = {
            "openai": cls.normalize_openai,
            "anthropic": cls.normalize_anthropic,
            "google": cls.normalize_google
        }
        return normalizers[provider](tool_call)

Now I can compare trajectories regardless of which LLM provider I’m using.

Step 3: Diff Trajectories and Detect Drifts

The core detection logic:

drift_detector.py
from dataclasses import dataclass
from typing import List, Optional, Any
from enum import Enum


class DriftType(Enum):
    TOOL_ORDER_CHANGED = "tool_order_changed"
    STEP_SKIPPED = "step_skipped"
    STEP_ADDED = "step_added"
    PARAMETERS_CHANGED = "parameters_changed"
    EXECUTION_TIME_CHANGED = "execution_time_changed"


@dataclass
class DriftReport:
    drift_type: DriftType
    description: str
    baseline_value: Any
    current_value: Any
    severity: str  # "low", "medium", "high"


def diff_trajectories(
    baseline: AgentTrajectory,
    current: AgentTrajectory,
    tolerance_config: dict = None
) -> List[DriftReport]:
    """Compare current execution against baseline, detect drifts."""


    drifts = []
    config = tolerance_config or {
        "execution_time_threshold": 0.5,  # 50% deviation
        "parameter_tolerance": {}  # Exact match by default
    }


    # Check tool order
    baseline_tools = [tc["tool_name"] for tc in baseline["tool_calls"]]
    current_tools = [tc["tool_name"] for tc in current["tool_calls"]]


    if baseline_tools != current_tools:
        # Detect skipped steps
        for i, tool in enumerate(baseline_tools):
            if tool not in current_tools:
                drifts.append(DriftReport(
                    drift_type=DriftType.STEP_SKIPPED,
                    description=f"Step '{tool}' was skipped",
                    baseline_value=baseline_tools,
                    current_value=current_tools,
                    severity="high"
                ))


        # Detect added steps
        for tool in current_tools:
            if tool not in baseline_tools:
                drifts.append(DriftReport(
                    drift_type=DriftType.STEP_ADDED,
                    description=f"Unexpected step '{tool}' was added",
                    baseline_value=baseline_tools,
                    current_value=current_tools,
                    severity="medium"
                ))


        # Detect order changes
        if len(baseline_tools) == len(current_tools) and baseline_tools != current_tools:
            drifts.append(DriftReport(
                drift_type=DriftType.TOOL_ORDER_CHANGED,
                description="Tool execution order changed",
                baseline_value=baseline_tools,
                current_value=current_tools,
                severity="medium"
            ))


    # Check parameter changes
    for i, (b_tc, c_tc) in enumerate(zip(baseline["tool_calls"], current["tool_calls"])):
        if b_tc["parameters"] != c_tc["parameters"]:
            drifts.append(DriftReport(
                drift_type=DriftType.PARAMETERS_CHANGED,
                description=f"Parameters for {b_tc['tool_name']} changed",
                baseline_value=b_tc["parameters"],
                current_value=c_tc["parameters"],
                severity="low"
            ))


    # Check execution time
    baseline_time = baseline["execution_time_ms"]
    current_time = current["execution_time_ms"]
    threshold = config["execution_time_threshold"]


    if abs(current_time - baseline_time) / baseline_time > threshold:
        drifts.append(DriftReport(
            drift_type=DriftType.EXECUTION_TIME_CHANGED,
            description=f"Execution time deviated by >{threshold*100}%",
            baseline_value=f"{baseline_time:.2f}ms",
            current_value=f"{current_time:.2f}ms",
            severity="low"
        ))


    return drifts


def assert_no_drift(
    baseline: AgentTrajectory,
    current: AgentTrajectory,
    allowed_severities: List[str] = None
):
    """Assert that trajectory hasn't drifted beyond allowed thresholds."""
    allowed = allowed_severities or []  # No drifts allowed by default
    drifts = diff_trajectories(baseline, current)


    critical_drifts = [
        d for d in drifts
        if d.severity not in allowed
    ]


    if critical_drifts:
        report = "\n".join([
            f"  - {d.drift_type.value}: {d.description}"
            for d in critical_drifts
        ])
        raise AssertionError(f"Behavior drift detected:\n{report}")

Step 4: Continuous Monitoring in CI/CD

The final piece: running this in my pipeline.

monitor.py
import json
from pathlib import Path
from typing import Dict, List
from datetime import datetime


class AgentBehaviorMonitor:
    """Monitor agent behavior over time, alert on drift."""


    def __init__(self, baseline_dir: str = "baselines"):
        self.baseline_dir = Path(baseline_dir)
        self.baseline_dir.mkdir(exist_ok=True)


    def record_baseline(self, agent, test_cases: List[dict]):
        """Record baselines for a set of test cases."""
        baselines = {}


        for test in test_cases:
            query = test["query"]
            trajectory = capture_trajectory(agent, query)
            baselines[test["name"]] = trajectory


        # Save all baselines
        for name, trajectory in baselines.items():
            save_baseline(trajectory, name)


        return baselines


    def check_drift(self, agent, test_cases: List[dict]) -> Dict[str, List[DriftReport]]:
        """Run test cases and check for drift against baselines."""
        results = {}


        for test in test_cases:
            name = test["name"]
            baseline = load_baseline(name)
            current = capture_trajectory(agent, test["query"])


            drifts = diff_trajectories(baseline, current)
            results[name] = drifts


            if drifts:
                self._alert(name, drifts)


        return results


    def _alert(self, test_name: str, drifts: List[DriftReport]):
        """Alert on detected drifts."""
        high_severity = [d for d in drifts if d.severity == "high"]


        if high_severity:
            print(f"[ALERT] High-severity drift in {test_name}:")
            for drift in high_severity:
                print(f"  {drift.description}")
                print(f"  Baseline: {drift.baseline_value}")
                print(f"  Current: {drift.current_value}")


# Usage in CI/CD pipeline
monitor = AgentBehaviorMonitor()


test_cases = [
    {"name": "calculate_percentage", "query": "Calculate 25% of 400"},
    {"name": "search_docs", "query": "Find documentation about agent routing"},
    {"name": "multi_step_task", "query": "Research and summarize latest AI papers"}
]


# Record baseline once when agent is working correctly
# monitor.record_baseline(agent, test_cases)


# Run in CI to detect drift
drift_results = monitor.check_drift(agent, test_cases)


# Fail CI if critical drift detected
for test_name, drifts in drift_results.items():
    critical = [d for d in drifts if d.severity == "high"]
    if critical:
        raise Exception(f"Critical behavior drift detected in {test_name}")

What I Learned

After implementing this, I caught issues I never would have seen before:

  1. A model update caused tool order changes - My agent started calling search before validate after an OpenAI model update. The outputs looked fine, but edge cases were failing.

  2. Prompt changes skipped validation steps - A “small” prompt tweak caused the agent to skip a safety check. Output still looked valid.

  3. Tool changes affected execution patterns - When I updated a tool’s parameters, the agent started using a different code path entirely.

The key insight: output quality tests are not enough. You need to test the execution path, not just the result.

Getting Started

If you’re building LLM agents, here’s my recommended approach:

  1. Identify critical user flows - What are the most important things your agent does?
  2. Record baselines when behavior is correct - Capture trajectories during development
  3. Add drift detection to CI - Run comparison on every commit
  4. Set severity thresholds - Not all drift is bad, but skipped steps are usually bad
  5. Alert on high-severity drifts - Get notified immediately when critical behavior changes

Most production issues with agents aren’t quality problems. They’re behavior drift problems. Start treating your agent’s execution path as a first-class artifact, and you’ll catch silent degradations before your users do.

Final Words + More Resources

My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me

Here are also the most important links from this article along with some further resources that will help you in this scope:

Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!

Comments

Tags: