LangChain 与 LangGraph：一个管搭积木，一个管怎么跑

2026-11-11 · #Agent#LLM#框架

前面几篇讲 Agent 时我一直不绑定框架，但真动手做项目，绕不开“要不要用 LangChain”这个问题。更让人困惑的是它还有个 LangGraph，两者到底什么关系？这篇就讲这个。代码基于 2025 年 10 月 1.0 之后的版本¹（写这篇时 langchain 1.4、langgraph 1.2），1.0 之前的 LLMChain、AgentExecutor 那一套不再展开。

两层东西

一句话：LangChain 管“用什么搭”，LangGraph 管“怎么跑”。

• LangChain：统一的模型接口、消息、工具、结构化输出，加一个现成的 create_agent；
• LangGraph：底层的编排框架和运行时，把流程画成图来执行，并把持久化、中断、流式做成一等公民²。

关键在于：create_agent 返回的就是一张 LangGraph 图。 它天然带着 LangGraph 的能力；不够用时不用换框架，往下走一层自己画图就行。

这个结构是演化出来的。早期 LangChain 擅长把提示词、检索、模型“串成链”，但 Agent 是个循环，AgentExecutor 把循环藏进黑盒，想加审批、换停止条件都很难。2024 年的 LangGraph 干脆让开发者把循环画出来³。1.0 把事情理顺了：老代码挪进 langchain-classic⁴，Agent 统一到 create_agent（取代 create_react_agent），新增中间件，LangGraph 承诺 2.0 前不再有破坏性改动²。

LangChain：积木和 create_agent

from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain.agents import create_agent

model = init_chat_model("anthropic:claude-sonnet-4-6")   # 换厂商只换字符串

@tool
def get_weather(city: str) -> str:
    """查询某个城市当前的天气，返回天气状况、气温和降水概率。"""
    return f"{city}：小雨，18°C，降水概率 80%"

agent = create_agent(model=model, tools=[get_weather], system_prompt="你是简洁的出行助手。")
result = agent.invoke({"messages": [{"role": "user", "content": "北京出门要带伞吗？"}]})
print(result["messages"][-1].content)

• init_chat_model 统一了各家模型的接口，消息上的 content_blocks 把推理、文本、工具调用整理成统一格式⁴；
• @tool 把函数名、文档字符串、类型标注变成工具定义。它省掉的是 JSON Schema，工具设计 的功夫一点都省不掉；
• create_agent 就是 最小的 Agent 循环：调模型、执行工具、追加结果，直到模型不再请求工具。

要结构化结果，加一个 response_format，结果在 result["structured_response"] 里⁵。模型支持原生结构化输出时用 ProviderStrategy，否则用 ToolStrategy（把 schema 包装成一个工具），直接传 schema 类型会自动选。

中间件

真实项目总要在循环里加东西：压缩上下文、审批危险操作、失败重试、切备用模型。1.0 用中间件在固定位置开放钩子⁶：

常见需求都有现成的：SummarizationMiddleware（压缩）、HumanInTheLoopMiddleware（审批）、ModelCallLimitMiddleware / ToolCallLimitMiddleware（限流）、ModelFallbackMiddleware（降级）、ToolRetryMiddleware（重试）、PIIMiddleware（脱敏）。自己写也很简单，比如把工具异常变成模型能读懂的提示：

from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage

@wrap_tool_call
def errors_as_feedback(request, handler):
    try:
        return handler(request)
    except Exception as e:
        return ToolMessage(content=f"工具失败：{e}。请检查参数后重试。",
                           tool_call_id=request.tool_call["id"])

agent = create_agent(model=model, tools=[...], middleware=[errors_as_feedback])

这就是 Harness 那篇讲的钩子：把确定性的逻辑插进不确定的循环。

人工审批值得单独看一眼，因为它用到了 LangGraph 的持久化⁷：

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

agent = create_agent(
    model=model, tools=[search_orders, issue_refund],
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"issue_refund": True})],
    checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": "ticket-1024"}}
agent.invoke({"messages": [{"role": "user", "content": "订单 A17 退款"}]}, config)
# 停在 issue_refund 之前……审批人同意后，用同一个 thread_id 接着跑：
agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config)

LangGraph：把流程画成图

create_agent 的结构是固定的：一个模型节点、一个工具节点、一个循环。流程有明确分支、要并行、有多个角色、有些步骤根本不需要 LLM 时，就该下沉到 LangGraph 了。用 Anthropic 的话说⁸，这时你要的是“工作流”而不是“Agent”，或者两者的混合。

LangGraph 只有三个概念⁹：State（共享状态）、Node（读状态、返回更新的函数）、Edge（决定下一步）。

import operator
from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    question: str
    category: str
    answer: str
    log: Annotated[list[str], operator.add]   # reducer：追加而不是覆盖

def classify(state: State):
    c = "billing" if "退款" in state["question"] else "tech"
    return {"category": c, "log": [f"分类：{c}"]}

def billing(state: State): return {"answer": "已提交退款申请。", "log": ["退款流程"]}
def tech(state: State):    return {"answer": "请先重启应用。", "log": ["技术支持"]}

def route(state: State) -> Literal["billing", "tech"]:
    return state["category"]

builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("billing", billing)
builder.add_node("tech", tech)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route)
builder.add_edge("billing", END)
builder.add_edge("tech", END)
graph = builder.compile()

