第03章 ReAct循环：思考-行动-观察的迭代
第03章 ReAct循环：思考-行动-观察的迭代
“真正的智慧不在于一次做对，而在于思考、尝试、观察，然后修正。这是科学方法的本质，也是Agent规划的核心。” —— Yao et al., ReAct: Synergizing Reasoning and Acting in Language Models, 2022
ReAct（Reasoning + Acting）是目前最广泛使用的 Agent 规划范式。它的核心思想极其简洁：Think → Act → Observe，循环往复，直到任务完成。
这个循环看起来朴素，却解决了 LLM 面临的核心难题：如何在不确定的环境中，通过与外部世界的交互，逐步推进目标。
3.1 ReAct 的三元组：Thought-Action-Observation
# ====================================================
# ReAct 核心结构：Thought-Action-Observation 三元组
# ====================================================
from dataclasses import dataclass
from typing import Optional, Any
from enum import Enum
import json
class StepType(Enum):
THOUGHT = "thought" # 思考：Agent的推理过程
ACTION = "action" # 行动：调用工具或执行操作
OBSERVATION = "observation" # 观察：工具返回的结果
@dataclass
class ReActStep:
"""ReAct循环中的一个步骤"""
step_type: StepType
content: str # 步骤内容
tool_name: Optional[str] = None # ACTION步骤使用的工具
tool_input: Optional[dict] = None # ACTION步骤的工具输入
tool_output: Optional[Any] = None # OBSERVATION步骤的工具输出
step_index: int = 0
@dataclass
class ReActTrace:
"""完整的ReAct执行轨迹"""
goal: str
steps: list # List[ReActStep]
final_answer: Optional[str] = None
success: bool = False
def to_prompt_context(self) -> str:
"""将轨迹格式化为LLM的上下文"""
lines = [f"目标：{self.goal}\n"]
for step in self.steps:
if step.step_type == StepType.THOUGHT:
lines.append(f"思考：{step.content}")
elif step.step_type == StepType.ACTION:
tool_input_str = json.dumps(step.tool_input, ensure_ascii=False)
lines.append(f"行动：{step.tool_name}({tool_input_str})")
elif step.step_type == StepType.OBSERVATION:
lines.append(f"观察：{step.content}")
return "\n".join(lines)
# ---- 一个完整的 ReAct 执行轨迹示例 ----
def create_example_trace() -> ReActTrace:
"""展示ReAct处理'查询今天上海天气'的完整轨迹"""
trace = ReActTrace(goal="告诉我今天上海的天气情况，以及是否适合户外运动")
trace.steps = [
ReActStep(
step_type=StepType.THOUGHT,
content="我需要查询今天上海的天气。我应该使用天气查询工具获取当前温度、天气状况和风力等信息，然后根据这些数据判断是否适合户外运动。",
step_index=1,
),
ReActStep(
step_type=StepType.ACTION,
content="调用天气查询工具",
tool_name="weather_search",
tool_input={"city": "上海", "date": "today"},
step_index=2,
),
ReActStep(
step_type=StepType.OBSERVATION,
content='{"city": "上海", "date": "2026-04-30", "temp_high": 22, "temp_low": 15, "condition": "多云转晴", "wind": "东南风3级", "humidity": 65, "aqi": 48}',
tool_output={"city": "上海", "temp_high": 22, "condition": "多云转晴"},
step_index=3,
),
ReActStep(
step_type=StepType.THOUGHT,
content="天气数据已获取：22℃，多云转晴，东南风3级，AQI 48（优）。温度适宜，空气质量好，风力不大，非常适合户外运动。我可以直接给出答案了。",
step_index=4,
),
]
trace.final_answer = (
"今天上海天气：多云转晴，最高22℃，最低15℃，东南风3级，空气质量优（AQI 48）。\n"
"✅ **非常适合户外运动**：温度舒适，空气质量好，风力适中。建议穿一件薄外套，早晚注意保暖。"
)
trace.success = True
return trace
trace = create_example_trace()
print(trace.to_prompt_context())
print(f"\n最终答案：{trace.final_answer}")
3.2 实现完整的 ReAct Agent
# ====================================================
# 完整 ReAct Agent 实现
# ====================================================
import asyncio
import json
import re
from typing import Dict, Callable, Awaitable
from openai import AsyncOpenAI
client = AsyncOpenAI()
class ReActAgent:
"""
基于 Thought-Action-Observation 循环的 Agent
"""
def __init__(
self,
tools: Dict[str, Callable],
model: str = "gpt-4o",
max_iterations: int = 10,
):
self.tools = tools # 工具字典：工具名 → 异步函数
self.model = model
self.max_iterations = max_iterations
self.system_prompt = self._build_system_prompt()
def _build_system_prompt(self) -> str:
"""构建 ReAct 系统提示词"""
tools_desc = "\n".join([
f"- {name}: {func.__doc__ or '无描述'}"
for name, func in self.tools.items()
])
return f"""你是一个使用ReAct模式工作的智能助手。
可用工具：
{tools_desc}
工作模式：
每次响应时，你必须按照以下格式之一输出：
格式A（还需要继续）：
思考：[你对当前情况的分析和下一步计划]
行动：tool_name
输入：{{"param1": "value1", "param2": "value2"}}
格式B（任务完成）：
思考：[为什么任务已经完成]
最终答案：[给用户的完整回答]
规则：
1. 每次只能执行一个工具调用
2. 必须先思考，再行动
3. 行动后等待观察结果，再决定下一步
4. 如果某个工具调用失败，思考原因并尝试不同的方法
5. 最多执行{self.max_iterations}次行动后必须给出最终答案
"""
async def run(self, goal: str) -> ReActTrace:
"""
执行 ReAct 循环直到任务完成
Returns:
完整的执行轨迹
"""
trace = ReActTrace(goal=goal)
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": f"请帮我完成以下目标：{goal}"},
]
for iteration in range(self.max_iterations):
print(f"\n--- 迭代 {iteration + 1} ---")
# 调用 LLM 获取下一步
response = await client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.1,
)
llm_output = response.choices[0].message.content
print(f"LLM输出：\n{llm_output}")
# 解析 LLM 输出
parsed = self._parse_llm_output(llm_output)
if parsed["type"] == "final_answer":
# 任务完成
thought_step = ReActStep(
step_type=StepType.THOUGHT,
content=parsed["thought"],
step_index=len(trace.steps) + 1,
)
trace.steps.append(thought_step)
trace.final_answer = parsed["answer"]
trace.success = True
print(f"\n✅ 任务完成：{parsed['answer']}")
return trace
elif parsed["type"] == "action":
# 执行工具调用
thought_step = ReActStep(
step_type=StepType.THOUGHT,
content=parsed["thought"],
step_index=len(trace.steps) + 1,
)
trace.steps.append(thought_step)
action_step = ReActStep(
step_type=StepType.ACTION,
content=f"调用 {parsed['tool_name']}",
tool_name=parsed["tool_name"],
tool_input=parsed["tool_input"],
step_index=len(trace.steps) + 1,
)
trace.steps.append(action_step)
# 执行工具
observation = await self._execute_tool(
parsed["tool_name"],
parsed["tool_input"]
)
obs_step = ReActStep(
step_type=StepType.OBSERVATION,
content=str(observation),
tool_output=observation,
step_index=len(trace.steps) + 1,
)
trace.steps.append(obs_step)
# 将工具结果添加到对话历史
messages.append({"role": "assistant", "content": llm_output})
messages.append({
"role": "user",
"content": f"观察：{str(observation)}\n\n请继续。"
})
else:
# 解析失败，让LLM重试
messages.append({"role": "assistant", "content": llm_output})
messages.append({
"role": "user",
"content": "我无法解析你的输出格式。请严格按照'思考→行动→输入'或'思考→最终答案'的格式输出。"
})
# 达到最大迭代次数
trace.final_answer = "已达到最大迭代次数，任务可能未完全完成。"
trace.success = False
return trace
def _parse_llm_output(self, output: str) -> dict:
"""解析LLM输出，提取思考、行动或最终答案"""
# 检查是否是最终答案
final_match = re.search(r"最终答案[：:]\s*(.+)", output, re.DOTALL)
if final_match:
thought_match = re.search(r"思考[：:]\s*(.+?)(?=最终答案)", output, re.DOTALL)
return {
"type": "final_answer",
"thought": thought_match.group(1).strip() if thought_match else "",
"answer": final_match.group(1).strip(),
}
# 解析行动
thought_match = re.search(r"思考[：:]\s*(.+?)(?=行动[：:])", output, re.DOTALL)
action_match = re.search(r"行动[：:]\s*(\w+)", output)
input_match = re.search(r"输入[：:]\s*(\{.+?\})", output, re.DOTALL)
if action_match and input_match:
try:
tool_input = json.loads(input_match.group(1))
except json.JSONDecodeError:
tool_input = {}
return {
"type": "action",
"thought": thought_match.group(1).strip() if thought_match else "",
"tool_name": action_match.group(1),
"tool_input": tool_input,
}
return {"type": "unknown", "raw": output}
async def _execute_tool(self, tool_name: str, tool_input: dict) -> Any:
"""执行工具调用"""
if tool_name not in self.tools:
return f"错误：工具 '{tool_name}' 不存在"
try:
tool_func = self.tools[tool_name]
result = await tool_func(**tool_input)
return result
except Exception as e:
return f"工具执行失败：{str(e)}"
# ---- 定义工具并测试 ReAct Agent ----
async def weather_search(city: str, date: str = "today") -> dict:
"""查询指定城市的天气信息"""
# 模拟天气API（实际使用真实API）
mock_data = {
"上海": {"temp_high": 22, "temp_low": 15, "condition": "多云转晴", "aqi": 48},
"北京": {"temp_high": 18, "temp_low": 8, "condition": "晴", "aqi": 62},
}
return mock_data.get(city, {"error": f"找不到城市 {city} 的天气数据"})
async def web_search(query: str) -> str:
"""搜索互联网获取信息"""
# 模拟搜索结果
return f"搜索'{query}'的结果：找到3篇相关文章，主要内容为..."
async def calculator(expression: str) -> float:
"""计算数学表达式"""
try:
# 注意：生产环境应使用安全的表达式求值，而非eval
import ast
return float(ast.literal_eval(expression))
except Exception as e:
return f"计算错误：{e}"
async def main():
agent = ReActAgent(
tools={
"weather_search": weather_search,
"web_search": web_search,
"calculator": calculator,
},
max_iterations=5,
)
goal = "查询上海今天的天气，并计算如果最高气温22℃换算成华氏度是多少"
trace = await agent.run(goal)
print("\n=== 执行轨迹摘要 ===")
print(f"总步骤数: {len(trace.steps)}")
print(f"执行成功: {trace.success}")
print(f"最终答案: {trace.final_answer}")
asyncio.run(main())
3.3 ReAct 的终止条件与循环控制
# ====================================================
# 终止条件设计：防止无限循环
# ====================================================
from typing import Optional
import time
class TerminationController:
"""
ReAct循环的终止条件控制器
多种终止条件并行检测，任一满足即停止循环
"""
def __init__(
self,
max_iterations: int = 15,
max_time_seconds: float = 120,
max_tool_calls: int = 20,
repetition_window: int = 3, # 检测最近N步是否有重复行为
):
self.max_iterations = max_iterations
self.max_time_seconds = max_time_seconds
self.max_tool_calls = max_tool_calls
self.repetition_window = repetition_window
self.start_time = time.time()
self.iteration_count = 0
self.tool_call_count = 0
self.recent_actions: list = [] # 记录最近的行动
def should_terminate(self, trace: ReActTrace) -> tuple:
"""
检查是否应该终止循环
Returns:
(应该终止, 终止原因)
"""
self.iteration_count += 1
elapsed = time.time() - self.start_time
# 条件1：达到最大迭代次数
if self.iteration_count >= self.max_iterations:
return True, f"达到最大迭代次数 ({self.max_iterations})"
# 条件2：超时
if elapsed >= self.max_time_seconds:
return True, f"执行超时 ({elapsed:.1f}秒)"
# 条件3：工具调用次数过多
action_steps = [s for s in trace.steps if s.step_type == StepType.ACTION]
if len(action_steps) >= self.max_tool_calls:
return True, f"工具调用次数过多 ({len(action_steps)}次)"
# 条件4：检测循环行为（相同工具+相同输入重复出现）
if len(action_steps) >= self.repetition_window:
recent = action_steps[-self.repetition_window:]
recent_signatures = [
f"{s.tool_name}:{json.dumps(s.tool_input, sort_keys=True)}"
for s in recent
]
if len(set(recent_signatures)) < len(recent_signatures):
return True, "检测到重复行为（Agent陷入循环）"
return False, None
def get_summary(self) -> dict:
"""获取执行统计摘要"""
return {
"iterations": self.iteration_count,
"elapsed_seconds": round(time.time() - self.start_time, 2),
"tool_calls": self.tool_call_count,
}
# ---- 带终止控制的增强版 ReAct ----
async def react_with_termination_control(
agent: ReActAgent,
goal: str,
termination_config: Optional[dict] = None
) -> ReActTrace:
"""
带完善终止控制的 ReAct 执行
"""
config = termination_config or {}
controller = TerminationController(
max_iterations=config.get("max_iterations", 10),
max_time_seconds=config.get("max_time", 60),
max_tool_calls=config.get("max_tools", 15),
)
trace = await agent.run(goal) # 内部已有基本控制
summary = controller.get_summary()
print(f"\n执行统计：{summary}")
return trace
# ---- ReAct vs. 简单LLM对比实验 ----
async def compare_react_vs_simple():
"""对比ReAct和直接LLM调用的效果"""
goal = "找出Python中最快的JSON解析库，并比较它们的性能差异"
# 方法1：直接LLM（一次性回答）
print("=== 方法1：直接LLM（一次调用）===")
direct_response = await client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": goal}],
)
print(f"直接回答：\n{direct_response.choices[0].message.content[:300]}...")
print("（无法查询最新数据，答案可能过时）")
# 方法2：ReAct（搜索+分析）
print("\n=== 方法2：ReAct（工具增强）===")
agent = ReActAgent(
tools={"web_search": web_search},
max_iterations=4,
)
trace = await agent.run(goal)
print(f"ReAct答案：{trace.final_answer}")
print(f"使用了 {len([s for s in trace.steps if s.step_type == StepType.ACTION])} 次工具调用")
# 对比：ReAct的答案基于实际搜索，更可靠
asyncio.run(compare_react_vs_simple())
本章小结
-
ReAct 的核心是 Thought-Action-Observation 三元组：思考给出方向，行动获取信息，观察更新认知。这个循环让 Agent 能够在不确定环境中逐步逼近目标。
-
与直接 LLM 调用的本质区别：ReAct 允许 Agent 在执行中途获取新信息，每一步的"观察"都可以改变后续决策，而不是基于训练时的静态知识一次性回答。
-
终止条件是不可或缺的安全机制：最大迭代次数、超时、重复行为检测，三重保障确保 Agent 不会进入无限循环。忽略这一点是初学者最常犯的错误。
-
解析 LLM 输出是工程关键：格式化的系统提示（Thought/Action/Observation 格式）加上健壮的正则解析，决定了 ReAct 的可靠性。
-
ReAct 是后续所有架构的基础：第4章的 Plan-and-Execute、第5章的依赖图、第6章的动态重规划，都建立在 ReAct 的思想基础上。
# 核心行动：在你的项目中实现一个最小可用的 ReAct Agent
import asyncio
async def minimal_react_demo():
"""最小可用的 ReAct 实现，5分钟可以跑起来"""
# 1. 定义你的工具（替换为真实工具）
async def my_tool(query: str) -> str:
"""你的工具描述"""
return f"工具返回结果: {query}"
# 2. 创建 Agent
agent = ReActAgent(
tools={"my_tool": my_tool},
max_iterations=5,
)
# 3. 运行
trace = await agent.run("你的任务目标")
print(f"结果: {trace.final_answer}")
asyncio.run(minimal_react_demo())
本章提示词模板
【模板1：ReAct系统提示词（可直接使用）】
你是一个使用ReAct（推理+行动）模式的智能助手。
可用工具：
{tools_list}
你必须按照以下格式输出（每次响应只能是其中一种）：
需要使用工具时：
思考：[分析当前情况，说明为什么需要这个工具]
行动：[工具名称]
输入：{"参数名": "参数值"}
任务完成时：
思考：[总结已获得的所有信息，确认任务已完成]
最终答案：[给用户的完整、清晰的回答]
重要规则：
- 每次只调用一个工具
- 必须先思考，再行动
- 根据观察结果调整你的思考方向
- 如果工具失败，思考原因并尝试其他方法
【模板2：ReAct调试提示词】
以下ReAct执行轨迹出现了问题，请分析原因：
执行轨迹：
{trace_content}
问题症状：{problem_description}
请分析：
1. Agent在第几步开始偏离正确方向？原因是什么？
2. 哪个思考步骤出现了逻辑错误？
3. 工具调用的参数是否正确？
4. 应该如何修改系统提示词或工具设计来避免此类问题？
给出具体的改进建议。