AI Agent 调试实战：从 Prompt 追踪到执行回放的系统化排障方法
·
AI Agent 调试实战：从 Prompt 追踪到执行回放的系统化排障方法
一、Agent 的"黑盒困境"：为什么它总是不听话？
AI Agent 的调试难度远超传统软件。传统程序的 Bug 可以通过断点、日志、堆栈追踪来定位，但 Agent 的行为由 LLM 的推理决定，而 LLM 的推理过程是不可解释的。你只能看到"输入了什么"和"输出了什么"，中间的推理链路是一个黑盒。
某次生产排障中，一个客服 Agent 在处理退款请求时，错误地调用了"创建订单"工具而非"发起退款"工具。问题出在哪里？是 Prompt 中的工具描述不够清晰？是用户意图被误解？还是 LLM 在多轮对话中丢失了上下文？没有结构化的调试手段，只能靠猜测反复修改 Prompt，效率极低。
Agent 调试的核心挑战有三：第一，LLM 输出的不确定性——同样的输入可能产生不同输出；第二，多步推理的链式失败——前面一步出错，后续步骤全部偏离；第三，工具调用的副作用——错误调用可能修改了真实数据，无法简单重试。
二、Agent 调试体系与执行追踪架构
flowchart TB
subgraph Agent["Agent 执行层"]
INPUT["用户输入"]
LLM_CALL["LLM 推理"]
TOOL_CALL["工具调用"]
OBSERVE["观察结果"]
OUTPUT["最终输出"]
end
subgraph Trace["执行追踪层"]
SPAN["Span 追踪<br/>每步记录输入/输出/耗时"]
TREE["执行树<br/>可视化推理链路"]
REPLAY["回放引擎<br/>基于 Trace 重现执行"]
end
subgraph Debug["调试工具层"]
D1["Prompt Diff<br/>对比修改前后的 Prompt"]
D2["工具 Mock<br/>替换真实工具为模拟"]
D3["断言检查<br/>关键步骤插入校验"]
D4["A/B 测试<br/>对比不同 Prompt 效果"]
end
INPUT --> LLM_CALL --> TOOL_CALL --> OBSERVE --> OUTPUT
LLM_CALL --> SPAN
TOOL_CALL --> SPAN
OBSERVE --> SPAN
SPAN --> TREE
TREE --> REPLAY
D1 -.-> LLM_CALL
D2 -.-> TOOL_CALL
D3 -.-> OBSERVE
D4 -.-> LLM_CALL
调试体系的核心是"让每一步都可观测、可回放、可对比"：
- Span 追踪：Agent 执行的每一步（LLM 调用、工具调用、状态变更）都记录为一个 Span，包含输入、输出、耗时和 Token 消耗。
- 执行树：将 Span 组织为树形结构，可视化展示推理链路，快速定位哪一步出了问题。
- 回放引擎：基于 Trace 数据重现执行过程，无需重新调用 LLM，支持在任意步骤插入断点。
- 工具 Mock：将真实工具替换为模拟实现，避免调试时的副作用。
三、生产级 Agent 调试代码实现
3.1 执行追踪器
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Optional
@dataclass
class Span:
"""执行追踪的最小单元"""
span_id: str
parent_id: Optional[str]
span_type: str # llm_call / tool_call / state_change
name: str # 步骤名称
input_data: Any # 输入数据
output_data: Any = None # 输出数据
start_time: float = 0.0
end_time: float = 0.0
duration_ms: float = 0.0
token_usage: dict = field(default_factory=dict)
status: str = "running" # running / success / error
error: Optional[str] = None
metadata: dict = field(default_factory=dict)
class AgentTracer:
"""Agent 执行追踪器"""
def __init__(self):
self.spans: list[Span] = []
self.trace_id: str = str(uuid.uuid4())
self._active_spans: dict[str, Span] = {}
def start_span(
self,
span_type: str,
name: str,
input_data: Any,
parent_id: Optional[str] = None,
) -> Span:
"""开始一个新的 Span"""
span = Span(
span_id=str(uuid.uuid4()),
parent_id=parent_id,
span_type=span_type,
name=name,
input_data=input_data,
start_time=time.time(),
)
self.spans.append(span)
self._active_spans[span.span_id] = span
return span
def end_span(
self,
span_id: str,
output_data: Any,
token_usage: dict = None,
error: Optional[str] = None,
):
"""结束一个 Span"""
span = self._active_spans.get(span_id)
if not span:
return
span.end_time = time.time()
span.duration_ms = (span.end_time - span.start_time) * 1000
span.output_data = output_data
span.token_usage = token_usage or {}
if error:
span.status = "error"
span.error = error
else:
span.status = "success"
del self._active_spans[span_id]
def get_execution_tree(self) -> dict:
"""构建执行树，用于可视化展示"""
span_map = {s.span_id: s for s in self.spans}
root_spans = [s for s in self.spans if s.parent_id is None]
def build_tree(span: Span) -> dict:
children = [
build_tree(s) for s in self.spans
if s.parent_id == span.span_id
]
return {
"span_id": span.span_id,
"type": span.span_type,
"name": span.name,
"status": span.status,
"duration_ms": round(span.duration_ms, 2),
"input": str(span.input_data)[:200],
"output": str(span.output_data)[:200] if span.output_data else None,
"error": span.error,
"children": children,
}
return {
"trace_id": self.trace_id,
"total_spans": len(self.spans),
"total_duration_ms": sum(s.duration_ms for s in self.spans),
"total_tokens": sum(
s.token_usage.get("total_tokens", 0) for s in self.spans
),
"tree": [build_tree(s) for s in root_spans],
}
def find_error_spans(self) -> list[Span]:
"""快速定位所有出错的 Span"""
return [s for s in self.spans if s.status == "error"]
def find_slow_spans(self, threshold_ms: float = 2000) -> list[Span]:
"""快速定位耗时超过阈值的 Span"""
return [s for s in self.spans if s.duration_ms > threshold_ms]
3.2 可追踪的 Agent 执行器
class TraceableAgent:
"""带追踪能力的 Agent 执行器"""
def __init__(self, llm_client: Any, tools: list, tracer: AgentTracer):
self.llm = llm_client
self.tools = {t.name: t for t in tools}
self.tracer = tracer
async def run(self, query: str, max_steps: int = 10) -> dict:
"""执行 Agent 推理循环，每步都记录追踪信息"""
root_span = self.tracer.start_span(
"agent_run", "agent_execution", {"query": query}
)
messages = [{"role": "user", "content": query}]
step_count = 0
while step_count < max_steps:
step_count += 1
# 记录 LLM 调用
llm_span = self.tracer.start_span(
"llm_call",
f"step_{step_count}_llm",
{"messages_count": len(messages)},
parent_id=root_span.span_id,
)
try:
response = await self.llm.chat(
messages=messages,
tools=list(self.tools.values()),
)
self.tracer.end_span(
llm_span.span_id,
output_data={
"content": response.content[:500],
"tool_calls": [
{"name": tc.name, "args": str(tc.arguments)[:200]}
for tc in response.tool_calls
],
},
token_usage=response.usage,
)
except Exception as e:
self.tracer.end_span(
llm_span.span_id,
output_data=None,
error=str(e),
)
break
# 如果没有工具调用，说明 LLM 给出了最终回答
if not response.tool_calls:
self.tracer.end_span(
root_span.span_id,
output_data={"answer": response.content},
)
return {"answer": response.content, "steps": step_count}
# 执行工具调用
for tc in response.tool_calls:
tool_span = self.tracer.start_span(
"tool_call",
f"step_{step_count}_tool_{tc.name}",
{"tool": tc.name, "args": str(tc.arguments)[:200]},
parent_id=llm_span.span_id,
)
try:
tool = self.tools.get(tc.name)
if not tool:
raise ValueError(f"未知工具: {tc.name}")
result = await tool.run(**tc.arguments)
self.tracer.end_span(
tool_span.span_id,
output_data=str(result)[:500],
)
# 将工具结果加入消息列表
messages.append({
"role": "tool",
"name": tc.name,
"content": str(result),
})
except Exception as e:
self.tracer.end_span(
tool_span.span_id,
output_data=None,
error=str(e),
)
messages.append({
"role": "tool",
"name": tc.name,
"content": f"工具调用失败: {str(e)}",
})
# 达到最大步数，强制结束
self.tracer.end_span(
root_span.span_id,
output_data={"answer": "达到最大执行步数", "steps": step_count},
error="max_steps_reached",
)
return {"answer": "执行超时", "steps": step_count}
3.3 工具 Mock 与回放引擎
class ToolMock:
"""工具 Mock，用于调试时避免副作用"""
def __init__(self, name: str, responses: dict[str, Any]):
self.name = name
self.responses = responses # 参数哈希 -> 预设响应
self.call_log: list[dict] = []
async def run(self, **kwargs) -> Any:
"""返回预设响应，记录调用日志"""
param_key = self._hash_params(kwargs)
self.call_log.append({
"params": kwargs,
"param_key": param_key,
"timestamp": time.time(),
})
if param_key in self.responses:
return self.responses[param_key]
# 未预设的参数组合，返回默认响应
return {"mocked": True, "tool": self.name, "params": kwargs}
def _hash_params(self, params: dict) -> str:
"""将参数转为可哈希的字符串"""
import json
return json.dumps(params, sort_keys=True)
class ReplayEngine:
"""基于 Trace 数据的回放引擎"""
def __init__(self, tracer: AgentTracer):
self.tracer = tracer
def replay_step(self, step_index: int) -> dict:
"""回放指定步骤，返回该步骤的输入和输出"""
llm_spans = [s for s in self.tracer.spans if s.span_type == "llm_call"]
if step_index >= len(llm_spans):
return {"error": f"步骤 {step_index} 不存在"}
span = llm_spans[step_index]
return {
"step": step_index,
"name": span.name,
"input": span.input_data,
"output": span.output_data,
"duration_ms": span.duration_ms,
"token_usage": span.token_usage,
"status": span.status,
"error": span.error,
}
def diff_steps(self, step_a: int, step_b: int) -> dict:
"""对比两个步骤的差异，用于定位变化点"""
span_a = self.replay_step(step_a)
span_b = self.replay_step(step_b)
return {
"step_a": span_a,
"step_b": span_b,
"duration_diff_ms": (
span_b.get("duration_ms", 0) - span_a.get("duration_ms", 0)
),
"token_diff": (
span_b.get("token_usage", {}).get("total_tokens", 0)
- span_a.get("token_usage", {}).get("total_tokens", 0)
),
}
def find_divergence(self, other_tracer: "AgentTracer") -> Optional[int]:
"""找到两个 Trace 首次出现差异的步骤索引"""
spans_a = [s for s in self.tracer.spans if s.span_type == "llm_call"]
spans_b = [s for s in other_tracer.spans if s.span_type == "llm_call"]
for i in range(min(len(spans_a), len(spans_b))):
if spans_a[i].output_data != spans_b[i].output_data:
return i
return None
四、Agent 调试体系的代价与边界
追踪数据的存储成本：每次 Agent 执行产生的 Trace 数据可能包含数 KB 到数十 KB 的文本。高频场景下，Trace 数据的存储成本不容忽视。建议对 Trace 数据设置保留策略——成功执行的 Trace 保留 7 天，失败执行的 Trace 保留 30 天。
回放引擎的局限性：回放只能复现已发生的执行路径，无法模拟"如果这步选择了另一个工具会怎样"。要实现这种"假设分析"，需要引入 LLM 重新推理，成本和延迟都会增加。
工具 Mock 的维护成本：Mock 响应需要手动维护，当真实工具的接口变更时，Mock 可能过时。建议从 Trace 数据中自动生成 Mock 响应，减少手动维护。
执行树的复杂度：当 Agent 执行超过 10 步时，执行树会变得非常复杂，难以直观理解。建议提供"折叠"视图——默认只展示 LLM 调用步骤，工具调用步骤按需展开。
| 调试手段 | 定位效率 | 实施成本 | 适用场景 |
|---|---|---|---|
| Span 追踪 | 高 | 低 | 所有 Agent |
| 执行树可视化 | 高 | 中 | 多步推理 Agent |
| 回放引擎 | 中 | 高 | 线上故障复现 |
| 工具 Mock | 中 | 中 | 开发调试阶段 |
| Trace Diff | 高 | 低 | Prompt 变更验证 |
五、总结
Agent 调试的核心是"让不可解释的推理过程变得可观测"。Span 追踪记录每一步的输入、输出和耗时，执行树将推理链路可视化，回放引擎支持故障复现，工具 Mock 避免调试副作用。四者协同，才能从"猜 Prompt"的低效循环中跳出来，进入"看数据、定位问题、验证修复"的工程化调试模式。
落地路线建议：第一步，在 Agent 执行器中集成 Span 追踪，确保每次执行都有完整的 Trace 数据；第二步，实现执行树可视化，快速定位出错步骤；第三步，对高频故障场景建立 Trace 数据集，用于回归测试；第四步，引入工具 Mock 和回放引擎，支持离线调试。调试体系的投入产出比是递增的——初期只需 Span 追踪就能解决 80% 的问题，后续按需引入更高级的调试手段。
这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。
更多推荐
11
0- 0
已为社区贡献9条内容
所有评论(0)