【AI应用开发】Agent 无限 loop（反复调用同一个工具）如何规避？

【AI应用开发】Agent 无限 loop反复调用同一个工具如何规避1. Agent Loop 的四种模式模式A: 完全重复 — query_order(id001) × 8 次参数完全相同 模式B: 渐进重复 — search_docs(退货)→search_docs(退货流程)→search_docs(如何退货) 模式C: 乒乓循环 — query_order → search_kb → query_order → search_kb → ... 模式D: 发散循环 — 不断换工具但都不解决问题广度优先式尝试2. 根因分析根因说明LLM 不记忆历史上下文截断后忘记早期步骤工具返回无完成标记LLM 不知道什么算够了Prompt 鼓励过度探索如果不够详细请再次查询是坏指令工具设计缺陷缺少分页标记LLM 以为还有下页3. 七层防御体系第1层: 步数上限 (max_steps15) ← 最简单有效 第2层: 重复检测熔断 (相同调用≤2次) ← 模式A专用 第3层: Token 预算 (超80%预警) ← 兜底保护 第4层: 语义收敛检测 (余弦相似度0.95) ← 模式B/C/D 第5层: 智能终止信号 (完成标记/答案已覆盖) ← 主动终止 第6层: 超时控制 (全局120s/单步30s) ← 最后防线 第7层: 监控告警 (实时指标自动降级) ← 持续改进4. 第一层步数硬限制classStepLimiter:def__init__(self,default_max15,adaptiveTrue):self.limits{single_query:3,# 单次查询multi_step:8,# 多步执行research:20,# 深度研究}self.default_maxdefault_maxdefget_limit(self,task_typeNone):returnself.limits.get(task_type,self.default_max)defcheck(self,step,max_steps):ifstepmax_steps:returnFalse,达最大步数请立即给最终答案ifstepmax_steps*0.8:returnTrue,f仅剩{max_steps-step}步请尽快总结returnTrue,5. 第二层重复检测熔断classDuplicateDetector:def__init__(self):self.history[]self.max_exact2# 完全相同最多2次self.max_similar3# 相似调用最多3次self.circuit_openFalsedefcheck(self,step,tool_name,args,result_hash):# 检测1: 完全相同same[hforhinself.historyifh[tool]tool_nameandh[args]args]iflen(same)self.max_exact:self.circuit_openTruereturnf⛔ 熔断: {tool_name} 相同参数已调{len(same)}次# 检测2: 乒乓 ABABiflen(self.history)3:last4_names[h[tool]forhinself.history[-3:]][tool_name]iflast4_names[0]last4_names[2]andlast4_names[1]last4_names[3]:returnf⚠️ 检测到乒乓循环:{last4_names[0]}↔{last4_names[1]}# 检测3: 结果停滞same_tool[hforhinself.historyifh[tool]tool_name]iflen(same_tool)2andall(h[result_hash]result_hashforhinsame_tool[-2:]):return⚠️ 结果连续不变继续调用无意义self.history.append({step:step,tool:tool_name,args:args,result_hash:result_hash})returnNone6. 第三层Token 预算管理classTokenBudget:def__init__(self,max_total16000,reserve_output2000):self.max_totalmax_total self.reservereserve_output self.used0defcan_continue(self):remainingself.max_total-self.usedifremainingself.reserve:returnFalse,Token预算耗尽必须立即生成最终答案禁止调用工具ifself.used/self.max_total0.8:returnTrue,Token已用80%请精简returnTrue,defspend(self,tokens):self.usedtokens7. 第四层语义收敛检测importnumpyasnpclassSemanticConvergence:检测 Agent 输出是否还在取得进展def__init__(self,embed_model,threshold0.95):self.embedembed_model self.thresholdthreshold self.history[]# [(step, embedding)]defcheck(self,step,output_text):embself.embed.encode(output_text[:500])self.history.append((step,emb))iflen(self.history)3:returnTrue,recent[h[1]forhinself.history[-3:]]sims[]foriinrange(len(recent)-1):simnp.dot(recent[i],recent[i1])/(np.linalg.norm(recent[i])*np.linalg.norm(recent[i1]))sims.append(sim)ifall(sself.thresholdforsinsims):returnFalse,连续步骤结果高度相似信息增益趋零请停止并总结returnTrue,8. 第五层智能终止信号classTerminationManager:defshould_terminate(self,context):综合判断是否该停止reasons[]# 1. LLM 主动说完成了last_msgself._last_assistant_msg(context[messages])iflast_msgandany(winlast_msgforwin[已完成,以上就是,总结]):reasons.append(LLM声明完成)# 2. 问题已被覆盖回答ifself._question_covered(context):reasons.append(问题已完整回答)# 3. 工具返回完成标记forrincontext[tool_results][-2:]:ifisinstance(r.get(result),dict):ifr[result].get(completion_marker,{}).get(task_completed):reasons.append(f工具{r[tool]}标记完成)# 4. 用户取消last_userself._last_user_msg(context[messages])iflast_userandany(cinlast_userforcin[取消,停止,算了]):reasons.append(用户取消)returnlen(reasons)0,; .join(reasons)ifreasonselsedef_last_assistant_msg(self,msgs):forminreversed(msgs):ifm.get(role)assistantandm.get(content):returnm[content]returnNonedef_last_user_msg(self,msgs):forminreversed(msgs):ifm.get(role)user:returnm.get(content,)returndef_question_covered(self,context):# 简化: 检查问题关键词是否在结果中出现qcontext.get(original_question,)all_textjson.dumps([r.get(result,)forrincontext[tool_results]],ensure_asciiFalse)keywords[wforwinqif\u4e00w\u9fff]ifkeywords:coveredsum(1forkinset(.join(keywords)[:10])ifkinall_text)returncovered7returnFalse9. 第六层超时控制importasyncioclassTimeoutGuard:def__init__(self,global_timeout120,step_timeout30,llm_timeout20):self.global_timeoutglobal_timeout self.step_timeoutstep_timeout self.llm_timeoutllm_timeoutasyncdefrun(self,agent_fn,user_input):try:returnawaitasyncio.wait_for(agent_fn(user_input),timeoutself.global_timeout)exceptasyncio.TimeoutError:return{success:False,result:处理超时请缩小问题范围重新提问。,reason:global_timeout}10. 综合防御框架classLoopSafeAgent:集成所有防御机制的 Agentdef__init__(self,llm,tools,embed_model):self.llmllm self.toolstools self.step_limiterStepLimiter()self.dup_detectorDuplicateDetector()self.token_budgetTokenBudget()self.convergenceSemanticConvergence(embed_model)self.terminationTerminationManager()self.timeoutTimeoutGuard()self.terminated_earlyFalse# ← 关键标记是否提前终止asyncdefrun(self,user_input,task_typeNone):state{messages:[],tool_results:[],current_step:0,original_question:user_input}max_stepsself.step_limiter.get_limit(task_type)forstepinrange(1,max_steps1):state[current_step]step# 1. 步数检查can,warningself.step_limiter.check(step,max_steps)ifnotcan:self.terminated_earlyTruereturnself._force_summarize(state,warning)ifwarning:state[messages].append({role:system,content:warning})# 2. Token 预算检查can,msgself.token_budget.can_continue()ifnotcan:returnself._force_summarize(state,msg)# 3. 执行步骤responseawaitself._execute_step(state)# 4. 如果调用了工具记录并检测ifresponse.tool_calls:fortcinresponse.tool_calls:resultawaitself._call_tool(tc)# 重复检测loop_warningself.dup_detector.check(step,tc.name,tc.args,hashlib.md5(json.dumps(result).encode()).hexdigest())ifloop_warning:state[messages].append({role:system,content:f⚠️{loop_warning}})ifself.dup_detector.circuit_open:self.terminated_earlyTrue# ← 熔断强制终止returnself._force_summarize(state,工具熔断)# 语义收敛检测progress,msgself.convergence.check(step,str(result)[:500])ifnotprogress:state[messages].append({role:system,content:f⚠️{msg}})state[tool_results].append({step:step,tool:tc.name,args:tc.args,result:result})# 5. 终止条件检查should_stop,reasonself.termination.should_terminate(state)ifshould_stopandresponse.finish_reasonstop:returnresponse.content# 更新 Token 预算估算self.token_budget.spend(len(str(state[messages][-1]))//4)# 达到最大步数returnself._force_summarize(state,达到最大步数)def_force_summarize(self,state,reason):强制总结已获取的信息ifnotstate[tool_results]:return抱歉处理过程中遇到问题请重新描述您的问题。summary_prompt(f执行被终止原因:{reason}。f请基于以下已获取的信息给用户一个完整的回答\n\njson.dumps([{步骤:r[step],工具:r[tool],结果:str(r[result])[:300]}forrinstate[tool_results]],ensure_asciiFalse,indent2))returnself.llm.chat(summary_prompt)核心思想不要把防止 Loop 仅当作 Prompt 工程问题——必须在代码层面建立多层防线。LLM 不会主动停下来要靠系统强制停下来。