LangGraph 术语汇总

使用说明 本术语表按 LangGraph 核心模块分类,提供技术定义、LangGraph 应用场景和通俗易懂的解释。建议配合实际代码示例学习。

1️⃣ 核心架构(Core Architecture)

术语	技术定义	LangGraph 应用	通俗解释	重要程度
StateGraph	有向状态图,管理状态转换和节点执行	LangGraph 核心类,定义 Agent 工作流	就像一张流程图,每个方框是一个处理步骤,箭头表示流转方向	⭐⭐⭐⭐⭐
Node(节点)	图中的计算单元,接收状态并返回更新	函数签名:`def node(state: State) -> State`	工作流中的一个具体操作,比如"调用 AI"、"查询数据库"	⭐⭐⭐⭐⭐
Edge(边)	连接节点的有向连接,定义执行顺序	`add_edge("node_a", "node_b")`	流程图的箭头,告诉程序"完成这步后去哪里"	⭐⭐⭐⭐⭐
Conditional Edge	根据状态动态选择下一节点的边	`add_conditional_edges(node, routing_func)`	分叉路口,根据条件决定走哪条路(如:成功→继续,失败→重试)	⭐⭐⭐⭐⭐
START / END	图的入口和出口特殊节点	`add_edge(START, "first_node")`	流程的起点和终点,就像程序的 main 函数和 return	⭐⭐⭐⭐⭐
Cycle(循环)	图中存在的回路,支持迭代执行	Agent 可以反复调用工具直到完成任务	允许流程"回头再来",比如 AI 多轮对话或重试机制	⭐⭐⭐⭐⭐
Compiled Graph	编译后的可执行图对象	`graph.compile(checkpointer=memory)`	把流程图"打包"成可运行的程序	⭐⭐⭐⭐⭐

2️⃣ 状态管理(State Management)

术语	技术定义	LangGraph 应用	通俗解释	重要程度
State	在节点间传递的数据字典	存储消息历史、上下文、工具调用结果等	程序的"记忆",记录所有发生过的事情	⭐⭐⭐⭐⭐
State Schema	TypedDict 定义的状态结构	`class AgentState(TypedDict): messages: list`	状态的"数据格式说明书",定义有哪些字段	⭐⭐⭐⭐⭐
Reducer	状态字段的合并策略函数	`Annotated[list, operator.add]`	定义如何合并来自多个节点的数据(加法、覆盖等)	⭐⭐⭐⭐⭐
MessagesState	内置的消息管理状态类	自带 `add_messages` reducer	专门管理对话历史的状态,自动处理消息追加	⭐⭐⭐⭐⭐
add_messages	消息列表的智能 reducer	支持消息去重、覆盖、删除	聪明地追加消息,避免重复,支持更新和删除	⭐⭐⭐⭐⭐
RemoveMessage	标记要删除的消息	`return {"messages": [RemoveMessage(id=msg_id)]}`	从对话历史中"擦除"某条消息	⭐⭐⭐⭐
Annotated	Python 类型注解,附加元数据	`Annotated[list, operator.add]` 定义 reducer	给类型加"标签",告诉系统这个字段的特殊处理规则	⭐⭐⭐⭐
Multiple Schemas	不同节点使用不同状态结构	全局状态 + 节点私有状态	不同部门有不同表格,但共享部分关键信息	⭐⭐⭐⭐
Private State	仅在特定节点可见的状态字段	子图内部状态,不暴露给外部	"私房钱",只有特定部分能看到和使用	⭐⭐⭐

3️⃣ 消息系统(Message System)

术语	技术定义	LangGraph 应用	通俗解释	重要程度
HumanMessage	用户输入的消息类型	`messages.append(HumanMessage(content="你好"))`	用户说的话	⭐⭐⭐⭐⭐
AIMessage	AI 模型返回的消息类型	包含 LLM 回复和 `tool_calls`	AI 的回答	⭐⭐⭐⭐⭐
SystemMessage	系统指令消息	定义 AI 角色和行为规则	给 AI 的"使用说明书",告诉它该怎么做	⭐⭐⭐⭐⭐
ToolMessage	工具执行结果消息	返回工具调用的输出	工具(函数)返回的结果	⭐⭐⭐⭐⭐
tool_calls	AI 请求调用的工具列表	`[{"name": "search", "args": {"query": "天气"}}]`	AI 说"我需要调用这个工具来帮我"	⭐⭐⭐⭐⭐
tool_call_id	工具调用的唯一标识符	关联 AIMessage 和 ToolMessage	给每次工具调用一个"身份证号"	⭐⭐⭐⭐
trim_messages	按策略裁剪消息历史	保留最近 N 条或最近 K tokens	删除老旧消息,节省 token 成本	⭐⭐⭐⭐
filter_messages	按条件过滤消息	移除特定类型或满足条件的消息	筛选出需要的消息,过滤掉不要的	⭐⭐⭐⭐
merge_message_runs	合并连续同角色消息	将多条 AIMessage 合并为一条	把连续的同一个人说的话合并成一段	⭐⭐⭐

4️⃣ 工具系统(Tool System)

