3.1 子图 Subgraph:封装与复用

当图变大,你需要把一块逻辑封装成"子图",像搭乐高一样组合。回忆第 1 章:编译后的图是个 Runnable,所以一个图可以直接当节点塞进另一个图。

复制# 子图:一个独立的小图(比如一个专精的检索 Agent)
sub_builder = StateGraph(SubState)
sub_builder.add_node(...)
subgraph = sub_builder.compile()

# 父图:直接把编译好的子图当成一个节点
parent = StateGraph(ParentState)
parent.add_node("retriever_agent", subgraph)   # ← 子图作为节点
parent.add_node("writer", writer_node)

关键:父子图的 State 怎么对接

3.2 Send:并行 map-reduce(动态扇出)

第 1 章讲了 Send 的原理,这里给一个完整可跑的 map-reduce:把一个主题列表,并行生成多个结果,再归并、选最优。

复制import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

class State(TypedDict):
    topics: list
    drafts: Annotated[list, operator.add]   # 累积通道:收集并行结果
    best: str

def write_one(state: dict):           # map:处理单个主题(局部 state)
    return {"drafts": [f"《{state['topic']}》的草稿…"]}

def dispatch(state: State):           # 运行时扇出:有几个主题就并行几路
    return [Send("write_one", {"topic": t}) for t in state["topics"]]

def reduce_best(state: State):         # reduce:所有草稿到齐后选最优
    return {"best": max(state["drafts"], key=len)}

b = StateGraph(State)
b.add_node("write_one", write_one)
b.add_node("reduce_best", reduce_best)
b.add_conditional_edges(START, dispatch, ["write_one"])
b.add_edge("write_one", "reduce_best")   # 所有并行分支汇到 reduce
b.add_edge("reduce_best", END)
graph = b.compile()
graph.invoke({"topics": ["AI", "露营", "做饭"]})

3.3 Command:动态路由 + 跨图跳转

原理见第 1 章。这里给多智能体里最常用的一招:子 Agent 干完,用 Command 直接把控制权交回父图的 supervisor。

复制from langgraph.types import Command
from typing import Literal

def research_agent(state) -> Command[Literal["supervisor"]]:
    result = do_research(state)
    return Command(
        update={"messages": [result]},   # 写回结果
        goto="supervisor",              # 交回主管,由它决定下一步
        graph=Command.PARENT,           # 跳到【父图】的 supervisor(跨图)
    )

对比记忆:条件边适合"纯看状态的简单分支";Command适合"决策与状态更新耦合、或要跨图跳"的场景。第 4 章多智能体案例会大量用到。

3.4 跨会话长期记忆:Store

第 2 章的 Checkpointer 是会话内记忆(同一 thread)。但"记住这个用户上个月说他喜欢露营"是跨会话的长期记忆——这要用 Store。

复制from langgraph.store.memory import InMemoryStore

store = InMemoryStore()   # 生产用 PostgresStore,可带向量检索
graph = builder.compile(checkpointer=memory, store=store)

# 节点里通过注入的 store 读写长期记忆(按 namespace 隔离用户)
def node(state, *, store):
    ns = ("user-42", "prefs")
    store.put(ns, "hobby", {"value": "露营"})    # 写
    items = store.search(ns)                       # 读/检索
    return {...}

3.5 生产级持久化

开发用 MemorySaver(进程内,重启即丢)。上生产要换成落盘/数据库的 checkpointer。

复制# SQLite(单机/轻量)
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("checkpoints.sqlite") as cp:
    graph = builder.compile(checkpointer=cp)

# Postgres(生产/多实例共享)
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string("postgresql://...") as cp:
    cp.setup()                       # 首次建表
    graph = builder.compile(checkpointer=cp)

异步服务用对应的 AsyncSqliteSaver / AsyncPostgresSaver + ainvoke/astream。多实例部署时,Postgres 版让任意实例都能恢复同一会话——这是水平扩展的关键。

3.6 流式细节:不止打字机

第 2 章提了 stream,这里讲生产里真正要用的两个深入点。

① 组合 stream_mode

复制# 同时要"每步状态"和"逐 token",传一个列表
for mode, chunk in graph.stream(inputs, config,
        stream_mode=["updates", "messages"]):
    if mode == "messages":
        token, meta = chunk
        print(token.content, end="")   # 前端打字机
    elif mode == "updates":
        print(chunk)                    # 哪个节点改了啥,用于状态条

② astream_events:细到每个 LLM/工具调用

需要"工具开始执行""LLM 第一个 token"这类细粒度事件(做复杂前端、埋点)时,用 astream_events,它吐出整棵 Runnable 调用树的事件流。

3.7 部署上线

编译后的图就是个 Runnable,部署有三条路,由轻到重:

复制pip install "langgraph-cli[inmem]"
# 项目根放一个 langgraph.json 声明图入口,然后:
langgraph dev        # 本地起服务 + Studio 可视化调试