ReAct范式
关键词：思考-行动-观察，循环
工作流程
通过提示工程来引导模型，使得其每一步都遵循
- 先思考：分析当前情况、分解任务、制定下一步计划，或者反思上一步的结果
- 再行动：智能体决定采取的具体动作，通常是调用一个外部工具
- 最后观察：从外部工具返回的结果
循环往复，将新的观察结果添加到历史记录中，知道认为已经找到了最终答案，最后输出结果。
适用场景
- 需要外部知识的任务：如查询天气
- 需要精确计算的任务：将数学问题交给计算器工具
- 需要与API交互的任务：如操作数据库
实现方式
环境配置
以搜索为例，可以使用SerpApi，该网站提供结构化的Google搜索结果，返回的结果更精确
先安装库
pip install google-search-results
然后在SerpApi注册一个免费账户，获取apikey，同样添加到.env文件中
# .env file
# ... (保留之前的LLM配置)
SERPAPI_API_KEY="YOUR_SERPAPI_API_KEY"
工具的定义
调用serpapi来搜索
def search(query: str) -> str:
try:
api_key = os.getenv("SERPAPI_API_KEY")
if api_key is None:
raise ValueError("缺少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"])
]
return "\n\n".join(snippets)
return f"对不起，没有找到关于 '{query}' 的信息。"
except Exception as e:
return f"搜索时发生错误: {e}"
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}
print(f"工具 '{name}' 已注册")
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()
])
利用ToolExecutor来管理工具
ReAct实现
class ReAct_Agent:
#思考一下需要哪些东西，首先肯定是一个大模型接口，然后是工具
def __init__(self, llm_client : LLMs, tool_executor: ToolExecutor, max_steps : int = 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 = []
cur_step = 0
while cur_step < self.max_steps:
cur_step += 1
print(f"----------第{cur_step}步----------")
tools_desc = self.tool_executor.getAvailableTools()
history_str = "\n".join(self.history)
prompt = REACT_PROMPT_TEMPLATE.format(
tools = tools_desc,
question = question,
history = history_str
)
messages = [{"role" : "user", "content" : prompt}]
response_text = self.llm_client.think(messages=messages)
if not response_text:
print("错误：大模型未能返回有效响应")
break
thougt, action = self._parse_output(response_text)
# print(action)
if thougt:
print(f"思考:{thougt}")
if not action:
print("警告：未能解析出有效的Action，流程终止")
break
if action.startswith("Finish"):
final_answer = self._parse_action_input(action)
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
如果使用这个模板，需要在Prompt里说明让大模型严格按照以下格式来回答，不然提取不到思考过程和行为。
Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。
Action: 你决定采取的行动，必须是以下格式之一：
- `{{tool_name}}[{{tool_input}}]`：调用一个可用工具。
- `Finish[最终答案]`：当你认为已经获得最终答案时。
- 当你收集到足够的信息，能够回答用户的最终问题时，你必须在`Action:`字段后使用 `Finish[最终答案]` 来输出最终答案。
Plan-and-Solve范式
工作流程
该范式将流程解耦未两个步骤
- 规划阶段：智能体在接收到用户的问题后，将问题进行分解，并指定行动计划。
- 执行阶段：智能体按照之前的计划，严格执行。
计划器的实现
class Planner:
def __init__(self, llm_client : LLMs):
self.llm_client = llm_client
def plan(self, question : str)->list[str]:
prompt = PLANNER_PROMPT_TEMPLATE.format(question)
messages = [{"role" : "user", "content" : prompt}]
print("----------正在生成计划----------")
response_text = self.llm_client.think(messages=messages) or ""
print("计划已生成：\n{response_text}")
try:
plan_str = response_text.split("```python")[1].split("```")[0].strip()
plan = ast.literal_eval(plan_str)
return plan if isinstance(plan, list) else []
except(ValueError, SyntaxError, IndexError) as e:
print(f"解析计划时出错：{e}")
print(f"原始响应：{response_text}")
return []
except Exception as e:
print("解析计划时发生未知错误：{e}")
return []
该模板建议搭配以下prompt：
"""
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。
问题: {question}
请严格按照以下格式输出你的计划,```python与```作为前后缀是必要的:
```python
["步骤1", "步骤2", "步骤3", ...]
```
"""
执行器与状态管理
class Executor:
def __init__(self, llm_client : LLMs):
self.llm_client = llm_client
def execute(self, question : str, plan : list[str]) -> str:
history = ""
print("\n----------正在执行计划----------")
for i , step in enumerate(plan):
print(f"正在执行步骤{i+1}/{len(plan)}: {step}")
prompt = EXECUTOR_PROMPT_TEMPLATE.format(
question = question,
plan = plan,
history = history if history else "无",
current_step = step
)
messages = [{"role" : "user", "content" : prompt}]
response_text = self.llm_client.think(messages=messages) or ""
history += f"步骤{i+1}: {step}\n结果：{response_text}\n\n"
print(f"步骤{i+1}已完成，结果：{response_text}")
final_answer = response_text
return final_answer
该模板建议搭配以下prompt
"""
你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。
你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
请你专注于解决“当前步骤”，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。
# 原始问题:
{question}
# 完整计划:
{plan}
# 历史步骤与结果:
{history}
# 当前步骤:
{current_step}
请仅输出针对“当前步骤”的回答:
"""
Plan-and-Solve实现
class Plan_and_Solve:
def __init__(self, llm_client : LLMs):
self.llm_client = llm_client
self.planner = Planner(self.llm_client)
self.executor = Executor(self.llm_client)
def run(self, question : str):
print("\n----------开始处理问题----------\n问题：{question}")
plan = self.planner.plan(question)
if not plan:
print("\n----------任务终止----------\n无法生成有效的行动计划")
return
final_answer = self.executor.execute(question, plan)
print(f"----------任务完成----------\n最终结果：{final_answer}")
Reflection
工作流程
- 初步执行：拿到任务后，智能体先初步生成一个解决方案
- 反思：智能体调用一个独立的大模型作为评审员，来从多个维度评审之前生成的解决方案
- 优化：智能体将初稿和反思作为新的上下文，再次调用大模型进行修订
记忆模块
from typing import List, Dict, Any, Optional
class Memory:
def __init__(self):
self.records : List[Dict[str, Any]] = []
def add_record(self, record_type: str, content: str):
record = {"type" : record_type, "content" : content}
self.records.append(record)
print(f"记忆已更新，新增一条'{record_type}' 记录。")
def get_trajectory_parts(self) -> str:
trajectory_parts = []
for record in self.records:
if record['type'] == 'execution':
trajectory_parts.append(f"----------上一轮尝试(代码)----------\n{record['content']}")
elif record['type'] == 'reflection':
trajectory_parts.append(f"----------评审反馈----------\n{record['content']}")
return "\n\n".join(trajectory_parts)
def get_last_execution(self) -> Optional[str]:
for record in self.records:
if record['type'] == 'execution':
return record['content']
return None
Reflection实现
class Reflection:
def __int__(self, llm_client : LLMs, max_iteration = 3):
self.llm_client = llm_client
self.memory = Memory()
self.max_iteration = max_iteration
def run(self, task : str):
print("\n----------开始处理任务----------\n")
print("\n----------正在进行初始尝试----------\n")
initial_prompt = INITIAL_PROMPT_TEMPLATE.format(task = task)
initial_code = self._get_llm_response(initial_prompt)
self.memory.add_record("execution", initial_code)
for i in range(self.max_iteration):
print(f"\n--- 第 {i+1}/{self.max_iterations} 轮迭代 ---")
print("\n-> 正在进行反思...")
last_code = self.memory.get_last_execution()
reflect_prompt = REFINE_PROMPT_TEMPLATE.format(task = task, code = last_code)
feedback = self._get_llm_response(reflect_prompt)
self.memory.add_record("reflection", feedback)
if "无需改进" in feedback:
print("\n 反思认为代码已无需改进，任务完成。")
break
print("\n正在进行优化")
refine_prompt = REFINE_PROMPT_TEMPLATE.format(
task=task,
last_code_attempt=last_code,
feedback=feedback
)
refined_code = self._get_llm_response(refine_prompt)
self.memory.add_record("execution", refined_code)
final_code = self.memory.get_last_execution()
print(f"\n--- 任务完成 ---\n最终生成的代码:\n```python\n{final_code}\n```")
return final_code
def _get_llm_response(self, prompt : str) -> str:
messages = [{"role" : "user", "content" : prompt}]
response_text = self.llm_client.think(messages=messages) or ""
return response_text
如果使用该模板
INITIAL_PROMPT_TEMPLATE建议为:
"""
你是一位资深的Python程序员。请根据以下要求，编写一个Python函数。
你的代码必须包含完整的函数签名、文档字符串，并遵循PEP 8编码规范。
要求: {task}
请直接输出代码，不要包含任何额外的解释。
"""
REFLECT_PROMPT_TEMPLATE建议为:
"""
你是一位极其严格的代码评审专家和资深算法工程师，对代码的性能有极致的要求。
你的任务是审查以下Python代码，并专注于找出其在<strong>算法效率</strong>上的主要瓶颈。
# 原始任务:
{task}
# 待审查的代码:
```python
{code}
```
请分析该代码的时间复杂度，并思考是否存在一种<strong>算法上更优</strong>的解决方案来显著提升性能。
如果存在，请清晰地指出当前算法的不足，并提出具体的、可行的改进算法建议（例如，使用筛法替代试除法）。
如果代码在算法层面已经达到最优，才能回答“无需改进”。
请直接输出你的反馈，不要包含任何额外的解释。
"""
REFINE_PROMPT_TEMPLATE建议为:
"""
你是一位资深的Python程序员。你正在根据一位代码评审专家的反馈来优化你的代码。
# 原始任务:
{task}
# 你上一轮尝试的代码:
{last_code_attempt}
评审员的反馈：
{feedback}
请根据评审员的反馈，生成一个优化后的新版本代码。
你的代码必须包含完整的函数签名、文档字符串，并遵循PEP 8编码规范。
请直接输出优化后的代码，不要包含任何额外的解释。
"""
总结
- ReAct优势在于环境适应能力和纠错能力
- Plan-and-Solve优势在于结构性和稳定性
- Reflection优势在于能显著提升解决方案的质量
