2026年3月5日

自主智能体系统的调试与可观测性

悄无声息地失败的自主智能体比没有智能体更糟糕。当传统函数抛出异常时，你会得到一个堆栈跟踪。当智能体在二十次工具调用和三次模型调用中走错了路，你得到的是一个错误答案——没有任何明显的解释。

调试智能体需要不同的思维模型。系统并不是在执行一条确定性路径；它在做一系列决策。可观测性意味着捕获这些决策——不仅仅是输入和输出，还有连接它们的推理过程。

为何传统调试对智能体无效

标准日志记录捕获的是发生了什么。智能体可观测性需要捕获为什么——模型得出了什么结论，选择了哪个工具以及原因，以及它正在处理什么中间状态。

失败模式也各有不同：

无声幻觉：智能体自信地给出错误答案，却不发出任何不确定性信号。
决策漂移：每一步在局部看起来都合理，但整个序列偏离了目标。
工具误用：智能体调用了正确的工具，但参数微妙地错误。
无限循环：智能体陷入不断重试失败方法的循环。
上下文污染：早期步骤的错误输出污染了所有后续推理。

这些情况都不会产生异常。它们产生的是错误行为，只有在重建完整执行追踪时才能看到。

智能体决策的结构化日志

第一步是用结构化日志包装每次智能体交互。不要记录原始API响应——记录语义事件。

import json
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Any
import anthropic

client = anthropic.Anthropic()

@dataclass
class AgentEvent:
    trace_id: str
    step: int
    event_type: str      # "llm_call", "tool_call", "tool_result", "decision", "error"
    model: str | None
    input_tokens: int | None
    output_tokens: int | None
    latency_ms: float | None
    content: dict[str, Any]
    timestamp: float

def log_event(event: AgentEvent):
    print(json.dumps(asdict(event)))  # 替换为你的日志目标

class TracedAgent:
    def __init__(self, trace_id: str | None = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self.step = 0
        self.tools = []

    def add_tool(self, name: str, description: str, input_schema: dict):
        self.tools.append({
            "name": name,
            "description": description,
            "input_schema": input_schema
        })

    def call(self, messages: list[dict], system: str = "") -> str:
        self.step += 1
        start = time.monotonic()

        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            system=system,
            tools=self.tools,
            messages=messages
        )

        latency_ms = (time.monotonic() - start) * 1000

        log_event(AgentEvent(
            trace_id=self.trace_id,
            step=self.step,
            event_type="llm_call",
            model="claude-opus-4-6",
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            latency_ms=latency_ms,
            content={
                "stop_reason": response.stop_reason,
                "text_blocks": [b.text for b in response.content if b.type == "text"],
                "tool_calls": [
                    {"name": b.name, "input": b.input}
                    for b in response.content if b.type == "tool_use"
                ]
            },
            timestamp=time.time()
        ))

        return response

每次LLM调用现在都会发出一个结构化事件，包含追踪ID、步骤编号、token计数、延迟和模型决策。

构建完整追踪

单行日志是不够的——你需要完整的执行追踪，将每个决策与其结果连接起来：

from typing import Callable

def run_traced_agent(
    task: str,
    tools: dict[str, Callable],
    tool_schemas: list[dict],
    system: str,
    max_steps: int = 20,
) -> dict:
    agent = TracedAgent()
    for schema in tool_schemas:
        agent.add_tool(**schema)

    messages = [{"role": "user", "content": task}]
    trace = {"trace_id": agent.trace_id, "task": task, "steps": []}
    step_count = 0

    while step_count < max_steps:
        step_count += 1
        response = agent.call(messages, system=system)

        step_record = {
            "step": step_count,
            "stop_reason": response.stop_reason,
            "model_output": [],
            "tool_results": []
        }

        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    step_record["model_output"].append(block.text)
            trace["steps"].append(step_record)
            trace["final_answer"] = step_record["model_output"][-1] if step_record["model_output"] else ""
            break

        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                step_record["model_output"].append({
                    "tool": block.name,
                    "input": block.input
                })

                tool_fn = tools.get(block.name)
                if not tool_fn:
                    result = f"错误：未知工具 '{block.name}'"
                else:
                    try:
                        result = tool_fn(**block.input)
                    except Exception as e:
                        result = f"工具错误：{e}"

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result)
                })
                step_record["tool_results"].append({
                    "tool": block.name,
                    "result_preview": str(result)[:200]
                })

        trace["steps"].append(step_record)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

    else:
        trace["error"] = f"超过最大步骤数 ({max_steps})"

    return trace

