核心概念 #
概述 #
LangGraph 的核心是将 Agent 工作流建模为**图（Graph）**结构。通过组合节点（Nodes）和边（Edges），你可以创建复杂的有状态、可循环的工作流。
text
┌─────────────────────────────────────────────────────────────┐
│ LangGraph 核心架构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ │ State │ │
│ │ 状态 │ │
│ └────┬────┘ │
│ │ │
│ ┌────────────┼────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │ Node │ │ Node │ │ Node │ │
│ │ 节点 │ │ 节点 │ │ 节点 │ │
│ └────┬───┘ └────┬───┘ └────┬───┘ │
│ │ │ │ │
│ └────────────┼────────────┘ │
│ │ │
│ ┌────┴────┐ │
│ │ Edge │ │
│ │ 边 │ │
│ └─────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
三大核心组件 #
LangGraph 使用三个关键组件来定义 Agent 的行为：
1. State（状态） #
状态是一个共享的数据结构，代表应用程序的当前快照。它可以在图的节点之间传递和更新。
python
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
documents: list
current_step: str
2. Nodes（节点） #
节点是编码 Agent 逻辑的函数。它们接收当前状态作为输入，执行计算或副作用，然后返回更新后的状态。
python
def agent_node(state: State):
response = llm.invoke(state["messages"])
return {"messages": [response]}
3. Edges（边） #
边是根据当前状态确定下一个执行节点的函数。它们可以是条件分支或固定转换。
python
def should_continue(state: State):
if state["messages"][-1].tool_calls:
return "tools"
return END
图（Graph） #
StateGraph #
StateGraph 是 LangGraph 的主要图类，它以用户定义的状态对象为参数。
python
from langgraph.graph import StateGraph
class State(TypedDict):
messages: list
count: int
graph = StateGraph(State)
图的编译 #
编译图是一个重要步骤，它会：
- 验证图结构（检查孤立节点等）
- 配置运行时参数（如检查点、断点）
python
app = graph.compile()
编译选项 #
python
from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(
checkpointer=MemorySaver(), # 添加持久化
interrupt_before=["sensitive_node"], # 在特定节点前中断
interrupt_after=["review_node"], # 在特定节点后中断
)
状态（State）详解 #
状态 Schema #
状态 Schema 定义了图的数据结构，可以使用 TypedDict 或 Pydantic 模型。
python
from typing import TypedDict, Annotated
from operator import add
from pydantic import BaseModel
class TypedState(TypedDict):
messages: Annotated[list, add]
count: int
class PydanticState(BaseModel):
messages: list
count: int = 0
Reducer 函数 #
Reducer 定义了如何将节点的更新应用到状态上。
text
┌─────────────────────────────────────────────────────────────┐
│ Reducer 工作原理 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 默认 Reducer（覆盖）： │
│ state = {"count": 1} │
│ update = {"count": 2} │
│ result = {"count": 2} # 直接覆盖 │
│ │
│ 自定义 Reducer（如 add）： │
│ state = {"items": [1, 2]} │
│ update = {"items": [3]} │
│ result = {"items": [1, 2, 3]} # 合并 │
│ │
└─────────────────────────────────────────────────────────────┘
内置 Reducer #
python
from typing import Annotated
from operator import add
class State(TypedDict):
items: Annotated[list, add]
messages: Annotated[list, add_messages]
自定义 Reducer #
python
def custom_reducer(left: list, right: list) -> list:
return left + [x for x in right if x not in left]
class State(TypedDict):
unique_items: Annotated[list, custom_reducer]
多 Schema 支持 #
LangGraph 支持为图定义不同的输入、输出和内部 Schema。
python
from typing import TypedDict
class InputState(TypedDict):
user_input: str
class OutputState(TypedDict):
graph_output: str
class OverallState(TypedDict):
user_input: str
internal_data: str
graph_output: str
graph = StateGraph(
OverallState,
input_schema=InputState,
output_schema=OutputState
)
节点（Nodes）详解 #
节点函数签名 #
节点是 Python 函数，可以接收以下参数：
python
from langchain_core.runnables import RunnableConfig
from langgraph.runtime import Runtime
def node_function(
state: State, # 必需：当前状态
config: RunnableConfig, # 可选：配置信息
runtime: Runtime, # 可选：运行时上下文
):
return {"key": "value"} # 返回状态更新
添加节点 #
python
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))
graph.add_node(lambda state: {"count": state["count"] + 1})
特殊节点 #
text
┌─────────────────────────────────────────────────────────────┐
│ 特殊节点 │
├─────────────────────────────────────────────────────────────┤
│ │
│ START 节点： │
│ - 虚拟节点，表示图的入口 │
│ - 用于定义第一个执行的节点 │
│ │
│ END 节点： │
│ - 虚拟节点，表示图的出口 │
│ - 当流程到达 END 时，图执行结束 │
│ │
│ 使用示例： │
│ graph.add_edge(START, "first_node") │
│ graph.add_edge("last_node", END) │
│ │
└─────────────────────────────────────────────────────────────┘
节点缓存 #
LangGraph 支持节点级别的缓存，避免重复计算。
python
from langgraph.types import CachePolicy
from langgraph.cache.memory import InMemoryCache
def expensive_node(state: State):
import time
time.sleep(2)
return {"result": state["input"] * 2}
graph.add_node("expensive", expensive_node, cache_policy=CachePolicy(ttl=60))
app = graph.compile(cache=InMemoryCache())
边（Edges）详解 #
普通边 #
普通边表示固定的转换，从一个节点直接到另一个节点。
python
from langgraph.graph import START, END
graph.add_edge(START, "agent")
graph.add_edge("agent", "tools")
graph.add_edge("tools", END)
条件边 #
条件边根据状态动态决定下一个节点。
python
def route(state: State):
if state["needs_tools"]:
return "tools"
return END
graph.add_conditional_edges(
"agent",
route,
["tools", END] # 可能的目标节点
)
条件边映射 #
可以使用字典映射返回值到节点名称：
python
def route(state: State):
if state["type"] == "math":
return "math_agent"
elif state["type"] == "code":
return "code_agent"
return "general_agent"
graph.add_conditional_edges(
"router",
route,
{
"math_agent": "math_agent",
"code_agent": "code_agent",
"general_agent": "general_agent"
}
)
入口点 #
入口点定义图的起始节点。
python
from langgraph.graph import START
graph.add_edge(START, "first_node")
graph.set_entry_point("first_node")
条件入口点 #
python
from langgraph.graph import START
def route_entry(state: State):
if state["urgent"]:
return "priority_handler"
return "normal_handler"
graph.add_conditional_edges(START, route_entry)
执行模型 #
Super-step（超步） #
LangGraph 基于 Pregel 模型，使用"超步"概念：
text
┌─────────────────────────────────────────────────────────────┐
│ Super-step 执行模型 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Super-step 1: │
│ ┌─────────┐ │
│ │ Node A │ ───> 执行并产生消息 │
│ └─────────┘ │
│ │
│ Super-step 2: │
│ ┌─────────┐ ┌─────────┐ │
│ │ Node B │ │ Node C │ ───> 并行执行 │
│ └─────────┘ └─────────┘ │
│ │
│ Super-step 3: │
│ ┌─────────┐ │
│ │ Node D │ ───> 合并 B 和 C 的结果 │
│ └─────────┘ │
│ │
│ 规则： │
│ - 同一超步内的节点并行执行 │
│ - 不同超步的节点顺序执行 │
│ - 所有节点完成且无消息传递时结束 │
│ │
└─────────────────────────────────────────────────────────────┘
并行执行 #
当一个节点有多个出边时，目标节点会并行执行：
python
graph.add_edge("splitter", "worker_1")
graph.add_edge("splitter", "worker_2")
graph.add_edge("splitter", "worker_3")
worker_1, worker_2, worker_3 会并行执行
递归限制 #
防止无限循环：
python
result = app.invoke(
{"messages": [...]},
config={"recursion_limit": 25}
)
Command 对象 #
Command 是一个多功能原语，用于控制图的执行。
python
from langgraph.types import Command
def my_node(state: State) -> Command:
return Command(
update={"foo": "bar"}, # 状态更新
goto="next_node" # 控制流
)
Command 用途 #
text
┌─────────────────────────────────────────────────────────────┐
│ Command 用途 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 从节点返回： │
│ - update: 更新状态 │
│ - goto: 跳转到指定节点 │
│ │
│ 2. 输入到 invoke/stream： │
│ - resume: 在中断后恢复执行 │
│ │
│ 3. 从工具返回： │
│ - 结合状态更新和控制流 │
│ │
│ 4. 子图导航： │
│ - graph=Command.PARENT 跳转到父图 │
│ │
└─────────────────────────────────────────────────────────────┘
Send 对象 #
Send 用于动态创建边，常用于 Map-Reduce 模式。
python
from langgraph.types import Send
def map_node(state: State):
return [
Send("process", {"item": item})
for item in state["items"]
]
graph.add_conditional_edges("mapper", map_node)
Map-Reduce 示例 #
python
from langgraph.types import Send
def generate_topics(state: State):
return [Send("generate_joke", {"topic": t}) for t in state["topics"]]
def generate_joke(state: State):
return {"jokes": [llm.invoke(f"joke about {state['topic']}")]}
def combine_jokes(state: State):
return {"result": "\n".join(state["jokes"])}
graph = StateGraph(State)
graph.add_node("generate_topics", generate_topics)
graph.add_node("generate_joke", generate_joke)
graph.add_node("combine_jokes", combine_jokes)
graph.add_edge(START, "generate_topics")
graph.add_conditional_edges("generate_topics", generate_topics)
graph.add_edge("generate_joke", "combine_jokes")
graph.add_edge("combine_jokes", END)
运行时上下文 #
配置（Config） #
python
from langchain_core.runnables import RunnableConfig
def my_node(state: State, config: RunnableConfig):
thread_id = config["configurable"]["thread_id"]
return {"thread_id": thread_id}
result = app.invoke(
{"messages": [...]},
config={"configurable": {"thread_id": "123"}}
)
运行时上下文（Runtime） #
python
from dataclasses import dataclass
from langgraph.runtime import Runtime
@dataclass
class Context:
user_id: str
db_connection: any
graph = StateGraph(State, context_schema=Context)
def my_node(state: State, runtime: Runtime[Context]):
user_id = runtime.context.user_id
return {"user_id": user_id}
result = app.invoke(
{"messages": [...]},
context=Context(user_id="user-123", db_connection=db)
)
图迁移 #
LangGraph 支持图定义的迁移，即使使用了检查点。
text
┌─────────────────────────────────────────────────────────────┐
│ 图迁移支持 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 已完成的线程： │
│ ✅ 可以完全改变拓扑结构 │
│ ✅ 添加/删除/重命名节点和边 │
│ │
│ 中断中的线程： │
│ ✅ 可以添加节点和边 │
│ ⚠️ 不能重命名/删除节点（可能正在等待） │
│ │
│ 状态修改： │
│ ✅ 添加/删除状态键 │
│ ⚠️ 重命名状态键会丢失历史数据 │
│ ⚠️ 类型不兼容可能导致问题 │
│ │
└─────────────────────────────────────────────────────────────┘
下一步 #
现在你已经理解了 LangGraph 的核心概念，接下来学习 状态管理，深入了解如何设计和使用状态！
最后更新：2026-03-30