术语	技术定义	LangGraph 应用	通俗解释	重要程度
Tool	可被 AI 调用的 Python 函数	`@tool` 装饰器定义工具	AI 的"工具箱",里面是各种能力(搜索、计算等)	⭐⭐⭐⭐⭐
bind_tools()	将工具绑定到 LLM	`llm.bind_tools([search_tool, calculator])`	告诉 AI"你可以用这些工具"	⭐⭐⭐⭐⭐
ToolNode	自动执行工具调用的节点	`tools_node = ToolNode([search, calculator])`	专门负责执行工具的节点,AI 说"用工具"它就执行	⭐⭐⭐⭐⭐
tools_condition	判断是否需要调用工具的路由函数	检查 `tool_calls` 是否为空	检查 AI 是否想用工具,决定下一步	⭐⭐⭐⭐⭐
parallel_tool_calls	并行执行多个工具调用	AI 同时调用多个工具提升效率	同时做多件事,而不是排队一个个来	⭐⭐⭐⭐
Docstring	函数文档字符串	LLM 通过 docstring 理解工具用途	工具的"说明书",告诉 AI 这个工具是干什么的	⭐⭐⭐⭐

5️⃣ 记忆与持久化(Memory & Persistence)

术语	技术定义	LangGraph 应用	通俗解释	重要程度
Checkpointer	状态持久化接口	保存和恢复图执行状态	程序的"存档点",可以随时暂停和继续	⭐⭐⭐⭐⭐
MemorySaver	内存版 Checkpointer	状态存在内存,进程结束后消失	把状态记在"草稿纸"上,关机就没了	⭐⭐⭐⭐⭐
SqliteSaver	SQLite 数据库持久化	状态持久化到本地数据库	把状态记在"日记本"里,永久保存	⭐⭐⭐⭐⭐
PostgresSaver	PostgreSQL 持久化(生产级)	企业级多用户状态管理	把状态记在"银行保险箱"里,安全可靠	⭐⭐⭐⭐⭐
Thread	对话会话/线程	通过 `thread_id` 区分不同用户会话	每个用户有自己的"对话房间"	⭐⭐⭐⭐⭐
thread_id	线程的唯一标识符	`config = {"configurable": {"thread_id": "user_123"}}`	对话房间的"门牌号"	⭐⭐⭐⭐⭐
Checkpoint	某个时刻的状态快照	包含完整状态和元数据	游戏的"存档",记录某个时刻的所有信息	⭐⭐⭐⭐⭐
checkpoint_id	检查点的唯一标识符	用于 Time Travel 定位特定快照	存档的"时间戳",用来找到特定版本	⭐⭐⭐⭐
Store	长期记忆存储系统	跨会话持久化信息(用户偏好、知识)	AI 的"长期记忆",记住跨多次对话的信息	⭐⭐⭐⭐⭐
InMemoryStore	内存版 Store	快速但非持久化	把长期记忆记在"便签"上	⭐⭐⭐⭐
Namespace	Store 的命名空间	组织和隔离不同类型的记忆	给记忆分类,比如"用户信息"、"产品知识"等	⭐⭐⭐⭐

6️⃣ 人机协同(Human-in-the-Loop)

术语	技术定义	LangGraph 应用	通俗解释	重要程度
Breakpoint	程序执行的暂停点	`interrupt_before=["node_name"]`	在特定节点前"按下暂停键"	⭐⭐⭐⭐⭐
interrupt()	主动中断执行	在节点内调用 `interrupt()` 请求人工介入	让程序主动"举手"说"我需要帮助"	⭐⭐⭐⭐⭐
NodeInterrupt	带数据的中断异常	传递需要人类审核的信息	中断时附带一张"问卷"给人类	⭐⭐⭐⭐
update_state()	编辑图的当前状态	人工修改状态后恢复执行	人类"编辑"程序的记忆,然后继续运行	⭐⭐⭐⭐⭐
get_state()	获取当前状态快照	查看中断时的完整状态	查看程序当前的所有信息	⭐⭐⭐⭐⭐
get_state_history()	获取历史状态列表	查看图执行的所有历史快照	翻看程序的"历史记录"	⭐⭐⭐⭐
Time Travel	回到历史状态并分叉执行	从过去某个点重新开始	游戏的"读档"功能,回到过去重新玩	⭐⭐⭐⭐
Command	控制图执行的指令对象	`Command(resume="value", goto="node")`	给程序发"指令":继续、跳转、回退等	⭐⭐⭐⭐

7️⃣ 执行模式(Execution Modes)

术语	技术定义	LangGraph 应用	通俗解释	重要程度
invoke()	同步执行整个图	`result = graph.invoke(input, config)`	"一次性运行到底",等结果出来再继续	⭐⭐⭐⭐⭐
stream()	流式返回每个节点结果	`for chunk in graph.stream(input): ...`	边运行边输出,实时看到进度	⭐⭐⭐⭐⭐
astream()	异步流式执行	`async for chunk in graph.astream(input): ...`	异步版流式,不阻塞其他任务	⭐⭐⭐⭐
stream_mode	流式输出的粒度	`values`(完整状态)/`updates`(增量)/`messages`(仅消息)	控制输出什么内容:全部、变化、还是消息	⭐⭐⭐⭐
astream_events()	事件级异步流	最细粒度的流式,包含 token 级输出	看到最细节的过程,包括 AI 逐字输出	⭐⭐⭐⭐
Streaming	实时输出执行过程	适用于聊天应用的用户体验	让用户立即看到反馈,而不是等很久	⭐⭐⭐⭐⭐

