从原理到实战:AI Agent停止策略全解析,避免无限循环与资源浪费
本文深入探讨了AI Agent停止策略的设计思路与实现方法。AI Agent本质是一个大循环,若无合理停止机制,会导致无限循环浪费资源或过早停止无法完成任务。文章分析了六种常用停止策略:硬性限制、任务完成检测、显式停止信号、循环检测、错误累积和用户中断。结合OpenManus和Gemini CLI的源码,详细展示了如何通过terminate工具、状态机管理、声明式输出系统和三层循环检测机制实现有效
本文深入探讨了AI Agent停止策略的设计思路与实现方法。AI Agent本质是一个大循环,若无合理停止机制,会导致无限循环浪费资源或过早停止无法完成任务。文章分析了六种常用停止策略:硬性限制、任务完成检测、显式停止信号、循环检测、错误累积和用户中断。结合OpenManus和Gemini CLI的源码,详细展示了如何通过terminate工具、状态机管理、声明式输出系统和三层循环检测机制实现有效停止。强调实际应用中需组合多种策略形成多层防护,在"完成任务"与"防止失控"间找到平衡。
前排提示,文末有大模型AGI-CSDN独家资料包哦!
简单来讲,AI Agent 实现的的大逻辑就是一个大的循环 + 获取上下文 + 不停的 LLM 调用 + 工具的调用。
那么一个关键问题就出现了:这个循环什么时候应该停止?如果处理不当,Agent 可能会陷入无限循环,浪费计算资源,或者过早停止而无法完成任务。本文将深入探讨 AI Agent 停止策略的核心设计思路。
常用停止策略
AI Agent 停止策略无外乎以下几种情况:
1. 硬性限制
最简单粗暴的方法:
- 最大步数限制(比如最多循环 30 次)
- 执行时间限制(比如最多跑 5 分钟)
- API 调用次数限制(比如最多调 100 次)
- API 调用 Token 数限制
这种方法简单有效,但用户体验很差。经常出现任务做到一半被强制停止的情况。
2. 任务完成检测
让 LLM 判断任务是否完成:
# 每次循环后问 LLMresponse = llm.ask("任务是否已经完成?")if response == "是": stop()
3. 显式停止信号
给 Agent 一个专门的"停止"工具:
tools = [ "search", "calculate", "terminate" # 专门用来停止]
当 Agent 调用 terminate 工具时就停止。这个方法不错,但需要在 prompt 里教会 Agent 什么时候该调用它。
4. 循环检测
检测 Agent 是否在做重复的事:
- 连续多次调用同一个工具
- 动作序列出现循环模式(A→B→A→B…)
- 输出内容高度相似
5. 错误累积
连续失败多次就放弃:
if consecutive_errors > 3: stop("连续失败太多次")
6. 用户中断
让用户能随时喊停。
下面我们以 OpenManus 和 Gemini CLI 的源码来看一下他们是怎么做的。
OpenManus 的停止逻辑
OpenManus 的停止机制设计得比较完整,它用了一个多层防护的思路。
核心:terminate 工具
OpenManus 给每个 Agent 都配了一个 terminate 工具:
class Terminate(BaseTool): name: str = "terminate" description = """当请求已满足或无法继续时终止交互。 完成所有任务后,调用此工具结束工作。""" asyncdef execute(self, status: str) -> str: returnf"交互已完成,状态:{status}" 以上为示例,原始代码:from app.tool.base import BaseTool_TERMINATE_DESCRIPTION = """Terminate the interaction when the request is met OR if the assistant cannot proceed further with the task.When you have finished all the tasks, call this tool to end the work."""class Terminate(BaseTool): name: str = "terminate" description: str = _TERMINATE_DESCRIPTION parameters: dict = { "type": "object", "properties": { "status": { "type": "string", "description": "The finish status of the interaction.", "enum": ["success", "failure"], } }, "required": ["status"], } asyncdef execute(self, status: str) -> str: """Finish the current execution""" returnf"The interaction has been completed with status: {status}"
OpenManus 使用的是方案 3,把「何时停止」的决策权交给了 LLM。prompt 里会明确告诉 Agent:任务完成了就调用 terminate。
状态机管理
OpenManus 用状态机来管理 Agent 的生命周期:
class AgentState(Enum): IDLE = "idle" RUNNING = "running" FINISHED = "finished" ERROR = "error"
当检测到特殊工具(如 terminate)被调用时,会触发状态转换:
async def _handle_special_tool(self, name: str, result: Any): if name.lower() == "terminate": self.state = AgentState.FINISHED logger.info("🏁 任务完成!")
步数限制
不同类型的 Agent 有不同的步数上限:
# ToolCallAgent: 30 步# SWEAgent: 20 步# PlanningFlow: 可配置while self.current_step < self.max_steps and self.state != AgentState.FINISHED: self.current_step += 1 await self.step()if self.current_step >= self.max_steps: results.append(f"达到最大步数限制 ({self.max_steps})")
这是一个保底机制,防止 Agent 无限运行。
卡死检测
OpenManus 还会检测 Agent 是否卡住了:
def is_stuck(self) -> bool: # 检查是否有重复的 assistant 消息 # 如果最近的回复都一样,说明卡住了 recent_messages = self.get_recent_assistant_messages() if len(set(recent_messages)) == 1: return True return False
Planning Agent 的结束逻辑
1. 计划完成的判断机制
PlanningFlow 的结束判断并不是简单检查所有步骤是否完成:
# 在主执行循环中while True: # 获取当前需要执行的步骤 self.current_step_index, step_info = await self._get_current_step_info() # 如果没有更多活跃步骤,则结束计划 if self.current_step_index is None: result += await self._finalize_plan() break
2. 步骤状态检查逻辑
_get_current_step_info() 方法负责判断是否还有未完成的步骤:
# 查找第一个非完成状态的步骤for i, step in enumerate(steps): if i >= len(step_statuses): status = PlanStepStatus.NOT_STARTED.value else: status = step_statuses[i] # 如果步骤状态为活跃状态(未开始或进行中),返回该步骤 if status in PlanStepStatus.get_active_statuses(): return i, step_info# 如果没找到活跃步骤,返回 Nonereturn None, None
其中 get_active_statuses() 返回 ["not_started", "in_progress"],意味着只有当所有步骤都是 "completed" 或 "blocked" 状态时,计划才会结束。
3. 计划结束处理
当没有更多活跃步骤时,会调用 _finalize_plan() 方法:
async def _finalize_plan(self) -> str: """使用 LLM 生成计划完成总结""" plan_text = await self._get_plan_text() # 使用 LLM 生成总结 system_message = Message.system_message( "You are a planning assistant. Your task is to summarize the completed plan." ) user_message = Message.user_message( f"The plan has been completed. Here is the final plan status:\n\n{plan_text}\n\nPlease provide a summary of what was accomplished and any final thoughts." ) response = await self.llm.ask(messages=[user_message], system_msgs=[system_message]) return f"Plan completed:\n\n{response}"
Gemini CLI 的停止逻辑
Gemini CLI 的设计思路完全不同,它用了一个更优雅但也更复杂的方案。
subagent 的停止逻辑
1. 达到最大轮次(MAX_TURNS)
if (this.runConfig.max_turns && turnCounter >= this.runConfig.max_turns) { this.output.terminate_reason = SubagentTerminateMode.MAX_TURNS; break;}
这是最简单的保护机制,防止无限循环。
2. 执行超时(TIMEOUT)
let durationMin = (Date.now() - startTime) / (1000 * 60);if (durationMin >= this.runConfig.max_time_minutes) { this.output.terminate_reason = SubagentTerminateMode.TIMEOUT; break;}
注意这里检查了两次超时:
- 在调用 LLM 之前检查一次
- 在调用 LLM 之后又检查一次
这是因为 LLM 调用可能很耗时,要确保不会超时太多。
3. 用户中断(通过 AbortSignal)
if (abortController.signal.aborted) return;
这个检查出现在 stream 处理循环里,确保能及时响应用户的取消操作。
4. 错误异常(ERROR)
catch (error) { console.error('Error during subagent execution:', error); this.output.terminate_reason = SubagentTerminateMode.ERROR; throw error;}
任何未捕获的异常都会导致停止。
5. 目标完成(GOAL)
目标完成的判断分两种情况:
情况A:没有预定输出要求
if (!this.outputConfig || Object.keys(this.outputConfig.outputs).length === 0) { // 没有要求特定输出,LLM 不调用工具就认为完成了 if (functionCalls.length === 0) { this.output.terminate_reason = SubagentTerminateMode.GOAL; break; }}
情况B:有预定输出要求
// 检查是否所有要求的变量都已输出const remainingVars = Object.keys(this.outputConfig.outputs).filter( (key) => !(key in this.output.emitted_vars));if (remainingVars.length === 0) { this.output.terminate_reason = SubagentTerminateMode.GOAL; break;}
声明式输出系统的实现
声明式输出系统的核心是 outputConfig:
// 预先声明需要什么输出this.outputConfig = { outputs: { "summary": "string", "recommendations": "array", "risk_score": "number" }};// Agent 通过 self.emitvalue 工具来产生输出// 每次调用会把值存到 this.output.emitted_vars 里this.output.emitted_vars = { "summary": "这是总结...", "recommendations": ["建议1", "建议2"] // risk_score 还没输出};
系统会不断检查 emitted_vars 是否包含了所有 outputs 中声明的变量。只有全部输出了才认为目标完成。
Nudge 机制
Nudge(轻推)机制代码:
if (functionCalls.length === 0) { // LLM 停止调用工具了 // 检查是否还有变量没输出 const remainingVars = Object.keys(this.outputConfig.outputs).filter( (key) => !(key inthis.output.emitted_vars) ); if (remainingVars.length > 0) { // 还有变量没输出,"推"它一下 const nudgeMessage = `You have stopped calling tools but have not emitted the following required variables: ${remainingVars.join(', ')}. Please use the 'self.emitvalue' tool to emit them now, or continue working if necessary.`; // 把提醒作为新的用户消息发给 LLM currentMessages = [{ role: 'user', parts: [{ text: nudgeMessage }] }]; // 继续循环,不退出 }}
完整的 subagent 执行流程
开始 ↓while (true) { 检查是否超时/超轮次 → 是 → 退出 ↓ 否 调用 LLM ↓ LLM 返回工具调用? ├─ 是 → 执行工具 → 检查目标是否完成 │ ├─ 是 → 退出 │ └─ 否 → 继续循环 │ └─ 否(LLM 停止调用工具) ↓ 有预定输出要求吗? ├─ 没有 → 退出(认为完成) └─ 有 → 检查是否都输出了 ├─ 是 → 退出 └─ 否 → Nudge 提醒 → 继续循环}
三层循环检测机制
第一层:工具调用重复检测
这是最简单直接的检测,针对 Agent 反复调用相同工具的情况。
private checkToolCallLoop(toolCall: { name: string; args: object }): boolean { // 把工具名和参数一起哈希,生成唯一标识 const key = this.getToolCallKey(toolCall); if (this.lastToolCallKey === key) { // 和上次调用完全一样,计数+1 this.toolCallRepetitionCount++; } else { // 不一样,重置计数 this.lastToolCallKey = key; this.toolCallRepetitionCount = 1; } // 连续5次调用相同工具+相同参数 = 循环 if (this.toolCallRepetitionCount >= TOOL_CALL_LOOP_THRESHOLD) { returntrue; }}
触发条件:连续 5 次调用完全相同的工具(包括参数)。
这种检测很严格——必须是连续的、完全相同的调用。如果中间插入了其他工具调用,计数就会重置。
第二层:内容重复检测("咒语"检测)
这是最复杂的部分,用来检测 LLM 输出重复内容的情况,就像在念咒语一样。
private checkContentLoop(content: string): boolean { // 1. 先检查是否在特殊内容块中(代码块、表格、列表等) const numFences = (content.match(/```/g) ?? []).length; const hasTable = /(^|\n)\s*(\|.*\||[|+-]{3,})/.test(content); // ... 检查各种格式 // 在代码块中不检测循环(代码本来就可能有重复) if (this.inCodeBlock) { returnfalse; } // 2. 把新内容加入历史 this.streamContentHistory += content; // 3. 保持历史在 1000 字符以内 this.truncateAndUpdate(); // 4. 分析内容块是否重复 returnthis.analyzeContentChunksForLoop();}
核心算法是滑动窗口 + 哈希检测:
private analyzeContentChunksForLoop(): boolean { while (this.hasMoreChunksToProcess()) { // 提取 50 字符的块 const currentChunk = this.streamContentHistory.substring( this.lastContentIndex, this.lastContentIndex + CONTENT_CHUNK_SIZE // 50 ); // 计算哈希 const chunkHash = createHash('sha256').update(currentChunk).digest('hex'); // 检查这个块是否重复出现 if (this.isLoopDetectedForChunk(currentChunk, chunkHash)) { returntrue; } // 滑动窗口向前移动 1 个字符 this.lastContentIndex++; }}
判断循环的条件:
private isLoopDetectedForChunk(chunk: string, hash: string): boolean { const existingIndices = this.contentStats.get(hash); if (!existingIndices) { // 第一次见到这个块,记录位置 this.contentStats.set(hash, [this.lastContentIndex]); returnfalse; } // 验证内容确实相同(防止哈希碰撞) if (!this.isActualContentMatch(chunk, existingIndices[0])) { returnfalse; } existingIndices.push(this.lastContentIndex); // 需要出现至少 10 次 if (existingIndices.length < CONTENT_LOOP_THRESHOLD) { // 10 returnfalse; } // 关键:这 10 次必须距离很近(平均距离 ≤ 75 字符) const recentIndices = existingIndices.slice(-CONTENT_LOOP_THRESHOLD); const totalDistance = recentIndices[recentIndices.length - 1] - recentIndices[0]; const averageDistance = totalDistance / (CONTENT_LOOP_THRESHOLD - 1); const maxAllowedDistance = CONTENT_CHUNK_SIZE * 1.5; // 75 return averageDistance <= maxAllowedDistance;}
触发条件:同一个 50 字符的内容块,在很短的距离内重复出现 10 次。
第三层:LLM 智能检测
这是最高级的检测,用 AI 来判断 AI 是否陷入循环。
private async checkForLoopWithLLM(signal: AbortSignal) { // 取最近 20 轮对话 const recentHistory = this.config .getGeminiClient() .getHistory() .slice(-LLM_LOOP_CHECK_HISTORY_COUNT); // 20 // 清理历史(去掉悬空的函数调用等) const trimmedHistory = this.trimRecentHistory(recentHistory); // 让 Gemini Flash 模型分析 const result = awaitthis.config.getBaseLlmClient().generateJson({ contents: [...trimmedHistory, { role: 'user', parts: [{ text: taskPrompt }] }], schema: { type: 'object', properties: { reasoning: { type: 'string' }, confidence: { type: 'number' } // 0-1 之间 } }, model: DEFAULT_GEMINI_FLASH_MODEL, systemInstruction: LOOP_DETECTION_SYSTEM_PROMPT }); if (result['confidence'] > 0.9) { // 高置信度认为是循环 console.warn(result['reasoning']); returntrue; }}
触发时机:
async turnStarted(signal: AbortSignal) { this.turnsInCurrentPrompt++; if ( this.turnsInCurrentPrompt >= LLM_CHECK_AFTER_TURNS && // 至少 30 轮 this.turnsInCurrentPrompt - this.lastCheckTurn >= this.llmCheckInterval ) { this.lastCheckTurn = this.turnsInCurrentPrompt; return await this.checkForLoopWithLLM(signal); }}
- 必须执行超过 30 轮才开始检查(避免误判)
- 不是每轮都检查,有间隔(默认 3 轮)
- 间隔会根据置信度动态调整(5-15 轮)
// 动态调整检查频率this.llmCheckInterval = Math.round( MIN_LLM_CHECK_INTERVAL + // 5 (MAX_LLM_CHECK_INTERVAL - MIN_LLM_CHECK_INTERVAL) * (1 - result['confidence']) // 置信度越高,检查越频繁);
三种循环类型
系统定义了三种循环类型:
enum LoopType { CONSECUTIVE_IDENTICAL_TOOL_CALLS, // 连续相同工具调用 CHANTING_IDENTICAL_SENTENCES, // 重复输出相同内容 LLM_DETECTED_LOOP // LLM 检测到的逻辑循环}
每种都有不同的检测方法和触发条件。
这比较适合处理长对话场景,既能有效检测循环,又不会因为过于敏感而误判正常的迭代操作。
小结
AI Agent 的停止策略是一个容易被忽视但极其重要的技术问题。从原理上看,Agent 就是一个大循环,不断调用 LLM 和工具来完成任务,但如果没有合理的停止机制,就会出现无限循环浪费资源,或者过早停止无法完成任务的问题。常见的停止方案包括硬性限制(步数、时间、API调用次数)、任务完成检测、显式停止信号、循环检测、错误累积和用户中断等,实际应用中需要组合使用多种策略。
OpenManus 采用了相对简单直接的设计:给每个 Agent 配备 terminate 工具,让 LLM 自己决定何时停止,同时用状态机管理生命周期,配合步数限制作为保底,并确保无论如何停止都会正确清理资源。
而 Gemini CLI 的设计更加精巧,核心是声明式输出系统——预先定义需要什么输出,只有全部输出才算完成,如果 Agent 停止调用工具但还有变量未输出,系统会通过 Nudge 机制温和提醒;在循环检测上,Gemini 实现了三层防护:工具调用重复检测(连续5次相同调用)、内容重复检测(滑动窗口+哈希算法检测"咒语"现象)、以及用 LLM 分析对话历史判断是否陷入逻辑循环。
实践中的关键是不要依赖单一停止机制,要组合使用多种策略形成多层防护,给 LLM 明确的停止指引,为不同类型的停止原因提供清晰的用户反馈,并确保资源能够可靠清理。停止策略的本质是在"让 Agent 完成任务"和"防止失控"之间找到平衡点。
以上。
如何学习AI大模型 ?
“最先掌握AI的人,将会晚掌握AI的人有竞争优势,晚掌握AI的人比完全不会AI的人竞争优势更大”。 在这个技术日新月异的时代,不会新技能或者说落后就要挨打。
老蓝我作为一名在一线互联网企业(保密不方便透露)工作十余年,指导过不少同行后辈。帮助很多人得到了学习和成长。
我是非常希望可以把知识和技术分享给大家,但苦于传播途径有限,很多互联网行业的朋友无法获得正确的籽料得到学习的提升,所以也是整理了一份AI大模型籽料包括:AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、落地项目实战等 免费分享出来。
- AI大模型学习路线图
- 100套AI大模型商业化落地方案
- 100集大模型视频教程
- 200本大模型PDF书籍
- LLM面试题合集
- AI产品经理资源合集

大模型学习路线
想要学习一门新技术,你最先应该开始看的就是学习路线图,而下方这张超详细的学习路线图,按照这个路线进行学习,学完成为一名大模型算法工程师,拿个20k、15薪那是轻轻松松!

视频教程
首先是建议零基础的小伙伴通过视频教程来学习,其中这里给大家分享一份与上面成长路线&学习计划相对应的视频教程。文末有整合包的领取方式

技术书籍籽料
当然,当你入门之后,仅仅是视频教程已经不能满足你的需求了,这里也分享一份我学习期间整理的大模型入门书籍籽料。文末有整合包的领取方式

大模型实际应用报告合集
这套包含640份报告的合集,涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师,还是对AI大模型感兴趣的爱好者,这套报告合集都将为您提供宝贵的信息和启示。文末有整合包的领取方式

大模型落地应用案例PPT
光学理论是没用的,要学会跟着一起做,要动手实操,才能将自己的所学运用到实际当中去,这时候可以搞点实战案例来学习。文末有整合包的领取方式

大模型面试题&答案
截至目前大模型已经超过200个,在大模型纵横的时代,不仅大模型技术越来越卷,就连大模型相关的岗位和面试也开始越来越卷了。为了让大家更容易上车大模型算法赛道,我总结了大模型常考的面试题。文末有整合包的领取方式

领取方式
这份完整版的 AI大模型学习籽料我已经上传CSDN,需要的同学可以微⭐扫描下方CSDN官方认证二维码免费领取!
火山引擎开发者社区是火山引擎打造的AI技术生态平台,聚焦Agent与大模型开发,提供豆包系列模型(图像/视频/视觉)、智能分析与会话工具,并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长,新用户可领50万Tokens权益,助力构建智能应用。
更多推荐
所有评论(0)