AI Agent工程实战系列 · 第04篇 / 共10篇
ReAct vs Plan-and-Execute 怎么选,长任务的检查点设计
以及人工干预接口的正确实现方式
一个死循环的解剖
某天监控告警:一个Agent任务已经连续运行了47分钟,调用了238次LLM,花费了$12.3。
进去一看,Agent在做什么?
步骤1:搜索竞品A的价格信息 步骤2:搜索竞品B的价格信息 步骤3:对比价格,生成报告 → 工具调用失败(网络超时) → 重试步骤3 → 再次失败 → Agent决定:可能是数据不够,重新执行步骤1 → 重新搜索,数据和之前一样 → 再次执行步骤3 → 再次失败 → Agent决定:可能是数据不够,重新执行步骤1 ...(循环238次)原因很清楚:步骤3的失败是网络问题,重新执行步骤1根本解决不了问题。但Agent没有意识到这一点,它只知道"步骤3失败了,可能需要更多数据",于是陷入了无效的重试循环。
没有循环检测,没有最大步骤数限制,没有人工干预接口。三个都没有。
这是今天要解决的问题。
任务规划的两种主流范式
在写防死循环的代码之前,先把两种规划范式的区别说清楚——因为选错范式,会在架构层面埋下死循环的根因。
ReAct:边想边做
ReAct(Reason + Act)是最常见的Agent范式。每一步都是:思考当前状态 → 决定下一步行动 → 执行 → 观察结果 → 再思考。
用户:帮我查一下订单ORDER-001的物流状态,如果已发货就发邮件通知我 ReAct执行过程: 思考:需要先查订单状态 行动:call get_order(order_id="ORDER-001") 观察:{"status": "shipped", "tracking": "SF123456"} 思考:订单已发货,现在需要发邮件 行动:call send_email(to=user_email, subject="订单已发货", ...) 观察:{"success": true} 思考:任务完成 输出:已查询到订单状态为已发货,并已发送邮件通知。ReAct的优点:灵活,能根据中间结果动态调整;适合步骤不确定、需要根据结果决定下一步的任务。
ReAct的致命弱点:没有全局任务视图。Agent不知道自己走了多少步,不知道还有多少步,不知道有没有在绕圈子。这正是死循环的温床。
Plan-and-Execute:先规划再执行
Plan-and-Execute把任务分成两个阶段:先让LLM生成完整的执行计划,再逐步执行计划。
用户:帮我生成一份竞品分析报告 阶段一(规划): LLM生成计划: 步骤1:搜索竞品A的基本信息 步骤2:搜索竞品B的基本信息 步骤3:搜索竞品A的定价信息 步骤4:搜索竞品B的定价信息 步骤5:综合以上信息,生成对比报告 阶段二(执行): 按计划逐步执行,每步完成后更新状态 如果某步失败,不是"重新想",而是"执行备用方案或上报"Plan-and-Execute的优点:有全局视图,步骤有限,便于监控和中断恢复。
Plan-and-Execute的缺点:计划是静态的,遇到计划外的情况(中间结果和预期完全不同)需要重新规划,增加复杂度。
怎么选
用 ReAct 的场景: ✓ 任务步骤少(3步以内) ✓ 每步结果高度不确定,需要动态决策 ✓ 任务不需要人工审核中间步骤 ✓ 失败影响小,可以快速重试 用 Plan-and-Execute 的场景: ✓ 任务步骤多(5步以上) ✓ 任务有副作用(发邮件、修改数据) ✓ 需要人工审核某些关键步骤 ✓ 任务可能被中断,需要从断点恢复 ✓ 需要向用户展示进度 两者都不适合: ✗ 任务本质上是并行的 (应该拆成多个独立的小任务并行执行)防死循环的四道防线
不管用哪种规划范式,这四道防线都必须有。
防线一:最大步骤数硬限制
fromdataclassesimportdataclassfromtypingimportOptional,List,Dict,Any,Callableimporttime@dataclassclassExecutionLimits:"""Agent执行的硬限制"""max_steps:int=30# 最大执行步骤数max_llm_calls:int=50# 最大LLM调用次数max_tool_calls:int=40# 最大工具调用次数max_duration_seconds:int=300# 最大执行时间(5分钟)max_cost_usd:float=1.0# 最大花费(美元)max_retries_per_step:int=3# 每个步骤最大重试次数classExecutionGuard:""" 执行守卫:监控Agent的执行状态, 在触碰任何限制时强制中断 """def__init__(self,limits:ExecutionLimits):self.limits=limits self.reset()defreset(self):self.step_count=0self.llm_call_count=0self.tool_call_count=0self.start_time=time.time()self.total_cost_usd=0.0self.step_retry_counts:Dict[str,int]={}defcheck_before_step(self,step_id:str)->None:""" 每步执行前调用,任意条件触发则抛出异常中断执行 """self.step_count+=1violations=[]ifself.step_count>self.limits.max_steps:violations.append(f"超过最大步骤数限制 ({self.limits.max_steps}步)")elapsed=time.time()-self.start_timeifelapsed>self.limits.max_duration_seconds:violations.append(f"超过最大执行时间 ({self.limits.max_duration_seconds}秒)")ifself.total_cost_usd>self.limits.max_cost_usd:violations.append(f"超过最大花费限制 (${self.limits.max_cost_usd})")ifviolations:raiseExecutionLimitExceeded(violations=violations,step_count=self.step_count,elapsed_seconds=elapsed,total_cost=self.total_cost_usd)defcheck_retry(self,step_id:str)->None:"""检查单个步骤的重试次数"""count=self.step_retry_counts.get(step_id,0)+1self.step_retry_counts[step_id]=countifcount>self.limits.max_retries_per_step:raiseMaxRetriesExceeded(step_id=step_id,retry_count=count,max_retries=self.limits.max_retries_per_step)defrecord_llm_call(self,cost_usd:float=0.0):self.llm_call_count+=1self.total_cost_usd+=cost_usdifself.llm_call_count>self.limits.max_llm_calls:raiseExecutionLimitExceeded(violations=[f"超过最大LLM调用次数 ({self.limits.max_llm_calls}次)"],step_count=self.step_count,elapsed_seconds=time.time()-self.start_time,total_cost=self.total_cost_usd)defget_stats(self)->dict:return{"step_count":self.step_count,"llm_calls":self.llm_call_count,"tool_calls":self.tool_call_count,"elapsed_seconds":round(time.time()-self.start_time,1),"total_cost_usd":round(self.total_cost_usd,4),}防线二:循环检测
fromcollectionsimportCounterimporthashlibfromtypingimportListclassLoopDetector:""" 循环检测器:识别Agent是否在重复执行相同的操作 两种循环模式: 1. 精确循环:完全相同的工具调用参数重复出现 2. 模糊循环:相似的操作在短时间内重复出现 """def__init__(self,exact_repeat_threshold:int=2,window_size:int=10):self.exact_repeat_threshold=exact_repeat_threshold self.window_size=window_size self.call_history:List[str]=[]defrecord_tool_call(self,tool_name:str,params:dict)->None:"""记录一次工具调用"""# 生成调用的指纹(工具名+参数的哈希)call_repr=f"{tool_name}:{sorted(params