langgraph

LangGraph 有状态 Agent 的图编排与持久化框架。

补充说明

LangGraph 用节点和边描述 Agent 流程，状态在节点间传递，checkpointer 负责持久化和恢复。它适合需要循环、条件分支、人工审批和长任务恢复的场景，而不是简单的线性链。

核心概念：State、Node、Edge、checkpointer、thread_id、interrupt。MessagesState 是内置的消息状态，使用 add_messages 归并消息而不是覆盖。

安装与启动

pip install -U langgraph langchain-openai
pip install -U "langgraph-cli[inmem]"

常用示例

带持久化记忆的最小对话图，同一 thread_id 会累积上下文：

python - <<'PY'
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.memory import InMemorySaver

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

def chat(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

builder = StateGraph(MessagesState)
builder.add_node("chat", chat)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "user-42"}}
print(graph.invoke({"messages": [{"role": "user", "content": "你好"}]}, config)["messages"][-1].content)
print(graph.invoke({"messages": [{"role": "user", "content": "继续"}]}, config)["messages"][-1].content)
PY

流式输出 token，适合 Web 或终端：

python - <<'PY'
for chunk, metadata in graph.stream(
    {"messages": [{"role": "user", "content": "列出三个排查步骤"}]},
    config,
    stream_mode="messages",
):
    if chunk.content:
        print(chunk.content, end="", flush=True)
PY

人工审批节点用 interrupt 暂停，再由 Command(resume=...) 恢复：

python - <<'PY'
from langgraph.types import interrupt, Command

def approval(state: MessagesState):
    decision = interrupt({"question": "是否执行扩容？"})
    return {"messages": [{"role": "assistant", "content": f"审批结果: {decision}"}]}

# 在图中加入 approval 节点后：
# result = graph.invoke({"messages": [...]}, config)
# graph.invoke(Command(resume="yes"), config)
PY

易错点与排错

• 编译时没有传 checkpointer，却给 configurable.thread_id，会报缺少 thread_id 或状态无法恢复；两者必须同时配置。
• 自定义状态字段没有 reducer，后一个节点会覆盖前一个节点的列表；用 Annotated[list, operator.add] 或 add_messages。
• 节点直接修改 state 对象而不是返回增量字典，可能导致状态不更新或难以追踪；保持节点为纯函数式更新。
• 循环没有终止条件会撞 recursion_limit（默认约 25 步），表现为 GraphRecursionError；加条件边或在 config 中显式提高上限。
• InMemorySaver 只存在进程内，重启即丢失；生产要用 Postgres/SQLite checkpointer，并保证多副本共享存储。
• 异步 Web 服务里调用同步 invoke 会阻塞事件循环；改用 ainvoke/astream，并让节点里的模型调用也支持异步。
• 人工审批恢复时传错 thread_id 会恢复另一条会话；审批请求里应携带业务侧的任务 ID 并做校验。
• langgraph dev 是内存态开发服务器，不适合生产；生产部署要用持久化后端并配置鉴权。

读完自测

合上页面，先回答，再展开对照。

1. checkpointer 和 thread_id 的关系是什么？
2. 为什么自定义列表字段会被覆盖？
3. interrupt 恢复时要注意什么？

相关条目

• langchain
• crewai
• autogen
• mcp
• langfuse