从零手写一个 ReAct Agent：循环、观察与决策

为什么要手写

上一篇文章拆开了 Tool Calling 的机制，但"知道原理"和"能用原理做出东西"之间， 还隔着一段距离。最快的跨越方式，是自己不用框架写一遍。

这篇的目标很明确：只用 openai 这个库和一个 while 循环， 做出一个能查天气、能算数的 Agent。写完再回头看 LangChain，你会发现它做的一切都变得透明了。

ReAct 是什么

ReAct 是 Reasoning + Acting 的缩写，核心思想是让模型在"思考"和"行动"之间交替：

思考 → 行动 → 观察 → 思考 → 行动 → 观察 → ... → 最终回答

翻译成代码就是：把工具执行结果塞回上下文，再问一次模型， 直到模型说"我不需要工具了，这就是答案"。

第一步：定义工具

先用最朴素的方式定义两个工具——普通函数 + 一个手写的描述表：

import json
        import random

        def get_weather(city: str) -> str:
            # 真实项目里这里应该请求天气 API，先用假数据演示
            data = {"北京": ("晴", 18), "上海": ("多云", 22), "广州": ("小雨", 26)}
            desc, temp = data.get(city, ("未知", 20))
            return json.dumps({"city": city, "desc": desc, "temp": temp}, ensure_ascii=False)


        def calculate(expression: str) -> str:
            # 演示用，生产环境绝不要 eval 用户输入
            allowed = set("0123456789+-*/(). ")
            if not set(expression) <= allowed:
                return json.dumps({"error": "表达式包含非法字符"})
            try:
                return json.dumps({"result": eval(expression)})
            except Exception as e:
                return json.dumps({"error": str(e)})


        # 工具注册表：名字 -> (函数, 描述, 参数结构)
        TOOLS = {
            "get_weather": (
                get_weather,
                "查询指定城市的当前天气。当用户询问天气时使用。",
                {"type": "object",
                 "properties": {"city": {"type": "string", "description": "城市名，如 北京"}},
                 "required": ["city"]},
            ),
            "calculate": (
                calculate,
                "计算一个数学表达式。当用户需要精确计算时使用。",
                {"type": "object",
                 "properties": {"expression": {"type": "string", "description": "如 (18+22)/2"}},
                 "required": ["expression"]},
            ),
        }
        

注意这里的描述文本。它马上就要变成提示词的一部分—— 写得越清楚，模型选得越准。

第二步：把工具转成 API 需要的格式

def build_tool_schema():
            return [
                {
                    "type": "function",
                    "function": {
                        "name": name,
                        "description": desc,
                        "parameters": params,
                    },
                }
                for name, (_, desc, params) in TOOLS.items()
            ]
        

第三步：核心循环

这就是整个 Agent 的心脏，二十几行：

from openai import OpenAI

        client = OpenAI(
            api_key="sk-xxx",
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        )
        MODEL = "qwen-plus"


        def run_agent(user_input: str, max_steps: int = 6) -> str:
            messages = [
                {"role": "system", "content": "你是一个助手，需要时请调用工具。"},
                {"role": "user", "content": user_input},
            ]

            for step in range(max_steps):
                resp = client.chat.completions.create(
                    model=MODEL,
                    messages=messages,
                    tools=build_tool_schema(),
                )
                msg = resp.choices[0].message

                # 情况 A：模型不需要工具，直接给答案 —— 结束
                if not msg.tool_calls:
                    return msg.content

                # 情况 B：模型要调用工具
                # 先把模型这条"我要调用工具"的消息记进历史
                messages.append({
                    "role": "assistant",
                    "content": msg.content,
                    "tool_calls": [
                        {
                            "id": tc.id,
                            "type": "function",
                            "function": {
                                "name": tc.function.name,
                                "arguments": tc.function.arguments,
                            },
                        }
                        for tc in msg.tool_calls
                    ],
                })

                # 逐个执行，把结果回填
                for tc in msg.tool_calls:
                    name = tc.function.name
                    try:
                        args = json.loads(tc.function.arguments)
                    except json.JSONDecodeError:
                        result = json.dumps({"error": "参数不是合法 JSON"})
                    else:
                        fn = TOOLS.get(name, (None,))[0]
                        result = fn(**args) if fn else json.dumps({"error": f"未知工具 {name}"})

                    print(f"  [步骤 {step + 1}] 调用 {name}({tc.function.arguments})")
                    print(f"            → {result}")

                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": result,
                    })

            return "（达到最大步数，未能得出结论）"
        

第四步：跑起来

if __name__ == "__main__":
            print(run_agent("北京和上海哪个更热？差多少度？"))
        

输出大概是这样：

  [步骤 1] 调用 get_weather({"city": "北京"})
                    → {"city": "北京", "desc": "晴", "temp": 18}
          [步骤 1] 调用 get_weather({"city": "上海"})
                    → {"city": "上海", "desc": "多云", "temp": 22}
          [步骤 2] 调用 calculate({"expression": "22-18"})
                    → {"result": 4}

        上海更热，气温 22℃，比北京的 18℃ 高 4 度。
        

注意步骤 1 里模型一次返回了两个工具调用—— 这叫并行工具调用，模型可以在一轮里同时要求调用多个工具。 我的循环用 for tc in msg.tool_calls 正确处理了这种情况。

踩到的三个坑

坑一：忘了把 assistant 消息存进历史

第一次写的时候，我只把 role="tool" 的结果追加进 messages， 忘了先把模型那条 tool_calls 消息存进去。结果 API 直接报错：

Error: messages with role 'tool' must be a response to a preceding message with 'tool_calls'

道理很简单：工具结果是对某条消息的回复，那条消息必须在历史里， 而且 tool_call_id 要能对上。这个约束也解释了为什么框架要维护那么复杂的消息状态。

坑二：没有最大步数限制

如果模型陷入循环（一直调用同一个工具），没有 max_steps 就是无限烧钱。 我第一版忘了加，测试时看着 token 账单上涨才反应过来。 生产环境的 Agent 一定要有步数上限和超时。

坑三：参数 JSON 解析失败

模型偶尔会吐出不合法的 JSON（比如少了引号、多了逗号）。 所以 json.loads 必须包在 try 里，失败了要回填一条错误信息让模型自己修正， 而不是让程序崩掉。

现在再看 LangChain

手写完之后，再读 LangChain 的 create_agent，一切都很清楚了：

框架没有魔法，它只是把上面这些繁琐但必要的部分工程化了。 知道每一行在做什么之后，出问题时你就知道该去哪里查。

先手写，再用框架。这个顺序能省掉后面几个月的"不知道为什么不好使"。

小结

• ReAct = 循环「问模型 → 执行工具 → 回填结果」直到模型给出答案
• 关键约束：tool 消息必须对应之前的 tool_calls
• 必须有步数上限，否则会无限循环烧钱
• 模型可能并行请求多个工具，要能处理
• 框架封装的是工程细节，不是黑魔法