8️⃣ 高级模式(Advanced Patterns)

术语	技术定义	LangGraph 应用	通俗解释	重要程度
Subgraph	嵌套在主图中的子图	模块化复用工作流	大流程里包含小流程,像函数调用	⭐⭐⭐⭐⭐
Send API	动态发送消息到节点	`Send("node_name", state)`	运行时动态决定要执行哪些节点	⭐⭐⭐⭐⭐
Map-Reduce	并行处理后聚合结果	Map(分发)→ Reduce(汇总)	把任务拆分给多个"工人",最后汇总结果	⭐⭐⭐⭐⭐
Parallelization	同一步内并行执行多个节点	Fan-out + Fan-in 模式	同时做多件事,提升效率	⭐⭐⭐⭐⭐
Fan-out	一个节点分发到多个节点	任务拆分阶段	一个任务变成多个子任务	⭐⭐⭐⭐
Fan-in	多个节点汇总到一个节点	使用 Reducer 聚合结果	把多个结果合并成一个	⭐⭐⭐⭐
ReAct	Reasoning + Acting 循环模式	AI 思考→行动→观察→再思考	AI 像人一样"边想边做"	⭐⭐⭐⭐⭐
Agent	自主决策并使用工具的系统	基于 ReAct 实现,循环调用 LLM + Tools	能自己做决定、用工具解决问题的 AI	⭐⭐⭐⭐⭐
Multi-Agent	多个 Agent 协作	Supervisor 模式 / Hierarchical 模式	多个 AI 分工合作完成复杂任务	⭐⭐⭐⭐⭐
Supervisor Pattern	主管 Agent 协调多个工作 Agent	主管分配任务,工作者执行	一个"经理"AI 管理多个"员工"AI	⭐⭐⭐⭐

9️⃣ 记忆架构(Memory Architecture)

术语	技术定义	LangGraph 应用	通俗解释	重要程度
Semantic Memory	语义记忆/知识记忆	存储事实和概念(用户偏好、产品知识)	AI 的"知识库",记住各种事实	⭐⭐⭐⭐⭐
Episodic Memory	情节记忆/事件记忆	存储历史对话和事件	AI 的"日记",记住发生过什么	⭐⭐⭐⭐
Procedural Memory	程序性记忆	存储技能和流程(如何执行任务)	AI 的"技能",记住怎么做事	⭐⭐⭐⭐
Profile Schema	用户画像模式	单条记录持续更新(如用户资料)	一个人只有一张"档案卡",不断更新	⭐⭐⭐⭐⭐
Collection Schema	集合模式	多条记录累积(如笔记、事件)	一个人有多条"日记",不断添加	⭐⭐⭐⭐⭐
Trustcall	结构化输出 + 自我修正	用于更新 Memory Schema	AI 按格式输出,还能自己检查修正	⭐⭐⭐⭐
JSON Patch	增量更新 JSON 的标准	Profile 模式的增量更新机制	只改变化的部分,不重写整个文档	⭐⭐⭐⭐
put()	存储记忆的方法	`store.put(namespace, key, value)`	往记忆库"存入"一条记忆	⭐⭐⭐⭐⭐
get()	读取记忆的方法	`store.get(namespace, key)`	从记忆库"读取"一条记忆	⭐⭐⭐⭐⭐
search()	搜索记忆的方法	`store.search(namespace, filter=...)`	在记忆库里"查找"符合条件的记忆	⭐⭐⭐⭐⭐

🔟 生产部署(Production Deployment)

术语	技术定义	LangGraph 应用	通俗解释	重要程度
LangGraph Platform	官方托管平台	部署、管理、监控 LangGraph 应用	LangGraph 的"云服务器"	⭐⭐⭐⭐⭐
LangGraph Cloud	托管版 LangGraph 服务	无需自己管服务器	"全托管"服务,开箱即用	⭐⭐⭐⭐⭐
LangGraph Studio	可视化开发和调试工具	图形化查看和调试 Agent	Agent 的"开发者工具"	⭐⭐⭐⭐⭐
LangGraph CLI	命令行工具	`langgraph dev` / `langgraph build`	本地开发和构建的命令行	⭐⭐⭐⭐
langgraph.json	项目配置文件	定义图、依赖、环境变量	项目的"配置清单"	⭐⭐⭐⭐⭐
Assistant	可配置的 Agent 实例	同一图的不同配置版本	同一个 AI 的不同"人设"版本	⭐⭐⭐⭐⭐
assistant_id	Assistant 的唯一标识	区分不同的 Agent 配置	Assistant 的"身份证号"	⭐⭐⭐⭐
Run	一次完整的图执行	从输入到输出的完整过程	一次完整的"运行任务"	⭐⭐⭐⭐⭐
Double Texting	用户快速连续发消息	并发执行冲突处理	用户连续点"发送"按钮的处理策略	⭐⭐⭐⭐
Reject	拒绝新请求策略	返回 409 错误	"我正忙着,稍后再来"	⭐⭐⭐⭐
Enqueue	排队策略	新请求进入队列	"请排队,一个个来"	⭐⭐⭐⭐
Interrupt	中断当前执行策略	中断旧任务,执行新任务	"停下手头的,做新的"	⭐⭐⭐⭐
Rollback	回滚策略	撤销未完成的执行	"撤销刚才的,重新来"	⭐⭐⭐⭐
LangGraph SDK	客户端库	Python/JS 调用远程图	连接 LangGraph 服务的"客户端"	⭐⭐⭐⭐⭐
Remote Graph	远程部署的图	通过 API 调用	运行在服务器上的 Agent	⭐⭐⭐⭐⭐

