AI Agent 调试实战:链路追踪、Prompt 可视化与异常定位的系统方法
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[工具调用1br/name args] D -- E[工具返回1br/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_factorylambda: 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_factorydict) 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_factorylambda: uuid.uuid4().hex[:16]) agent_name: str user_input: str final_output: str spans: list[Span] field(default_factorylist) start_time: float field(default_factorytime.time) end_time: float 0 metadata: dict field(default_factorydict) 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_nameself.name, user_inputuser_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_typeSpanType.LLM_CALL, namefllm_call_iter_{iteration}, input_datamessages[-1], ) llm_span.start_time time.time() try: response self.llm.chat( messagesmessages, toolsself._get_tool_schemas(), temperature0.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_typeSpanType.TOOL_CALL, nametool_name, statusSpanStatus.ERROR, errorf检测到循环调用: {call_sig}, ) trace.add_span(loop_span) final_output 错误Agent 陷入工具调用循环 break # 记录工具调用 Span tool_span Span( span_typeSpanType.TOOL_CALL, nametool_name, input_datatool_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[i1] tool_names[i2]: 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: fLLM 调用耗时 {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: fAgent 执行了 {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} , fAgent: {trace.agent_name}, fInput: {trace.user_input[:100]}..., fDuration: {trace.total_duration_ms:.0f}ms, fTokens: {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} [{i1}] {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(fFinal 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 对比时需要考虑随机性。建议固定temperature0用于调试生产环境使用低温度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 在生产环境中就是黑盒。