前面几篇我已经把 Function Calling、ReAct、Workflow 这些概念拆开讲过了，但如果一直停留在概念层，很容易产生一种“好像懂了，但自己写不出来”的错觉。

所以这篇我不再继续抽象讨论，而是直接回到代码：用一个极简的 Python 项目，从零实现一个最小可运行的 ReAct Agent。

完整的代码在文章最后。

这个实现只保留 3 个最核心的部分：

1. 一个模型节点 call_model
2. 一个工具节点 ToolNode
3. 一条 模型 -> 工具 -> 模型 的循环边

它不追求“功能很多”，而是追求“结构清楚”。你看完后，应该能回答这几个问题：

• ReAct 的最小执行闭环到底长什么样？
• tool_calls 是怎么把流程从模型切到工具节点的？
• 为什么说 ReAct 本质上就是一个 思考 -> 行动 -> 观察 -> 再思考 的循环？

一、项目目录

这个最小项目只有 4 个核心文件：

ReAct/
├── graph.py
├── state.py
├── tools.py
└── main.py

它们各自的职责非常明确：

• graph.py：定义 ReAct 图结构
• state.py：定义状态结构
• tools.py：定义工具
• main.py：提供命令行交互入口

二、先看状态：为什么只有一个 messages

在最小 ReAct 实现里，State 可以非常简单：

class InputState(TypedDict):
    messages: list[AnyMessage]


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

这里最关键的是：

• messages 是整个 Agent 的主状态
• 节点不是返回完整状态，而是返回 {"messages": [...]} 这样的局部更新
• add_messages 会把新消息自动合并进已有历史

这也是为什么 ReAct 模板通常不需要一开始就设计很多业务字段。因为在最小实现里，消息历史本身就足够支撑模型循环推理。

三、工具：让模型具备最基本的行动能力

为了让这个 Agent 有最小“行动能力”，我只保留了 3 个工具：这就是tools.py的部分

@tool
def get_current_time() -> str:
    """获取当前本地时间。适合回答现在几点、今天几号这类问题。"""


@tool
def save_note(content: str) -> str:
    """把一条笔记追加保存到本地 notes.md 文件。"""


@tool
def read_note() -> str:
    """读取本地 notes.md 的内容。适合查看、总结或复用已保存的笔记。"""

这 3 个工具分别代表了三类典型能力：

• 查询类：get_current_time
• 写入类：save_note
• 读取类：read_note

也就是说，哪怕是最小 ReAct Agent，也已经具备了：

• 访问外部信息
• 改变外部状态
• 再次读取外部状态

这比“只会聊天”的模型已经前进了一大步。

四、模型节点：ReAct 循环的大脑

graph.py 里的核心函数是 call_model：

def call_model(state: State) -> dict:
    llm = get_llm().bind_tools(TOOLS)
    response = llm.invoke(
        [SystemMessage(content=SYSTEM_PROMPT), *state["messages"]]
    )
    return {"messages": [response]}

这里发生了 3 件关键的事：

1. 加载模型
2. 通过 bind_tools(TOOLS) 给模型绑定工具能力
3. 把系统提示词和历史消息一起发给模型

最重要的一点是：
模型节点不直接执行工具。它只负责输出下一条 AIMessage。

这条 AIMessage 可能有两种情况：

• 没有 tool_calls：说明模型决定直接回答
• 有 tool_calls：说明模型决定先调用工具

这就是 ReAct 里的“Reason”阶段。

五、路由函数：决定下一步去哪

接下来是这个很小但很关键的函数：

def route_model_output(state: State) -> str:
    last_message = state["messages"][-1]
    if not getattr(last_message, "tool_calls", None):
        return END
    return "tools"

它的作用可以压缩成一句话：

检查模型刚输出的最后一条消息里有没有 tool_calls，如果有，就去工具节点；如果没有，就结束。

这就是 ReAct 里的“Act or Finish”分叉点。

六、图结构：最小 ReAct loop 是怎么连起来的

整个图的骨架非常简单：

builder = StateGraph(State, input_schema=InputState)

builder.add_node("call_model", call_model)
builder.add_node("tools", ToolNode(TOOLS))

builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", route_model_output)
builder.add_edge("tools", "call_model")

把它翻译成流程图，就是：

START
  -> call_model
      -> 如果没有 tool_calls，结束
      -> 如果有 tool_calls，进入 tools
  -> tools
  -> call_model
  -> ...

这就是一个最小可运行的 ReAct Loop。

注意这里的重点不是“循环本身”，而是：

• 模型负责决定是否行动
• 工具负责执行行动
• 工具结果回到消息历史
• 模型基于新观察继续推理

这正是 Thought -> Action -> Observation -> Thought 的代码化形式。

七、命令行入口：把循环真正跑起来

最后是 main.py：

def main():
    graph = build_graph()
    messages = []

    while True:
        user_input = input("请输入你的问题（输入 exit 退出）：").strip()

        if user_input.lower() == "exit":
            print("已退出。")
            break

        if not user_input:
            print("输入不能为空。")
            continue

        messages.append(HumanMessage(content=user_input))

        result = graph.invoke({"messages": messages})
        messages = result["messages"]

        final_message = messages[-1]
        print("\nAgent 回复：")
        print(final_message.content)

