跳转至

第1章 LangGraph 概览

学习目标

  • 理解 LangGraph 的设计动机与整体架构
  • 厘清 LangGraph 与 LangChain 的关系与边界
  • 掌握"图即工作流"的核心理念,区分链式编排与图式编排
  • 能够安装 LangGraph 并运行一个最小的 Hello World 图
  • 判断何时该用 LangGraph、何时用更轻量的方案

问题导入

当你用 LangChain 写过几个 Agent 后,多半会遇到这样的困境:链(Chain)是线性的,可 Agent 的真实行为却是"想一想→调个工具→再想想→换个工具"的循环过程。为了表达循环、分支、回退,你不得不嵌套一堆 if、回调、状态字典,代码迅速变成一团乱麻。更麻烦的是,你很难把"当前在哪一步、为什么走到这里"讲给同事听。

LangGraph 正是为解决这个问题而生。它把工作流抽象成一张有向图:节点是函数,边是控制流,状态在节点之间显式传递。循环、分支、并行、人在回路,都变成图上的一类边或一类节点。本章将带你建立对 LangGraph 的整体认知,并用 20 行代码跑通第一个图。

核心概念

1. 框架目的与架构

LangGraph 是 LangChain 团队推出的有状态、可循环的 Agent 编排框架。它的核心抽象是 StateGraph:开发者用节点(Node)描述"做什么",用边(Edge)描述"下一步去哪",用一个共享的 State 对象承载"当前数据"。图被编译后成为可执行的 Runnable,可被多次调用、流式消费、持久化与中断恢复。

架构上 LangGraph 自下而上分为三层:

  • 底层运行时:负责图编译、节点调度、状态合并、循环控制。
  • 检查点与持久化层:通过 Checkpointer 将每一步状态写入内存/SQLite/Postgres,支持断点续跑与回放。
  • 高层 API 与集成层:langgraph.prebuilt 提供开箱即用的 create_react_agent、ToolNode 等,并与 LangChain 的 LLM、工具、检索器无缝衔接。

2. 与 LangChain 的关系

LangChain 提供了 LCEL(LangChain Expression Language)用于构建有向无环的链式流水线,擅长"输入→检索→拼 Prompt→调模型→输出"这类线性流程。但当流程需要循环(如 ReAct 的"思考-行动-观察"循环)、多 Agent 协作、人在回路审批时,LCEL 表达力不足。

LangGraph 并非要取代 LangChain,而是其补充:

维度 LangChain (LCEL) LangGraph
拓扑 有向无环链 (DAG) 支持循环的有向图
状态 隐式传递 显式 State 对象,可合并
控制流 顺序为主 条件边、循环、并行
持久化 需自行实现 内置 Checkpointer
适用场景 RAG、简单链 Agent、多 Agent、长流程

二者共享同一套工具与模型抽象:LangChain 的 ChatModel、Tool、Retriever 可直接在 LangGraph 节点中使用。

3. 核心理念:图即工作流

LangGraph 最关键的认知转变是:把工作流画成图,而不是写成嵌套代码。这样做有三个好处:

  • 可读性:图的结构就是控制流,一眼看清"有哪些步骤、怎么跳转"。
  • 可控性:循环和分支是一等公民,不必用 while + break 模拟。
  • 可观测性:每经过一个节点,状态都会被检查点记录,便于调试与回放。

4. 安装与快速开始

pip install -U langgraph langchain-openai

最小化安装只需 langgraph;若要调用 OpenAI 模型,再加 langchain-openai。LangGraph 0.2+ 要求 Python 3.9 及以上。

5. 适用场景

  • ReAct 风格 Agent:需要"思考-行动-观察"循环。
  • 多 Agent 系统:Supervisor 分派任务、专家 Agent 协作。
  • 人在回路:高风险操作前需人工审批。
  • 长运行任务:需要断点续跑、状态回放。
  • 复杂审批流:多级审批、条件回退。

不适合的场景:纯单轮问答、简单 RAG 检索——这些用 LangChain LCEL 或直接调用 API 更轻量。

原理图

flowchart TB
    subgraph 应用层
        APP[业务代码]
    end
    subgraph LangGraph 核心
        COMPILE[图编译器<br/>build_compile]
        RUNTIME[运行时调度器]
        STATE[(共享 State)]
        CHECK[Checkpointer<br/>内存/SQL/Postgres]
    end
    subgraph 节点层
        N1[LLMNode]
        N2[ToolNode]
        N3[AgentNode]
        N4[自定义节点]
    end
    subgraph LangChain 生态
        LLM[ChatModel]
        TOOLS[Tools]
        RETRIEVER[Retriever]
    end

    APP -->|invoke/stream| COMPILE
    COMPILE --> RUNTIME
    RUNTIME -->|读/写| STATE
    RUNTIME -->|每步存档| CHECK
    RUNTIME -->|调度| N1
    RUNTIME --> N2
    RUNTIME --> N3
    RUNTIME --> N4
    N1 --> LLM
    N2 --> TOOLS
    N3 --> LLM
    N3 --> TOOLS
    N4 --> RETRIEVER
    CHECK -.->|断点续跑| RUNTIME

图:LangGraph 架构分层。应用层调用编译后的图,运行时调度节点并读写共享 State,每一步状态由 Checkpointer 持久化,节点内部可复用 LangChain 生态的模型与工具。

示例代码

下面是一个最小的 LangGraph Hello World:构建一个两节点图,展示"定义状态→加节点→加边→编译→运行"的完整闭环。

import os
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

# 1. 定义状态:使用 TypedDict 描述共享数据结构
#    add_messages 是一个 reducer,用于追加而非覆盖消息列表
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 2. 初始化模型
llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
)

