快速入门:什么是 LangGraph

面试官:「聊聊 LangGraph 吧。」

你:「它是一个框架,用来构建 LLM 应用的工作流。」

面试官:「和 LangChain 有什么区别?」

你:「LangChain 是线性链条,LangGraph 是有向图,支持循环和中断。」

🤔 这一章,从零开始搞懂 LangGraph。

1.1 什么是 LangGraph

LangGraph 是 LangChain 团队推出的有状态、多步骤工作流框架。如果说 LangChain 是「一条流水线」,那 LangGraph 就是「一张导航图」。

图 1-1:LangChain vs LangGraph 架构对比
LangChain:线性链式 Prompt LLM Output 单向执行,无法循环 LangGraph:有向图 START LLM Tool END 条件分支
面试要点

LangGraph 本质上是一个有向图,每个节点可以是 LLM、工具或任意 Python 函数,边定义了节点间的流向。

1.2 为什么需要 LangGraph

当你的 AI 应用需要多步骤推理循环迭代人工介入时,LangChain 的线性链式调用就不够用了。

场景LangChainLangGraph
简单问答✅ 适合✅ 可以
ReAct 推理⚠️ 需特殊处理✅ 原生支持
多 Agent 协作❌ 不支持✅ 完整支持
人机交互❌ 不支持✅ 支持中断恢复
状态持久化❌ 不支持✅ Checkpointer
图 1-2:State 在节点间的流转过程
初始 State messages=[] city="" weather=None extract_city Node 更新 city 字段 State 更新 city = "北京" messages 不变 fetch_weather Node 更新 weather 字段 最终 State ✅ weather 更新 END

1.3 核心概念

1.3.1 State(状态)

State 是贯穿整个图的共享数据容器。你可以定义任意结构的状态:

python
from typing import TypedDict, Annotated from operator import add class AgentState(TypedDict): # 自动累加消息(不覆盖,追加) messages: Annotated[list, add] # 当前城市 city: str # 天气信息(可为 None) weather: str | None
Reducer 是什么?

Annotated[list, add] 告诉 LangGraph:每次更新 messages 时,用 add 函数合并。这样新消息会追加到历史,而不是覆盖。

1.3.2 Nodes(节点)

节点是处理函数,接收当前状态,返回要更新的字段:

python
# 节点函数签名:输入 State,返回 dict def extract_city(state: AgentState) -> dict: user_msg = state["messages"][-1].content city = extract_from_text(user_msg) # 简单解析 return {"city": city} def fetch_weather(state: AgentState) -> dict: weather = call_weather_api(state["city"]) return {"weather": weather}
图 1-3:Node 执行模型 — 输入 State,输出差量更新
输入 State messages city weather Node 函数 def node_fn(state): → 读取 state["city"] → 返回 {"city": "北京"} 差量更新 {"city": "北京"} 仅返回变更字段
关键理解

Node 只返回它想更新的字段,不需要返回整个 State。LangGraph 会用 Reducer 自动合并更新。比如只改了 city,messages 保持不变。

1.3.3 Edges(边)

边定义节点之间的流向

python
from langgraph.graph import StateGraph, START, END # 构建图 graph = StateGraph(AgentState) # 添加节点 graph.add_node("extract_city", extract_city) graph.add_node("fetch_weather", fetch_weather) # 添加边 graph.add_edge(START, "extract_city") # 起点 graph.add_edge("extract_city", "fetch_weather") # 普通边 graph.add_edge("fetch_weather", END) # 终点
图 1-2:LangGraph 三要素
State 状态定义 + Reducer Nodes 处理函数 Edges 节点流向 StateGraph = Nodes + Edges

1.4 环境安装

bash
# 基础安装 pip install langgraph langchain-openai # 可选:状态持久化后端 pip install langgraph-checkpoint # 验证版本 python -c "import langgraph; print(langgraph.__version__)"
版本要求

LangGraph 需要 Python 3.10+LangChain 0.1.20+。建议使用虚拟环境隔离依赖。

1.5 第一个 LangGraph 应用

实战:构建一个天气问答助手,从用户问题中提取城市并查询天气。

步骤 1:定义状态

python
from typing import TypedDict, Annotated from operator import add class WeatherState(TypedDict): messages: Annotated[list, add] city: str weather: str | None

步骤 2:定义节点

python
def extract_city(state: WeatherState): last_msg = state["messages"][-1].content city = extract_city_from_text(last_msg) return {"city": city} def fetch_weather(state: WeatherState): weather = weather_api.get(state["city"]) return {"weather": weather}

步骤 3:构建图

python
from langgraph.graph import StateGraph, START, END graph = StateGraph(WeatherState) graph.add_node("extract_city", extract_city) graph.add_node("fetch_weather", fetch_weather) graph.add_edge(START, "extract_city") graph.add_edge("extract_city", "fetch_weather") graph.add_edge("fetch_weather", END) app = graph.compile() # 编译后才能执行

步骤 4:执行

python
result = app.invoke({ "messages": [{"role": "user", "content": "北京今天天气怎么样?"}], "city": "", "weather": None }) print(result["weather"])
图 1-4:LangGraph 执行流程 — compile() + invoke() 全链路
① 定义 State Nodes Edges ② 构建图 StateGraph() .add_node() .add_edge() ③ 编译 .compile() → app 对象 (可序列化) ④ 执行 app.invoke() app.stream() → 结果 State

1.6 本章小结

  • LangGraph 是有状态的多步骤工作流框架
  • 核心三要素:State(状态)Nodes(节点)Edges(边)
  • State 使用 Annotated[type, reducer] 定义,reducer 决定如何合并更新
  • 图必须编译(compile)后才能执行
  • 与 LangChain 的关系:LangGraph 构建在 LangChain 之上,专注复杂工作流
下章预告

第 2 章我们将深入学习 State 的高级用法:条件更新、多个 Reducer、以及如何设计复杂的状态结构。