ReAct(Reasoning and Acting) 是一种思考与行动紧密结合的范式，让智能体边想边做，动态调整。
工作流程
在ReAct诞生之前，模型解决问题的方式主要包括两种：
- 推理型：“思维链(Chain-of-Thought)”引导模型进行复杂的逻辑推理。
- 行动型：模型直接输出要执行的动作。
ReAct将推理与行动显式结合了起来，形成一个“思考-行动-观察”的循环。
- Thought：智能体分析当前情况，分解任务制定下一步计划或者反思上一步结果的结论。
- Action：智能体采取的具体动作，如工具调用等。
- Observation：执行动作后智能体返回的结果。
具体来说，在每个时间步 $t$，智能体的策略（即大语言模型 $\pi$）会根据初始问题 $q$ 和之前所有步骤的”行动-观察”历史轨迹 $((a_1, o_1), \dots, (a_{t-1}, o_{t-1}))$，来生成当前的思考 $th_t$ 和行动 $a_t$：
$$
(th_t, a_t) = \pi(q, (a_1, o_1), \dots, (a_{t-1}, o_{t-1}))
$$
随后，环境中的工具 $T$ 会执行行动 $a_t$，并返回一个新的观察结果 $o_t$：
$$
o_t = T(a_t)
$$
实现一个ReAct智能体
系统提示词设计
REACT_PROMPT_PROMPT_TEMPLATE = """
你是一个智能小助手，可用工具如下
{tools}
Thought: 你的思考过程，用户分析问题，并产出下一步行动。
Action: 你决定采取的行动，必须是以下格式之一。
- `[{tool_input}]`: 调用可用的参数。
- `Finish[最终答案]`: 当你认为已经获得最终答案时，给出最终回到。
现在开始:
Question: {question}
History: {history}
"""
这个提示词定义了LLM与Agent之间的交互规范：
- 角色定义
- 工具清单
- 格式规约
- 动态上下文
核心循环实现
"""
ReAct 智能体框架
=========================
基于 ReAct（推理+行动）范式的智能体实现，支持 LLM 推理与外部工具调用的协同工作。
"""
import os
import re
from typing import Dict, Any, List, Optional, Callable
from serpapi import SerpApiClient
from dotenv import load_dotenv
from openai import OpenAI
# 从 .env 文件加载环境变量
load_dotenv()
# ─── ReAct 提示词模板 ──────────────────────────────────────────────
# 该模板定义了 LLM 遵循 ReAct 推理模式的指令格式：思考→行动→观察循环
REACT_PROMPT_TEMPLATE = """你是一个使用工具回答问题的助手。
可用工具：
{tools}
问题：{question}
历史步骤：
{history}
请严格按照以下格式回复：
Thought: [你对下一步操作的推理]
Action: [工具名][工具输入] 或 Finish[最终答案]
"""
class HelloAgentsLLM:
"""
支持流式输出的 LLM 客户端封装。
负责处理认证、模型配置和实时 token 流式传输，
为智能体提供交互式体验。
"""
def __init__(
self,
model: Optional[str] = None,
api_key: Optional[str] = None,
base_url: Optional[str] = None,
timeout: Optional[int] = None
):
"""
初始化 LLM 客户端，参数缺失时自动回退到环境变量。
参数:
model: 模型标识符（回退到 MODEL 环境变量）
api_key: API 认证密钥（回退到 API_KEY 环境变量）
base_url: 自定义 API 端点（回退到 BASE_URL 环境变量）
timeout: 请求超时时间（秒），默认 60s
异常:
ValueError: 回退后仍有必要参数缺失时抛出
"""
# 解析参数，优先使用传入值，否则从环境变量读取
self.model = model or os.getenv("MODEL")
resolved_base_url = base_url or os.getenv("BASE_URL")
resolved_api_key = api_key or os.getenv("API_KEY")
resolved_timeout = int(timeout or os.getenv("LLM_TIMEOUT", "60"))
# 校验所有必要参数是否齐全
if not all([self.model, resolved_api_key, resolved_base_url]):
missing = []
if not self.model: missing.append("MODEL")
if not resolved_api_key: missing.append("API_KEY")
if not resolved_base_url: missing.append("BASE_URL")
raise ValueError(f"缺少必要参数: {', '.join(missing)}")
# 初始化兼容 OpenAI 协议的客户端
self.client = OpenAI(
api_key=resolved_api_key,
base_url=resolved_base_url,
timeout=resolved_timeout
)
def think(self, messages: List[Dict[str, str]], temperature: float = 0.0) -> Optional[str]:
"""
生成响应并实时流式输出 token。
参数:
messages: OpenAI 格式的对话历史
temperature: 采样温度（0.0 = 确定性输出）
返回:
完整生成的文本，请求失败时返回 None
"""
try:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
stream=True # 启用逐 token 流式传输
)
# 收集流式 token 的同时实时显示
collected_content: List[str] = []
for chunk in response:
content = chunk.choices[0].delta.content or ""
print(content, end="", flush=True) # 实时打印到控制台
collected_content.append(content)
return "".join(collected_content)
except Exception as e:
print(f"\n[LLM 错误] {e}")
return None
def search(query: str) -> str:
"""
使用 SerpAPI Google 引擎的网络搜索工具。
搜索信息并从多种结果格式（答案框、知识图谱等）中提取最相关的答案。
参数:
query: 搜索查询字符串
返回:
提取的答案文本，搜索失败时返回错误信息
"""
try:
api_key = os.getenv("SERPAPI_API_KEY")
if not api_key:
return "[错误] .env 中未配置 SERPAPI_API_KEY"
# 配置搜索参数，针对中文地区优化
params = {
"engine": "google",
"q": query,
"api_key": api_key,
"gl": "cn", # 地理位置：中国
"hl": "zh-cn" # 语言：简体中文
}
client = SerpApiClient(params)
results = client.get_dict()
# 从响应的多个可能位置提取答案
# 优先级：answer_box_list > answer_box > knowledge_graph
if "answer_box_list" in results:
return "\n".join(str(item) for item in results["answer_box_list"])
if "answer_box" in results and "answer" in results["answer_box"]:
return results["answer_box"]["answer"]
if "knowledge_graph" in results and "description" in results["knowledge_graph"]:
return results["knowledge_graph"]["description"]
return "[未找到相关信息]"
except Exception as e:
return f"[搜索错误] {str(e)}"
class ToolExecutor:
"""
智能体工具的注册表和执行器。
管理工具的注册、查找，并提供格式化描述以便注入 LLM 提示词。
"""
def __init__(self):
# 存储结构：{工具名: {"description": 描述字符串, "func": 可调用函数}}
self.tools: Dict[str, Dict[str, Any]] = {}
def register_tool(self, name: str, description: str, func: Callable) -> None:
"""
注册或更新工具到执行器。
参数:
name: 唯一工具标识符
description: 供 LLM 理解的人类可读描述
func: 工具被调用时执行的可调用函数
"""
if name in self.tools:
print(f"[工具更新] 覆盖已有工具: {name}")
self.tools[name] = {
"description": description,
"func": func
}
def get_tool(self, name: str) -> Optional[Callable]:
"""
按名称获取工具函数。
返回:
找到的工具函数，未找到时返回 None
"""
return self.tools.get(name, {}).get("func")
def get_available_tools(self) -> str:
"""
生成格式化的工具列表，用于注入 LLM 提示词。
返回:
"- 工具名 : 描述" 格式的换行分隔字符串
"""
if not self.tools:
return "(暂无可用工具)"
return "\n".join([
f"- {name}: {info['description']}"
for name, info in self.tools.items()
])
class ReActAgent:
"""
ReAct（推理+行动）智能体实现。
执行 思考-行动-观察 循环：LLM 推理下一步操作 → 选择工具执行 →
将结果纳入后续推理，直到得出最终答案。
"""
def __init__(
self,
llm_client: HelloAgentsLLM,
tool_executor: ToolExecutor,
max_steps: int = 5
):
"""
初始化智能体，绑定 LLM 客户端和工具注册表。
参数:
llm_client: 已配置的 LLM 实例，用于推理
tool_executor: 包含已注册函数的工具注册表
max_steps: 放弃前的最大推理迭代次数
"""
self.llm_client = llm_client
self.tool_executor = tool_executor
self.max_steps = max_steps
self.history: List[str] = []
def run(self, question: str) -> Optional[str]:
"""
对给定问题执行 ReAct 推理循环。
循环交替执行：
1. LLM 生成 思考 + 行动
2. 执行行动（工具调用或结束）
3. 记录观察结果并重复
参数:
question: 用户提问
返回:
成功时返回最终答案字符串，超过最大步数时返回 None
"""
self.history = []
for step in range(1, self.max_steps + 1):
print(f"\n{'='*60}")
print(f"第 {step}/{self.max_steps} 步")
print(f"{'='*60}")
# 构建包含当前上下文的提示词
tools_desc = self.tool_executor.get_available_tools()
history_str = "\n".join(self.history) if self.history else "(尚无历史步骤)"
prompt = REACT_PROMPT_TEMPLATE.format(
tools=tools_desc,
question=question,
history=history_str
)
messages = [{"role": "user", "content": prompt}]
# 获取 LLM 响应
response_text = self.llm_client.think(messages=messages)
if not response_text:
print("\n[错误] LLM 返回空响应")
break
# 从响应中解析思考和行动
thought, action = self._parse_output(response_text)
if thought:
print(f"\n 思考: {thought}")
if not action:
print("\n[错误] 未能从响应中解析出行动")
break
# 检查终止条件
if action.startswith("Finish"):
match = re.match(r"Finish\[(.*)\]", action)
if match:
final_answer = match.group(1).strip()
print(f"\n✅ 最终答案: {final_answer}")
return final_answer
else:
print("\n[错误] Finish 格式无效")
break
# 执行工具
tool_name, tool_input = self._parse_action(action)
if not tool_name or not tool_input:
print(f"\n[错误] 行动解析失败: {action}")
self.history.append(f"Action: {action}")
self.history.append("Observation: [解析错误 - 行动格式无效]")
continue
print(f"\n 工具: {tool_name}")
print(f" 输入: {tool_input}")
tool_function = self.tool_executor.get_tool(tool_name)
if not tool_function:
observation = f"[错误] 工具 '{tool_name}' 不存在"
print(f" {observation}")
else:
try:
observation = tool_function(tool_input)
print(f" ✅ 结果: {str(observation)[:200]}...")
except Exception as e:
observation = f"[工具错误] {str(e)}"
print(f" ❌ {observation}")
# 将行动和观察记录到历史中
self.history.append(f"Action: {action}")
self.history.append(f"Observation: {observation}")
print(f"\n️ 已达最大步数 ({self.max_steps})，未能完成")
return None
@staticmethod
def _parse_output(text: str) -> tuple[Optional[str], Optional[str]]:
"""
从 LLM 响应中提取 思考 和 行动 部分。
使用带前瞻断言的正则表达式定位各部分边界。
参数:
text: LLM 原始响应文本
返回:
(思考内容, 行动内容) 元组，任一部分可能为 None
"""
# 匹配 Thought: 到 \nAction: 或字符串末尾之间的所有内容
thought_match = re.search(
r"Thought:\s*(.*?)(?=\nAction:|$)",
text,
re.DOTALL | re.IGNORECASE
)
# 匹配 Action: 之后到字符串末尾的所有内容
action_match = re.search(
r"Action:\s*(.*?)$",
text,
re.DOTALL | re.IGNORECASE
)
thought = thought_match.group(1).strip() if thought_match else None
action = action_match.group(1).strip() if action_match else None
return thought, action
@staticmethod
def _parse_action(action_text: str) -> tuple[Optional[str], Optional[str]]:
"""
解析工具调用语法：工具名[工具输入]
参数:
action_text: LLM 输出的原始行动字符串
返回:
(工具名, 工具输入) 元组，解析失败时返回 (None, None)
"""
match = re.match(r"(\w+)\[(.*)\]", action_text, re.DOTALL)
if match:
return match.group(1), match.group(2)
return None, None
# ─── 使用示例 ───────────────────────────────────────────────────────
if __name__ == "__main__":
# 初始化各组件
llm = HelloAgentsLLM()
executor = ToolExecutor()
# 注册搜索工具
executor.register_tool(
name="search",
description="使用 Google 搜索网络信息",
func=search
)
# 创建并运行智能体
agent = ReActAgent(
llm_client=llm,
tool_executor=executor,
max_steps=5
)
# 执行查询
result = agent.run("法国的首都是哪里？")
print(f"\n结果: {result}")
ReAct 的特点与局限性
特点
- 高可解释性：Thought链会清晰展示智能体每一步的心路历程，这让智能体的每个行动的原因都一目了然。
- 动态规划与纠错能力：ReAct走一步看一步，遇到错误的Observation时可以自行调整和修正重试。
- 工具协同：ReAct范式天然将大模型与外部工具关联，突破了单一LLM在知识实效性、计算准确性方面的固有局限。
局限性
- 对LLM自身能力的依赖：依赖LLM具有逻辑推理、指令遵循和格式化输出能力。否则容易产生错误规划。
- 执行效率问题：完成任务需要多次调用LLM执行一个循序渐进的推理步骤。
- 提示词高依赖：依赖一个精心设计的提示词模版，一旦提示词变化，会影响LLM的行为导致Agent无法正常工作。
- 局部最优：步进式的智能体缺乏全局、长远的规划，它可能因为眼前的Observation选择一个看似正确单长远来看并非最优的路径，甚至导致原地打转。
调试ReAct 智能体
- 检查原始提示词
- 分析原始输出
- 验证工具的输入与输出
- 尝试不同prompt
- 尝试不同模型或参数