AI Agent 调试实战：链路追踪、Prompt 可视化与异常定位的系统方法

一、Agent 调试的黑盒困境

AI Agent 系统的调试难度远高于传统软件。传统程序的执行路径是确定性的，输入相同则输出相同，断点加日志基本能定位问题。Agent 的执行路径由 LLM 的输出决定，而 LLM 的输出具有随机性——同一个 Prompt 两次调用可能产生不同的工具调用序列、不同的推理路径、不同的最终结果。

生产环境中的典型调试场景：一个数据分析 Agent，用户输入"分析上周的销售趋势"，Agent 应该调用数据库查询工具获取数据，再调用图表工具生成可视化。但实际运行中，Agent 有时跳过数据查询直接编造数据，有时调用错误的工具，有时陷入工具调用的无限循环。这些问题的根因可能在 Prompt 设计、工具描述、上下文管理、LLM 温度参数等多个环节，传统调试手段无法快速定位。

更隐蔽的问题是"静默错误"：Agent 调用了正确的工具，但传递了错误的参数，返回了不完整的数据，LLM 基于不完整数据生成了看似合理但实际错误的结论。这类错误不会抛异常，只有在业务结果出现偏差时才会被发现。

二、Agent 调试的架构：全链路可观测

Agent 调试的核心思路是全链路可观测：记录 Agent 执行的每一步（思考、工具调用、工具返回、最终输出），构建完整的执行轨迹（Trace），支持回放和对比分析。

graph TB
    subgraph Agent 执行链路
        A[用户输入] --> B[LLM 推理<br/>思考/决策]
        B --> C{需要工具调用?}
        C -->|是| D[工具调用1<br/>name + args]
        D --> E[工具返回1<br/>result + latency]
        E --> B
        C -->|否| F[最终输出]
    end

    subgraph 可观测层
        G[Span 记录<br/>每步的输入/输出/耗时] --> H[Trace 聚合<br/>完整执行轨迹]
        H --> I[异常检测<br/>循环/超时/参数错误]
        I --> J[回放与对比<br/>重现问题 / A/B 对比]
    end

    style A fill:#e1f5fe
    style B fill:#fff3e0
    style D fill:#e8f5e9
    style F fill:#f3e5f5
    style H fill:#ffebee

关键设计要素：

• Span：记录单步操作（LLM 调用或工具调用）的输入、输出、耗时、token 消耗。
• Trace：一次 Agent 执行的完整 Span 链路，包含所有步骤的时序关系。
• 异常检测规则：自动识别常见异常模式（循环调用、参数缺失、输出格式错误）。

三、生产级代码：Agent 调试框架实现

3.1 Span 与 Trace 数据结构

import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional
import json


class SpanType(Enum):
    LLM_CALL = "llm_call"
    TOOL_CALL = "tool_call"
    TOOL_RESULT = "tool_result"


class SpanStatus(Enum):
    OK = "ok"
    ERROR = "error"
    TIMEOUT = "timeout"


@dataclass
class Span:
    """单步操作记录"""
    span_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
    trace_id: str = ""
    parent_id: Optional[str] = None
    span_type: SpanType = SpanType.LLM_CALL
    name: str = ""
    status: SpanStatus = SpanStatus.OK

    # 输入输出
    input_data: Any = None
    output_data: Any = None
    error: Optional[str] = None

    # 性能指标
    start_time: float = 0
    end_time: float = 0
    token_usage: dict = field(default_factory=dict)

    @property
    def duration_ms(self) -> float:
        if self.end_time and self.start_time:
            return (self.end_time - self.start_time) * 1000
        return 0

    def to_dict(self) -> dict:
        return {
            "span_id": self.span_id,
            "trace_id": self.trace_id,
            "parent_id": self.parent_id,
            "type": self.span_type.value,
            "name": self.name,
            "status": self.status.value,
            "input": _safe_serialize(self.input_data),
            "output": _safe_serialize(self.output_data),
            "error": self.error,
            "duration_ms": round(self.duration_ms, 2),
            "token_usage": self.token_usage,
        }


