5. 用 LangGraph 构建 Agent：把 ReAct 画成图

这是什么：LangGraph = 把 Agent 画成一张「状态图」的框架：节点负责干活，边决定走向，状态在节点之间传。

为什么用：第 4 章的 ReAct 循环，手写出来是 for 循环里套 if。当 Agent 变复杂——要分支、要并行、要暂停等人确认、要能存盘续跑——这种写法就吃力了。

如何用：本章从最小图开始，逐步加节点、加条件边、加持久化，最后用 create_agent 一行搞定标准 ReAct。

5.1 换个视角：Agent 是一张状态机

        ┌──────────────┐
        │              │◄────────────┐
        ▼              │             │
   ┌─────────┐   有工具调用?  ┌──────────┐
   │  agent  │ ──是──►  │   tools   │──观察写回──┘
   │  节点    │ ◄─────────  │   节点    │
   └─────────┘   无调用?     └──────────┘
        │
        │ 结束
        ▼
      END

节点（node）：一个干活的函数，接收状态、返回状态更新
边（edge）：决定下一步去哪个节点
条件边：根据状态决定走哪条边（上面那个「有工具调用?」就是）
状态（state）：节点之间传递的数据

这就是 ReAct 循环的图形版：「思考」是 agent 节点，「行动」是 tools 节点，「观察」写回状态那条边。

5.2 安装

pip install -U langgraph langchain langchain-openai

版本说明：LangGraph 1.x（当前 1.2.x）+ LangChain 1.x（当前 1.4.x）要求 Python ≥ 3.10，站点基线 3.15 可用。 ⚠️ v1 的重要变化：langgraph.prebuilt.create_react_agent 已废弃，改用 LangChain 的 create_agent（见 5.7）。网上很多老教程还在用旧写法，注意区分。

5.3 第一个图：三行代码跑通

import os
from typing import Annotated, TypedDict

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph, add_messages

model = init_chat_model("gpt-4o-mini", api_key=os.environ["LLM_API_KEY"])


class State(TypedDict):
    """图里流通的数据：messages 是一条消息列表"""
    messages: Annotated[list, add_messages]


def agent_node(state: State):
    """节点：把消息喂给模型，返回模型的新消息"""
    resp = model.invoke(state["messages"])
    return {"messages": [resp]}


builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)

graph = builder.compile()

result = graph.invoke({"messages": [HumanMessage("用一句话介绍什么是大语言模型")]})
print(result["messages"][-1].content)

你会看到（示意）：

大语言模型是一种通过海量文本训练、能理解和生成人类语言的模型。

5.4 状态与 reducer

节点返回的只是状态的一部分，LangGraph 负责合并。合并规则由 reducer 决定：

写法	效果
`messages: list`	新值覆盖旧值（只剩最后一条）
`messages: Annotated[list, add_messages]`	新消息追加到旧消息后面，还能自动给消息分配 id

from langgraph.graph import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 追加，而不是覆盖

Agent 场景基本都要用 add_messages——否则每轮的新消息会把历史覆盖掉，模型立刻失忆。

5.5 条件边：ReAct 的循环就在这里

tools_condition 是 LangGraph 提供的现成条件：检查最后一条消息里有没有工具调用——有就走向 tools 节点，没有就结束。

from langgraph.prebuilt import ToolNode, tools_condition

builder.add_conditional_edges("agent", tools_condition)

自定义条件也很简单，返回下一个节点名即可：

def route(state: State):
    last = state["messages"][-1]
    if last.get("tool_calls"):
        return "tools"
    return "finish"          # 字符串就是下一个节点名

builder.add_conditional_edges("agent", route, {"tools": "tools", "finish": END})

5.6 用图实现一个 ReAct Agent

把第 4 章的 ReAct 完整搬进图里：

import os
from typing import Annotated, TypedDict

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph, add_messages
from langgraph.prebuilt import ToolNode, tools_condition


@tool
def get_weather(city: str) -> str:
    """查询指定城市的当前天气。城市名用中文。"""
    fake = {"北京": "晴，26℃", "上海": "多云，29℃"}
    return fake.get(city, f"没查到 {city} 的天气")


@tool
def get_tip(weather: str) -> str:
    """根据天气描述给出穿衣建议。"""
    if "雨" in weather:
        return "记得带伞"
    return "穿薄外套就行"


model = init_chat_model("gpt-4o-mini", api_key=os.environ["LLM_API_KEY"])


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


def agent_node(state: State):
    system = SystemMessage(content="你是一个助手，需要信息就调用工具，别凭空编。")
    resp = model.invoke([system] + state["messages"])
    return {"messages": [resp]}


builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode([get_weather, get_tip]))

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)   # 有调用 → tools，否则 → END
builder.add_edge("tools", "agent")                        # 工具结果写回状态 → 再思考

graph = builder.compile()

result = graph.invoke({
    "messages": [HumanMessage("北京天气怎么样？需要穿什么？")]
})
print(result["messages"][-1].content)

对照第 4 章的手写版，你会发现：

手写 ReAct	LangGraph 版
`for` 循环	图的边
`if msg.tool_calls`	`tools_condition` 条件边
`messages.append(...)`	状态 + `add_messages` reducer
`max_steps` 兜底	递归上限 `recursion_limit`

循环逻辑是 LangGraph 帮你画的，代码反而更短。

5.7 快捷方式：create_agent

标准 ReAct 图，LangGraph 提供了现成的构造函数，一行顶上面三十行：

from langchain.agents import create_agent

agent = create_agent(
    model,
    tools=[get_weather, get_tip],
    system_prompt="你是一个助手，需要信息就调用工具，别凭空编。",
)

result = agent.invoke({"messages": [HumanMessage("北京天气怎么样？")]})
print(result["messages"][-1].content)

⚠️ 老教程里的 from langgraph.prebuilt import create_react_agent 在 v1 已废弃，参数名也从 prompt= 改成了 system_prompt=。看到旧写法就别抄了。

什么时候自己画图？ 标准 ReAct 用 create_agent 就够；出现下面这些需求时，自己画：

需要多个 Agent 分工（一个查资料、一个写代码、一个审核）
某个节点要并行跑好几个工具
流程里有固定步骤（先检索、再总结、再校验）
需要人在中间某步拍板

5.8 持久化：断点续跑与多轮记忆

给图挂上 checkpointer，它就会自动保存每一步的状态：

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user-001"}}

# 第一次
graph.invoke({"messages": [HumanMessage("我叫小青蛙")]}, config)

# 同一个 thread_id 再来一次 —— 记得住刚才说过的话
graph.invoke({"messages": [HumanMessage("我叫什么？")]}, config)

thread_id 就是「会话」的标识：同一个 id 共享历史，不同 id 互相隔离。

想跨进程存盘，把 MemorySaver 换成官方提供的数据库版 checkpointer 即可（SQLite / Postgres 等）。

5.9 人在回路：让 Agent 停下来问你

危险操作（发邮件、删文件、付款）不该让模型自己拍板。LangGraph 的 interrupt() 可以在任意节点挂起，等人确认后再继续：

from langgraph.types import Command, interrupt


def sensitive_node(state: State):
    decision = interrupt({          # 挂起，等外部输入
        "question": "确认要发送这封邮件吗？",
        "draft": state["messages"][-1].content,
    })
    if decision == "yes":
        return {"messages": []}      # 继续往下走
    return {"messages": []}          # 取消

恢复时用 Command(resume="yes") 把人的答复送回去。这比第 8 章那种「在 Python 里 input() 等确认」优雅得多——因为状态是被存下来的，换个进程、隔一天再回来也能继续。

5.10 常见翻车现场

报错 / 现象	原因	修法
每轮都失忆	状态没用 `add_messages`，新值覆盖了旧值	`Annotated[list, add_messages]`
`RecursionError` / 死循环	图里没有出口	给条件边加终止分支，必要时设 `recursion_limit`
工具不被调用	模型没识别出工具	检查 `@tool` 的 docstring 写没写清
`create_react_agent` 导入报错	v1 已废弃	换 `from langchain.agents import create_agent`
换个进程就没记忆	只用了 `MemorySaver`（在内存里）	换数据库版 checkpointer
`interrupt` 抛异常	恢复时没带 `Command(resume=...)`	用 `graph.invoke(..., Command(resume="yes"))` 恢复

5.11 动手改

把 5.3 的最小图加上第二个节点和一个条件边，做一个「先查资料再回答」的两步流程
给 5.6 的 ReAct 图挂上 MemorySaver，验证同一个 thread_id 能不能记住上一轮
在 tools 节点前插一个 interrupt()，体验「Agent 停下来问你要授权」

5.12 小结

LangGraph 把 Agent 变成状态图：节点干活，边决定走向，状态在节点间传
状态要用 Annotated[list, add_messages]，否则每轮覆盖历史
条件边就是 ReAct 的循环：tools_condition 是现成的判断
标准 ReAct 用 create_agent 一行搞定；复杂流程再自己画图
挂上 checkpointer 就有断点续跑和多轮记忆，thread_id 是会话标识
interrupt() 让流程能停下来等人拍板，比在代码里 input() 强得多

下一章：给 Agent 装上手，工具描述写得好，Agent 才用得对 🐸