孔良1. ReAct范式：思考与行动的完美结合
ReAct（Reasoning + Acting）是一种将推理（Reasoning）与行动（Acting）相结合的智能体工作范式。它通过"思考-行动-观察"的循环机制，让智能体能够像人类一样逐步解决问题。这种范式特别适合需要多步骤推理和外部工具调用的复杂任务场景。
1.1 ReAct的核心工作流程
ReAct的每一步都遵循固定的三步循环：
-
思考（Thought）：智能体的"内心独白"。在这个阶段，智能体会分析当前情况、分解任务、制定下一步计划，或者反思上一步的结果。思考的质量直接影响后续行动的有效性。
-
行动（Action）：智能体决定采取的具体动作。通常是通过特定语法调用一个外部工具，例如
Search['苹果最新款手机']。行动是将思考转化为实际操作的桥梁。 -
观察（Observation）：执行Action后从外部工具返回的结果。这个结果会被添加到上下文中，供下一轮思考使用。观察的质量决定了智能体能否获取有效信息来修正其策略。
这个循环会不断重复，直到智能体认为已经获得满意的结果，或者达到预设的最大循环次数。在这个过程中，上下文信息会不断积累，帮助智能体做出更准确的判断。
1.2 ReAct与传统方法的对比
与传统的大语言模型直接生成答案的方式相比，ReAct具有明显优势：
- 可解释性：通过Thought环节，我们可以清楚地了解智能体的思考过程
- 动态调整：根据Observation结果，智能体可以实时调整策略
- 工具整合：可以灵活调用各种外部工具来弥补大模型的局限性
- 准确性：通过多轮验证，最终结果更加可靠
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. ReAct的编码实现
2.1 环境准备与基础配置
要实现一个完整的ReAct智能体，我们需要准备以下环境：
bash复制pip install openai python-dotenv google-search-results
建议使用Python 3.10或更高版本。我们需要三个核心组件：
- 大语言模型客户端（用于思考）
- 工具执行器（用于行动）
- ReAct智能体（协调整个流程）
2.1.1 配置环境变量
创建.env文件存储敏感信息：
ini复制# LLM配置
LLM_API_KEY="YOUR-API-KEY"
LLM_MODEL_ID="YOUR-MODEL"
LLM_BASE_URL="YOUR-URL"
# SerpApi配置
SERPAPI_API_KEY="YOUR_SERPAPI_API_KEY"
2.1.2 LLM客户端实现
python复制import os
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict
load_dotenv()
class HelloAgentsLLM:
def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None, timeout: int = None):
self.model = model or os.getenv("LLM_MODEL_ID")
apiKey = apiKey or os.getenv("LLM_API_KEY")
baseUrl = baseUrl or os.getenv("LLM_BASE_URL")
timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60))
if not all([self.model, apiKey, baseUrl]):
raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。")
self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout)
def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
try:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
stream=True,
)
collected_content = []
for chunk in response:
content = chunk.choices[0].delta.content or ""
collected_content.append(content)
return "".join(collected_content)
except Exception as e:
print(f"调用LLM API时发生错误: {e}")
return None
2.2 工具的定义与实现
工具是智能体与外部世界交互的接口。我们需要实现一个搜索工具和一个工具执行器。
2.2.1 搜索工具实现
python复制import os
from serpapi import SerpApiClient
def search(query: str) -> str:
try:
api_key = os.getenv("SERPAPI_API_KEY")
if not api_key:
return "错误:SERPAPI_API_KEY未配置"
params = {
"engine": "google",
"q": query,
"api_key": api_key,
"gl": "cn",
"hl": "zh-cn",
}
client = SerpApiClient(params)
results = client.get_dict()
if "answer_box_list" in results:
return "\n".join(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"]
if "organic_results" in results and results["organic_results"]:
snippets = [
f"[{i + 1}] {res.get('title', '')}\n{res.get('snippet', '')}"
for i, res in enumerate(results["organic_results"][:3])
]
return "\n\n".join(snippets)
return f"未找到关于'{query}'的信息"
except Exception as e:
return f"搜索时发生错误: {e}"
2.2.2 工具执行器
python复制from typing import Dict, Any
class ToolExecutor:
def __init__(self):
self.tools: Dict[str, Dict[str, Any]] = {}
def registerTool(self, name: str, description: str, func: callable):
if name in self.tools:
print(f"警告:工具'{name}'已存在，将被覆盖。")
self.tools[name] = {"description": description, "func": func}
def getTool(self, name: str) -> callable:
return self.tools.get(name, {}).get("func")
def getAvailableTools(self) -> str:
return "\n".join([
f"- {name}: {info['description']}"
for name, info in self.tools.items()
])
2.3 ReAct智能体核心实现
python复制import re
from typing import List, Dict
class ReActAgent:
REACT_PROMPT_TEMPLATE = """
请注意，你是一个有能力调用外部工具的智能助手。
可用工具如下:
{tools}
请严格按照以下格式进行回应:
Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。
Action: 你决定采取的行动，必须是以下格式之一:
- `{{tool_name}}[{{tool_input}}]`:调用一个可用工具。
- `Finish[最终答案]`:当你认为已经获得最终答案时。
"""
def __init__(self, llm_client, tool_executor, max_steps=5):
self.llm_client = llm_client
self.tool_executor = tool_executor
self.max_steps = max_steps
self.history = []
def run(self, question: str):
self.history = []
current_step = 0
while current_step < self.max_steps:
current_step += 1
# 1. 格式化提示词
tools_desc = self.tool_executor.getAvailableTools()
history_str = "\n".join(self.history)
prompt = self.REACT_PROMPT_TEMPLATE.format(
tools=tools_desc,
question=question,
history=history_str
)
# 2. 调用LLM进行思考
messages = [{"role": "user", "content": prompt}]
response_text = self.llm_client.think(messages=messages)
if not response_text:
print("LLM未能返回有效响应")
break
# 3. 解析LLM的输出
thought, action = self._parse_output(response_text)
if thought:
print(f"思考: {thought}")
if not action:
print("未能解析出有效的Action")
break
# 4. 执行Action
if action.startswith("Finish"):
final_answer = re.match(r"Finish\[(.*)\]", action).group(1)
print(f"最终答案: {final_answer}")
return final_answer
tool_name, tool_input = self._parse_action(action)
if not tool_name or not tool_input:
continue
print(f"行动: {tool_name}[{tool_input}]")
tool_function = self.tool_executor.getTool(tool_name)
if not tool_function:
observation = f"错误:未找到名为'{tool_name}'的工具"
else:
observation = tool_function(tool_input)
print(f"观察: {observation}")
self.history.append(f"Action: {action}")
self.history.append(f"Observation: {observation}")
print("已达到最大步数")
return None
def _parse_output(self, text: str):
thought_match = re.search(r"Thought:\s*(.*?)(?=\nAction:|$)", text, re.DOTALL)
action_match = re.search(r"Action:\s*(.*?)$", text, re.DOTALL)
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
def _parse_action(self, action_text: str):
match = re.match(r"(\w+)\[(.*)\]", action_text, re.DOTALL)
if match:
return match.group(1), match.group(2)
return None, None
3. ReAct实战演示
3.1 完整运行示例
python复制from llm_client import HelloAgentsLLM
from tool_executor import ToolExecutor
from ReAct import ReActAgent
from search_tool import search
def main():
# 初始化组件
llm_client = HelloAgentsLLM()
tool_executor = ToolExecutor()
# 注册工具
search_description = "一个网页搜索引擎。用于回答关于时事、事实等最新信息。"
tool_executor.registerTool("Search", search_description, search)
# 创建智能体
agent = ReActAgent(
llm_client=llm_client,
tool_executor=tool_executor,
max_steps=30
)
# 运行智能体
questions = [
"苹果手机的最新型号是什么",
"2024年巴黎奥运会的举办时间是什么时候",
"Python中如何读取CSV文件"
]
for question in questions:
print(f"\n问题: {question}")
answer = agent.run(question)
print(f"最终答案: {answer}")
if __name__ == "__main__":
main()
3.2 典型执行流程分析
以"苹果手机的最新型号是什么"为例，智能体的执行流程如下：
-
第一轮循环：
- Thought：分析问题，决定使用搜索引擎查找最新型号
- Action：调用Search工具查询"苹果手机最新型号"
- Observation：获取到包含iPhone 17 Pro等信息的搜索结果
-
第二轮循环：
- Thought：评估搜索结果，发现信息可能不准确，决定查询官网
- Action：调用Search工具查询"苹果官网 iPhone 最新型号"
- Observation：获取官网信息
-
第三轮循环：
- Thought：综合分析所有信息，确认最新型号
- Action：使用Finish输出最终答案
3.3 执行结果示例
code复制问题: 苹果手机的最新型号是什么
思考: 用户询问苹果手机的最新型号是什么。这是一个关于当前产品信息的问题...
行动: Search[苹果手机最新型号]
观察: [1] iPhone - 机型比较...
思考: 根据搜索结果，出现了iPhone 17 Pro等型号...
行动: Search[苹果官网 iPhone 最新型号]
观察: [1] iPhone...
思考: 综合分析所有信息...
最终答案: 根据苹果官方信息，目前苹果手机的最新型号包括iPhone 17 Pro、iPhone 17和全新亮相的iPhone Air。
4. ReAct的优缺点分析
4.1 优势分析
-
透明可解释：
- 通过Thought环节，我们可以清楚了解智能体的思考过程
- 每个决策都有明确的理由，便于调试和优化
-
动态适应性：
- 根据Observation结果实时调整策略
- 能够处理复杂、多变的现实问题
-
工具整合能力：
- 可以灵活调用各种外部工具
- 弥补了大模型在实时性、准确性等方面的局限
-
结果可靠性：
- 通过多轮验证确保最终答案的准确性
- 减少了"幻觉"问题的发生
4.2 局限性
-
依赖大模型质量：
- 思考环节的质量直接影响整体表现
- 需要较强的大语言模型支持
-
执行效率问题：
- 多轮调用带来较高的延迟
- 每次调用都产生API成本
-
提示词敏感性：
- 提示词设计对效果影响很大
- 需要精心设计和调试
-
错误累积风险：
- 前一步的错误可能导致后续偏离
- 需要设计有效的错误恢复机制
5. 优化与实践建议
5.1 性能优化策略
-
缓存机制：
- 对常见问题的结果进行缓存
- 减少不必要的API调用
-
并行执行：
- 对可以并行的Action进行优化
- 减少整体响应时间
-
超时控制：
- 设置合理的超时限制
- 避免长时间卡在某个步骤
5.2 可靠性提升方法
-
输入验证：
- 对Action的输出进行验证
- 确保符合预期格式
-
错误处理：
- 设计完善的错误处理流程
- 提供有意义的错误信息
-
结果验证：
- 对最终答案进行验证
- 可以设计二次确认机制
5.3 扩展应用场景
-
复杂问题求解：
- 多步骤的数学问题
- 需要查证的复杂事实
-
自动化流程：
- 结合RPA实现自动化
- 处理复杂的业务流程
-
数据分析：
- 结合数据分析工具
- 实现智能数据分析
在实际应用中，ReAct范式展现出了强大的潜力，特别是在需要结合推理和外部工具的场景下。通过合理的优化和扩展，可以构建出更加智能、可靠的AI应用系统。