1️⃣ Python 技术基础

术语	技术定义	LangGraph 应用	通俗解释	重要程度
TypedDict	带类型注解的字典	定义 State Schema	给字典加"类型标签",增强类型检查	⭐⭐⭐⭐⭐
Annotated	附加元数据的类型注解	定义 Reducer	给类型加"额外信息"	⭐⭐⭐⭐
Literal	字面量类型	限定字段为特定值	只能是这几个值,别的不行	⭐⭐⭐⭐
Optional	可选类型(可为 None)	`Optional[str]`	这个值可能不存在	⭐⭐⭐⭐
Pydantic	数据验证库	定义复杂 Schema 和验证规则	数据的"格式检查器"	⭐⭐⭐⭐⭐
async/await	异步编程关键字	异步执行图和工具	"异步"编程,不阻塞等待	⭐⭐⭐⭐
Callable	可调用对象类型	定义节点和工具签名	表示"这是个函数或方法"	⭐⭐⭐⭐
Dataclass	数据类装饰器	简化状态类定义	快速创建数据结构	⭐⭐⭐⭐

💡 使用建议

初学者优先级:先掌握 ⭐⭐⭐⭐⭐(5星)术语
按模块学习:跟随课程顺序,逐个模块掌握术语
结合代码:每个术语都应配合实际代码示例理解
对比学习:注意"技术定义"、"LangGraph 应用"和"通俗解释"的差异
查阅参考:遇到术语时回到本汇总快速查找

术语总数:约 120+ 个核心术语 覆盖范围:从基础架构到生产部署的完整体系 更新日期:2025-01-13

📘 LangGraph 核心概念详细解读

本节提供所有重要术语的深度解析,每个概念都配有简单示例说明。

🔵 1. StateGraph (状态图)

字面意思

State + Graph = "状态图"

State:状态、数据 Graph:图结构(节点+边)

为什么叫 StateGraph?

因为这是一个以状态为核心的图结构:

状态在节点间流转
每个节点可以修改状态
通过边来控制状态的流转路径

在 LangGraph 里的作用

StateGraph 是构建所有工作流的基础类,它:

定义状态的结构(State Schema)
添加处理节点(Nodes)
定义流转路径(Edges)
编译成可执行的应用

简单示例

python

from langgraph.graph import StateGraph
from typing import TypedDict

# 1. 定义状态结构
class MyState(TypedDict):
    message: str
    count: int

# 2. 创建状态图
graph = StateGraph(MyState)

# 3. 添加节点
def process(state):
    return {"count": state["count"] + 1}

graph.add_node("process", process)

# 4. 添加边
graph.add_edge(START, "process")
graph.add_edge("process", END)

# 5. 编译
app = graph.compile()

# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

👍 总结:StateGraph = "用图结构管理状态流转的容器"

🔵 2. Node (节点)

字面意思

Node = "节点"

在图论中,节点是图的基本组成单元。

为什么叫 Node?

因为在图结构中:

Node 是处理单元
通过 Edge(边)连接
每个 Node 接收状态、处理、返回更新

在 LangGraph 里的作用

Node 是图中的函数/处理单元:

接收当前状态
执行特定逻辑(调用 AI、查询数据库等)
返回状态更新

简单示例

python

# 节点就是一个函数
def my_node(state: MyState) -> dict:
    """
    节点函数的标准格式:
    - 输入: 完整的 state
    - 输出: 要更新的字段(部分更新)
    """
    new_message = f"处理: {state['message']}"
    return {"message": new_message}

# 添加到图中
graph.add_node("my_node", my_node)

👍 总结:Node = "接收状态、执行逻辑、返回更新的函数"

🔵 3. Edge (边)

字面意思

Edge = "边"

在图论中,边连接两个节点。

为什么叫 Edge?

因为它定义了:

节点之间的连接
执行的顺序
状态的流转路径

在 LangGraph 里的作用

Edge 控制执行流程:

普通边:从 A 直接到 B
条件边:根据状态决定去哪里

简单示例

python

# 普通边:固定路径
graph.add_edge("node_a", "node_b")  # a → b

# 条件边:动态路由
def router(state):
    if state["count"] > 10:
        return "end"
    return "continue"

graph.add_conditional_edges(
    "node_a",      # 从哪个节点
    router,        # 路由函数
    {
        "end": END,
        "continue": "node_b"
    }
)

👍 总结:Edge = "定义节点间流转路径的箭头"

🔵 4. State (状态)

字面意思

State = "状态"

状态是程序在某个时刻的数据快照。

为什么叫 State?

因为它:

记录当前的所有信息
在节点间传递和更新
就像程序的"记忆"

在 LangGraph 里的作用

State 是:

数据的容器(字典)
节点间通信的媒介
程序执行的"记忆"

