ReAct Agent Loop 原理解析
一、什么是 ReAct？
ReAct = Reasoning + Acting（推理 + 行动）
ReAct 是一种让 AI Agent 能够通过"思考-行动-观察"的循环来解决复杂问题的范式。它结合了推理（Reasoning）和行动（Acting），让 Agent 不仅能够执行任务，还能解释为什么要这样做。
核心思想
Reasoning (推理): 思考要做什么
↓
Acting (行动): 执行工具完成任务
↓
Observation (观察): 查看执行结果
↓
回到 Reasoning (继续推理)
二、Agent Loop 工作原理
Agent Loop 是 ReAct 的具体实现，是一个循环处理流程，Agent 通过不断重复以下步骤来完成用户任务。
循环流程图
┌─────────────────────────────────────────┐
│ Agent Loop (循环) │
├─────────────────────────────────────────┤
│ 1. 接收用户消息 │
│ 2. 构建上下文 (历史/记忆/技能/性格) │
│ 3. 调用 LLM 生成回复 │
│ 4. 解析 LLM 响应 │
│ ├─ 有工具调用? → 执行工具 │
│ └─ 无工具调用? → 返回答案 │
│ 5. 将工具结果反馈给 LLM │
│ 6. 回到步骤 3，继续循环 (直到完成) │
└─────────────────────────────────────────┘
三、具体示例
假设用户问：“今天天气怎么样？”
Agent Loop 执行过程：
【迭代 1】
────────────────────────────
用户: 今天天气怎么样？
构建上下文:
- 系统提示: 你是AI助手，名叫小C...
- 历史对话: 无
- 智能记忆: 无
- 技能说明: 可用web_search工具
调用 LLM:
LLM 思考: 我需要查询天气信息
LLM 决定: 调用 web_search 工具
LLM 输出: 工具调用 {"tool": "web_search", "args": {"query": "今天天气 北京"}}
工具调用:
- 工具: web_search
- 参数: {"query": "今天天气 北京"}
观察结果:
- 返回: "北京今天晴，气温15-25℃，空气质量优"
────────────────────────────
【迭代 2】
────────────────────────────
将结果反馈给 LLM:
消息: 工具执行结果: 北京今天晴，气温15-25℃，空气质量优
调用 LLM:
LLM 思考: 已获取天气信息，可以回答用户
LLM 决定: 不需要再调用工具，直接回复
最终回复:
"北京今天天气晴朗，气温在15-25摄氏度之间，空气质量优，适合外出活动。"
────────────────────────────
循环结束
四、代码实现
1. 基础版 Agent Loop
class AgentLoop:
"""Agent 核心循环类"""
def __init__(self, provider, tools, max_iterations=20):
self.provider = provider # LLM 提供商
self.tools = tools # 工具注册表
self.max_iterations = max_iterations
async def process_message(self, message):
"""处理用户消息"""
# 1. 构建初始消息
messages = [
{"role": "system", "content": "你是AI助手..."},
{"role": "user", "content": message}
]
# 2. 开始循环
iteration = 0
final_answer = None
while iteration < self.max_iterations:
iteration += 1
# 3. 调用 LLM
response = await self.provider.chat(
messages=messages,
tools=self.tools.get_definitions()
)
# 4. 检查是否有工具调用
if response.has_tool_calls:
# 5. 执行工具
for tool_call in response.tool_calls:
result = await self.tools.execute(
tool_call.name,
tool_call.arguments
)
# 6. 将结果反馈给 LLM
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# 7. 继续循环
continue
# 8. 无工具调用，得到最终答案
final_answer = response.content
break
return final_answer
2. 增强版 Agent Loop（带流式响应和重试）
class AgentLoop:
"""增强版 Agent 循环类"""
def __init__(
self,
provider,
tools,
max_iterations=25,
max_retries=3,
retry_delay=1.0
):
self.provider = provider
self.tools = tools
self.max_iterations = max_iterations
self.max_retries = max_retries
self.retry_delay = retry_delay
async def process_message(self, message, cancel_token=None):
"""处理用户消息并生成流式响应"""
# 1. 构建上下文
messages = [
{"role": "system", "content": "你是AI助手..."},
{"role": "user", "content": message}
]
# 2. 开始循环
iteration = 0
total_tool_calls = 0
final_content = ""
while iteration < self.max_iterations:
# 检查取消
if cancel_token and cancel_token.is_cancelled:
return
iteration += 1
# 3. 调用 LLM（流式）
tool_calls_buffer = []
async for chunk in self.provider.chat_stream(
messages=messages,
tools=self.tools.get_definitions()
):
if chunk.is_content and chunk.content:
final_content += chunk.content
yield chunk.content # 实时推送
if chunk.is_tool_call and chunk.tool_call:
tool_calls_buffer.append(chunk.tool_call)
# 4. 执行工具（带重试）
for tool_call in tool_calls_buffer:
# 检查取消
if cancel_token and cancel_token.is_cancelled:
return
total_tool_calls += 1
# 重试机制
result = None
for attempt in range(self.max_retries):
try:
result = await self.tools.execute(
tool_call.name,
tool_call.arguments
)
logger.debug(f"Tool {tool_call.name} executed successfully")
break
except Exception as e:
logger.warning(f"Tool {tool_call.name} failed (attempt {attempt+1})")
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delay)
# 5. 添加结果到消息列表
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"name": tool_call.name,
"content": result or "工具执行失败"
})
# 6. 判断是否继续
if not tool_calls_buffer:
break # 无工具调用，结束
# 7. 保存会话
if final_content:
self.session_manager.save(message, final_content)
五、关键特性
1. 循环次数限制
防止无限循环，通常限制在 20-50 次：
self.max_iterations = 25
while iteration < self.max_iterations:
iteration += 1
# ... 执行逻辑
2. 工具调用链
用户问题 → 工具1 → 结果1 → 工具2 → 结果2 → ... → 最终答案
3. 上下文累积
messages = [
{"role": "system", "content": "..."},
{"role": "user", "content": "查询天气"},
{"role": "assistant", "content": "...", "tool_calls": [...]},
{"role": "tool", "content": "天气结果"},
{"role": "assistant", "content": "根据天气结果..."} # LLM基于工具结果继续思考
]
4. 流式响应
async for chunk in self.provider.chat_stream(messages, tools):
if chunk.is_content:
yield chunk.content # 实时返回给用户
5. 取消支持
if cancel_token and cancel_token.is_cancelled:
logger.info("Agent loop cancelled")
return
六、循环终止条件
Agent Loop 在以下情况终止：
| 条件 | 说明 |
|---|---|
| 无工具调用 | LLM 决定不再需要工具，直接给出答案 |
| 达到迭代上限 | 防止无限循环，强制停止 |
| 用户取消 | 用户主动取消任务 |
| 发生错误 | 工具执行失败且重试后仍失败 |
七、复杂示例
用户问题: “帮我创建一个Python爬虫，抓取新闻网站的标题”
【迭代 1】
LLM: 我需要先查看用户的文件结构
工具: list_dir(path=".")
【迭代 2】
LLM: 我需要创建爬虫文件
工具: write_file(path="crawler.py", content="import requests...")
【迭代 3】
LLM: 我需要测试爬虫是否工作
工具: exec(command="python crawler.py")
【迭代 4】
LLM: 爬虫运行成功，已完成任务
最终答案: "已创建 crawler.py，测试成功，可以抓取新闻标题了。"
八、不同框架的实现对比
NanoBot - 简洁同步
while iteration < max_iterations:
response = await provider.chat(messages, tools)
if response.has_tool_calls:
for tool_call in response.tool_calls:
result = await tools.execute(tool_call.name, args)
messages.append(tool_result)
else:
break
CountBot - 增强异步流式
while iteration < max_iterations:
async for chunk in provider.chat_stream(messages, tools):
if chunk.is_content:
yield chunk.content # 流式推送
if chunk.is_tool_call:
tool_calls_buffer.append(chunk.tool_call)
for tool_call in tool_calls_buffer:
for attempt in range(max_retries): # 重试机制
try:
result = await tools.execute(tool_name, args)
break
except Exception:
await asyncio.sleep(retry_delay)
OpenClaw - Pi框架 + 多级重试
// 32-160次重试，带Auth Profile轮换
const maxRetries = resolveMaxRunRetryIterations(profileCount);
for (let attempt = 0; attempt < maxRetries; attempt++) {
try {
const result = await runEmbeddedAttempt({
messages,
tools,
thinkingLevel
});
return result;
} catch (error) {
if (isFailoverError(error)) {
await markAuthProfileFailure(profile);
continue; // 切换到下一个Auth Profile
}
}
}
九、最佳实践
1. 工具设计原则
- 单一职责: 每个工具只做一件事
- 参数清晰: 使用清晰的参数名称和描述
- 错误处理: 返回有意义的错误信息
- 幂等性: 相同输入产生相同输出
2. 上下文管理
- 历史限制: 限制历史消息数量，避免上下文溢出
- 记忆总结: 定期总结对话，压缩上下文
- 技能加载: 按需加载技能，减少提示词长度
3. 性能优化
- 异步并发: 使用异步IO提高并发性能
- 缓存机制: 缓存工具结果，避免重复调用
- 超时控制: 设置工具执行超时，防止阻塞
4. 安全考虑
- 输入验证: 验证工具参数，防止注入攻击
- 权限控制: 限制工具访问范围
- 审计日志: 记录所有工具调用
十、总结
Agent Loop 是 AI Agent 的心脏，它：
- 持续循环: 不断推理和行动
- 工具调用: 根据需要调用工具
- 上下文累积: 记录每一步的结果
- 最终决策: 判断何时给出最终答案
简单来说：
Agent Loop = 思考 → 行动 → 观察 → 再思考，直到完成任务
核心优势
- 灵活性: 可以处理各种复杂任务
- 可解释性: 每一步都可以追溯和解释
- 可扩展性: 通过添加工具扩展能力
- 自主性: 能够自主决策和执行
应用场景
- 代码生成: 编写、测试、调试代码
- 数据分析: 查询、分析、可视化数据
- 任务自动化: 执行复杂的自动化任务
- 知识检索: 搜索、总结、整合信息
- 问题解决: 分析问题、制定方案、执行解决
ReAct Agent Loop 是现代 AI Agent 的核心架构，它让 AI 不仅能够回答问题，还能够思考、行动、解决实际问题。
