Agent 行为追踪每一步都要记下来才能事后复盘一、Agent 说我已经完成了但你根本不知道它做了什么一个典型的 Agent 交互用户发出指令 → Agent 进行 3 次工具调用 → 最终返回结果。表面看起来正常但事后复盘时你会发现 Agent 的第 2 次工具调用参数有个偏差导致获取了错误的数据而 Agent 自己没发现这个错误基于错误数据继续推理最后给出了看起来合理但实际错误的结论。更糟糕的是如果没有记录中间步骤你永远不知道这个错误发生过。Agent 的自主性在带来灵活性的同时也带来了可解释性的黑洞。传统的 LLM 应用是输入→输出的单步模式调试相对简单。Agent 是思考→行动→观察→思考…的多步循环每一步的微小偏差都可能被后续步骤放大形成复合错误。Agent 的行为追踪需要回答三个问题每一步做出了什么决策思维链追踪工具调用的参数和返回值是什么工具链追踪哪个决策导致了最终的错误根因分析二、Agent 执行的状态机可观测的是状态转换而非最终结果stateDiagram-v2 [*] -- Planning: 接收用户指令 Planning -- Reasoning: 制定执行计划 Reasoning -- ToolCall: 需要工具调用 Reasoning -- Response: 直接回答 ToolCall -- Observation: 工具返回结果 Observation -- Reasoning: 分析工具结果 Observation -- Error: 工具调用失败 Error -- Reasoning: 重试或调整 Error -- Fallback: 超过重试次数 Response -- [*]: 完成 Fallback -- Response: 降级回答 note right of Observation 关键观测点 - 工具调用参数是否正确 - 返回结果是否匹配预期 - 是否注意到关键信息 end note note right of Error 关键观测点 - 失败原因分类 - 重试策略是否有效 - 是否需要人工介入 end note在状态转换的每一步都需要记录当前状态Planning → Reasoning → ToolCall → Observation 中的哪一个状态转换的原因为什么决定调用这个工具为什么认为任务已完成状态转换的结果工具返回了什么Agent 对返回结果的理解是什么。三、Agent Trace 采集器不侵入业务逻辑的埋点方案import json import time import uuid from dataclasses import dataclass, field from typing import List, Dict, Any, Optional from enum import Enum class AgentState(Enum): Agent 执行状态枚举 PLANNING planning REASONING reasoning TOOL_CALL tool_call OBSERVATION observation ERROR error RESPONSE response FALLBACK fallback dataclass class AgentStep: Agent 执行步骤记录 step_id: str field(default_factorylambda: str(uuid.uuid4())[:8]) state: AgentState timestamp: float field(default_factorytime.time) # 设计原因每个步骤记录详细的输入输出 # 不做省略或摘要完整保留方便事后回放 input_data: Dict[str, Any] field(default_factorydict) output_data: Dict[str, Any] field(default_factorydict) # 设计原因记录 LLM 调用的 Token 消耗 # 用于成本和效率分析也用于检测 Token 消耗异常 tokens_used: int 0 # 设计原因记录与本步骤相关的元信息 # 如模型名称、temperature、工具名称等 metadata: Dict[str, Any] field(default_factorydict) dataclass class AgentTrace: 一次完整 Agent 会话的追踪记录 trace_id: str field(default_factorylambda: str(uuid.uuid4())[:16]) session_id: str user_query: str steps: List[AgentStep] field(default_factorylist) start_time: float field(default_factorytime.time) end_time: Optional[float] None final_answer: Optional[str] None # 设计原因记录 Agent 的自我评估 # Agent 对自己回答的信心程度对于后续分析非常关键 self_assessment: Optional[str] None class AgentTracer: Agent 行为追踪器 设计原因使用装饰器模式不侵入 Agent 的核心业务逻辑 通过 TraceContext 在 Agent 执行过程中自动采集步骤信息 def __init__(self, storage_backendNone): # 设计原因支持可插拔的存储后端 # 开发环境用本地文件生产环境用远程存储如 Elasticsearch self.storage storage_backend or LocalFileStorage() self.active_traces: Dict[str, AgentTrace] {} def start_trace(self, session_id: str, user_query: str) - AgentTrace: 开始一次追踪 trace AgentTrace(session_idsession_id, user_queryuser_query) self.active_traces[trace.trace_id] trace return trace def record_step( self, trace_id: str, state: AgentState, input_data: Dict, output_data: Dict, metadata: Dict None, tokens_used: int 0 ): 记录一个执行步骤 设计原因记录是原子的——要么全记录要么全不记录 避免部分记录导致的幽灵步骤有头无尾的记录 if trace_id not in self.active_traces: # 设计原因未注册 trace_id 表示追踪未正确初始化 # 不应静默忽略应该抛出错误或记录告警 raise ValueError(f未找到 Trace: {trace_id}请先调用 start_trace()) step AgentStep( statestate, input_datainput_data, output_dataoutput_data, metadatametadata or {}, tokens_usedtokens_used ) self.active_traces[trace_id].steps.append(step) def end_trace(self, trace_id: str, final_answer: str, self_assessment: str None): 结束追踪并持久化 trace self.active_traces.pop(trace_id) trace.end_time time.time() trace.final_answer final_answer trace.self_assessment self_assessment # 设计原因持久化时做一次统计汇总 # 将原始步骤数据转化为分析指标方便查询 summary self._summarize(trace) self.storage.save({ trace: trace, summary: summary, timestamp: time.time() }) return trace def _summarize(self, trace: AgentTrace) - Dict: 从原始步骤中提取汇总指标 total_tokens sum(s.tokens_used for s in trace.steps) tool_calls [s for s in trace.steps if s.state AgentState.TOOL_CALL] errors [s for s in trace.steps if s.state AgentState.ERROR] return { trace_id: trace.trace_id, duration_seconds: (trace.end_time - trace.start_time), total_steps: len(trace.steps), total_tool_calls: len(tool_calls), total_errors: len(errors), total_tokens: total_tokens, # 设计原因统计状态转换序列如 Planning→Reasoning→ToolCall # 不同任务的典型转换序列不同异常序列是排查问题的线索 state_sequence: [s.state.value for s in trace.steps], } class LocalFileStorage: 本地文件存储后端 def __init__(self, base_path: str ./agent_traces): import os self.base_path base_path os.makedirs(base_path, exist_okTrue) def save(self, data: Dict): trace_id data[trace][trace_id] filepath f{self.base_path}/{trace_id}.json # 设计原因使用 dataclass 的 __dict__ 序列化 # 对于复杂类型如 Enum需要自定义 JSON encoder with open(filepath, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2, defaultstr)四、追踪数据的使用与存储取舍全量记录还是采样Agent 行为的完整追踪数据量大需要考虑存储策略。全量记录优点不遗漏任何异常事后可以分析任意一次用户交互缺点存储成本高每天数百万次 Agent 调用 × 每 Trace 10KB 数十 GB/天适用金融、医疗等对可追溯性要求极高的行业。采样记录优点存储成本可控保留长期趋势数据缺点低概率的异常行为可能被采样遗漏适用对可追溯性要求一般的通用场景。分层存储策略推荐热数据近 7 天全量存储支持实时查询和回放温数据7~30 天只保留汇总指标和异常 Trace支持趋势分析冷数据30 天以上仅保留统计摘要用于长期容量规划。额外考虑追踪数据包含用户输入隐私合规是硬约束。需要在采集阶段就对敏感信息做脱敏处理PII 屏蔽而非事后删除。五、总结Agent 的行为追踪需要记录每一步的状态转换、输入输出和决策依据。追踪器应采用非侵入设计如装饰器模式不修改 Agent 的核心业务逻辑。Agent 的状态机视角将执行过程分解为可观测的状态转换序列根因分析通过回溯状态转换找到最早偏离预期的步骤。存储策略需要在全量信息和存储成本之间权衡——分层存储热/温/冷是工程实践中的平衡方案。追踪数据中的用户输入涉及隐私合规需在采集阶段做脱敏处理。