简单示例

python

from typing import TypedDict, Annotated
import operator

class MyState(TypedDict):
    # 简单字段:直接覆盖
    current_step: str

    # 带 reducer 的字段:累加
    count: Annotated[int, operator.add]

    # 消息列表:智能合并
    messages: Annotated[list, add_messages]

# 节点返回状态更新
def node1(state):
    return {
        "current_step": "step1",
        "count": 1  # 如果之前是 5,会变成 6(因为 operator.add)
    }

👍 总结:State = "在节点间流转的数据字典"

🔵 5. Annotated (带注解的)

字面意思

Annotated = "带注解的、被标注的"

来自 Python 的 typing.Annotated。

为什么叫 Annotated?

因为:

python

Annotated[T, metadata]

T:原本的类型(如 list, int)
metadata:附加的"注解/元数据"

字面理解:在类型上加一层标签/说明

在 LangGraph 里的作用

Annotated 用于给状态字段添加 Reducer(合并策略):

python

from typing import Annotated
import operator

class State(TypedDict):
    # 不带注解:直接覆盖
    name: str

    # 带注解:使用 operator.add 累加
    count: Annotated[int, operator.add]

    # 带注解:使用 add_messages 智能合并
    messages: Annotated[list, add_messages]

简单示例

python

# 示例 1:数字累加
class State(TypedDict):
    score: Annotated[int, operator.add]

# 节点 1 返回
return {"score": 10}  # state["score"] = 10

# 节点 2 返回
return {"score": 5}   # state["score"] = 15 (累加!)

# 示例 2:消息合并
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 节点 1 返回
return {"messages": [HumanMessage(content="你好")]}

# 节点 2 返回
return {"messages": [AIMessage(content="你好呀")]}

# 最终 state["messages"] = [HumanMessage(...), AIMessage(...)]

👍 总结:Annotated = "给类型加标签,告诉框架如何合并这个字段"

🔵 6. Reducer (归约器/合并器)

字面意思

Reducer = "减少者、归约器"

来自函数式编程,将多个值"归约"成一个值。

为什么叫 Reducer?

因为它定义了如何合并多个值:

多个节点可能同时更新同一字段
Reducer 决定最终的值是什么

在 LangGraph 里的作用

Reducer 控制状态字段的更新策略:

operator.add:累加(数字、列表)
add_messages:智能合并消息
自定义函数:任意合并逻辑

简单示例

python

import operator

# 内置 reducer
count: Annotated[int, operator.add]  # 累加

# 自定义 reducer
def merge_dicts(left, right):
    """合并两个字典"""
    return {**left, **right}

data: Annotated[dict, merge_dicts]

# 使用示例
class State(TypedDict):
    score: Annotated[int, operator.add]

# 节点 A 返回
{"score": 10}

# 节点 B 返回
{"score": 5}

# 合并后:state["score"] = 15

👍 总结:Reducer = "定义如何合并多个更新的函数"

🔵 7. Checkpointer (检查点保存器)

字面意思

Checkpoint + er = "检查点 + 执行者"

Checkpoint:检查点、存档点

为什么叫 Checkpointer?

因为它:

在执行过程中保存"检查点"
可以随时恢复到某个检查点
就像游戏的"存档系统"

在 LangGraph 里的作用

Checkpointer 提供:

持久化:保存状态到磁盘/数据库
恢复:从上次中断处继续
Time Travel:回到历史状态

简单示例

python

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver

# 内存版(开发用)
memory = MemorySaver()

# 持久化版(生产用)
db = SqliteSaver.from_conn_string("checkpoints.db")

# 编译时指定
app = graph.compile(checkpointer=memory)

# 使用 thread_id 区分不同会话
config = {"configurable": {"thread_id": "user_123"}}

# 第一次调用
app.invoke({"messages": [...]}, config)

# 第二次调用(会记住上次的状态)
app.invoke({"messages": [...]}, config)

👍 总结:Checkpointer = "保存和恢复执行状态的存档系统"

🔵 8. Thread (线程/会话)

字面意思

Thread = "线程、会话"

在这里不是操作系统的线程,而是"对话线程"。

为什么叫 Thread?

因为:

每个用户有独立的"对话线程"
通过 thread_id 区分
就像论坛的"帖子"

在 LangGraph 里的作用

Thread 用于:

隔离不同用户的状态
保存独立的对话历史
支持多用户并发

简单示例

python

# 用户 A 的对话
config_a = {"configurable": {"thread_id": "user_alice"}}
app.invoke({"messages": [("user", "你好")]}, config_a)

# 用户 B 的对话
config_b = {"configurable": {"thread_id": "user_bob"}}
app.invoke({"messages": [("user", "嗨")]}, config_b)

# 两个用户的状态完全独立!

👍 总结:Thread = "每个用户独立的对话房间"

🔵 9. Breakpoint (断点)

字面意思

Break + Point = "中断点"

程序调试中的断点。

为什么叫 Breakpoint?

因为:

程序执行到这里会暂停
等待人工介入
然后可以继续

在 LangGraph 里的作用

Breakpoint 实现人机协作:

在关键节点前暂停
让人类审核/修改
然后继续执行

简单示例

python

# 编译时设置断点
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["human_review"]  # 在这个节点前暂停
)

