Overview
- ReAct: 推理和行动协同的范式
- ReAct: 模拟实现
- ReAct: 平衡推理和行动
ReAct: 推理和行动协同的范式
- ReAct 提出：让 LLM 同时生成 “推理轨迹” (指导行动方向) 和 “行动” (获取外部反馈更新推理) ，形成 “推理→行动→环境反馈→更新推理”的闭环
- 双行动空间模型:
- 原有行动空间 (A) ：与外部环境交互的具体操作，如 “调用维基 API 检索实体”, “点击网页按钮”, “在文本游戏中移动到桌子旁”，行动会触发环境反馈 (Observation，如检索结果, 页面内容, 游戏场景变化) ；
- 新增语言推理空间 (L) ：即 “推理轨迹 (Thought) “，是模型用自然语言表达的思考过程，不直接影响外部环境，仅用于：
- 分解任务目标 (如 “要回答这个问题，需先查 A，再验证 B”) ；
- 跟踪进度 (如 “已找到 A 的信息，下一步需查 B”) ；
- 处理异常 (如 “检索 A 失败，换相似关键词 A’重试”) ；
- 注入常识 (如 “台灯通常放在桌子或书架上，优先查这些位置”) ；
- 解析环境反馈 (如 “检索结果未提时间，需补充 Lookup 操作”) 。
- ReAct 实现方式: 基于冻结的大语言模型实现，无需大量微调，核心依赖 few-shot prompt
- 提示构造：在模型输入中插入 “人类标注的任务轨迹示例”，每个示例包含完整的 “任务描述→Thought→Action→Observation” 循环，让模型学习 “何时生成推理, 何时生成行动”；示例格式 (以 HotPotQA 为例) ：
Question: 科罗拉多造山带东段延伸区域的海拔范围是多少？
Thought 1: 需先查"科罗拉多造山带"，找到其东段延伸的区域，再查该区域的海拔。
Action 1: Search[科罗拉多造山带]
Observation 1: 科罗拉多造山带是科罗拉多及周边地区的造山运动，东段延伸至大平原 (High Plains) ，又称中央平原造山带。
Thought 2: 已确定延伸区域是大平原，下一步需查"大平原的海拔范围"。
Action 2: Search[High Plains (United States)]
Observation 2: 大平原东西向海拔从 1800 英尺升至 7000 英尺。
Thought 3: 海拔范围已找到，可结束任务。
Action 3: Finish[1800 to 7000 ft]
ReAct 模拟实现
import logging
import sys
import time
import os
from datetime import datetime
from typing import List, Dict, Any, Union
from gtts import gTTS
import gradio as gr
from langchain_openai import ChatOpenAI
from langchain.schema.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
from openai import OpenAI
# -------------------------
# 模型配置
# -------------------------
chat = ChatOpenAI(
model_name="Qwen/Qwen3-30B-A3B-Instruct-2507",
openai_api_key="",
openai_api_base="https://api.siliconflow.cn/v1",
temperature=1.0,
streaming=True
)
# user_input = "你好啊"
# # 如果你只想拿到最终完整文本，可以：
# # print("\n=== 完整 Response ===")
# # print(response.content)
# def test_api(user_input):
# langchain_messages = [
# HumanMessage(content=user_input)
# ]
# full_response = ""
# # 假设 chat 是 ChatOpenAI 对象，langchain_messages 是你的消息列表
# for chunk in chat.stream(langchain_messages):
# if chunk.content: # chunk.content 可能为 None
# new_content = chunk.content
# full_response += new_content
# print(new_content, end="", flush=True) # 实时打印新增部分
# # print(full_response)
# return full_response
# # test_api(user_input)
# -------------------------
# 工具 / 环境模拟
# -------------------------
def env_run(action: str) -> str:
if not action:
return "Observation: 无操作"
action_lower = action.lower()
if "天气" in action_lower:
return "今天旧金山晴，22°C"
elif "时间" in action_lower:
return "当前北京时间 11:00"
elif "计算" in action_lower:
return "计算结果: 42"
elif "最终答案" in action_lower or "返回最终结果" in action_lower or "输出最终答案" in action_lower:
# 直接把历史 Observation 汇总为最终答案
return "Final Answer: 今天旧金山晴，22°C"
return f"Observation: 无法理解的 Action -> {action}"
# -------------------------
# 解析模型输出
# -------------------------
def parse_output(output: str):
thought, action, final_answer = None, None, None
for line in output.split("\n"):
line = line.strip()
if not line:
continue
# 不区分大小写，更宽容
if line.lower().startswith("thought:"):
thought = line.split(":", 1)[1].strip()
elif line.lower().startswith("action:"):
action = line.split(":", 1)[1].strip()
# 将空操作或 None 标准化
if action.lower() in ["none", "无操作", "无动作", ""]:
action = None
elif line.lower().startswith("final answer:"):
final_answer = line.split(":", 1)[1].strip()
return thought, action, final_answer
# -------------------------
# 初始 Prompt 模板
# -------------------------
system_message = SystemMessage(
content=f"""你是一个智能助手，会在多轮循环中回答问题。
请严格遵循以下规则：
1. 每轮只输出两项：
- Thought: 本轮你的思考
- Action: 本轮你打算执行的操作 (可调用工具或 API)
2. 当你已收集到足够信息可以回答用户问题时，请生成:
Final Answer: <最终答案>
3. 不要在 Action 中输出最终答案。
4. 你将收到 Observation，Observation 是唯一可靠的信息来源。
5. **你的最终答案只能依据 Observation 中的内容，不得编造, 推测或补充未在 Observation 中出现的信息。**
"""
)
user_question = "旧金山今天的天气如何？"
# -------------------------
# 历史消息
# -------------------------
messages = [
system_message,
HumanMessage(content=user_question)
]
# -------------------------
# ReAct 循环
# -------------------------
max_steps = 5
for step in range(max_steps):
# 调用模型
response = chat.invoke(messages)
output = response.content
print(f"\n----------- Step {step+1} 模型输出 ------------\n{output}")
thought, action, final_answer = parse_output(output)
# 判断是否满足终止条件
# 1. 模型返回 final_answer
# 2. Observation 已经包含可识别结果 (这里用简单关键字判断)
# 3. 模型没有可执行的 Action
terminate = False
if final_answer:
terminate = True
elif action is None or action.lower() in ["无操作", "none", ""]:
terminate = True
final_answer = "任务已完成，没有进一步操作"
elif step == max_steps - 1:
terminate = True
final_answer = "达到最大步数，未获取完整答案"
if terminate:
print("\n=== 最终答案 / 循环结束提示 ===")
print(final_answer)
break
# 如果没终止，执行 Action 获取 Observation
observation = env_run(action)
print(f"Observation: {observation}")
# 如果 Observation 是最终答案，循环结束
if observation.startswith("Final Answer:"):
final_answer = observation[len("Final Answer:"):].strip()
print("\n=== 最终答案 ===")
print(final_answer)
break
# 将 Observation 反馈给模型，形成下一轮 prompt
messages.append(AIMessage(content=f"{output}"))
# 注意，这里采用 ToolMessage 相比 AIMessage 能够减低模型幻觉
messages.append(ToolMessage(content=f"Observation: {observation}", tool_call_id="sdf"))
# messages.append(AIMessage(content=f"Observation: {observation}"))
# print(messages)
ReAct: 平衡推理和行动
我们实际在 ReAct 做 Agent 的时候, 并非固定生成 “推理 - 行动:”序列，而是根据任务的 “推理需求”和 “行动复杂度” 动态调整推理轨迹的密度：
知识密集型推理任务 (如多跳问答 HotPotQA, 事实核查 FEVER) ：
- 需求：需频繁结合外部知识 (如维基) 验证推理，推理需 “密集”；
- 模式：强制 “Thought→Action→Observation”循环，每步行动前必有推理指导，确保检索方向不偏离；
- 行动空间：设计轻量 API (如 Search[实体] 检索维基前 5 句, Lookup[关键词] 模拟 Ctrl+F, Finish[答案] 结束任务)
交互式决策任务 (如文本游戏 ALFWorld, 网页购物 WebShop) ：
- 需求：行动步骤多 (如游戏需 50+ 步, 购物需多轮点击) ，推理无需每步都有，避免冗余；
- 模式：让模型自主决定推理时机 (“稀疏推理”) ，仅在关键节点生成 Thought，如 “任务开始分解目标””行动卡住时调整策略””确认是否购买时判断属性匹配度”；
- 行动空间：任务专属操作 (如游戏的 go to desk 1, 购物的 click [商品ID]) 。
Reference
[1]. ReAct: Synergizing Reasoning and Acting in Language Models.
转载请注明来源 goldandrabbit.github.io