
Agent 故障复盘按状态机还原工具调用链Agent 故障复盘应按状态机还原每次模型响应、工具入参和状态迁移。先找出第一次偏离预期的位置再讨论 Prompt 或重试策略避免只盯着最后一条报错。问题现象与排查入口可以构造一条失败链路OCR 工具返回无法解析的字符串Agent 随后调用格式化工具格式化工具再次返回解析错误。测试要检查状态机能否识别重复错误并停止调用。两者相互传递错误后Agent 会重复自我对话和工具调用直到外层超时才停止。影响大小取决于重试上限、上下文长度和并发量应通过调用轨迹统计轮次与 Token而不是预设一个事故数字。这类失效说明多模态 Agent 若缺少边界隔离和异常阻断一个格式解析错误可能沿工具调用链扩散影响其他任务。错误回包如果没有终止分支Agent 可能反复重试。防线重点是标准化错误、限制轮次并在重试前裁剪不再需要的多模态载荷。故障排查根因定位与 Agent 幻觉链路推导复盘故障根因主要暴露了三个维度的工程设计漏洞没有对 Tool 回包建立 Schema 隔离工具执行失败时直接将原始 Python 堆栈 Trace 扔给了模型导致模型产生“我可以自己修复代码”的幻觉。缺乏单次会话的 Token 与轮次硬性封顶前端发起请求后后端 Agent 引擎没有设定单次 Interaction 的最大轮数Max Rounds。多模态上下文未切断在多轮重试过程中每一次重试都带着完整的原始高分辨率图片 Base64 编码导致 Prefill 阶段的 Token 开销呈现持续增加。通过分析错误日志可以看到 Agent 引擎在发生死循环时的内存与 Token 消耗走势[23:41:02] Step 1: User upload image (Tokens: 1450) - Tool: OCR_Extract (Success) [23:41:05] Step 2: Agent Tool Call - Tool: Format_JSON (Error: Invalid \u0000 char) [23:41:08] Step 3: Agent Retrying - Tool: Format_JSON (Error: Invalid \u0000 char) ... (中间重复 28 次带图全量上下文重试) [23:42:15] Step 32: Agent Retrying - Tokens: 48,200/req! - System Timeout!确定性防线与自动恢复重构实践可以在 Agent 运行时增加错误分类、重试上限和上下文裁剪。每项规则都应有对应的失败用例证明它能终止重复调用且不会误删必要上下文。在重构后的代码中引入了ErrorBucket异常桶与多模态 Payload 脱敏机制。一旦检测到 Tool 返回错误及时将高开销的多模态图片从后续重试 Context 中抹去仅保留文本简报。import time import json from typing import Dict, Any, List class ResilientAgentEngine: def __init__(self, max_allowed_steps: int 5, max_tool_errors: int 2): self.max_allowed_steps max_allowed_steps self.max_tool_errors max_tool_errors def run_agent_loop(self, session_id: str, initial_text: str, image_bytes: Optional[bytes] None) - Dict[str, Any]: step_count 0 tool_error_count 0 context_history: List[Dict[str, Any]] [] # 初始帧挂载图像信息 if image_bytes: context_history.append({ role: user, content: initial_text, has_image: True, image_token_cost: 1200 # 估算图片 Token 占用 }) else: context_history.append({role: user, content: initial_text}) while step_count self.max_allowed_steps: step_count 1 print(f[{session_id}] 执行 Step {step_count}/{self.max_allowed_steps}...) # 模拟 LLM 决策过程 llm_decision self._simulate_llm_step(context_history, step_count) if llm_decision[type] FINAL_ANSWER: return {status: SUCCESS, answer: llm_decision[content], steps: step_count} # 如果决定调用 Tool if llm_decision[type] TOOL_CALL: tool_name llm_decision[tool_name] tool_args llm_decision[tool_args] tool_result self._execute_tool_safely(tool_name, tool_args) if not tool_result[success]: tool_error_count 1 print(fWarning: 工具 [{tool_name}] 报错 ({tool_error_count}/{self.max_tool_errors})) # 精简上下文发生错误时移除高开销的图像数据防止 Token 爆炸 self._sanitize_context_history(context_history) if tool_error_count self.max_tool_errors: print(fError: 触发工具错误熔断停止 Agent Loop) return { status: DEGRADED_FALLBACK, answer: 抱歉系统在处理您的图片时遇到格式异常已为您转接人工客服。, reason: fTool [{tool_name}] 连续报错熔断 } # 回传给模型经过脱敏的标准化错误描述严禁直接丢 Raw StackTrace context_history.append({ role: tool, name: tool_name, content: fTool Execution Failed: {tool_result[clean_error]}. Please try alternative approaches. }) else: context_history.append({role: tool, name: tool_name, content: json.dumps(tool_result[data])}) return {status: TIMEOUT_EXCEEDED, answer: 处理超时请稍后重试。, steps: step_count} def _sanitize_context_history(self, history: List[Dict[str, Any]]): for msg in history: if msg.get(has_image): msg[has_image] False msg[content] (注: 原始图像数据因重试已被清理) def _simulate_llm_step(self, history: List[Dict[str, Any]], step: int) - Dict[str, Any]: # 演示用逻辑 if step 1: return {type: TOOL_CALL, tool_name: OCR_Tool, tool_args: {file: invoice.jpg}} return {type: TOOL_CALL, tool_name: Format_Tool, tool_args: {raw: bad_data}} def _execute_tool_safely(self, name: str, args: Dict[str, Any]) - Dict[str, Any]: # 模拟频繁报错的工具 return {success: False, clean_error: INVALID_CHARACTER_ENCODING}长效治理与监控防线建设多模态 Agent 上线前还要落实四项持续治理要求治理防线物理措施监控指标与报警阈值熔断限速闸门并发上限、超时与熔断记录监控指标与报警阈值多模态 Payload 瘦身第一轮推理后在 Context 中将图片数据转换为特征 URL 或纯文本摘要记录监控指标与报警阈值错误回传标准化封装统一的 Tool Exception 拦截层只给 LLM 返回标准化 Error Code从调用预算和历史基线确定全局预算熔断器配置中心实时计算 API 消费总额达到单小时阈值即暂停 Agent 调用记录监控指标与报警阈值故障复盘不应当是寻找责任人的问责会而是完善工程体系的契机。Agent 的非确定性输出应由 Schema、权限和状态机限制。上线前覆盖格式错误、工具超时和重复调用并确认人工接管路径。