简单来讲，AI Agent 实现的的大逻辑就是一个大的循环 + 获取上下文 + 不停的 LLM 调用 + 工具的调用。
那么一个关键问题就出现了：这个循环什么时候应该停止？如果处理不当，Agent 可能会陷入无限循环，浪费计算资源，或者过早停止而无法完成任务。本文将深入探讨 AI Agent 停止策略的核心设计思路。
常用停止策略
AI Agent 停止策略无外乎以下几种情况：
1. 硬性限制
最简单粗暴的方法：
-
最大步数限制（比如最多循环 30 次） -
执行时间限制（比如最多跑 5 分钟） -
API 调用次数限制（比如最多调 100 次） -
API 调用 Token 数限制
这种方法简单有效，但用户体验很差。经常出现任务做到一半被强制停止的情况。
2. 任务完成检测
让 LLM 判断任务是否完成：
# 每次循环后问 LLM
response = 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 = """当请求已满足或无法继续时终止交互。
完成所有任务后，调用此工具结束工作。"""
async def execute(self, status: str) -> str:
return f"交互已完成，状态：{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"],
}
async def execute(self, status: str) -> str:
"""Finish the current execution"""
return f"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
# 如果没找到活跃步骤，返回 None
return 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 in this.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) {
return true;
}
}
触发条件：连续 5 次调用完全相同的工具（包括参数）。
这种检测很严格——必须是连续的、完全相同的调用。如果中间插入了其他工具调用，计数就会重置。
第二层：内容重复检测（”咒语”检测）
这是最复杂的部分，用来检测 LLM 输出重复内容的情况，就像在念咒语一样。
private checkContentLoop(content: string): boolean {
// 1. 先检查是否在特殊内容块中（代码块、表格、列表等）
const numFences = (content.match(/```/g) ?? []).length;
const hasTable = /(^|\n)\s*(\|.*\||[|+-]{3,})/.test(content);
// ... 检查各种格式
// 在代码块中不检测循环（代码本来就可能有重复）
if (this.inCodeBlock) {
return false;
}
// 2. 把新内容加入历史
this.streamContentHistory += content;
// 3. 保持历史在 1000 字符以内
this.truncateAndUpdate();
// 4. 分析内容块是否重复
return this.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)) {
return true;
}
// 滑动窗口向前移动 1 个字符
this.lastContentIndex++;
}
}
判断循环的条件：
private isLoopDetectedForChunk(chunk: string, hash: string): boolean {
const existingIndices = this.contentStats.get(hash);
if (!existingIndices) {
// 第一次见到这个块，记录位置
this.contentStats.set(hash, [this.lastContentIndex]);
return false;
}
// 验证内容确实相同（防止哈希碰撞）
if (!this.isActualContentMatch(chunk, existingIndices[0])) {
return false;
}
existingIndices.push(this.lastContentIndex);
// 需要出现至少 10 次
if (existingIndices.length < CONTENT_LOOP_THRESHOLD) { // 10
return false;
}
// 关键：这 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 = await this.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']);
return true;
}
}
触发时机：
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 完成任务”和”防止失控”之间找到平衡点。
以上。