节点只返回要改的字段；没标 reducer 的字段默认覆盖，标了的按 reducer 合并。消息列表用的是内置的 add_messages（追加新消息、按 ID 替换旧消息），MessagesState 就是只有这一个字段的状态。

超步：为什么需要 reducer

LangGraph 按 Pregel¹⁰ 的 BSP 模型一轮一轮执行，每轮叫一个超步³：

同一超步里的节点并行执行、读同一份快照；全部结束后用 reducer 按固定顺序合并，再存检查点。所以并行是免费的（多条出边就并行），结果是确定的（没有竞态），每一步都能停（检查点在步与步之间）。并行节点写同一个字段时，没有 reducer 就会丢数据。

运行时才知道要并行几份时，用 Send 做 map-reduce；想在节点里一边更新状态一边决定去向，返回 Command(update=..., goto=...)：

from langgraph.types import Send

def fan_out(state):   # 每篇文档发一个 Send，summaries 字段用 operator.add 汇总
    return [Send("summarize", {"doc": d}) for d in state["docs"]]

builder.add_conditional_edges("load_docs", fan_out, ["summarize"])

这基本就是 子 Agent 那篇说的“拆开分发、并行去做”。

检查点：真正的核心

画图不难，难的是让一个要跑几十分钟、要等人、随时会崩的流程能停、能续。这才是 LangGraph 和“自己写个 while 循环”的本质区别。编译时传 checkpointer，调用时带 thread_id¹¹：

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "user-42"}}   # 同一个 ID 就是同一段会话
graph.invoke(inputs, config)

开发用 InMemorySaver，生产换 SqliteSaver 或 PostgresSaver，接口一样。写入频率用 durability 控制："exit" 只在结束时写，"async" 异步写，"sync" 每步写完再继续。

有了每一步的检查点，就能时间旅行¹²：

history = list(graph.get_state_history(config))            # 从新到旧
before = next(s for s in history if s.next == ("billing",))
graph.invoke(None, before.config)                           # 从这里回放
fork = graph.update_state(before.config, {"category": "tech"})
graph.invoke(None, fork)                                    # 改了状态，分叉出一条新分支

调试 40 步的 Agent 时，不用从头重跑，回到出错前那一步改了再跑就行。

interrupt：停下来等人

from langgraph.types import interrupt, Command

def approve_refund(state):
    ok = interrupt({"question": f"确认退款 {state['amount']} 元？"})
    return {"status": "refunded" if ok == "yes" else "rejected"}

graph.invoke({"amount": 199}, config)          # 停下，result["__interrupt__"] 里是问题
graph.invoke(Command(resume="yes"), config)    # resume 的值就是 interrupt() 的返回值

有一个反直觉的坑：恢复时节点从头重新执行，而不是从 interrupt() 那行接着跑¹³。所以 interrupt() 之前的副作用必须幂等（否则邮件会发两遍），不要用 try/except 包住它，传的内容要能 JSON 序列化。最稳妥的做法是把 interrupt() 放在节点开头。

流式和其他

• 流式¹⁴：stream_mode="updates" 推每个节点的更新，适合做进度；"messages" 推 LLM 的 token，适合打字机效果；"custom" 推自定义进度。
• Functional API¹⁵：不想画图时，用 @entrypoint 和 @task 写普通的 Python 控制流，一样有检查点和中断；完成的 @task 恢复时不会重跑。
• Store¹⁶：检查点按 thread_id 隔离，跨会话的 长期记忆 要用 Store，按命名空间存取（store.put / store.search），节点里通过 runtime.store 访问。

再往上一层：Deep Agents

能干长活的 Agent 需要一整套 Harness：规划、文件系统、子 Agent、上下文压缩。LangChain 把这套打包成了 Deep Agents，官方直接叫它“Agent Harness”¹⁷：

from deepagents import create_deep_agent

agent = create_deep_agent(model=model, tools=[internet_search],
                          system_prompt="你是一名研究员，负责写带引用的调研报告。")

它在 create_agent 之上用中间件预装了文件工具（read_file、write_file、edit_file、grep 等）、派发子 Agent 的 task 工具和上下文压缩，可以对照 Harness 那篇来看。

怎么选

也要知道代价：出问题时要在多层抽象间定位，小版本发得很勤（记得锁版本），简单场景直接用 SDK 反而更可控。LangGraph 的价值主要不在画图，而在运行时：不需要持久化、中断、流式的话，画成图好处不多；一旦需要“停下来等人审批，三天后接着跑”，自己实现的成本远高于学一遍 LangGraph。

小结

1. LangChain 提供积木和 create_agent，LangGraph 是底下的运行时，create_agent 本身就是一张图；
2. 中间件是 create_agent 的扩展点，常见需求都有现成的；
3. LangGraph = 状态 + 节点 + 边，按超步执行，reducer 让并行结果确定；
4. 检查点是核心：会话、断点续跑、时间旅行、人工介入都靠它；interrupt() 恢复时节点会重跑；
5. 从 create_agent 开始，结构不够用再下沉到 LangGraph，长任务看 Deep Agents。

LangChain 回答“Agent 由什么组成”，LangGraph 回答“一个会循环、会并行、会停下等人、会崩溃的流程怎么可靠地跑完”。 后一个问题，才是真正值得交给框架的部分。

参考文献