不使用框架从零开始创建 ReAct AI Agent
一、为什么模型需要工具
LLM 的能力边界，在需要实时或外部信息的问题上体现得很明显：
- 问今天天气，它答不上来：训练数据有截止时间，模型也不知道"今天"是几号；
- 算
123 * 456 + 789这类多步算术，中间步骤容易出错，数字越大越明显； - 实时信息查询、接口调用这类操作，模型根本不具备执行能力。
提示词工程能缓解一部分问题，但解决不了根本：聊天模型只能生成文字，不能"做事"，这是架构决定的。
Agent 的思路是绕过这个限制：给模型配几件工具，让它回答之前先查一查、算一算，拿到真实结果之后再组织答案。天气问题调天气接口，算术问题调计算器，是否使用工具由模型自己判断。
之所以自己实现而不是直接用框架，原因在于框架把最关键的机制封装起来了：模型是怎么知道该调用工具的？工具返回的结果又是如何回到模型上下文里的？这两个问题不弄清楚，用框架始终隔一层。自己写一遍，代码量其实不大。
二、ReAct 的工作原理
ReAct 来自 2022 年的论文 ReAct: Synergizing Reasoning and Acting in Language Models（Yao et al.）。核心思路是让模型把推理过程显式写出来，并在推理中间插入工具调用，形成如下循环：
Thought: 我缺什么信息，下一步怎么办
Action: 调哪个工具、传什么参数
Observation: 工具返回了什么
（重复……直到能回答为止）
Final Answer: 最终答案与更早的 Chain-of-Thought（思维链）相比：CoT 引导模型"多想几步再回答"，但推理中的猜想没有机会被外部验证；ReAct 让"想"与"做"交替进行，模型说"北京现在 25 度"不作数，天气接口返回 18 度才算数，Observation 会把它拉回现实。
这套循环能工作，依赖一个朴素的机制：模型在训练语料中见过大量"先思考后行动"的文本格式，提示词再给出充分示例后，模型会按 Thought / Action / PAUSE / Observation 的固定格式输出。于是模型的"意图"变成可解析的文本，程序只需要做机械的工作：定位 Action: 行、取出工具名和参数、调用真实的 Python 函数、把结果拼成 Observation: 塞回对话。下一轮模型就能看到上一步的真实结果，决定下一步做什么。
关于 PAUSE：它是一个语义信号，告诉模型"Action 已经发出，程序正在执行工具，等待 Observation 返回即可"，避免模型在 Action 之后自行编造结果。
三、项目结构
react-agent-from-scratch/
├── agent.py # 核心：ReAct 循环、工具注册、记忆管理
├── web_app.py # Streamlit 网页版界面
├── prompts/
│ ├── system_prompt.txt # 协议提示词（含工具清单占位符和示例）
│ └── summary_prompt.txt # 记忆压缩用的提示词
├── tools/
│ ├── base_tool.py # 工具基类
│ ├── calculator.py # 计算器
│ ├── weather.py # 天气（OpenWeatherMap）
│ ├── web_search.py # 网页搜索（Tavily）
│ └── wikipedia.py # 百科（wikipedia-api）
└── utils/
└── message.py # 消息类，就是个 dict 子类仓库地址：https://github.com/zwzhangyu/ai-agent-lab
当前项目目录：awesome-agentic-ai-zh/react-agent-from-scratch
提示词独立放在 prompts/ 目录而不是硬编码在代码里，修改协议不用动逻辑。工具是可插拔的：新增能力只需在 tools/ 下添加一个继承 BaseTool 的类，Agent 代码无需修改。
四、核心循环：几十行足以跑通
在分析仓库代码之前，先看一个去掉工程细节的最小版本，把循环机制看清楚。
import re
from openai import OpenAI
client = OpenAI(api_key="...", base_url="...")
# 玩具计算器，仅演示协议用
TOOLS = {
"calculator": {"fn": lambda q: str(eval(q)), "desc": "执行数学计算"},
}
SYSTEM_PROMPT = """你是一个会使用工具的助手，按下面的循环工作：
Thought: 思考接下来做什么
Action: calculator: 数学表达式
PAUSE
Observation: 工具返回的结果
...循环...
当你得到最终答案时，输出：
Final Answer: 你的答案"""
def run(query: str) -> str:
messages = [{"role": "user", "content": query}]
for _ in range(10): # 迭代上限，防止死循环
resp = client.chat.completions.create(
model="qwen-max",
messages=[{"role": "system", "content": SYSTEM_PROMPT}] + messages,
).choices[0].message.content
print("=== LLM 输出 ===\n", resp)
if "Final Answer:" in resp: # ① 模型给出最终答案
return resp.split("Final Answer:")[-1].strip()
m = re.search(r"Action:\s*(\w+):\s*(.+)", resp)
if not m: # ② 输出不符合协议格式
return "无法解析模型的行动指令"
name, arg = m.group(1), m.group(2).strip()
observation = TOOLS[name]["fn"](arg) # ③ 执行真实函数
messages.append({ # ④ 结果回填给模型
"role": "user",
"content": f"Observation: {observation}",
})
return "超过最大迭代次数，未能得到答案"
print(run("23 * 17 等于多少？"))这段代码虽然短，Agent 的机制已经完整：步骤①判断是否得到最终答案；步骤②从输出文本中提取工具名和参数；步骤③执行真实函数；步骤④把结果回填到消息历史，让循环继续。
实际决定行为的是提示词和示例，而不是循环代码本身。模型以什么格式输出、何时调用工具，都由 prompt 决定，程序只负责忠实执行。
五、完整实现
5.1 工具接口
为了让工具可插拔、可被模型理解，先约定一个抽象基类。每个工具定义三样东西：name（模型调用时的标识）、description（给模型的说明）、run(query)（字符串输入、结果输出）：
from abc import ABC, abstractmethod
class BaseTool(ABC):
"""所有工具的抽象基类。"""
def __init__(self):
self.name = "" # 唯一名称，模型通过它调用
self.description = "" # 描述，注入 prompt 帮助模型决策
@abstractmethod
def run(self, query: str) -> str:
"""输入字符串，返回执行结果。"""
raise NotImplementedError5.2 四个工具
| 工具 | 功能 | 数据来源 | 输入格式 | 返回 |
|
| 数学计算 | 本地执行 | JSON： | 结果 |
|
| 实时天气 | OpenWeatherMap | 城市名，如 | 自然语言描述 |
|
| 实时搜索 | Tavily | 搜索词 | 标题/摘要/URL 列表 |
|
| 百科查询 | wikipedia-api | 词条名 | 标题 + 摘要 |
四个工具分别演示了 Agent 工具的不同形态：本地计算、外部 HTTP API、搜索服务、百科知识库。逐个看实现。
先看 CalculatorTool。计算器需要明确的二元运算和参数，自由文本表达不清楚，所以提示词强制模型输出 JSON，代码端负责解析与校验：
def run(self, query: str) -> str:
data = json.loads(query) # 解析失败 → 返回错误字符串
operation = data["operation"]
params = data["params"] # 校验 {"a": .., "b": ..}
if hasattr(self, operation): # 白名单方法分发
method = getattr(self, operation) # add / subtract / multiply ...
return str(method(**params))
return f"Error: Unknown operation '{operation}'"参数解析、校验、执行和异常处理都收敛在工具内部，任何失败都以字符串返回，而不是抛异常中断整个循环。
再看 WeatherTool，它是外部 HTTP API 的典型：请求 OpenWeatherMap（units=metric 返回摄氏温度），校验响应状态，把字段整理成自然语言返回：
def run(self, query):
if not query or not query.strip():
return "Error: City name cannot be empty."
api_key = os.getenv("OPENWEATHER_API_KEY", "")
if not api_key:
return "Error: OPENWEATHER_API_KEY not configured."
url = f"{self.base_url}?q={query}&appid={api_key}&units=metric"
try:
response = requests.get(url, timeout=5)
if response.status_code != 200:
return f"Error: Unable to fetch weather data. Server responded with {response.status_code}"
data = response.json()
if "main" not in data or "weather" not in data:
return f"Could not find weather data for '{query}'. Please check the city name."
temperature = data["main"]["temp"]
description = data["weather"][0]["description"]
humidity = data["main"]["humidity"]
wind_speed = data["wind"]["speed"]
return (f"The temperature in {query} is {temperature}°C. "
f"The weather is {description}. "
f"The humidity is {humidity}%. "
f"The wind speed is {wind_speed} m/s.")
except requests.exceptions.RequestException as req_err:
return f"Request failed: {str(req_err)}"这段代码体现了两点。一是返回刻意整理成完整句子：Observation 的"读者"是模型而不是程序，把结构化 JSON 翻译成自然语言，模型下一轮推理会更顺畅，所以工具返回值应尽量易读。二是错误处理：key 缺失、请求失败、城市名找不到，全部以字符串返回而不是抛异常，模型看到 Error: ... 后会自行修正重试（比如把拼音城市名换成英文）。
WebSearchTool 是反向的例子，返回保持结构化。Tavily 客户端在首次调用时惰性初始化，这样即使 API key 未配置，工具也能正常注册、不会在启动阶段报错；搜索结果整理成字典列表：
def run(self, query: str) -> str:
client = self._get_client() # 惰性初始化，key 未配置时返回 None
if not client:
return [{"error": "TAVILY_API_KEY not configured."}]
search_results = client.search(query=query, max_results=2)
formatted_results = []
for result in search_results["results"]:
formatted_results.append({
"title": result.get("title"),
"content": result.get("content"),
"url": result.get("url"),
"score": result.get("score"),
})
return formatted_results if formatted_results else [{"error": "No results found."}]WikipediaTool 更简单：用 wikipedia-api 按词条名取页（默认英文版），页面存在则返回标题与摘要，否则返回错误字典。
四个工具的返回值类型并不统一（字符串、字典、列表都有）。Agent 不要求统一类型：拼接 Observation 时会统一做 str() 转换，内容可打印即可。两种形态各有用途，天气这种"结论型"结果适合自然语言，搜索结果这种"素材型"结果适合结构化数据，模型需要从中筛选引用。
5.3 工具的自动发现
Agent 通过扫描目录获知可用工具。register_tools() 用 pkgutil 遍历 tools/ 包下的模块并动态导入，把继承 BaseTool 的类实例化后登记到字典：
def register_tools(self):
tool_modules = [name for _, name, _ in pkgutil.iter_modules(["tools"])]
for module_name in tool_modules:
module = importlib.import_module(f"tools.{module_name}")
for attr_name in dir(module):
attr = getattr(module, attr_name)
if isinstance(attr, type) and issubclass(attr, BaseTool) and attr is not BaseTool:
tool = attr()
self.tools[tool.name.lower()] = tool随后 get_tools() 把每个工具的 name 和 description 拼成清单，通过模板注入 system prompt：
- calculator: Performs mathematical calculations. Supports: add, subtract, multiply, divide, power, modulus.
- weather: Fetches current weather conditions for a specified city.
- web_search: Searches the web for up-to-date and real-time information.
- wikipedia: Searches Wikipedia for general knowledge and factual information.模型对"有哪些工具、参数怎么传"的认知全部来自这份清单。新增工具只需添加文件，Agent 本体无需修改，比在代码里维护注册列表省事；代价是行为相对隐式，目录结构本身成了注册表。
5.4 提示词模板：协议的定义
prompts/system_prompt.txt 是整个系统的核心，节选如下：
You run in a loop of Thought, Action, PAUSE, and Observation until you obtain a final answer.
Use the following format:
Thought: you should always think about what to do next
Action: the action to take, must be one of: {tools}
PAUSE
... wait for observation ...
Observation: the result of the action
When you have a final answer, output:
Final Answer: your answer here
Important rules:
1. For greetings or farewells, respond directly without using any tools.
2. If you already know the answer, respond directly with a Final Answer.
3. If you need external information, use the appropriate tool.
...
Action format for calculator (MUST use JSON):
Action: calculator: {{"operation": "add", "params": {{"a": 5, "b": 3}}}}这里有两个容易踩坑的地方。
一是规则 1、2。寒暄和常识性问题不需要调用工具，模板明确要求"知道答案就直接答"，避免模型无谓地发起工具调用，既慢又费 token。
二是花括号转义。模板中的 {tools}、{date} 是留给代码用 str.format() 填充的占位符，因此模板里字面的 JSON 花括号必须写成 {{ }} 双重转义，格式化后才会还原成 { }。如果想避开这个坑，也可以换用 f-string 或普通字符串替换来完成注入。
模板后半段给出了完整的 few-shot 示例（(7 + 2) * 4 先加后乘，两次 Action、两次 Observation，最后给出 Final Answer）。示例质量直接决定模型对协议的遵循程度，这是文本协议方案中最值得调整的部分。
5.5 主循环：think / determine_action / execute_action
循环被拆成三个方法（代码为节选，determine_action 中 calculator 的 JSON 特判分支略去，完整实现见 agent.py）：
def think(self):
"""核心推理步骤：调 LLM、记录输出、解析并执行行动。"""
self.current_iteration += 1
if self.current_iteration > self.max_iterations: # 迭代上限兜底
self.add_message("assistant",
"I'm sorry, but I couldn't find a satisfactory answer...")
return
prompt = self.system_prompt.format(tools=self.get_tools(),
date=datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
if self.old_chats_summary: # 注入历史摘要
prompt += f"\n\nOld messages summary:\n{self.old_chats_summary}"
response = self.get_llm_response(prompt)
self.add_message("assistant", response) # 回答进入消息历史
self.format_output(response) # 终端着色打印
self.determine_action(response)
def determine_action(self, response):
"""解析 LLM 输出：发现 Final Answer 就终止，否则提取 Action 并执行。"""
if "Final Answer:" in response:
return # 终局，不再递归
action_start = response.find("Action:")
if action_start == -1:
print("[WARN] No action or final answer found.")
return
action_line = response[action_start:].split("\n")[0].strip()
action_parts = action_line.replace("Action:", "").strip().split(":", 1)
tool_name, query = action_parts[0].strip().lower(), action_parts[1].strip()
self.execute_action(tool_name, query)
def execute_action(self, tool_name, query):
tool = self.tools.get(tool_name)
if tool:
result = tool.run(query) # 调用真实工具
observation = f"Observation: {tool_name} tool output: {result}"
self.add_message("system", observation) # 观察回填消息历史
self.think() # 递归进入下一轮
else:
error_msg = f"Error: Tool '{tool_name}' not found"
self.add_message("system", error_msg) # 错误也回填，让模型自纠
self.think()think() 负责一轮推理：迭代计数、组装 prompt、调用 LLM、把回答存入消息历史、解析输出。发现 Final Answer: 则停止；否则把 Action: 行解析成工具名和参数，交给 execute_action() 执行，将 Observation 存入消息历史后递归调用 think() 进入下一轮。
实现中有几个取舍需要说明。
关于递归。用 while 写循环更常规，仓库里两种写法都存在：CLI 版 agent.py 使用递归，每轮处理逻辑一致，递归是最直接的表达，配合 10 次迭代上限，栈深度可控；网页版 web_app.py 为了逐轮刷新界面，用 while 重写了同一套解析。循环的具体写法是次要的，协议、工具表和消息历史才是 ReAct 的核心。
关于 Observation 的角色。代码把观察结果作为 system 角色消息追加进历史，是一种省事的写法。严格来说 system 消息应只出现在对话开头，部分 API 会校验甚至拒绝中段的 system 消息，这里使用的 DashScope 兼容端点可以接受。规范做法是拼进 user 消息，或使用 API 原生的 tool 消息。学习项目可以接受，但需要了解这个取舍。
关于错误处理。工具名不存在、格式错误、工具内部异常，都以文本形式回填给模型。模型看到 Error: Tool 'xxx' not found 后，通常会在下一轮自行修正，换工具或换表述重试。把错误作为 Observation 反馈而不是直接中断，让 Agent 具备基本的自纠能力。
5.6 长对话与记忆管理
消息历史会随多轮对话和 Observation 的累积不断变长，token 费用和上下文上限都是实际问题。仓库采用最直接的摘要压缩策略：当 user 消息超过 3 条且总 token 超过 1000 时，把最早几轮对话交给模型，按 summary_prompt 的要求压缩成一段话，删除原文；摘要累积在 old_chats_summary 中，每轮推理时拼在 system prompt 之后重新注入：
def memory_management(self, chat_history):
user_messages = [msg for msg in chat_history if msg["role"] == "user"]
if (len(user_messages) > self.messages_to_summarize
and self.num_tokens_from_messages(chat_history) > self.max_messages_tokens):
start, end = self.get_indices(chat_history) # 定位最早 N 轮用户消息区间
chats = chat_history[start:end]
new_summary = self.summarize_old_chats(chats) # 让 LLM 生成一段摘要
self.old_chats_summary = f"{self.old_chats_summary} {new_summary}".strip()
del self.messages[start:end] # 删除原文，释放上下文token 数量通过 tiktoken 估算。该方案的本质是用信息密度换取上下文长度，代价是摘要会丢失细节，例如用户之前明确提到的某个数字。更精细的做法是向量检索，按相关性召回历史片段而不是一刀切压缩，工程复杂度会高一个量级。
5.7 运行方式：CLI 与 Web 界面
同一个 Agent 提供两种运行方式。运行前先做环境准备：进入项目目录后创建并激活虚拟环境，执行 pip install -r requirements.txt 安装依赖；在仓库的 .env 中配置 DASHSCOPE_API_KEY（参考 .env.example，模型走 DashScope 的 OpenAI 兼容接口，默认 qwen-max，可用环境变量 LLM_MODEL 覆盖）。详细步骤见同目录 README.md。
第一种是终端交互模式，直接运行 agent.py，启动后以对话方式输入问题，支持 quit / exit 退出、reset 清空会话历史：
python agent.py第二种是 Streamlit Web 界面，浏览器访问 http://localhost:8501：
streamlit run web_app.pyCLI 模式用 colorama 给 Thought、Action、PAUSE、Final Answer 分别着色，推理过程在终端直接可见。Web 版在侧边栏逐条展示推理过程（Thought / Action / PAUSE / Observation / Final Answer），主区是聊天界面。
实现上有一些重复：终端版通过 print 输出思考过程，网页版无法捕获 stdout，所以 web_app.py 用 while 重写了解析逻辑，每解析一步写入 st.session_state 触发界面刷新。同一份协议逻辑在仓库里存在两份，维护时需要注意同步。
5.8 运行示例
以 README 中的测试问题为例（calculator 的结果可以精确验算），下面是终端里的示意输出：
User: 计算 123 * 456 + 789
[ASSISTANT]:
Thought: 我需要先计算 123 * 456 的乘法结果。
[ACTION]: calculator: {"operation": "multiply", "params": {"a": 123, "b": 456}}
[PAUSE]:
[SYSTEM]: Observation: calculator tool output: 56088
[ASSISTANT]:
Thought: 乘法结果是 56088，接下来再与 789 相加。
[ACTION]: calculator: {"operation": "add", "params": {"a": 56088, "b": 789}}
[PAUSE]:
[SYSTEM]: Observation: calculator tool output: 56877
[FINAL ANSWER]: 123 * 456 + 789 = 56877验算：123 × 456 = 56088，加 789 等于 56877，结果正确。多步计算问题被拆成两次独立的工具调用，每一步都基于真实计算结果，而不是模型的心算。
天气查询的流程（示意输出，具体数值以接口返回为准）：
User: 北京今天天气怎么样？
[ASSISTANT]:
Thought: 天气是实时信息，我需要调用 weather 工具查询北京的情况。
[ACTION]: weather: Beijing
[PAUSE]:
[SYSTEM]: Observation: weather tool output: The temperature in Beijing is 18.0°C. The weather is clear sky. The humidity is 40%. The wind speed is 3.0 m/s.
[FINAL ANSWER]: 北京今天天气晴朗，气温 18 摄氏度，湿度 40%，风力不大。注意模型把城市名从中文"北京"翻译成了 Beijing 再传给工具，这是提示词示例与工具描述共同作用的结果。
搜索类问题走同样的循环，区别在于 Observation 是结构化的结果列表，模型从中筛选后再总结作答（示意）：
User: 搜索 Python 3.13 的新特性
[ASSISTANT]:
Thought: 这需要最新信息，我的知识可能过时，调用 web_search 搜索。
[ACTION]: web_search: Python 3.13 new features
[PAUSE]:
[SYSTEM]: Observation: web_search tool output: [{'title': "What's New In Python 3.13", 'content': '...', 'url': 'https://docs.python.org/3/whatsnew/3.13.html', 'score': ...}]
[FINAL ANSWER]: 根据搜索结果，Python 3.13 的主要新特性包括……
仓库地址：https://github.com/zwzhangyu/ai-agent-lab
当前项目目录：awesome-agentic-ai-zh/react-agent-from-scratch
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
6
0- 0
已为社区贡献6条内容
所有评论(0)