Agent 工作流巡检给工具调用加边界和熔断Agent 工作流需要温和但明确的边界工具参数先校验循环次数有上限外部服务超时后能降级。巡检脚本负责暴露状态不应悄悄替用户执行高风险修复。新手在 Prompt 与 Tool 定义中最常踩的雷区许多开发者在刚接触 Agent 架构时往往沉迷于给模型设定宏大的“角色设定”却忽视了基础约束的建立。在长期的上线运营与故障排查中可以梳理出新手容易跌入的四大误区----------------------------------------------------------------------- | Agent 常见设计误区与后果对照表 | ----------------------------------------------------------------------- | 常见误区 | 典型代码/Prompt 表现 | 运营灾难后果 | ---------------------------------------------------------------------- | 角色描述模糊 | 你是一个极其聪明全能的助手 | 行为不可控越权 | | 缺乏 JSON 格式强校验 | 请输出 JSON包含 action | 解析报错导致死循环| | 未限制最大递归轮数 | while True: step() | 死循环刷爆 API 额度| | 假设工具调用百分百成功 | tools[action_name](args) | 抛错挂起状态丢失| ----------------------------------------------------------------------尤其是在定义 Tool 参数 schema 时如果缺乏明确的约束条件例如数字范围、字符串正则表达式、非空检查大模型容易在尝试填充参数时出现语义漂移。这种漂移一旦进入循环就会引发极难排查的连续报错。Agent 运行状态决策与自愈防线为了防止 Agent 在运行过程中失控需要为其建立一套全生命周期的状态机监控在每一个 Tool 执行完成后进行止损断言校验。Python 可部署的 Agent 状态机与止损检查表引擎下面的状态机示例展示 Prompt 字段校验、步骤上限、重复调用检测和工具异常分支。接入真实服务前还需补充持久化、并发控制与失败恢复测试。import json import logging from typing import Dict, Any, Callable, List, Optional from dataclasses import dataclass, field logging.basicConfig(levellogging.INFO, format%(asctime)s - [%(levelname)s] %(message)s) logger logging.getLogger(AgentCircuitBreaker) dataclass class AgentState: task_id: str max_steps: int 5 max_tool_failures: int 3 current_step: int 0 failed_tool_calls: int 0 history_actions: List[str] field(default_factorylist) is_halted: bool False halt_reason: Optional[str] None class ToolExecutionError(Exception): pass class SafeAgentRunner: def __init__(self, state: AgentState): self.state state self.tools: Dict[str, Callable] {} self._register_default_tools() def _register_default_tools(self): 注册内置工具函数 self.tools[fetch_user_journal] self._mock_fetch_journal self.tools[analyze_sentiment] self._mock_analyze_sentiment def _mock_fetch_journal(self, date_str: str) - str: if not date_str or len(date_str) ! 10: raise ToolExecutionError(参数错误date_str 格式必须为 YYYY-MM-DD) return 今天阳台上的花开了做了一顿丰盛的午饭心情格外舒畅。 def _mock_analyze_sentiment(self, text: str) - str: if not text: raise ToolExecutionError(参数错误解析文本不能为空) return self.sentiment_model.summarize(text) def parse_llm_json_output(self, raw_output: str) - Dict[str, Any]: 严格的 JSON 解析防护 try: parsed json.loads(raw_output) if action not in parsed or action_input not in parsed: raise ValueError(缺失必备字段 action 或 action_input) return parsed except Exception as e: logger.warning(fLLM 输出 JSON 解析失败: {str(e)} | 原文本: {raw_output}) raise def check_loop_repetition(self, action_key: str) - bool: 检测死循环调用行为 self.state.history_actions.append(action_key) # 如果连续三次采取完全相同的 Action if len(self.state.history_actions) 3: recent self.state.history_actions[-3:] if recent[0] recent[1] recent[2]: return True return False def step(self, simulated_llm_response: str) - Tuple[bool, str]: 单步执行状态演进与止损断言 if self.state.is_halted: return True, fAgent 已挂起终止执行。原因: {self.state.halt_reason} self.state.current_step 1 logger.info(f[{self.state.task_id}] 推进至第 {self.state.current_step}/{self.state.max_steps} 步) # 1. 递归深度检查 if self.state.current_step self.state.max_steps: self.state.is_halted True self.state.halt_reason EXCEEDED_MAX_STEPS_LIMIT logger.error(f[{self.state.task_id}] 触发止损超出最大允许步骤上限) return True, 抱歉梳理过程耗时过长已为您生成阶段性摘要。 # 2. 解析与格式校验 try: action_data self.parse_llm_json_output(simulated_llm_response) except Exception: self.state.failed_tool_calls 1 if self.state.failed_tool_calls self.state.max_tool_failures: self.state.is_halted True self.state.halt_reason MAX_PARSING_FAILURES_REACHED return True, 无法准确理解格式需求系统已自动挂起安全止损。 return False, 输出格式有误正在请求模型修正格式... action action_data[action] action_input action_data[action_input] action_key f{action}:{json.dumps(action_input, ensure_asciiFalse)} # 3. 死循环监测 if self.check_loop_repetition(action_key): self.state.is_halted True self.state.halt_reason INFINITE_LOOP_DETECTED logger.error(f[{self.state.task_id}] 触发止损检测到完全相同的死循环操作 {action_key}) return True, 检测到连续重复动作已自动中断运行。 # 4. 安全执行工具 if action finish: return True, f任务顺利完成! 最终结论: {action_input.get(final_answer, )} if action not in self.tools: self.state.failed_tool_calls 1 logger.warning(f未知工具名: {action}) return False, f未找到名为 {action} 的工具 try: tool_func self.tools[action] result tool_func(**action_input) logger.info(f工具 [{action}] 执行成功, 结果样本: {result[:20]}...) return False, f工具返回结果: {result} except Exception as err: self.state.failed_tool_calls 1 logger.error(f工具 [{action}] 运行异常: {str(err)}) if self.state.failed_tool_calls self.state.max_tool_failures: self.state.is_halted True self.state.halt_reason MAX_TOOL_EXCEPTIONS_REACHED return True, 依赖底层工具发生多次故障已安全停止。 return False, f工具执行出错: {str(err)} # 验证止损检查表引擎机制 if __name__ __main__: print( 测试场景 1: 检测死循环自动止损 ) state1 AgentState(task_idtask_001, max_steps5) runner1 SafeAgentRunner(state1) # 模拟 LLM 连续吐出相同的畸形/重复指令 bad_llm_output {action: fetch_user_journal, action_input: {date_str: 2026-08-15}} for i in range(4): done, msg runner1.step(bad_llm_output) print(f步骤 {i1} 状态: Done{done} | 反馈信息: {msg}) if done: break为智能体设定可验证的执行边界构建富有温情的 AI 产品并不意味着要放任智能体去进行不受约束的自由游走。恰恰相反越是贴近生活的核心应用越需要用最严谨的工程边界去呵护它。Prompt 变量校验、循环上限和工具异常捕获应一起进入测试。触发断路后保留任务状态并把后续动作交给用户或人工处理避免静默重试继续消耗预算。