当前位置：首页 > news >正文

ReAct Agent 原生代码实现（纯Python实现）

news 2025/8/8 9:35:44

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"] = stoptry: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 textdef 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_stateif 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'] = keywordlookup_state['results'] = [s for s in sentences if keyword.lower() in s.lower()]lookup_state['index'] = 0if lookup_state['index'] >= len(lookup_state['results']):return "没有更多结果。", lookup_stateelse:result = lookup_state['results'][lookup_state['index']]obs = f"(结果 {lookup_state['index'] + 1} / {len(lookup_state['results'])}) {result}"lookup_state['index'] += 1return 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 = textaction = ""# 清理重复标记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 = Nonelookup_state = {'keyword': None, 'results': [], 'index': 0}answer = Noneif 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_strif verbose:print(step_str)if answer:breakif not answer:answer = "无法确定答案"if verbose:print("=" * 50)print(f"最终答案：{answer}")return {'answer': answer, 'trajectory': prompt}if __name__ == "__main__":react_solve("ChatGPT是哪个公司的？这个公司有哪些子公司与产品？")