这里做的事也很直接：

1. 读取用户输入
2. 包装成 HumanMessage
3. 调用 graph.invoke(...)
4. 用返回的新 messages 覆盖本地消息历史
5. 输出最后一条模型回复

这就让整个 ReAct 图从“结构定义”变成了“可交互程序”。

八、这个最小 ReAct 实现的局限

当然，这份代码只是教学版最小实现，它还有很多没有覆盖的工程能力。

比如：

• 没有 Checkpoint，进程一断就丢状态
• 没有 Interrupt，不能人工中断恢复
• 没有独立业务字段，全部状态都靠 messages
• 没有完整错误处理和工具失败恢复
• 没有 observability / middleware / 权限控制

也就是说，这份实现回答的是：

ReAct 最小闭环怎么跑起来？

而不是：

 Agent 系统该怎么设计？

这两者是不同层级的问题。

九、完整代码

tools.py:

from datetime import datetime
from pathlib import Path

from langchain_core.tools import tool

NOTES_PATH = Path(__file__).with_name("notes.md")


@tool
def get_current_time() -> str:
    """获取当前本地时间。适合回答现在几点、今天几号这类问题。"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


@tool
def save_note(content: str) -> str:
    """把一条笔记追加保存到本地 notes.md 文件。"""
    with NOTES_PATH.open("a", encoding="utf-8") as file:
        file.write(content.strip() + "\n")
    return f"笔记已保存到 {NOTES_PATH.name}"


@tool
def read_note() -> str:
    """读取本地 notes.md 的内容。适合查看、总结或复用已保存的笔记。"""
    if not NOTES_PATH.exists():
        return "还没有笔记文件。"

    content = NOTES_PATH.read_text(encoding="utf-8").strip()
    return content or "笔记文件为空。"

state.py:

from typing import Annotated
from typing_extensions import TypedDict

from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages


class InputState(TypedDict):
    messages: list[AnyMessage]


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

graph.py：

import os

from langchain_core.messages import SystemMessage
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from pydantic import SecretStr

from state import InputState, State
from tools import get_current_time, read_note, save_note

TOOLS = [get_current_time, save_note, read_note]

SYSTEM_PROMPT = """
你是一个最小可用的 ReAct Agent。

你的工作原则：
1. 能直接回答时就直接回答。
2. 需要外部信息或持久化操作时再调用工具。
3. 调用工具后，基于工具结果继续推理并给出最终答复。
4. 回复尽量简洁、明确。
""".strip()


def get_llm() -> ChatOpenAI:
    api_key = os.getenv("DEEPSEEK_API_KEY")
    base_url = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
    model = os.getenv("DEEPSEEK_MODEL", "deepseek-chat")

    if not api_key:
        raise ValueError(
            "缺少环境变量 DEEPSEEK_API_KEY。请先设置 API Key 再运行。"
        )

    return ChatOpenAI(
        model=model,
        api_key=SecretStr(api_key),
        base_url=base_url,
        temperature=0,
    )


def call_model(state: State) -> dict:
    llm = get_llm().bind_tools(TOOLS)
    response = llm.invoke(
        [SystemMessage(content=SYSTEM_PROMPT), *state["messages"]]
    )
    return {"messages": [response]}


def route_model_output(state: State) -> str:
    last_message = state["messages"][-1]
    if not getattr(last_message, "tool_calls", None):
        return END
    return "tools"


def build_graph():
    builder = StateGraph(State, input_schema=InputState)

    builder.add_node("call_model", call_model)
    builder.add_node("tools", ToolNode(TOOLS))

    builder.add_edge(START, "call_model")
    builder.add_conditional_edges("call_model", route_model_output)
    builder.add_edge("tools", "call_model")

    return builder.compile()

tips:这里要设置你自己的api_key

main.py:

from graph import build_graph
from langchain_core.runnables import RunnableConfig

def main():
    graph = build_graph()
    config: RunnableConfig = {
    "configurable": {
        "thread_id": "react-memory-demo-1"
    }
    }
    messages=[]
    while True:
        user_input = input("请输入你的问题（输入 exit 退出）：").strip()

        if user_input.lower() == "exit":
            print("已退出。")
            break

        messages.append(("user", user_input))

        result = graph.invoke(
            {
                "messages": messages
            },
            config=config
        )

        messages = result["messages"]

        final_message = messages[-1]
        print("\nAgent 回复：")
        print(final_message.content)
        print()
if __name__ == "__main__":
    main()

十、总结

这个最小 ReAct 项目最想说明的一点是：

ReAct 并不神秘。它的本质，就是让模型在消息历史基础上不断判断：现在该不该行动、该调用什么工具、拿到结果后还要不要继续下一轮。

放到代码里，它最终只落成了几件事：

• 一个 messages 状态
• 一个模型节点
• 一个工具节点
• 一条 模型 -> 工具 -> 模型 的循环边

如果说之前《ReAct 是什么》讲的是概念层的“推理-行动循环”，那么这篇代码拆解讲的就是：

这个循环在最小实现里到底长什么样。

等把这个最小版本吃透以后，再去看更复杂的 Workflow、Checkpoint、Middleware、State 设计，就不会再觉得那些东西是凭空出现的“高级黑盒”，而会知道它们只是建立在这个最小循环之上的工程扩展。