从零开始掌握 LangGraph:构建下一代 AI Agent 的终极指南

摘要:当 LangChain 的线性链(Chain)无法满足复杂的业务逻辑时,我们需要更强大的工具。本文将带你深入理解 LangGraph 的核心概念,通过实战代码掌握状态管理、循环逻辑与人机协作,开启构建复杂 AI Agent 的大门。


1. 为什么我们需要 LangGraph?

在 LLM 应用开发的早期,LangChain 凭借其 Chain(链)的概念迅速流行。它非常适合处理线性的任务流程:输入 -> 提示词 -> 模型 -> 输出

然而,现实世界的应用场景往往不是线性的。想象一下你要构建一个自主 Agent

  • 它需要思考(Thinking)
  • 它需要调用工具(Tool Use)
  • 如果工具报错,它需要重试(Retry Loop)
  • 在关键操作前,它需要人类确认(Human-in-the-loop)

这种包含循环(Cycles)条件分支持久化状态的逻辑,用传统的线性 Chain 很难优雅地实现。

这就是 LangGraph 登场的时刻。它是 LangChain 生态的一部分,专门用于构建有状态的(Stateful)多角色(Multi-actor) 的应用程序。简单来说,它把 LLM 应用从“流水线”变成了“流程图”。


2. LangGraph 的核心概念

理解 LangGraph,只需要掌握三个核心要素。你可以把它想象成在画一张流程图:

2.1 状态(State)

这是 LangGraph 的灵魂。整个图共享一个状态对象(通常是一个 TypedDict 或 Pydantic 模型)。每个节点都可以读取和更新这个状态。

  • 比喻:就像游戏中的“存档”,记录着当前的对话历史、变量值和执行进度。

2.2 节点(Nodes)

节点是执行具体工作的 Python 函数。

  • 它可以是调用 LLM。
  • 它可以是执行一段代码。
  • 它可以是调用外部 API。
  • 关键点:节点接收当前状态,返回更新后的状态。

2.3 边(Edges)

边决定了流程的走向。

  • 普通边:从节点 A 无条件走到节点 B。
  • 条件边:根据节点 A 的输出结果,决定走到节点 B 还是节点 C(例如:如果意图是“查询天气”走 A,否则走 B)。

3. 快速上手:Hello LangGraph

让我们通过一个简单的代码示例来感受 LangGraph 的工作流。我们将构建一个简单的“加法器 Agent",它可以连续接收数字并累加。

3.1 环境准备

pip install langgraph langchain-core

3.2 定义状态与节点

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

# 1. 定义状态
class AgentState(TypedDict):
    numbers: Annotated[List[int], operator.add] # 使用 operator.add 实现列表自动合并
    total: int

# 2. 定义节点函数
def input_number(state: AgentState):
    # 模拟用户输入一个数字
    new_num = 5 
    print(f"输入数字:{new_num}")
    return {"numbers": [new_num]}

def calculate_sum(state: AgentState):
    current_sum = sum(state['numbers'])
    print(f"当前总和:{current_sum}")
    return {"total": current_sum}

# 3. 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("input", input_number)
workflow.add_node("calculate", calculate_sum)

# 添加边 (设置入口和流程)
workflow.set_entry_point("input")
workflow.add_edge("input", "calculate")
workflow.add_edge("calculate", END)

# 编译
app = workflow.compile()

3.3 运行

initial_state = {"numbers": [], "total": 0}
result = app.invoke(initial_state)
print(result)
# 输出:{'numbers': [5], 'total': 5}

虽然这个例子很简单,但它展示了 State 如何在节点间传递,以及 Graph 如何编排逻辑。


4. 进阶特性:LangGraph 的杀手锏

真正让 LangGraph 强大的是以下三个高级特性,它们解决了生产级 AI 应用的核心痛点。

4.1 循环与 Agent 逻辑

在 LangGraph 中,你可以轻松创建循环。这是构建 ReAct Agent(推理 - 行动)的基础。

  • 逻辑:LLM 思考 -> 决定调用工具 -> 执行工具 -> 返回结果给 LLM -> LLM 再次思考
  • 实现:通过条件边(Conditional Edges),如果 LLM 认为任务未完成,将流程指回 LLM 节点。

4.2 持久化与时间旅行(Persistence & Time Travel)

LangGraph 内置了 Checkpointer

  • 断点续传:如果程序崩溃,可以从上一步的状态继续执行,无需重新计算。
  • 人机协作:在敏感操作前暂停流程,等待人类批准。
  • 时间旅行:你可以查看历史状态快照,甚至回滚到之前的某个步骤重新修改参数运行。这对于调试 Agent 极其有用。
from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# 运行时传入 thread_id 即可保存状态
config = {"configurable": {"thread_id": "user_123"}}
app.invoke(input_state, config)

4.3 多 Agent 协作(Multi-Agent)

你可以将不同的 LangGraph 图嵌套在一起。例如:

  • 主管节点:负责分发任务。
  • 子图 A:负责写代码。
  • 子图 B:负责测试代码。
  • 子图 C:负责写文档。

这种模块化设计让复杂的系统变得可维护。


5. LangChain vs LangGraph:我该选哪个?

特性 LangChain (Chains) LangGraph
控制流 线性为主 (DAG) 支持循环 (Cycles)
状态管理 隐式,传递较难 显式,全局共享 State
持久化 需自行实现 内置 Checkpointer
人机交互 较复杂 原生支持中断/恢复
适用场景 简单问答、RAG 流水线 复杂 Agent、工作流自动化

建议:如果你的应用只是简单的“检索 - 生成”,LangChain 足够且更轻量。如果你需要构建能自主规划、多步执行、需要人工干预的 Agent,LangGraph 是必选项


6. 学习建议与避坑指南

在深入学习 LangGraph 的过程中,我总结了几条经验:

  1. 状态设计是关键:不要把所有数据都塞进 State。只存放需要在节点间共享的、持久化的数据。临时变量请在节点函数内部处理。
  2. 善用 LangSmith:LangGraph 的执行轨迹非常复杂,配合 LangSmith 的可视化追踪(Tracing),你能清晰地看到每一步的状态变化,调试效率提升 10 倍。
  3. 注意类型安全:尽量使用 Pydantic 模型来定义 State,这样 IDE 能提供自动补全,减少字段拼写错误。
  4. 避免无限循环:在编写条件边逻辑时,务必设置最大迭代次数(recursion_limit),防止 Agent 陷入死循环消耗 Token。
Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