大模型只会“说话”，而 AI Agent（智能体） 才会“办事”。本文不套用任何 Agent 框架，用纯 Python 从零手写一个 ReAct 智能体：它以“推理（Reason）→ 行动（Act）→ 观察（Observe）”的循环自主调用工具、积累记忆，直到给出最终答案。读完你就能理解 LangChain、AutoGPT 这类产品的底层骨架，并知道生产环境该补哪些保险。
一、ReAct 到底是什么
ReAct 是 “Reasoning + Acting” 的缩写。传统提示词让模型一次生成完整答案，遇到“北京现在几度”这类需要实时数据的问题就只能瞎编。ReAct 智能体 把任务拆成循环：模型先“想”下一步该做什么（Thought），再“做”一个动作（调用工具），拿到“观察”结果后继续想，直到能回答为止。这种“想一步、做一步、看一步”的范式，是当下绝大多数自主 Agent 的通用底座。
它和 Function Calling 实战 里讲的“工具调用”是同一套能力，区别在于：Function Calling 强调“模型如何正确地调用单个函数”，而 ReAct 强调“多步决策与状态流转”的工程编排。理解了 ReAct，你就能在 Spring AI 或 Dify 里更清楚地知道框架替你做了什么。
二、为什么要从零手写，而不是直接用框架
框架（LangChain、CrewAI、AutoGen）能让你 10 行代码跑起来，但也会把核心循环藏进黑盒：你不知道它什么时候停、为什么选这个工具、记忆怎么攒的。手写一遍的好处是——每一个 Token 的决策路径你都看得见，排障、加护栏、控成本都更从容。下面就是一个“最小可运行”的 Agent，不依赖任何第三方 Agent 库，只需一个 OpenAI 兼容的对话接口（本地 Ollama、vLLM 或云端都行）。
三、核心循环：Think → Act → Observe
3.1 先定义工具：函数 + JSON Schema
Agent 的“能力”来自工具。每个工具是一个普通 Python 函数，外加一份给模型看的 JSON Schema 描述。模型不会直接跑代码，它只返回“我想调用哪个工具、参数是什么”，由我们的循环负责真正执行。
import math, os, requests
# 工具 1：简单计算器
def calc(expression: str) -> str:
# 仅做四则运算，避免 eval 的安全风险
allowed = set("0123456789+-*/(). ")
if not set(expression) <= allowed:
return "error: 非法字符"
try:
return str(eval(expression, {"__builtins__": {}}, {}))
except Exception as e:
return f"error: {e}"
# 工具 2：模拟“查天气”（真实场景换成天气 API）
def get_weather(city: str) -> str:
fake = {"北京": 21, "上海": 25, "深圳": 28}
return f"{city} 当前气温 {fake.get(city, '未知')} 摄氏度"
# 工具注册表：name -> (函数, JSON Schema)
TOOLS = {
"calc": (
calc,
{
"type": "function",
"function": {
"name": "calc",
"description": "对数学表达式求值，支持 + - * / 和括号",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "如 (1+2)*3"}
},
"required": ["expression"],
},
},
},
),
"get_weather": (
get_weather,
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询城市当前气温",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
},
},
),
}
3.2 记忆（Memory）设计
Agent 的“智能”很大程度来自记忆：它必须记得自己前面想过什么、工具返回了什么，否则每一步都是“失忆”的。记忆 本质上就是一段不断追加的对话历史，既喂给模型当上下文，也保留工具结果。
class Memory:
def __init__(self):
self.turns = [] # 每条记录: {"role":.., "content":..}
def add(self, role, content):
self.turns.append({"role": role, "content": content})
def history(self):
return self.turns
mem = Memory()
mem.add("system", "你是一个会调用工具的助手，遇到不会的问题就用工具查。")
mem.add("user", "北京现在多少度？如果比上海热，告诉我温差是多少。")
四、动手实现 ReAct 循环
下面是整个 Agent 的心脏。我们用 OpenAI 兼容客户端发起对话，并在返回里检查是否包含 tool_calls；有就执行工具、把结果写回记忆，没有就说明模型已给出最终答案，循环结束。
from openai import OpenAI
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY", "sk-xxx"),
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)
MAX_STEPS = 6 # 防止无限循环
def run_agent(question: str) -> str:
mem.add("user", question)
tool_schemas = [v[1] for v in TOOLS.values()]
for step in range(MAX_STEPS):
resp = client.chat.completions.create(
model=os.getenv("MODEL", "gpt-4o-mini"),
messages=mem.history(),
tools=tool_schemas,
tool_choice="auto",
)
msg = resp.choices[0].message
mem.add("assistant", msg.content or "")
# 没有工具调用 = 模型认为可以收尾了
if not msg.tool_calls:
return msg.content
# 逐个执行模型请求的工具
for call in msg.tool_calls:
fn_name = call.function.name
args = json.loads(call.function.arguments)
fn, _ = TOOLS[fn_name]
observation = fn(**args) # 真正执行
mem.add("tool", f"[{fn_name}] -> {observation}")
return "超过最大步数，仍未给出答案。"
五、让 Agent 真正“会思考”的提示词
ReAct 的效果高度依赖提示词。一个有效写法是要求模型按固定格式输出，让“思考”和“动作”分离，方便我们解析。下面这段 system 提示让模型先写 Thought 再决定 Action：
SYSTEM_PROMPT = """你是一个 ReAct 智能体。每步严格按以下格式输出：
Thought: 你现在的推理
Action: 要调用的工具名（calc / get_weather）
Action Input: 工具参数（JSON）
当你能回答时，输出：
Final Answer: 最终回答
你只能使用提供的工具，不要编造结果。"""
六、跑一次完整对话
把上面的零件拼起来，问“北京比上海热多少”时，Agent 的记忆会这样生长：
user : 北京现在多少度？比上海热多少？
assistant : Thought: 需要先查两座城市的气温。
Action: get_weather Input: {"city":"北京"}
tool : [get_weather] -> 北京 当前气温 21 摄氏度
assistant : Thought: 再查上海。
Action: get_weather Input: {"city":"上海"}
tool : [get_weather] -> 上海 当前气温 25 摄氏度
assistant : Thought: 上海 25 度比北京 21 度热，应纠正用户。
Final Answer: 实际上上海（25℃）比北京（21℃）热 4℃，北京并不更热。
可以看到，Agent 自己纠正了问题里的错误前提——这正是“推理+行动”闭环比单次问答强的地方。想让模型能力更扎实，配合 Embedding 选型 做检索、用 RAG 评估 量化效果，是常见的进阶路线。
七、生产环境必须加的 4 道保险
Demo 跑通只是开始。把 Agent 放线上，下面这些风险必须提前兜住，否则轻则烧钱、重则“跑飞”调用了不该调的接口。
| 风险 | 表现 | 对策 |
|---|---|---|
| 无限循环 | 模型反复调用工具不收尾 | 设 MAX_STEPS，超限强制返回 |
| 工具越权 | 调用了未注册/危险函数 | 白名单校验 fn_name 必须在 TOOLS |
| 提示词注入 | 用户输入诱导泄露系统指令 | 工具结果做沙箱隔离与长度裁剪 |
| 成本失控 | 长对话 token 暴涨 | 记忆摘要化、对历史做窗口截断 |
八、它和 MCP、Dify 是什么关系
手写 Agent 让你吃透原理，但真实项目里往往会“借力”：用 MCP 协议 把外部工具标准化成可插拔服务，让 Agent 动态发现能力；用 Dify 这类平台把编排、记忆、评估做成可视化流水线。底层循环仍是本文讲的 ReAct——区别只是工具从“本地函数”变成了“远程 MCP Server”，记忆从“列表”变成了“向量库 + 摘要”。
九、把示例工具换成真实 API
示例里的 get_weather 是写死的。真实项目里工具就是普通的 HTTP 接口——只要遵循“函数签名 + JSON Schema”的契约，上面那段 ReAct 循环一行都不用改。下面把天气换成一次真实调用（端点与 Key 换成你自己的即可）：
def get_weather(city: str) -> str:
try:
r = requests.get(
"https://api.example.com/weather",
params={"city": city, "token": os.getenv("WX_TOKEN")},
timeout=5,
)
r.raise_for_status()
return f"{city} 当前气温 {r.json()['temp']} 摄氏度"
except Exception as e:
return f"查询失败: {e}"
TOOLS["get_weather"] = (get_weather, TOOLS["get_weather"][1])
关键在于工具函数保持“纯输入输出”：所有网络细节、异常、重试都在函数内部消化，对外只返回一段干净的字符串。这样模型拿到的永远是可读结果，不会因为 JSON 格式错乱而“理解崩坏”。这也是为什么在 Function Calling 实战 里我们反复强调“工具描述要准确、参数要可校验”。
十、记忆太长怎么办：摘要化与窗口截断
多轮工具调用后，messages 会迅速膨胀，既烧 token 又可能撑爆上下文窗口。常见做法是给 Memory 加一个“压缩”开关：当估算 token 超过阈值，就把最早的若干轮交给模型 summarise 成一段摘要，再替换原内容。
def estimate_tokens(turns):
return sum(len(t["content"]) // 2 for t in turns) # 中文约 2 字/token
def compress_memory(mem, client, model, threshold=3000):
if estimate_tokens(mem.turns) <= threshold:
return
old, mem.turns = mem.turns[:-4], []
prompt = "用一句中文概括以下对话要点：\n" + "\n".join(
f"{t['role']}: {t['content']}" for t in old
)
summary = client.chat.completions.create(
model=model, messages=[{"role": "user", "content": prompt}]
).choices[0].message.content
mem.turns = [{"role": "system", "content": "历史摘要: " + summary}] + old[-4:]
另一种更轻量的策略是滑动窗口：只保留最近 N 轮，外加一条持久化的“任务目标”摘要。对大多数客服、问答类 Agent 已经够用，且几乎零额外成本。当对话涉及长期偏好时，再把摘要定期写入外部存储（如 向量库）做长期记忆。
十一、让 Agent 更稳：重试、超时与流式
工具调用本质是 I/O，必然会失败。给每个工具包一层超时与重试，能显著降低“跑飞”概率；同时务必限制单工具最大耗时，避免一个慢接口拖垮整轮循环。
import functools, requests
def with_timeout(seconds=5):
def deco(fn):
@functools.wraps(fn)
def wrap(*a, **k):
return requests.get # 占位：实际用 concurrent 或 signal 设超时
return wrap
return deco
# 更稳妥：直接在请求层设 timeout，并在循环里捕获后重试一次
def safe_call(fn, *a, retries=1, **k):
for i in range(retries + 1):
try:
return fn(*a, **k)
except Exception as e:
if i == retries:
return f"工具执行失败: {e}"
time.sleep(1)
如果面向终端用户，建议把 Thought / Action 以流式方式逐步回传，让用户看到 Agent “正在思考”，体验远好于等几秒后一次性吐字。配合支持流式的对话接口，每个中间步骤都是一个 SSE 事件，前端按需渲染即可。
十二、小结
一个能干活儿的 AI Agent 并不神秘：它 = 一个会 Function Calling 的模型 + 一份工具注册表 + 一段不断追加的 记忆 + 一个“想—做—看”的循环。本文用不到 80 行 Python 跑通了核心，你接下来可以加上流式输出、并行工具调用、错误重试，甚至把工具换成真实 API。理解了 ReAct，再去读任何 Agent 框架，都会变成“原来它只是把我手写的循环封装了一层”。
