ReAct模式
--- title: "ReAct模式" description: "深入理解ReAct推理与行动模式的原理，掌握思考-行动-观察循环的实现方法" tags: ["ReAct", "推理与行动", "思考循环", "智能体"] category: "llm" icon: "🧠"
ReAct模式
ReAct简介
ReAct（Reasoning + Acting）是一种让大语言模型结合推理和行动的提示框架。它通过让模型交替进行"思考"（Thought）、"行动"（Action）和"观察"（Observation）的循环，来解决复杂的任务。
ReAct的核心思想是：模型应该先思考问题，决定采取什么行动，然后观察行动的结果，再基于结果继续思考，如此循环直到完成任务。
核心概念
1. 思考-行动-观察循环
import re
from typing import Dict, List, Tuple
class ReActAgent:
def __init__(self, llm, tools):
self.llm = llm
self.tools = {tool.name: tool for tool in tools}
self.trajectory = []
def build_prompt(self, question: str, history: List[Tuple]) -> str:
"""构建ReAct提示"""
prompt = f"""Answer the following questions as best you can. You have access to the following tools:
{self._format_tools()}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{', '.join(self.tools.keys())}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {question}"""
# 添加历史记录
for thought, action, action_input, observation in history:
prompt += f"""
Thought: {thought}
Action: {action}
Action Input: {action_input}
Observation: {observation}"""
return prompt
def _format_tools(self) -> str:
"""格式化工具描述"""
tool_descriptions = []
for name, tool in self.tools.items():
tool_descriptions.append(f"{name}: {tool.description}")
return "\n".join(tool_descriptions)
def parse_response(self, response: str) -> Dict:
"""解析LLM响应"""
result = {
"thought": "",
"action": "",
"action_input": "",
"final_answer": ""
}
# 提取Thought
thought_match = re.search(r"Thought:\s*(.+?)(?=Action:|$)", response, re.DOTALL)
if thought_match:
result["thought"] = thought_match.group(1).strip()
# 提取Action
action_match = re.search(r"Action:\s*(.+?)(?=Action Input:|$)", response, re.DOTALL)
if action_match:
result["action"] = action_match.group(1).strip()
# 提取Action Input
action_input_match = re.search(r"Action Input:\s*(.+?)(?=Observation:|$)", response, re.DOTALL)
if action_input_match:
result["action_input"] = action_input_match.group(1).strip()
# 提取Final Answer
final_match = re.search(r"Final Answer:\s*(.+?)$", response, re.DOTALL)
if final_match:
result["final_answer"] = final_match.group(1).strip()
return result
def execute_action(self, action: str, action_input: str) -> str:
"""执行动作"""
if action in self.tools:
return self.tools[action]._run(action_input)
return f"Error: Unknown action '{action}'"
def run(self, question: str, max_iterations: int = 10) -> str:
"""运行ReAct循环"""
history = []
for i in range(max_iterations):
print(f"\n--- 迭代 {i+1} ---")
# 构建提示并获取响应
prompt = self.build_prompt(question, history)
response = self.llm.predict(prompt)
print(f"LLM响应:\n{response}")
# 解析响应
parsed = self.parse_response(response)
# 如果有最终答案，返回
if parsed["final_answer"]:
print(f"最终答案: {parsed['final_answer']}")
return parsed["final_answer"]
# 执行动作
if parsed["action"]:
print(f"思考: {parsed['thought']}")
print(f"动作: {parsed['action']}")
print(f"输入: {parsed['action_input']}")
observation = self.execute_action(parsed["action"], parsed["action_input"])
print(f"观察: {observation}")
# 记录历史
history.append((
parsed["thought"],
parsed["action"],
parsed["action_input"],
observation
))
else:
# 如果没有动作，直接返回思考结果
return parsed["thought"]
return "达到最大迭代次数"
2. 带反思的ReAct
class ReflectiveReAct(ReActAgent):
def __init__(self, llm, tools):
super().__init__(llm, tools)
self.reflection_history = []
def reflect(self, trajectory: List[Tuple]) -> str:
"""对执行过程进行反思"""
trajectory_str = "\n".join([
f"思考: {t[0]}\n动作: {t[1]}\n结果: {t[3]}"
for t in trajectory
])
prompt = f"""你是一个智能助手，正在反思刚才的执行过程。
执行轨迹：
{trajectory_str}
请分析：
1. 哪些步骤是有效的？
2. 哪些步骤可以改进？
3. 是否需要调整策略？
反思结果："""
reflection = self.llm.predict(prompt)
self.reflection_history.append(reflection)
return reflection
def run_with_reflection(self, question: str, max_iterations: int = 10,
reflect_every: int = 3) -> str:
"""带反思的ReAct"""
history = []
for i in range(max_iterations):
print(f"\n--- 迭代 {i+1} ---")
# 构建提示
prompt = self.build_prompt(question, history)
# 添加反思上下文
if self.reflection_history:
recent_reflections = self.reflection_history[-2:]
prompt += "\n\n之前的反思：\n" + "\n".join(recent_reflections)
# 获取响应
response = self.llm.predict(prompt)
parsed = self.parse_response(response)
# 检查完成
if parsed["final_answer"]:
return parsed["final_answer"]
# 执行
if parsed["action"]:
observation = self.execute_action(parsed["action"], parsed["action_input"])
history.append((
parsed["thought"],
parsed["action"],
parsed["action_input"],
observation
))
# 定期反思
if (i + 1) % reflect_every == 0 and len(history) > 0:
print("\n=== 执行反思 ===")
reflection = self.reflect(history)
print(f"反思: {reflection}")
return "达到最大迭代次数"
工具定义
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
class SearchInput(BaseModel):
query: str = Field(description="搜索查询")
class SearchTool(BaseTool):
name = "search"
description = "在互联网上搜索信息。输入应该是搜索查询字符串。"
args_schema = SearchInput
def _run(self, query: str) -> str:
# 模拟搜索结果
return f"搜索'{query}'的结果：找到了相关信息"
async def _arun(self, query: str) -> str:
return self._run(query)
class CalculatorInput(BaseModel):
expression: str = Field(description="要计算的数学表达式")
class CalculatorTool(BaseTool):
name = "calculator"
description = "计算数学表达式。输入应该是数学表达式字符串。"
args_schema = CalculatorInput
def _run(self, expression: str) -> str:
try:
result = eval(expression)
return str(result)
except Exception as e:
return f"计算错误: {str(e)}"
async def _arun(self, expression: str) -> str:
return self._run(expression)
实际应用示例
from langchain_openai import ChatOpenAI
def example_react_usage():
"""ReAct使用示例"""
# 初始化LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)
# 创建工具
tools = [SearchTool(), CalculatorTool()]
# 创建Agent
agent = ReActAgent(llm, tools)
# 运行
question = "北京今天天气怎么样？如果温度超过30度，帮我计算需要喝多少升水（按体重70kg计算）"
answer = agent.run(question)
print(f"\n最终答案: {answer}")
if __name__ == "__main__":
example_react_usage()
总结
ReAct模式通过思考-行动-观察的循环，让LLM能够有效地利用工具解决复杂问题。这种模式是构建高效AI Agent的基础，广泛应用于问答系统、任务自动化等场景。