从零手写 AI Agent：ReAct 智能体实战

大模型只会“说话”，而 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 框架，都会变成“原来它只是把我手写的循环封装了一层”。