Agent 故障复盘:按状态机还原工具调用链
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(f"Warning: 工具 [{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(f"Error: 触发工具错误熔断,停止 Agent Loop!") return { "status": "DEGRADED_FALLBACK", "answer": "抱歉,系统在处理您的图片时遇到格式异常,已为您转接人工客服。", "reason": f"Tool [{tool_name}] 连续报错熔断" } # 回传给模型经过脱敏的标准化错误描述,严禁直接丢 Raw StackTrace context_history.append({ "role": "tool", "name": tool_name, "content": f"Tool 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、权限和状态机限制。上线前覆盖格式错误、工具超时和重复调用,并确认人工接管路径。