# 执行
app.invoke(input, config)

# 此时程序暂停,可以:
state = app.get_state(config)
print(state.values)  # 查看当前状态

# 修改状态
app.update_state(config, {"approved": True})

# 继续执行
app.invoke(None, config)

👍 总结:Breakpoint = "让程序暂停,等待人工介入的点"

🔵 10. Tool (工具)

字面意思

Tool = "工具"

就是字面意思的"工具"。

为什么叫 Tool?

因为:

AI 可以"使用"这些工具
就像人使用锤子、扳手
Tool 是 AI 的"能力扩展"

在 LangGraph 里的作用

Tool 让 AI 能够:

搜索信息
执行计算
调用 API
查询数据库

简单示例

python

from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """
    搜索互联网信息

    Args:
        query: 搜索关键词
    """
    # 实际的搜索逻辑
    return f"关于 {query} 的结果..."

# 绑定到 LLM
tools = [search]
llm_with_tools = llm.bind_tools(tools)

# AI 可以决定是否调用工具
response = llm_with_tools.invoke([
    ("user", "搜索 LangGraph 教程")
])

# response 可能包含 tool_calls
if response.tool_calls:
    print("AI 想调用工具:", response.tool_calls)

👍 总结:Tool = "AI 可以调用的函数/能力"

🔵 11. ReAct (推理+行动)

字面意思

Reasoning + Acting = "推理 + 行动"

为什么叫 ReAct?

因为这是一个循环模式:

Reason(推理):AI 思考该做什么
Act(行动):AI 调用工具执行
Observe(观察):查看工具结果
重复 1-3 直到完成任务

在 LangGraph 里的作用

ReAct 是 Agent 的核心模式:

用户提问 → AI 思考 → 调用工具 → 观察结果 → 再思考 → ...

简单示例

python

from langgraph.prebuilt import create_react_agent

# 一行代码创建 ReAct Agent
agent = create_react_agent(llm, tools=[search, calculator])

# 执行
agent.invoke({
    "messages": [("user", "北京今天天气如何?")]
})

# 内部流程:
# 1. AI: "我需要搜索北京天气" → tool_calls: [search("北京天气")]
# 2. 工具执行: "晴天,25度"
# 3. AI 观察结果: "好的,我知道了"
# 4. AI 回复用户: "北京今天晴天,25度"

👍 总结:ReAct = "AI 边思考边行动的循环模式"

🔵 12. MessagesState (消息状态)

字面意思

Messages + State = "消息 + 状态"

为什么叫 MessagesState?

因为这是专门管理消息列表的状态类:

内置 messages 字段
自带 add_messages reducer
简化对话管理

在 LangGraph 里的作用

MessagesState 是最常用的状态类:

python

from langgraph.graph import MessagesState

# 等价于:
class MessagesState(TypedDict):
    messages: Annotated[list, add_messages]

简单示例

python

from langgraph.graph import StateGraph, MessagesState

# 直接使用 MessagesState
graph = StateGraph(MessagesState)

def chatbot(state):
    # state["messages"] 是消息列表
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

graph.add_node("chatbot", chatbot)

👍 总结:MessagesState = "内置消息管理的状态类"

🔵 13. ToolNode (工具节点)

字面意思

Tool + Node = "工具 + 节点"

为什么叫 ToolNode?

因为这是专门执行工具调用的节点:

自动读取 tool_calls
执行对应工具
返回 ToolMessage

在 LangGraph 里的作用

ToolNode 简化工具执行:

简单示例

python

from langgraph.prebuilt import ToolNode

tools = [search, calculator]

# 创建工具节点
tool_node = ToolNode(tools)

# 添加到图
graph.add_node("tools", tool_node)

# AI 调用工具后自动执行
# input: AIMessage(tool_calls=[...])
# output: ToolMessage(content="结果")

👍 总结:ToolNode = "自动执行工具调用的节点"

🔵 14. Subgraph (子图)

字面意思

Sub + Graph = "子 + 图"

为什么叫 Subgraph?

因为:

大图里嵌套小图
就像函数调用
实现模块化复用

在 LangGraph 里的作用

Subgraph 用于:

封装复杂逻辑
模块化设计
复用工作流

简单示例

python

# 定义子图
sub_graph = StateGraph(SubState)
sub_graph.add_node("step1", ...)
sub_graph.add_node("step2", ...)
sub_app = sub_graph.compile()

# 嵌入主图
main_graph = StateGraph(MainState)
main_graph.add_node("subprocess", sub_app)  # 子图作为节点
main_graph.add_edge("start", "subprocess")

👍 总结:Subgraph = "可嵌套的子工作流"

🔵 15. Stream (流式输出)

字面意思

Stream = "流、流式"

就像水流一样连续输出。

为什么叫 Stream?

因为:

不是一次性返回全部结果
而是逐步输出
就像打字机逐字显示

在 LangGraph 里的作用

Stream 提供实时反馈:

简单示例

python

# invoke: 等全部完成
result = app.invoke(input, config)

# stream: 逐步返回
for chunk in app.stream(input, config):
    print(chunk)  # 每个节点完成后输出

# astream_events: token 级流式
async for event in app.astream_events(input, config):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

👍 总结:Stream = "逐步输出结果,提供实时反馈"

