一张图长什么样

LangGraph 的一切都围绕下面这张图展开：

   START
     │
     ▼
 ┌────────┐
 │  node  │──条件边──▶ 分支 A / 分支 B
 └────────┘
     │  ▲
     │  └────（循环）
     ▼
    END

用代码构建一张图，通常分为五步：

1. 定义状态（State）—— 图里流动的数据长什么样
2. 定义节点（Node）—— 每一步做什么
3. 连接边（Edge）—— 谁流向谁，什么条件下走哪条
4. 编译（compile）—— 校验并生成可执行对象
5. 运行 —— invoke / stream 触发执行

State：图里流动的数据

State 是贯穿整张图、所有节点共享的数据结构，用 TypedDict 定义。每个节点接收当前的 State，只返回要更新的那一部分，如何合并由 reducer 决定。

from typing import Annotated, TypedDict
import operator

class State(TypedDict):
    # 直接覆盖：节点返回什么，就替换成什么
    name: str
    # 累加：多个节点对同一个 key 的更新会相加
    total: Annotated[int, operator.add]
    # 追加消息：add_messages 是给消息列表专用的 reducer
    messages: Annotated[list, add_messages]

三种典型 reducer：

写法	含义
不写 Annotated	直接覆盖
Annotated[int, operator.add]	对整数累加
Annotated[list, add_messages]	消息追加（最常用）

Node：图里的一个步骤

节点就是一个 Python 函数，接收 State，返回一个部分更新的 dict。

def my_node(state: State) -> dict:
    # 读状态、做点事（比如调 LLM）
    result = do_something(state["messages"])
    # 返回对状态的更新
    return {"messages": [result]}

Edge：节点之间的流转

边决定「执行完这个节点，下一步去哪」，分两种：

普通边（固定流转）

builder.add_edge(node_a, node_b)   # A 执行完 → B
builder.add_edge(START, node_a)    # 从起点进入 A
builder.add_edge(node_b, END)      # B 执行完结束

条件边（动态路由）

根据节点的运行结果决定下一步，核心是一个「路由函数」：

def router(state: State) -> str:
    last = state["messages"][-1]
    # 如果最后一条消息带有工具调用，就进入 tools 分支
    if getattr(last, "tool_calls", None):
        return "tools"
    # 否则结束
    return END

builder.add_conditional_edges(
    "chatbot",                       # 哪个节点之后做路由
    router,                          # 路由函数
    {"tools": "tools", END: END}     # 返回值 → 目标节点的映射
)

条件边 + 环，正是 Agent 循环（模型 → 工具 → 模型 → …）的实现基础。

START 与 END

• START：图的入口，每个图必须有一条从 START 出发的边
• END：图的出口，执行到这里就结束

一个最小可运行的例子

下面的图没有任何 LLM，只是两个普通节点，用来演示整体结构：

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    count: int

def increment(state: State) -> dict:
    return {"count": state["count"] + 1}

def double(state: State) -> dict:
    return {"count": state["count"] * 2}

builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_node("double", double)

builder.add_edge(START, "increment")
builder.add_edge("increment", "double")
builder.add_edge("double", END)

graph = builder.compile()

result = graph.invoke({"count": 1})
print(result)  # {'count': 4}

编译与运行

• builder.compile()：把「构建器」变成可执行的「编译图」，会校验节点的重复定义、边的合法性等
• graph.invoke(input)：一次性跑到结束，返回最终状态
• graph.stream(input)：按节点增量产出，适合长流程的实时展示
• graph.astream(input)：异步版本

小结

LangGraph 的抽象其实很精简：用 State 承载数据，用 Node 执行逻辑，用 Edge（含条件边）编排流转。理解了这三个概念加上 START/END，就掌握了它 80% 的用法。剩下的持久化、human-in-the-loop、流式等都属于在编译和运行阶段叠加的增强能力。

贡献者: Ra1n

上一页LangGraph

下一页用 LangGraph 构建一个 Agent