@dataclass
class Trace:
    """完整执行轨迹"""
    trace_id: str = field(default_factory=lambda: uuid.uuid4().hex[:16])
    agent_name: str = ""
    user_input: str = ""
    final_output: str = ""
    spans: list[Span] = field(default_factory=list)
    start_time: float = field(default_factory=time.time)
    end_time: float = 0
    metadata: dict = field(default_factory=dict)

    def add_span(self, span: Span) -> Span:
        span.trace_id = self.trace_id
        self.spans.append(span)
        return span

    @property
    def total_duration_ms(self) -> float:
        if self.end_time:
            return (self.end_time - self.start_time) * 1000
        return 0

    @property
    def total_tokens(self) -> int:
        return sum(
            s.token_usage.get("total", 0) for s in self.spans
        )

    def to_dict(self) -> dict:
        return {
            "trace_id": self.trace_id,
            "agent_name": self.agent_name,
            "user_input": self.user_input,
            "final_output": self.final_output,
            "spans": [s.to_dict() for s in self.spans],
            "total_duration_ms": round(self.total_duration_ms, 2),
            "total_tokens": self.total_tokens,
            "metadata": self.metadata,
        }


def _safe_serialize(data: Any) -> Any:
    """安全序列化，处理不可 JSON 化的对象"""
    try:
        json.dumps(data)
        return data
    except (TypeError, ValueError):
        return str(data)

3.2 可观测 Agent 执行器

class ObservableAgent:
    """带全链路追踪的 Agent 执行器"""

    def __init__(
        self,
        name: str,
        llm_client,
        tools: dict[str, callable],
        max_iterations: int = 10,
        trace_callback: Optional[callable] = None,
    ):
        self.name = name
        self.llm = llm_client
        self.tools = tools
        self.max_iterations = max_iterations
        self.trace_callback = trace_callback  # Trace 完成后的回调

    def run(self, user_input: str) -> dict:
        """执行 Agent，返回结果和 Trace"""
        trace = Trace(
            agent_name=self.name,
            user_input=user_input,
        )

        messages = [{"role": "user", "content": user_input}]
        final_output = ""
        tool_call_history: list[str] = []  # 检测循环调用

        for iteration in range(self.max_iterations):
            # 1. LLM 调用
            llm_span = Span(
                span_type=SpanType.LLM_CALL,
                name=f"llm_call_iter_{iteration}",
                input_data=messages[-1],
            )
            llm_span.start_time = time.time()

            try:
                response = self.llm.chat(
                    messages=messages,
                    tools=self._get_tool_schemas(),
                    temperature=0.1,  # 低温度，减少随机性
                )
                llm_span.output_data = response
                llm_span.status = SpanStatus.OK
            except Exception as e:
                llm_span.status = SpanStatus.ERROR
                llm_span.error = str(e)
                llm_span.end_time = time.time()
                trace.add_span(llm_span)
                break

            llm_span.end_time = time.time()
            llm_span.token_usage = {
                "prompt": response.get("usage", {}).get("prompt_tokens", 0),
                "completion": response.get("usage", {}).get("completion_tokens", 0),
                "total": response.get("usage", {}).get("total_tokens", 0),
            }
            trace.add_span(llm_span)

            # 2. 检查是否有工具调用
            tool_calls = response.get("tool_calls", [])
            if not tool_calls:
                # 无工具调用，LLM 给出最终回答
                final_output = response.get("content", "")
                break

            # 3. 执行工具调用
            assistant_msg = response
            messages.append(assistant_msg)

            for tc in tool_calls:
                tool_name = tc["function"]["name"]
                tool_args = tc["function"]["arguments"]

                # 循环调用检测
                call_sig = f"{tool_name}({tool_args})"
                tool_call_history.append(call_sig)
                if self._detect_loop(tool_call_history):
                    loop_span = Span(
                        span_type=SpanType.TOOL_CALL,
                        name=tool_name,
                        status=SpanStatus.ERROR,
                        error=f"检测到循环调用: {call_sig}",
                    )
                    trace.add_span(loop_span)
                    final_output = "错误：Agent 陷入工具调用循环"
                    break

                # 记录工具调用 Span
                tool_span = Span(
                    span_type=SpanType.TOOL_CALL,
                    name=tool_name,
                    input_data=tool_args,
                )
                tool_span.start_time = time.time()

                try:
                    result = self.tools[tool_name](**tool_args)
                    tool_span.output_data = result
                    tool_span.status = SpanStatus.OK
                except Exception as e:
                    tool_span.status = SpanStatus.ERROR
                    tool_span.error = str(e)
                    result = f"工具执行错误: {e}"

                tool_span.end_time = time.time()
                trace.add_span(tool_span)

                # 将工具结果加入消息
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc["id"],
                    "content": str(result),
                })

            else:
                continue  # 正常完成工具调用，进入下一轮
            break  # 循环检测触发，退出

        trace.final_output = final_output
        trace.end_time = time.time()

        # 回调通知（用于持久化或实时展示）
        if self.trace_callback:
            self.trace_callback(trace)

        return {
            "output": final_output,
            "trace": trace.to_dict(),
        }

    def _detect_loop(self, history: list[str]) -> bool:
        """检测循环调用：最近 3 次调用是否完全相同"""
        if len(history) < 3:
            return False
        return (
            history[-1] == history[-2] == history[-3]
        )

    def _get_tool_schemas(self) -> list[dict]:
        """获取工具的 JSON Schema 描述"""
        # 实际实现中，从工具的 docstring 或装饰器提取
        return []

