02 · LangGraph 入门

🎯 难度：中级⏱ 预计：2 小时# StateGraph# 状态# 路由

把手写 Agent 的循环展开成图：START → model → tools → model → END。模型返回工具请求走 tools，普通文本走 END。

先认识三个概念

• State：流程共享的数据，类似一个带约束的 store。
• Node：读取 state、执行操作、返回更新的函数。
• Edge：规定接下来运行哪个节点，可以是条件路由。

MessagesState 使用消息 reducer 合并更新，因此节点只需要返回新增消息，不必返回整个历史。把整段历史再次返回可能造成重复。

准备

单独创建项目，复制 config.py，创建同样的 .env：

uv init graph-lab --python 3.12
cd graph-lab
uv add "langgraph>=1,<2" "langchain-openai>=1,<2" "python-dotenv>=1,<2"

这个项目的依赖组合与基础课不同，不要将基础课 openai<2 强行同时锁进来。由 langchain-openai 解析其兼容的 SDK 版本，并提交自己的 uv.lock。

完整的图式 Agent

from typing import Literal
from langchain_core.messages import SystemMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from config import required, model_name

@tool
def add(a: int, b: int) -> int:
    """将两个整数相加。"""
    return a + b

llm = ChatOpenAI(
    model=model_name(), api_key=required("LLM_API_KEY"),
    base_url=required("LLM_BASE_URL"), timeout=30, max_retries=1,
).bind_tools([add])


def model_node(state: MessagesState):
    reply = llm.invoke([
        SystemMessage(content="你是计算助手，计算时使用 add 工具。"),
        *state["messages"],
    ])
    return {"messages": [reply]}


def tools_node(state: MessagesState):
    outputs = []
    for call in state["messages"][-1].tool_calls:
        if call["name"] != "add":
            result = "未知工具"
        else:
            try:
                result = str(add.invoke(call["args"]))
            except (ValueError, TypeError):
                result = "参数无效"
        outputs.append(ToolMessage(content=result, tool_call_id=call["id"]))
    return {"messages": outputs}


def route(state: MessagesState) -> Literal["tools", "__end__"]:
    return "tools" if state["messages"][-1].tool_calls else END

builder = StateGraph(MessagesState)
builder.add_node("model", model_node)
builder.add_node("tools", tools_node)
builder.add_edge(START, "model")
builder.add_conditional_edges("model", route)
builder.add_edge("tools", "model")
agent = builder.compile()

if __name__ == "__main__":
    result = agent.invoke(
        {"messages": [("user", "计算 123 加 456")]},
        config={"recursion_limit": 12},
    )
    print(result["messages"][-1].content)

运行 uv run graph_agent.py，应得到 579。recursion_limit 是图执行步数限制，不是 Token 或费用上限；触达时会抛错，应用要显式处理。

与手写循环逐项对照

手写运行器	LangGraph
messages list	MessagesState
调模型代码	model_node
dispatch + 追加工具结果	tools_node
if tool_calls	条件路由 route
for 循环回到开始	tools → model 的边

正式项目也可使用框架提供的 ToolNode 和 tools_condition；本课手动写出来，便于理解和审计控制点。

加入检查点

把 agent = builder.compile() 替换为：

from langgraph.checkpoint.memory import InMemorySaver
agent = builder.compile(checkpointer=InMemorySaver())

调用时带同一个 thread_id：

config = {"configurable": {"thread_id": "local-demo"}, "recursion_limit": 12}
agent.invoke({"messages": [("user", "计算 1 加 2")]}, config)
result = agent.invoke({"messages": [("user", "刚才结果是多少？")]}, config)

InMemorySaver 仅存在当前进程中，重启即丢失。它不是跨用户授权机制。上线时选择受支持的持久化 checkpointer，并在服务端绑定用户与 thread_id。

练习与验收

增加 multiply 工具，问「先算 3 加 4，再乘以 5」。打印最终 messages，检查两个工具调用和返回是否都有对应 ID。

进阶探索：在写文件节点前增加人工确认中断；阅读官方持久化与 interrupt 文档，不要只添加一句 Prompt 来模拟批准。