🔵 16. Saver vs Store (对比与联系)

核心区别

Saver (Checkpointer) 和 Store 是 LangGraph 中两个不同的持久化系统,它们服务于不同的目的:

对比维度	Saver (Checkpointer)	Store
用途	保存 workflow 状态	保存 agent 的长期记忆
数据类型	图的执行状态(State)	业务数据(用户偏好、知识库等)
生命周期	与对话会话绑定(thread_id)	跨会话持久化
自动管理	LangGraph 自动保存和恢复	需要手动 put/get/search
典型场景	断点续传、Time Travel	用户画像、知识库、历史记录

为什么需要两个系统?

Saver: 关注"程序怎么运行的"
- 保存每一步的状态
- 支持暂停和恢复
- 实现 Time Travel
Store: 关注"程序记住了什么"
- 存储业务知识
- 跨多次对话的记忆
- 可以被查询和更新

字面意思对比

Saver (保存器)

Saver = "Save + er" = "保存的东西"
保存的是执行快照,就像游戏存档

Store (仓库)

Store = "商店、仓库"
存储的是长期数据,就像数据库

简单示例

python

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.store.memory import InMemoryStore

# 1️⃣ Saver: 保存 workflow 状态
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=checkpointer)

# 每次调用自动保存状态
config = {"configurable": {"thread_id": "user_123"}}
app.invoke({"messages": [("user", "你好")]}, config)
app.invoke({"messages": [("user", "再见")]}, config)  # 记住上次对话

# 查看历史状态(Time Travel)
for state in app.get_state_history(config):
    print(state.values)

# 2️⃣ Store: 保存 agent 的长期记忆
store = InMemoryStore()

# 手动存储用户信息
store.put(
    namespace=("users", "user_123"),
    key="profile",
    value={
        "name": "Alice",
        "preferences": {"language": "zh", "theme": "dark"},
        "last_login": "2025-01-13"
    }
)

# 手动读取
profile = store.get(namespace=("users", "user_123"), key="profile")
print(f"用户名: {profile.value['name']}")

# 搜索记忆
active_users = store.search(
    namespace=("users",),
    filter={"last_login": {"$gte": "2025-01-01"}}
)

实际应用场景对比

场景 1: 客服聊天机器人

python

# Saver: 记录对话状态
# - 用户说了什么
# - AI 回复了什么
# - 当前在哪个节点
# - 调用了哪些工具

# Store: 记录用户信息
# - 用户姓名、联系方式
# - 历史订单
# - 偏好设置
# - 常见问题

场景 2: 个人助理 Agent

python

# Saver: 保存当前任务状态
app = graph.compile(checkpointer=PostgresSaver(...))

# 用户: "帮我规划一个日本旅行"
# Saver 自动保存:
# - 当前正在规划日本旅行
# - 已经查询了机票价格
# - 下一步要查酒店

# Store: 保存长期偏好
store.put(
    namespace=("users", "alice"),
    key="travel_preferences",
    value={
        "budget": "中等",
        "喜欢的城市": ["东京", "京都"],
        "避免": ["极限运动"]
    }
)

# 下次对话时,Agent 可以读取这些偏好
prefs = store.get(namespace=("users", "alice"), key="travel_preferences")
# "记得您喜欢东京和京都,我优先推荐这些地方"

何时同时使用?

大多数生产应用需要同时使用两者:

python

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore

# 1. 创建两个持久化系统
checkpointer = PostgresSaver(...)  # 保存 workflow 状态
store = PostgresStore(...)           # 保存长期记忆

# 2. 编译时传入
app = graph.compile(
    checkpointer=checkpointer,  # 自动管理状态
    store=store                  # Agent 可以主动读写
)

# 3. 在节点中使用 Store
def personalized_agent(state, *, store):
    # 读取用户偏好(从 Store)
    user_id = state["user_id"]
    preferences = store.get(("users", user_id), "preferences")

    # 基于偏好生成回复
    response = llm.invoke([
        SystemMessage(f"用户偏好: {preferences.value}"),
        *state["messages"]
    ])

    return {"messages": [response]}

类比理解

想象一个图书馆:

Saver (Checkpointer) = 借阅记录

记录你现在借了哪本书
借到第几页了
什么时候借的
下次来可以接着看

Store = 个人档案

你的姓名、联系方式
你喜欢的书籍类型
你的阅读历史
你的书评和笔记

常见实现对比

实现类	Saver 版本	Store 版本
内存版	MemorySaver	InMemoryStore
SQLite	SqliteSaver	(无官方实现)
PostgreSQL	PostgresSaver	PostgresStore

最佳实践

开发阶段

python

# 都用内存版,快速原型
checkpointer = MemorySaver()
store = InMemoryStore()

生产环境

python

# 都用 PostgreSQL,企业级可靠性
checkpointer = PostgresSaver(connection_string)
store = PostgresStore(connection_string)

数据规划

Saver: 不需要规划,LangGraph 自动管理

Store: 需要设计 namespace 和 schema

python

# 命名空间设计示例
("users", user_id)       → 用户信息
("memories", user_id)    → 对话记忆
("knowledge", domain)    → 知识库

总结对比表