3.3 异常检测与诊断报告

class AgentDiagnostics:
    """Agent 异常检测与诊断报告生成"""

    @staticmethod
    def analyze(trace: Trace) -> dict:
        """分析 Trace，生成诊断报告"""
        issues = []
        tool_spans = [
            s for s in trace.spans if s.span_type == SpanType.TOOL_CALL
        ]
        llm_spans = [
            s for s in trace.spans if s.span_type == SpanType.LLM_CALL
        ]

        # 1. 循环调用检测
        tool_names = [s.name for s in tool_spans]
        for i in range(len(tool_names) - 2):
            if tool_names[i] == tool_names[i+1] == tool_names[i+2]:
                issues.append({
                    "type": "loop",
                    "severity": "high",
                    "message": f"工具 {tool_names[i]} 连续调用 3 次以上",
                    "suggestion": "检查工具描述是否清晰，LLM 是否理解工具返回值",
                })

        # 2. 工具调用失败
        failed_tools = [
            s for s in tool_spans if s.status == SpanStatus.ERROR
        ]
        for ft in failed_tools:
            issues.append({
                "type": "tool_error",
                "severity": "medium",
                "message": f"工具 {ft.name} 执行失败: {ft.error}",
                "suggestion": "检查工具参数是否正确，工具实现是否有 bug",
            })

        # 3. LLM 调用超时
        slow_llm = [
            s for s in llm_spans if s.duration_ms > 10000
        ]
        for s in slow_llm:
            issues.append({
                "type": "slow_llm",
                "severity": "low",
                "message": f"LLM 调用耗时 {s.duration_ms:.0f}ms",
                "suggestion": "考虑减少上下文长度或使用更快的模型",
            })

        # 4. Token 消耗异常
        total_tokens = trace.total_tokens
        if total_tokens > 10000:
            issues.append({
                "type": "high_token_usage",
                "severity": "medium",
                "message": f"总 Token 消耗 {total_tokens}，可能存在上下文膨胀",
                "suggestion": "检查是否需要截断历史消息或压缩上下文",
            })

        # 5. 迭代次数过多
        if len(llm_spans) > 5:
            issues.append({
                "type": "too_many_iterations",
                "severity": "medium",
                "message": f"Agent 执行了 {len(llm_spans)} 轮迭代",
                "suggestion": "检查 Prompt 是否明确指导 Agent 何时停止",
            })

        return {
            "trace_id": trace.trace_id,
            "agent_name": trace.agent_name,
            "total_duration_ms": round(trace.total_duration_ms, 2),
            "total_tokens": total_tokens,
            "num_tool_calls": len(tool_spans),
            "num_llm_calls": len(llm_spans),
            "issues": issues,
            "health": "unhealthy" if any(
                i["severity"] == "high" for i in issues
            ) else "healthy",
        }