# 3. 定义节点:每个节点接收 state,返回要更新的字段
def chatbot(state: State) -> dict:
    """调用 LLM 生成回复,并把回复追加到 messages。"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}  # 仅返回增量,reducer 会自动合并

def echo(state: State) -> dict:
    """简单回显,演示多节点协作。"""
    last = state["messages"][-1].content
    return {"messages": [{"role": "user", "content": f"[已确认] {last}"}]}

# 4. 构建图:加节点、加边
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("echo", echo)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", "echo")
graph_builder.add_edge("echo", END)

# 5. 编译:把 builder 转为可执行的 Runnable
graph = graph_builder.compile()

# 6. 运行
if __name__ == "__main__":
    result = graph.invoke({
        "messages": [{"role": "user", "content": "用一句话介绍 LangGraph。"}],
    })
    # 输出最后一条消息
    print(result["messages"][-1].content)

运行后,chatbot 节点先调用 LLM 生成回复,echo 节点再对回复做一次包装,最终输出形如 [已确认] LangGraph 是一个用于构建有状态、可循环 Agent 工作流的图形化框架。。这就是 LangGraph 的最小闭环:状态 → 节点 → 边 → 编译 → 运行。

企业案例

案例 1:从 LangChain LCEL 迁移到 LangGraph

某金融科技公司原有客服链用 LCEL 实现:检索 → 拼 Prompt → 调模型 → 输出。当业务方要求"如果模型回答引用了不存在的条款,就回退到人工检索并重试"时,LCEL 难以优雅表达回退。团队用 LangGraph 重写:把"检索""生成""校验""回退"设为节点,用条件边根据校验结果决定走"输出"还是回"检索"。迁移后代码从 200 行嵌套回调缩减到 80 行声明式图定义,且每一步都有检查点,便于排查"为什么回退了"。

案例 2:可观测的 ReAct Agent

一家工具厂商用 LangGraph 内置的 create_react_agent 搭建运维 Agent。由于每步状态自动持久化,运维同学可以在 Agent 调用高风险命令(如重启服务)前插入断点,人工确认后继续。此前用裸 Python 实现时,"在哪一步暂停"全靠日志拼接,迁移后直接用 graph.get_state() 拿到完整快照。

常见问题

Q1:LangGraph 必须配合 LangChain 使用吗? 不必。LangGraph 本身只依赖 langgraph 包。节点里你可以用任意 LLM SDK(OpenAI、Anthropic、本地模型)。但配合 LangChain 的 ChatModel 与 Tool 能减少样板代码。

Q2:LangGraph 与 LangChain 的 AgentExecutor 是什么关系? AgentExecutor 是 LangChain 早期的 Agent 运行器,功能固定。LangGraph 的 create_react_agent 是其继任者,把同样的 ReAct 循环用图表达,可定制性更强。官方建议新项目用 LangGraph。

Q3:图编译(compile)这一步有什么用? 编译会校验图的合法性(如是否有死节点、是否可达 END),并生成优化后的调度计划。编译产物是线程安全的 Runnable,可被多次 invoke。

Q4:LangGraph 支持异步吗? 支持。节点可定义为 async def,调用时用 await graph.ainvoke(...)。流式场景用 async for chunk in graph.astream(...)。

Q5:和 CrewAI / AutoGen 相比如何? CrewAI、AutoGen 侧重多 Agent 角色协作与对话;LangGraph 更底层,提供图原语让你自己定义协作拓扑,可控性更高,但样板代码略多。三者可结合:用 LangGraph 做底层调度,上层套角色定义。

面试题

1. 为什么需要 LangGraph?它解决了 LangChain LCEL 的什么不足? 答:LCEL 本质是有向无环链,难以表达循环、条件回退、人在回路。LangGraph 以有向图为核心抽象,把循环与分支提升为一等公民,并内置状态管理与检查点,适合构建复杂 Agent 工作流。

2. LangGraph 中 State 的作用是什么? 答:State 是图在节点间共享的数据容器,由 TypedDict 定义。每个节点只返回要更新的字段,由 reducer 合并到 State。这样节点解耦,且每步状态可被检查点完整记录。

3. add_messages reducer 解决了什么问题? 答:如果不指定 reducer,节点返回的新消息会覆盖整个列表;add_messages 让新消息追加到原列表,符合多轮对话的语义。

4. 图编译(compile)是否必须?它做了什么? 答:必须。StateGraph 是构建器,compile() 才生成可执行的 Runnable。编译过程会校验图结构、绑定 reducer、生成调度计划,并允许注入 Checkpointer。

5. 举一个不该用 LangGraph 的场景。 答:单轮问答或简单 RAG——输入→检索→生成→输出,没有循环与分支,用 LCEL 或直接调用 API 更轻量,引入 LangGraph 反而增加心智负担。

练习题

  1. 在 Hello World 示例基础上,增加第三个节点 summary,用一句话总结 chatbot 与 echo 的输出,并画出对应的图结构。
  2. 把 echo 节点改成条件边:当 LLM 回复超过 50 字时走 summary,否则直接结束。提示:用 add_conditional_edges。
  3. 安装 langgraph 并运行本章示例,使用 graph.get_state() 打印一次调用后的状态,观察 messages 列表的变化。

小结

本章建立了对 LangGraph 的整体认知:它是一个以有向图为核心、支持循环与状态持久化的 Agent 编排框架,与 LangChain 互补而非替代。我们走通了"定义 State → 写节点 → 连边 → 编译 → 运行"的最小闭环。下一章将深入图形工作流,讲解节点、边、条件路由与并行执行的细节,让你能画出任意复杂度的控制流。