ReAct Agent 范式思想
核心思想如下：
- 给定 任务执行的指导思想、外部工具文本描述、Question+Thought+Action+Observation 示例、新的问题，然后拼接为提示词
- 将提示词发给LLM模型（如 ChatGPT），让其生成 Thought、Action 字符串
- 基于 Action 字符串调用对应的工具函数，将工具函数的返回的结果作为 Observation
- 然后再将新的 Thought、Action、Observation 填充到LLM提示词，让其继续生成 Thought、Action，直到 Action 为 Finish[最终答案] 时，取出“最终答案”返回
本文代码的来源
参考论文 《ReAct: Synergizing Reasoning and Acting in Language Models 》提供的源码项目：https://github.com/ysymyth/ReAct?tab=readme-ov-file，下面是我写的一份简化版的 ReAct 项目，实现了 ReAct 范式，供读者了解 ReAct Agent 范式实际执行过程，特别是 LLM 调用时的实际提示词。
代码如何使用
下面代码中只需要配置 OPENAI_API_KEY 、OPENAI_BASE_URL、MODEL_NAME 为自己的信息，然后就可以运行。
如果运行过程中缺少依赖，自行安装即可
pip install requests beautifulsoup4
（下面代码配合AI理解更佳）
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ReAct范式演示代码
基于论文《ReAct: Synergizing Reasoning and Acting in Language Models》
实现思考-行动-观察的交替循环
"""
import requests
import time
import re
from bs4 import BeautifulSoup
import json
# =============================================================================
# 配置部分，修改为自己可用的API_KEY和BASE_URL （通义千问的 qwen-turbo 就可以）
# =============================================================================
OPENAI_API_KEY = ""
OPENAI_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
MODEL_NAME = "qwen-turbo"
# =============================================================================
# LLM调用函数
# =============================================================================
def llm(prompt, stop=None, max_tokens=200, temperature=0):
"""调用LLM API"""
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {OPENAI_API_KEY}"
}
payload = {
"model": MODEL_NAME,
"messages": [{"role": "user", "content": prompt}],
"temperature": temperature,
"max_tokens": max_tokens
}
if stop:
payload["stop"] = stop
try:
resp = requests.post(OPENAI_BASE_URL, headers=headers, data=json.dumps(payload), timeout=60)
resp.raise_for_status()
data = resp.json()
if "choices" in data and len(data["choices"]) > 0:
return data["choices"][0]["message"]["content"]
else:
return "LLM调用失败"
except Exception as e:
print(f"LLM调用错误: {e}")
return "LLM调用失败"
# =============================================================================
# Wikipedia工具函数
# =============================================================================
def clean_str(text):
"""清理文本字符串"""
try:
return text.encode().decode("unicode-escape").encode("latin1").decode("utf-8")
except:
return text
def get_page_obs(page):
"""获取页面前5个句子"""
if not page:
return ""
paragraphs = [p.strip() for p in page.split("\n") if p.strip()]
sentences = []
for p in paragraphs:
sentences += p.split('. ')
sentences = [s.strip() + '.' for s in sentences if s.strip()]
return ' '.join(sentences[:5])
def search_wikipedia(entity):
"""搜索Wikipedia"""
search_url = f"https://en.wikipedia.org/w/index.php?search={entity.replace(' ', '+')}"
try:
response = requests.get(search_url, timeout=10)
soup = BeautifulSoup(response.text, features="html.parser")
result_divs = soup.find_all("div", {"class": "mw-search-result-heading"})
if result_divs:
result_titles = [clean_str(div.get_text().strip()) for div in result_divs]
return None, f"找不到[{entity}]。相似结果：{result_titles[:5]}。"
else:
page_elements = soup.find_all("p") + soup.find_all("ul")
page_texts = [p.get_text().strip() for p in page_elements]
if any("may refer to:" in p for p in page_texts):
return search_wikipedia(f"[{entity}]")
else:
page_content = ""
for p in page_texts:
if len(p.split(" ")) > 2:
page_content += clean_str(p) + "\n"
return page_content, get_page_obs(page_content)
except Exception as e:
return None, f"搜索出错：{str(e)}"
def lookup_in_page(page, keyword, lookup_state):
"""在页面中查找关键词"""
if not page:
return "没有页面内容可以查找。", lookup_state
if lookup_state.get('keyword') != keyword:
paragraphs = [p.strip() for p in page.split("\n") if p.strip()]
sentences = []
for p in paragraphs:
sentences += p.split('. ')
sentences = [s.strip() + '.' for s in sentences if s.strip()]
lookup_state['keyword'] = keyword
lookup_state['results'] = [s for s in sentences if keyword.lower() in s.lower()]
lookup_state['index'] = 0
if lookup_state['index'] >= len(lookup_state['results']):
return "没有更多结果。", lookup_state
else:
result = lookup_state['results'][lookup_state['index']]
obs = f"(结果 {lookup_state['index'] + 1} / {len(lookup_state['results'])}) {result}"
lookup_state['index'] += 1
return obs, lookup_state
# =============================================================================
# ReAct提示词模板
# =============================================================================
REACT_PROMPT = """解决问答任务时需要交替进行"思考（Thought）"、"行动（Action）"和"观察（Observation）"的步骤。
"思考"可以用来推理当前情况；"行动"可以是以下三种类型：
（1）Search[实体]：在维基百科上搜索该确切实体，如果存在则返回第一段文字；如果不存在，则返回一些相似的实体以供继续搜索。
（2）Lookup[关键词]：在当前文章中返回包含该关键词的下一句。
（3）Finish[答案]：返回最终答案并结束任务。
以下是一些示例：
Question：Colorado造山运动的东部区域延伸到的地区，其海拔范围是多少？
Thought 1：我需要搜索Colorado造山运动，找出其东部区域延伸到哪里，然后查找该区域的海拔范围。
Action 1：Search[Colorado orogeny]
Observation 1：Colorado造山运动是一次在科罗拉多州及周边地区发生的造山运动。
Thought 2：没有提到东部区域。所以我需要查找"eastern sector"。
Action 2：Lookup[eastern sector]
Observation 2：（结果1 / 1）东部区域延伸到了高原地带，称为中央平原造山运动（Central Plains orogeny）。
Thought 3：Colorado造山运动的东部区域延伸到了高原地带。所以我需要搜索"High Plains"并查找其海拔范围。
Action 3：Search[High Plains]
Observation 3：High Plains指的是两个不同的地貌区域之一：
Thought 4：我需要改为搜索"High Plains (United States)"。
Action 4：Search[High Plains (United States)]
Observation 4：High Plains是大平原（Great Plains）的一个子区域。从东向西，High Plains的海拔从约1800英尺升高至7000英尺（550至2130米）。
Thought 5：High Plains的海拔范围是约1800到7000英尺，所以答案是1800到7000英尺。
Action 5：Finish[1800 to 7000 ft]
Question：音乐人兼讽刺作家Allie Goertz写了一首关于《辛普森一家》中角色Milhouse的歌，这个角色是Matt Groening以谁命名的？
Thought 1：问题可以简化为"Milhouse是以谁命名的"。我只需要搜索Milhouse，看看他是以谁命名的。
Action 1：Search[Milhouse]
Observation 1：Milhouse Mussolini Van Houten是福克斯动画剧《辛普森一家》中的常设角色，由Pamela Hayden配音，Matt Groening创作。
Thought 2：这一段没有说Milhouse是以谁命名的，也许我可以查找"named after"。
Action 2：Lookup[named after]
Observation 2：（结果1 / 1）Milhouse是以美国总统Richard Nixon命名的，他的中间名是Milhous。
Thought 3：Milhouse是以美国总统Richard Nixon命名的，所以答案是Richard Nixon。
Action 3：Finish[Richard Nixon]
Question：以下哪一部纪录片是关于芬兰摇滚乐队的，《亚当·克莱顿·鲍威尔》还是《萨伊马之声》？
Thought 1：我需要搜索《亚当·克莱顿·鲍威尔》和《萨伊马之声》，找出哪部是关于芬兰摇滚乐队的纪录片。
Action 1：Search[Adam Clayton Powell]
Observation 1：找不到[Adam Clayton Powell]。相似项包括：['Adam Clayton Powell III', 'Seventh Avenue (Manhattan)', 'Adam Clayton Powell Jr. State Office Building', 'Isabel Washington Powell', 'Adam Powell', 'Adam Clayton Powell (film)', 'Giancarlo Esposito']
Thought 2：要找纪录片，我可以搜索"Adam Clayton Powell (film)"。
Action 2：Search[Adam Clayton Powell (film)]
Observation 2：《亚当·克莱顿·鲍威尔》是一部1989年由Richard Kilberg导演的美国纪录片。
该片讲述的是有影响力的非裔美国政治家亚当·克莱顿·鲍威尔的兴衰历史。之后作为PBS系列《美国经历》的内容播出。
Thought 3：《亚当·克莱顿·鲍威尔》是一部关于非裔美国政治家的纪录片，并非关于芬兰摇滚乐队的。因此，关于芬兰摇滚乐队的纪录片应该是《萨伊马之声》。
Action 3：Finish[The Saimaa Gesture]
"""
# =============================================================================
# 解析函数
# =============================================================================
def parse_thought_action(text, step_num):
"""解析LLM返回的思考和行动"""
text = text.strip()
# 移除重复前缀
thought_prefix = f"Thought {step_num}："
action_prefix = f"Action {step_num}："
if text.startswith(thought_prefix):
text = text[len(thought_prefix):].strip()
if text.startswith(thought_prefix):
text = text[len(thought_prefix):].strip()
# 解析
if f"\n{action_prefix}" in text:
parts = text.split(f"\n{action_prefix}", 1)
thought = parts[0].strip()
action = parts[1].strip()
elif action_prefix in text:
parts = text.split(action_prefix, 1)
thought = parts[0].strip()
action = parts[1].strip()
else:
action_pattern = rf"Action\s*{step_num}\s*[:：]\s*(.+?)(?:\n|$)"
action_match = re.search(action_pattern, text, re.IGNORECASE)
if action_match:
action = action_match.group(1).strip()
thought = text[:action_match.start()].strip()
else:
thought = text
action = ""
# 清理重复标记
thought = re.sub(rf"^(Thought\s*{step_num}\s*[:：]\s*)+", "", thought, flags=re.IGNORECASE).strip()
return thought, action
# =============================================================================
# ReAct主函数
# =============================================================================
def react_solve(question, max_steps=8, verbose=True):
"""使用ReAct范式解决问题"""
current_page = None
lookup_state = {'keyword': None, 'results': [], 'index': 0}
answer = None
if verbose:
print(f"问题：{question}")
print("=" * 50)
prompt = REACT_PROMPT + f"\nQuestion：{question}\n"
for i in range(1, max_steps + 1):
# 生成思考和行动
thought_action = llm(
prompt + f"Thought {i}：",
stop=[f"\nObservation {i}：", "\nQuestion：", f"\nThought {i+1}："],
max_tokens=200
)
# 解析
thought, action = parse_thought_action(thought_action, i)
# 执行行动
if action.startswith("Search[") and action.endswith("]"):
entity = action[len("Search["):-1]
current_page, obs = search_wikipedia(entity)
lookup_state = {'keyword': None, 'results': [], 'index': 0}
elif action.startswith("Lookup[") and action.endswith("]"):
keyword = action[len("Lookup["):-1]
obs, lookup_state = lookup_in_page(current_page, keyword, lookup_state)
elif action.startswith("Finish[") and action.endswith("]"):
answer = action[len("Finish["):-1]
obs = "任务完成。"
else:
obs = f"无效动作：{action}"
# 记录步骤
step_str = f"Thought {i}：{thought}\nAction {i}：{action}\nObservation {i}：{obs}\n"
prompt += step_str
if verbose:
print(step_str)
if answer:
break
if not answer:
answer = "无法确定答案"
if verbose:
print("=" * 50)
print(f"最终答案：{answer}")
return {'answer': answer, 'trajectory': prompt}
if __name__ == "__main__":
react_solve("ChatGPT是哪个公司的？这个公司有哪些子公司与产品？")