用 LangGraph 构建一个 Agent

约 901 字大约 3 分钟

LangGraphLangChainAI Agent

2026-09-01

目标

这篇文章手把手从零构建一个 ReAct Agent：它能理解用户意图，在需要时调用工具（比如查天气），拿到结果后再回答用户。整个过程是一个「模型 → 工具 → 模型 → …」的循环，直到模型认为可以给出最终答案为止。

环境准备

pip install langgraph langchain langchain-openai

设置 API Key（这里以 OpenAI 为例，换成任意 langchain 兼容的 chat model 都成立）：

export OPENAI_API_KEY="你的 key"

第一步：定义 State

Agent 全程围绕「消息列表」运转，我们定义一个只含 messages 字段的状态：

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

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

add_messages 这个 reducer 会智能地追加消息：普通消息直接 append，同一条消息的多次更新则合并，还能处理消息的覆盖与删除。

第二步：准备模型和工具

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

def get_weather(city: str) -> str:
    """查询指定城市的天气"""
    return f"{city}：今天晴，25°C"

tools = [get_weather]

# bind_tools 让模型知道有哪些工具可用，并能在回答中发出 tool_calls
llm_with_tools = llm.bind_tools(tools)

第三步：定义节点

两个节点：

chatbot：把当前消息交给带工具的模型
tools：真正执行模型请求的那些工具

from langgraph.prebuilt import ToolNode

def chatbot(state: State) -> dict:
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

tool_node = ToolNode(tools)

第四步：搭图 + 条件路由

from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_node("tools", tool_node)

builder.add_edge(START, "chatbot")

# 关键：模型有 tool_calls 就进 tools，否则结束
def should_continue(state: State) -> str:
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None):
        return "tools"
    return END

builder.add_conditional_edges(
    "chatbot",
    should_continue,
    {"tools": "tools", END: END},
)

# 工具执行完，回到模型，让模型基于结果继续
builder.add_edge("tools", "chatbot")

graph = builder.compile()

到这里，一个完整的 Agent 环就出来了：

START → chatbot ──有 tool_calls──▶ tools ──▶ chatbot ──▶ … ──无 tool_calls──▶ END
                └────────无 tool_calls────────────────────────────▶ END

第五步：运行

result = graph.invoke(
    {"messages": [("user", "北京今天天气怎么样？")]}
)

for m in result["messages"]:
    m.pretty_print()

执行过程大致是：

模型收到问题，判断需要查天气，发出 tool_calls: ["get_weather"]
进入 tools 节点，执行 get_weather("北京")，把结果作为一条 ToolMessage 追加
回到 chatbot，模型看到工具结果后组织出自然语言回答
模型这次没有 tool_calls，路由到 END，结束

快捷写法：create_react_agent

如果不想手动搭图，LangGraph 提供了一行代码创建 ReAct Agent：

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(llm, tools)

result = agent.invoke(
    {"messages": [("user", "北京今天天气怎么样？")]}
)

内部就是上面那套 StateGraph + 条件边的封装。

进阶能力速览

持久化（Checkpointer）

编译时传入一个 checkpointer，每个 super-step 的状态都会被存档：

from langgraph.checkpoint.memory import MemorySaver

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

# 通过 thread_id 区分会话
graph.invoke({"messages": [...]}, config={"configurable": {"thread_id": "1"}})

带来的好处：会话内多轮连续、中断后恢复、以及对历史 checkpoint 的 time-travel 回溯。

Human-in-the-loop

在节点里调用 interrupt() 可以把图暂停下来，把控制权交还外部，等人工确认后再继续：

from langgraph.types import interrupt

def approve_node(state: State) -> dict:
    value = interrupt("请人工确认是否继续执行？")
    return {"approved": value.lower() == "yes"}

流式输出

for chunk in graph.stream({"messages": [...]}):
    print(chunk)

stream 会按节点产出增量，配合流式模型可实现逐字输出。

小结

手动构建一个 LangGraph Agent 的完整路径是：定义 State → 定义节点 → 用条件边连出「模型 ↔ 工具」的环 → 编译运行。理解了这条路径，再叠加 checkpointer、interrupt、流式等进阶能力，就能应对真实生产中绝大多数的 Agent 编排需求。

Claude Code

OpenAI

Qwen Code

LangGraph

Langfuse