开源项目性能优化的复现方法若 Agent 在处理特殊 Markdown 时输出不完整的 Tool Calling JSON回退机制若将错误堆栈直接放回 Prompt 并无限重试可能重复产生错误调用、耗尽 Token 预算并创建重复工单。把大模型接入工作流自动化绝不能给它“无限重试”的自由。这一次失败的生产事故暴露了系统在链路 Trace 可观测性缺失与确定性格式防线不完善上的致命漏洞。死循环故障的证据链拆解与防线重构复盘这次事故根因在于缺少完整的上下文 Trace 追踪以及解析失败后缺乏硬性的控制闸门。完整的故障证据链排查如下生产级防死循环与格式自愈工作流代码为了彻底杜绝此类死循环我们需要在工作流引擎中注入三重防线结构化 JSON Schema 强校验、自动语法修复自动补齐缺失括号、以及带有硬性 Step 限制与幂等控频的执行器。下面的 Python 代码展示了包含格式自愈与死循环隔离的 Agent 工作流执行器import json import re import logging from typing import Dict, Any, Optional logging.basicConfig(levellogging.INFO, format%(asctime)s - [%(levelname)s] - %(message)s) class AgentExecutionError(Exception): Agent 执行内部错误 pass class SafeAgentWorkflowEngine: def __init__(self, max_steps: int 5, max_token_budget: int 10000): self.max_steps max_steps self.max_token_budget max_token_budget def auto_repair_json(self, raw_text: str) - Optional[Dict[str, Any]]: 第一重防线针对大模型输出常见缺陷的 JSON 容错与自愈修复器 cleaned raw_text.strip() # 1. 尝试直接解析 try: return json.loads(cleaned) except json.JSONDecodeError: pass # 2. 提取 Markdown 代码块中的 JSON 字符串 match re.search(rjson\s*(\{.*?\})\s*, cleaned, re.DOTALL) if match: try: return json.loads(match.group(1)) except json.JSONDecodeError: pass # 3. 容错修复尝试自动补齐缺少结尾右括号的情况 if cleaned.startswith({) and not cleaned.endswith(}): try: repaired cleaned } return json.loads(repaired) except json.JSONDecodeError: pass return None def validate_schema(self, data: Dict[str, Any], required_fields: list) - bool: 第二重防线确定性的 Schema 字段与数据类型校验 for field in required_fields: if field not in data: logging.error(fSchema validation failed: missing field {field}) return False return True def run_workflow_step(self, user_input: str, system_prompt: str) - Dict[str, Any]: 第三重防线带有硬性步数上限与死循环隔离的执行循环 step_count 0 used_tokens 0 execution_history [] # 记录已经尝试过的错误 Output Hash防止完全相同的损坏输出反复提交 seen_error_hashes set() current_prompt user_input while step_count self.max_steps: step_count 1 logging.info(fExecuting Workflow Step {step_count}/{self.max_steps}...) # 模拟调用大模型 API (生产环境替换为实际 SDK 调用) raw_response, estimated_tokens self._mock_llm_call(current_prompt, step_count) used_tokens estimated_tokens # 检查预算是否超限 if used_tokens self.max_token_budget: raise AgentExecutionError(fToken budget exceeded ({used_tokens} {self.max_token_budget}). Terminating.) # 1. 尝试自愈解析 JSON parsed_data self.auto_repair_json(raw_response) if parsed_data is None: response_hash hash(raw_response) if response_hash in seen_error_hashes: # 发现模型陷入了完全相同的错误反复吐字状态立刻死循环熔断 raise AgentExecutionError(Detected DEAD LOCK loop: Model returned identical corrupt response twice.) seen_error_hashes.add(response_hash) # 仅将干净的错误提示喂回严禁塞入长篇 Python Traceback 干扰模型 current_prompt ( Your previous response was NOT valid JSON. Please reply with strictly VALID JSON containing action and payload keys, no other text. ) continue # 2. Schema 硬性校验 if not self.validate_schema(parsed_data, [action, payload]): current_prompt JSON schema invalid: Missing action or payload fields. continue # 成功得到合格合规的结构化输出 logging.info(fWorkflow step succeeded at step {step_count}. Result: {parsed_data}) return { status: success, total_steps: step_count, used_tokens: used_tokens, data: parsed_data } # 超出最大步数限制强制触发人工告警与回滚 raise AgentExecutionError(fWorkflow reached max steps limit ({self.max_steps}) without valid output.) def _mock_llm_call(self, prompt: str, step: int) - tuple: 模拟模型吐字演示自愈过程 if step 1: # 模拟缺少右括号的坏 JSON return json\n{action: create_jira, payload: {summary: bug report , 1200 else: # 模拟自愈后返回正常 JSON return {action: create_jira, payload: {summary: bug report}}, 800 # 执行验证 if __name__ __main__: engine SafeAgentWorkflowEngine(max_steps3, max_token_budget5000) try: result engine.run_workflow_step( user_inputExtract bug from user feedback, system_promptYou are a helpful workflow assistant. ) print(Final Workflow Result:, json.dumps(result, indent2)) except AgentExecutionError as e: print(Workflow Intercepted by Safety Guard:, e)工作流治理的三条落地反思从一次死循环事故中汲取教训智能工作流在进入生产环境前应收口三项工程细节绝对不要把未过滤的 Exception Traceback 喂给大模型当解析器报错时塞回给模型的 Prompt 应是标准化、简短的规范约束说明如“请输出合法 JSON”而不是把几百行的 Python/Node 报错堆栈直接拼进 Context这会导致模型彻底迷失在上下文垃圾信息中。硬性限制 Step 上限与全局 Token 预算在 Agent 代码的while循环头部强制校验step_count MAX_STEPS。任何单个 Workflow 实例的 Tool Calling 步数严禁超过 5 轮。基于 OpenTelemetry 记录全链路 Trace每一次 LLM 调用都应生成包含trace_id和span_id的标准 Trace。将 Prompt 输入、Tool 名称、返回状态码全量上报到 Jaeger/Zipkin一旦某个trace_id下的 Span 数量超过 10立马在网关层报警拦截。用确定性的工程防线兜底大模型的随机输出工作流自动化才能真正从“展示Demo”走向“生产可用”。