特性	Saver (Checkpointer)	Store
📝 保存什么	Workflow 执行状态	业务数据和长期记忆
🔄 管理方式	LangGraph 自动管理	手动 put/get/search
⏱️ 生命周期	与 thread_id 绑定	永久存储,跨会话
🎯 典型用途	断点续传、Time Travel、多轮对话	用户画像、知识库、偏好设置
🔍 查询方式	get_state(), get_state_history()	get(), search()
🏗️ 生产推荐	PostgresSaver	PostgresStore
📦 示例数据	`{"messages": [...], "current_node": "..."}`	`{"name": "Alice", "preferences": {...}}`

👍 总结:

Saver = "记住对话进度"(短期、自动)
Store = "记住用户信息"(长期、手动)
两者配合使用,打造智能的、有记忆的 Agent!

💡 学习建议

先理解字面意思,再理解技术含义
结合示例代码,动手实践
对比使用场景,加深理解
由浅入深,先掌握核心概念

这些概念相互关联,形成了 LangGraph 的完整生态系统!

LangGraph 术语汇总 ​

1️⃣ 核心架构(Core Architecture) ​

2️⃣ 状态管理(State Management) ​

3️⃣ 消息系统(Message System) ​

4️⃣ 工具系统(Tool System) ​

5️⃣ 记忆与持久化(Memory & Persistence) ​

6️⃣ 人机协同(Human-in-the-Loop) ​

7️⃣ 执行模式(Execution Modes) ​

8️⃣ 高级模式(Advanced Patterns) ​

9️⃣ 记忆架构(Memory Architecture) ​

🔟 生产部署(Production Deployment) ​

1️⃣ Python 技术基础 ​

💡 使用建议 ​

📘 LangGraph 核心概念详细解读 ​

🔵 1. StateGraph (状态图) ​

字面意思 ​

为什么叫 StateGraph? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 2. Node (节点) ​

字面意思 ​

为什么叫 Node? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 3. Edge (边) ​

字面意思 ​

为什么叫 Edge? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 4. State (状态) ​

字面意思 ​

为什么叫 State? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 5. Annotated (带注解的) ​

字面意思 ​

为什么叫 Annotated? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 6. Reducer (归约器/合并器) ​

字面意思 ​

为什么叫 Reducer? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 7. Checkpointer (检查点保存器) ​

字面意思 ​

为什么叫 Checkpointer? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 8. Thread (线程/会话) ​

字面意思 ​

为什么叫 Thread? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 9. Breakpoint (断点) ​

字面意思 ​

为什么叫 Breakpoint? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 10. Tool (工具) ​

字面意思 ​

为什么叫 Tool? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 11. ReAct (推理+行动) ​

字面意思 ​

为什么叫 ReAct? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 12. MessagesState (消息状态) ​

字面意思 ​

为什么叫 MessagesState? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 13. ToolNode (工具节点) ​

字面意思 ​

为什么叫 ToolNode? ​

在 LangGraph 里的作用 ​

简单示例 ​

🔵 14. Subgraph (子图) ​

LangGraph 术语汇总

1️⃣ 核心架构(Core Architecture)

2️⃣ 状态管理(State Management)

3️⃣ 消息系统(Message System)

4️⃣ 工具系统(Tool System)

5️⃣ 记忆与持久化(Memory & Persistence)

6️⃣ 人机协同(Human-in-the-Loop)

7️⃣ 执行模式(Execution Modes)

8️⃣ 高级模式(Advanced Patterns)

9️⃣ 记忆架构(Memory Architecture)

🔟 生产部署(Production Deployment)

1️⃣ Python 技术基础

💡 使用建议

📘 LangGraph 核心概念详细解读

🔵 1. StateGraph (状态图)

字面意思

为什么叫 StateGraph?

在 LangGraph 里的作用

简单示例

🔵 2. Node (节点)

字面意思

为什么叫 Node?

在 LangGraph 里的作用

简单示例

🔵 3. Edge (边)

字面意思

为什么叫 Edge?

在 LangGraph 里的作用

简单示例

🔵 4. State (状态)

字面意思

为什么叫 State?

在 LangGraph 里的作用

简单示例

🔵 5. Annotated (带注解的)

字面意思

为什么叫 Annotated?

在 LangGraph 里的作用

简单示例

🔵 6. Reducer (归约器/合并器)

字面意思

为什么叫 Reducer?

在 LangGraph 里的作用

简单示例

🔵 7. Checkpointer (检查点保存器)

字面意思

为什么叫 Checkpointer?

在 LangGraph 里的作用

简单示例

🔵 8. Thread (线程/会话)

字面意思

为什么叫 Thread?

在 LangGraph 里的作用

简单示例

🔵 9. Breakpoint (断点)

字面意思

为什么叫 Breakpoint?

在 LangGraph 里的作用

简单示例

🔵 10. Tool (工具)

字面意思

为什么叫 Tool?

在 LangGraph 里的作用

简单示例

🔵 11. ReAct (推理+行动)

字面意思

为什么叫 ReAct?

在 LangGraph 里的作用

简单示例

🔵 12. MessagesState (消息状态)

字面意思

为什么叫 MessagesState?

在 LangGraph 里的作用

简单示例

🔵 13. ToolNode (工具节点)

字面意思

为什么叫 ToolNode?

在 LangGraph 里的作用

简单示例

🔵 14. Subgraph (子图)