评估闭环

为什么Agent调试如此困难 与传统软件不同，AI Agent的执行路径不是确定性的——同样的输入可能产生不同的执行路径和结果。这种非确定性使得传统调试方法（断点、单步执行）效果有限。Agent调试需要一套全新的方法论。 Agent调试的独特挑战 传统软件 AI Agent 确定性执行路径 非确定性，同一输入不同输出 错误即崩溃 错误可能"静默"——不崩溃但结果错误 逻辑可推断 决策基于LLM推理，难以追溯 单一系统 多工具调用、多轮对话、外部依赖 单元测试覆盖 需要语义级别的评估 Agent调试的三层体系 第一层：日志（Logs）—— 发生了什么 第二层：追踪（Traces）—— 为什么发生 第三层：评估（Evaluation）—— 发生得对不对 第一层：结构化日志体系 Agent日志设计原则 import json import time import uuid from enum import Enum from datetime import datetime class LogLevel(Enum): DEBUG = "DEBUG" INFO = "INFO" WARN = "WARN" ERROR = "ERROR" class AgentLogger: """生产级Agent结构化日志""" def __init__(self, agent_name): self.agent_name = agent_name def log(self, level, event, **fields): entry = { "timestamp": datetime.utcnow().isoformat(), "level": level.value, "agent": self.agent_name, "event": event, "trace_id": fields.get("trace_id"), "span_id": fields.get("span_id"), **fields } print(json.dumps(entry, ensure_ascii=False, default=str)) # 使用示例 logger = AgentLogger("research_agent") logger.log(LogLevel.INFO, "tool_call", trace_id="tr_abc123", span_id="sp_001", tool_name="web_search", tool_input={"query": "2026 AI芯片市场"}, tool_output={"results_count": 5}, duration_ms=1200, tokens_used=150 ) 关键日志事件类型 class AgentEventTypes: """Agent生命周期中的关键事件""" # 规划阶段 PLAN_CREATED = "plan_created" # Agent制定了执行计划 PLAN_REVISED = "plan_revised" # 计划被修改 GOAL_DECOMPOSED = "goal_decomposed" # 目标被分解 # 执行阶段 TOOL_CALL_START = "tool_call_start" # 工具调用开始 TOOL_CALL_END = "tool_call_end" # 工具调用结束 TOOL_CALL_ERROR = "tool_call_error" # 工具调用失败 TOOL_CALL_RETRY = "tool_call_retry" # 工具调用重试 # 推理阶段 LLM_CALL_START = "llm_call_start" # LLM调用开始 LLM_CALL_END = "llm_call_end" # LLM调用结束 REASONING_STEP = "reasoning_step" # 推理步骤 DECISION_MADE = "decision_made" # 做出决策 # 状态管理 CONTEXT_UPDATED = "context_updated" # 上下文更新 MEMORY_READ = "memory_read" # 读取记忆 MEMORY_WRITE = "memory_write" # 写入记忆 # 错误与异常 HALLUCINATION_DETECTED = "hallucination_detected" LOOP_DETECTED = "loop_detected" # 检测到循环 BUDGET_EXCEEDED = "budget_exceeded" # 预算超限 MAX_STEPS_REACHED = "max_steps_reached" # 达到最大步数 日志分析常见模式 def analyze_agent_logs(logs): """分析Agent日志，识别常见问题模式""" patterns = { # 模式1：工具调用循环 "tool_loop": detect_tool_loops(logs), # 模式2：LLM调用失败率 "llm_failure_rate": calculate_llm_failure_rate(logs), # 模式3：Token消耗异常 "token_anomaly": detect_token_anomalies(logs), # 模式4：延迟热点 "latency_hotspots": find_latency_hotspots(logs), # 模式5：幻觉信号 "hallucination_signals": detect_hallucination_signals(logs), } return patterns def detect_tool_loops(logs): """检测工具调用循环——Agent反复调用同一工具""" tool_calls = [l for l in logs if l["event"] == "tool_call_end"] loops = [] window = 5 # 检查窗口 for i in range(len(tool_calls) - window): window_calls = tool_calls[i:i+window] tool_names = [c["tool_name"] for c in window_calls] # 如果同一工具在窗口内被调用3次以上 from collections import Counter counts = Counter(tool_names) for tool, count in counts.items(): if count >= 3: loops.append({ "tool": tool, "count": count, "window_start": window_calls[0]["timestamp"], "severity": "high" if count >= 4 else "medium", }) return loops 第二层：全链路追踪 Agent追踪架构 class AgentTracer: """Agent全链路追踪系统""" def __init__(self): self.spans = [] def start_trace(self, agent_name, user_input, context=None): """开始一个新的追踪""" trace_id = f"tr_{uuid.uuid4().hex[:12]}" return { "trace_id": trace_id, "agent_name": agent_name, "user_input": user_input, "context": context, "start_time": time.time(), "spans": [], } def start_span(self, trace, span_name, span_type, parent_id=None): """开始一个Span""" span_id = f"sp_{uuid.uuid4().hex[:8]}" span = { "span_id": span_id, "parent_id": parent_id, "name": span_name, "type": span_type, # llm, tool, reasoning, memory "start_time": time.time(), "inputs": None, "outputs": None, "error": None, } trace["spans"].append(span) return span_id def end_span(self, trace, span_id, outputs=None, error=None): """结束一个Span""" for span in trace["spans"]: if span["span_id"] == span_id: span["end_time"] = time.time() span["duration_ms"] = (span["end_time"] - span["start_time"]) * 1000 span["outputs"] = outputs span["error"] = error break # 追踪使用示例 tracer = AgentTracer() trace = tracer.start_trace("research_agent", "分析2026年AI芯片市场") root_span = tracer.start_span(trace, "research_task", "root") # 规划阶段 plan_span = tracer.start_span(trace, "planning", "reasoning", root_span) plan = agent.plan("分析2026年AI芯片市场") tracer.end_span(trace, plan_span, outputs={"plan": plan}) # 执行阶段 for step in plan.steps: if step.type == "tool_call": tool_span = tracer.start_span(trace, f"tool:{step.tool}", "tool", root_span) result = agent.call_tool(step.tool, step.params) tracer.end_span(trace, tool_span, outputs=result) tracer.end_span(trace, root_span, outputs={"answer": agent.final_answer}) 追踪可视化：执行树 Trace: tr_abc123 | research_agent | 总耗时: 12.3s │ ├── [reasoning] planning (0.8s) │ └── 计划: 1.搜索数据 2.分析数据 3.生成报告 │ ├── [tool] web_search (1.2s) │ ├── query: "2026 AI芯片市场份额" │ └── results: 5条 │ ├── [tool] web_search (1.1s) │ ├── query: "NVIDIA Blackwell vs 国产芯片" │ └── results: 8条 │ ├── [llm] data_analysis (3.5s) │ ├── input_tokens: 2048 │ ├── output_tokens: 1024 │ └── cost: $0.03 │ ├── [tool] write_file (0.3s) │ └── output: report.md │ └── [llm] final_answer (5.4s) ├── input_tokens: 4096 ├── output_tokens: 2048 └── cost: $0.06 这种可视化让Agent的完整执行过程一目了然，是定位问题的关键工具。 ...