核心概念:State / Node / Edge

第一个 Graph:数数

别急着上 LLM,先看最小例子——一个把数字加 1 的图:

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

class State(TypedDict):
    n: int

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

g = StateGraph(State)
g.add_node("add", add_one)
g.add_edge(START, "add")
g.add_edge("add", END)

app = g.compile()
print(app.invoke({"n": 0}))  # {'n': 1}

这段 20 行代码里包含了所有要素:

• State:一个 TypedDict,声明数据 schema
• add_one:节点函数,读 state,返回要更新的字段
• StateGraph(State):用 State 的类型创建图
• add_node:注册节点,第一个参数是节点名(字符串)
• add_edge:连边,START / END 是内置的入口/出口节点
• compile():得到可调用的 app,.invoke() 跑完整条

State:数据流过图的载体

定义方式:3 种

# 方式 1: TypedDict(推荐,最常见)
class State(TypedDict):
    messages: list
    user_id: str
    step: int

# 方式 2: Pydantic(需要验证)
from pydantic import BaseModel
class State(BaseModel):
    messages: list
    user_id: str = ""

# 方式 3: Dataclass
from dataclasses import dataclass, field
@dataclass
class State:
    messages: list = field(default_factory=list)

Reducer:合并策略的关键

默认行为:节点返回 {"n": 5} 就把 state 里的 n 覆盖为 5。但如果要"追加"而不是"覆盖"怎么办?

from typing import Annotated
from operator import add

class State(TypedDict):
    # 不加 Annotated: 返回 {"log": ["x"]} 会覆盖整个 list
    # 加 Annotated[..., add]: 返回 {"log": ["x"]} 会 append
    log: Annotated[list, add]
    count: int          # 无 reducer → 覆盖

def step_a(s): return {"log": ["a done"], "count": 1}
def step_b(s): return {"log": ["b done"], "count": 2}

# 跑完: state == {"log": ["a done", "b done"], "count": 2}
# log 被累加,count 被覆盖

消息专用:add_messages

对话场景最常用的 reducer 是 add_messages。它比普通 add 聪明——支持按 id 更新消息、删除消息:

from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage

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

# 节点 A 返回一条新消息 → append
def node_a(s): return {"messages": [AIMessage(content="hi", id="m1")]}

# 节点 B 返回同 id 消息 → 替换(而不是追加)
def node_b(s): return {"messages": [AIMessage(content="fixed", id="m1")]}

Node:干活的函数

函数签名

# 节点本质是: (State) -> dict(要更新的字段)
def my_node(state: State) -> dict:
    # 1. 读 state
    q = state["messages"][-1].content

    # 2. 干事(调 LLM、查库、调 API...)
    reply = llm.invoke(q)

    # 3. 返回"增量更新"
    return {"messages": [reply]}
    # 不要返回完整 state,返回你想改的字段就行

节点也可以是对象 / 子图

# Runnable 对象(LangChain 的 Chain 也是 Runnable)
g.add_node("rewrite", rewrite_prompt | llm)

# Tool 节点(见 Ch4)
from langgraph.prebuilt import ToolNode
g.add_node("tools", ToolNode([search, calc]))

# 子图(见 Ch8)
g.add_node("sub", subgraph.compile())

异步节点

节点可以是 async def。跑的时候用 app.ainvoke() 或 app.astream():

async def fetch(state):
    data = await http.get(state["url"])
    return {"data": data}

result = await app.ainvoke({"url": "..."})

Edge:谁连谁

静态边

g.add_edge("a", "b")    # A 跑完必去 B
g.add_edge(START, "a")   # 图入口进 A
g.add_edge("z", END)     # Z 跑完结束

条件边(重点,见 Ch3 详解)

def router(state) -> str:
    if state["intent"] == "refund":
        return "refund_flow"
    elif state["intent"] == "query":
        return "query_flow"
    return END

g.add_conditional_edges("classify", router)

并行边(fan-out / fan-in)

# 从一个节点分裂到多个
g.add_edge("start", "search_web")
g.add_edge("start", "search_db")
# 两个都跑完 → 一起进 merge
g.add_edge("search_web", "merge")
g.add_edge("search_db", "merge")

LangGraph 自动并行执行同级节点,然后在汇合节点等齐。

Compile 与调用方式

app = g.compile(
    checkpointer=None,   # Ch5 讲
    interrupt_before=[],  # Ch6 讲
)

# 1. 同步一把梭
result = app.invoke({"n": 0})

# 2. 流式,逐节点看
for chunk in app.stream({"n": 0}):
    print(chunk)   # {"add": {"n": 1}}

# 3. 异步
result = await app.ainvoke({"n": 0})

可视化:一行画图

# 生成 Mermaid 代码
print(app.get_graph().draw_mermaid())

# 生成 PNG (要装 pygraphviz)
app.get_graph().draw_png("graph.png")

一个带 LLM 的最小例子

from langchain_openai import ChatOpenAI
from langgraph.graph.message import add_messages

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

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

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

g = StateGraph(State)
g.add_node("chat", chatbot)
g.add_edge(START, "chat")
g.add_edge("chat", END)
app = g.compile()

result = app.invoke({"messages": [{"role": "user", "content": "你是谁?"}]})
print(result["messages"][-1].content)

常见坑

坑	症状	解法
返回整个 state	覆盖了别的节点的更新	只返回要改的字段,让 reducer 合并
列表字段忘加 reducer	每个节点都覆盖 list	Annotated[list, add] 或 add_messages
节点返回 None	报错 "Invalid update"	不想更新就 return {}
节点名重复	报错 Node exists	节点名全局唯一
忘连 START	图跑不起来	至少要 add_edge(START, "xx")

本章小结