循环检测

无限循环是常见的失败模式。通过对每次LLM调用的工具调用模式进行指纹识别来检测它们：

def detect_loop(trace: dict, window: int = 4) -> bool:
    steps = trace["steps"]
    if len(steps) < window:
        return False

    def step_signature(step: dict) -> str:
        tools_called = sorted(
            t["tool"] if isinstance(t, dict) else t
            for t in step.get("model_output", [])
            if isinstance(t, dict) and "tool" in t
        )
        return "|".join(tools_called)

    recent = [step_signature(s) for s in steps[-window:]]
    if len(set(recent)) == 1 and recent[0]:
        return True

    if len(steps) >= 4:
        pattern = [step_signature(s) for s in steps[-4:]]
        if pattern[0] == pattern[2] and pattern[1] == pattern[3]:
            return True

    return False

生产环境中要追踪的指标

from collections import Counter

def compute_trace_metrics(trace: dict) -> dict:
    steps = trace["steps"]
    errors = [s for s in steps if "error" in s]

    tool_calls_by_name: Counter = Counter()
    for step in steps:
        for output in step.get("model_output", []):
            if isinstance(output, dict) and "tool" in output:
                tool_calls_by_name[output["tool"]] += 1

    return {
        "trace_id": trace["trace_id"],
        "total_steps": len(steps),
        "error_steps": len(errors),
        "tool_call_distribution": dict(tool_calls_by_name),
        "completed": "final_answer" in trace,
        "loop_detected": detect_loop(trace),
    }

需要告警的关键信号：

循环率 > 5% — 智能体陷入卡顿
工具错误率 > 阈值 — 某个工具已损坏
平均步骤数上升趋势 — 任务变得更难或提示词在退化
p99延迟峰值 — 模型端点变慢

OpenTelemetry集成

对于已经使用OpenTelemetry的团队，将智能体追踪作为span发出：

from opentelemetry import trace as otel_trace

tracer = otel_trace.get_tracer("agent")

def run_with_otel(task: str, tools: dict, tool_schemas: list, system: str):
    with tracer.start_as_current_span("agent.run") as root_span:
        root_span.set_attribute("agent.task", task[:200])

        agent = TracedAgent()
        for schema in tool_schemas:
            agent.add_tool(**schema)

        messages = [{"role": "user", "content": task}]

        for step in range(20):
            with tracer.start_as_current_span(f"agent.step.{step}") as step_span:
                response = agent.call(messages, system=system)
                step_span.set_attribute("llm.stop_reason", response.stop_reason)
                step_span.set_attribute("llm.input_tokens", response.usage.input_tokens)

                if response.stop_reason == "end_turn":
                    break

日志中的PII脱敏

智能体日志通常包含敏感数据。在发送到任何外部系统之前，对其进行脱敏处理：

import re

PII_PATTERNS = [
    (re.compile(r'\b[\w.+-]+@[\w-]+\.[a-z]{2,}\b'), '[邮箱]'),
    (re.compile(r'\b1[3-9]\d{9}\b'), '[手机号]'),
    (re.compile(r'\bsk-[a-zA-Z0-9]{20,}\b'), '[API密钥]'),
]

def redact(text: str) -> str:
    for pattern, replacement in PII_PATTERNS:
        text = pattern.sub(replacement, text)
    return text

三个最重要的指标

任务完成率 — 有多少比例的运行达到final_answer，而不是触发max_steps或错误。按任务类型建立基准。

每任务token成本 — 对所有步骤的input_tokens + output_tokens求和。追踪其随时间的变化。成本增加20%而完成率不变，通常表明提示词退化。

工具错误率 — error_steps / total_steps。此指标的峰值直接指向损坏的工具或API。

智能体系统中的可观测性不是可选的——它是你可以迭代改进的系统与只能在崩溃时重启的系统之间的区别。从结构化事件和追踪ID开始。添加循环检测。推送指标。当你第一次在生产环境中遇到故障并能够重建发生了什么，而不是猜测，这笔投资就得到了回报。