3.4 Trace 可视化输出

class TraceFormatter:
    """Trace 格式化输出，用于日志和调试"""

    @staticmethod
    def to_text(trace: Trace) -> str:
        """将 Trace 格式化为可读文本"""
        lines = [
            f"=== Agent Trace: {trace.trace_id} ===",
            f"Agent: {trace.agent_name}",
            f"Input: {trace.user_input[:100]}...",
            f"Duration: {trace.total_duration_ms:.0f}ms",
            f"Tokens: {trace.total_tokens}",
            "",
        ]

        for i, span in enumerate(trace.spans):
            icon = "🤖" if span.span_type == SpanType.LLM_CALL else "🔧"
            status_icon = "✅" if span.status == SpanStatus.OK else "❌"
            lines.append(
                f"  {icon} [{i+1}] {span.name} "
                f"{status_icon} ({span.duration_ms:.0f}ms)"
            )
            if span.error:
                lines.append(f"      Error: {span.error}")
            if span.span_type == SpanType.TOOL_CALL:
                lines.append(
                    f"      Input: {str(span.input_data)[:80]}"
                )
                if span.output_data:
                    lines.append(
                        f"      Output: {str(span.output_data)[:80]}"
                    )

        lines.append("")
        lines.append(f"Final Output: {trace.final_output[:200]}")
        return "\n".join(lines)

四、Agent 调试的权衡与边界

4.1 Trace 存储成本

每次 Agent 执行产生一个 Trace，包含多个 Span。每个 Span 记录输入输出，可能包含大量文本。按每次执行平均 5KB 计算，日活 10 万次的服务每天产生 500MB Trace 数据。建议设置保留策略（如 7 天热数据 + 30 天冷数据），仅对异常 Trace 做长期存储。

4.2 敏感数据脱敏

Trace 中可能包含用户输入的敏感信息（如身份证号、密码）。记录前必须做脱敏处理。建议在 Span 记录层统一脱敏，而非依赖业务层。

4.3 LLM 随机性的影响

同一输入多次执行可能产生不同结果，Trace 对比时需要考虑随机性。建议固定 temperature=0 用于调试，生产环境使用低温度（0.1~0.3）。对于必须复现的问题，记录 LLM 的 seed 参数（如 OpenAI 的 seed 字段）。

4.4 适用与禁用场景

场景	调试策略	原因
工具调用错误	Trace + 参数检查	定位参数构造问题
循环调用	循环检测 + Prompt 优化	根因在 Prompt 设计
输出质量差	Prompt A/B 对比	需要控制变量
延迟过高	Span 耗时分析	定位慢步骤
幻觉问题	工具返回值 vs 最终输出对比	检查 LLM 是否忽略工具结果

五、总结

Agent 调试的核心是全链路可观测。通过 Span 记录每步操作的输入、输出、耗时和状态，构建完整的 Trace 轨迹。异常检测自动识别循环调用、工具失败、Token 消耗异常等常见问题。诊断报告提供问题定位和修复建议。Trace 的存储成本需要通过保留策略控制，敏感数据必须脱敏。LLM 的随机性通过固定温度参数和 seed 来缓解。调试不是事后补救，而是 Agent 系统的基础设施——没有可观测性的 Agent 在生产环境中就是黑盒。