LangGraph 函数字典：从入门到入魂，一网打尽所有常用 API

引言

本文是 LangGraph 的“新华字典”，收录了所有你可能会用到的函数、类和方法。
每个函数都配有 代码示例 和 真实使用场景，让你不仅知道“怎么用”，更知道“什么时候用”。
风格：CSDN 老司机带路，幽默不废话，干货拉满！

🎯 写在前面：LangGraph 的函数体系

LangGraph 的函数可以分为几大家族：

• 图构建家族：StateGraph、add_node、add_edge、add_conditional_edges

• 状态管理家族：Annotated、add_messages、自定义 Reducer

• 执行家族：invoke、stream、ainvoke、astream

• 持久化家族：MemorySaver、SqliteSaver、Checkpointer

• 高级特性家族：interrupt_before/after、get_state、update_state

下面我们一个一个盘！

第一章：图构建函数 —— 搭骨架的“钢筋水泥”

StateGraph —— 一切的起点

函数签名：

class StateGraph(StateType):
    def __init__(self, state_schema: Type[StateType])

作用：创建一个状态图对象，是所有 LangGraph 应用的“地基”。
参数：state_schema 是你定义的状态结构（TypedDict 或 Pydantic Model）。

代码示例：

from typing import TypedDict
from langgraph.graph import StateGraph

class MyState(TypedDict):
    counter: int
    messages: list

graph = StateGraph(MyState)  # 开工！

使用场景：每次你要构建一个新的 Agent 或工作流，第一行代码就是它。

调侃：它就像你买了个毛坯房，后面的 add_node 就是在砌墙、装门。

add_node —— 给图加“动作演员”

函数签名：

graph.add_node(name: str, action: Callable[[State], dict])

作用：向图中添加一个节点。节点就是一个普通的 Python 函数，接收当前状态，返回要更新的状态字段。
参数：

• name：节点的名字，后面跳转时要用

• action：节点函数，输入 State，输出 dict（部分状态更新）

代码示例：

def increment_counter(state: MyState):
    return {"counter": state["counter"] + 1}

graph.add_node("increment", increment_counter)

使用场景：每个“动作”都是一个节点。比如“调用 LLM”、“执行工具”、“人工审核”，都可以拆成独立的节点。

调侃：节点就像公司里的员工，各干各的活，互不干扰。

add_edge —— 连接节点，规划“剧情走向”

函数签名：

graph.add_edge(start: str, end: str)

作用：添加一条从 start 节点到 end 节点的固定边。执行完 start 后，一定走 end。
参数：

• start：源节点名

• end：目标节点名（可以是 END）

代码示例：

from langgraph.graph import END

graph.add_edge("increment", "print_result")   # 加完数就去打印
graph.add_edge("print_result", END)           # 打印完就结束

使用场景：当流程是确定的、没有分支的时候，用固定边。比如“先 A 后 B 再结束”。

调侃：这就是你每天上班的固定路线：家 → 地铁 → 公司 → 家。

add_conditional_edges —— 给剧情加“选择题”

函数签名：

graph.add_conditional_edges(
    source: str,
    path: Callable[[State], str],
    path_map: Optional[dict] = None
)

作用：根据当前状态，动态决定下一个节点。
参数：

• source：源节点

• path：一个函数，输入 State，返回一个字符串（节点名）

• path_map：可选，把返回值映射到实际节点名

代码示例：

def should_continue(state: MyState):
    if state["counter"] > 10:
        return "end"
    else:
        return "increment"

graph.add_conditional_edges(
    source="increment",
    path=should_continue,
    path_map={"end": END, "increment": "increment"}  # 可选
)

使用场景：经典场景是 ReAct Agent 的判断——“如果 AI 想用工具，就走工具节点；否则结束”。

调侃：这就是你妈问你“考了多少分”——100 分奖励玩具，60 分奖励巴掌。

set_entry_point —— 指定“第一幕从哪开始”

函数签名：

graph.set_entry_point(node_name: str)

作用：设置图的入口节点。
参数：node_name 是你想最先执行的节点名。

代码示例：

graph.set_entry_point("increment")  # 从 increment 开始

使用场景：每个图必须有且只有一个入口点，不然图不知道从哪开始跑。

调侃：这就像电影的开场画面，没它观众一脸懵逼。

**compile** —— 把“图纸”变成“可执行程序”

函数签名：

app = graph.compile(checkpointer: Optional[Checkpointer] = None, interrupt_before: Optional[list] = None)

作用：编译图，返回一个可运行的 Runnable 对象。编译后才能调用 invoke 或 stream。
参数：

• checkpointer：持久化检查点（后面细讲）

• interrupt_before/after：在哪些节点前后中断（人机协同）

代码示例：

使用场景：图搭完后，最后一步就是编译。不编译就像没装引擎的汽车。

调侃：编译后的 app 就是你的“最终产品”，可以上线卖了。

第二章：状态管理函数 —— 让数据“活”起来

**Annotated** + 自定义 Reducer —— 控制状态如何合并

函数签名：

from typing import Annotated

class MyState(TypedDict):
    messages: Annotated[list, add_messages]  # 内置 reducer
    counter: Annotated[int, lambda left, right: left + right]  # 自定义

作用：给状态字段加一个“合并规则”。当多个节点返回同一个字段时，决定是覆盖还是累加。
关键点：

• add_messages：内置 reducer，自动把新消息追加到列表末尾

• 自定义 reducer：你可以写任意合并逻辑

代码示例：

from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]  # 追加消息
    total_tokens: Annotated[int, lambda x, y: x + y]  # 累加 token

使用场景：

• messages 必须用追加，否则会覆盖历史对话

• 计数器、总 token 数等用累加

• 用户配置用覆盖（只保留最新）

调侃：没有 Reducer 的状态就像没有规矩的会议室，谁都能把白板擦掉。

第三章：执行函数 —— 让图“跑起来”

**invoke** —— 同步执行，拿最终结果

函数签名：

result = app.invoke(input: dict, config: Optional[dict] = None)

作用：同步执行整个图，返回最终状态。
参数：

• input：初始状态（字典）

• config：配置，如 {"configurable": {"thread_id": "xxx"}}

代码示例：

result = app.invoke({"messages": [HumanMessage(content="你好")]})
print(result["messages"][-1].content)

使用场景：当你不关心中间步骤，只要最终答案时用。

调侃：这就像吃火锅，你只管最后捞肉，中间涮的过程不关心。

**stream** —— 流式执行，看中间步骤

函数签名：

for chunk in app.stream(input: dict, config: Optional[dict] = None, stream_mode: str = "values"):
    print(chunk)

作用：逐步执行，每完成一个节点就 yield 一个结果。
stream_mode 参数：

• "values"：yield 当前完整状态

• "updates"：只 yield 本次节点返回的更新（差异）

代码示例：

for step in app.stream({"counter": 0}, stream_mode="updates"):
    print(step)  # 输出：{'increment': {'counter': 1}} 等

使用场景：

• 调试：想看每个节点的输出

• UI 实时更新：比如 Agent 思考过程逐条显示

• 长任务：提前给用户反馈

调侃：这就像直播带货，每一步都给你看，不像 invoke 是录播。

**ainvoke** / **astream** —— 异步版本

函数签名：

作用：与同步版相同，但支持 async/await，适合 FastAPI、异步 Web 应用。

使用场景：你的 Web 服务用了 asyncio，或者想在事件循环里调用。

第四章：持久化函数 —— 让 Agent 拥有“记忆”

**MemorySaver** / **SqliteSaver** —— 保存状态

函数签名：

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

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

作用：自动保存每个节点执行后的状态，支持通过 thread_id 恢复会话。
区别：

• MemorySaver：存内存，程序重启就丢

• SqliteSaver：存 SQLite 文件，永久保存

代码示例：

# 使用 SQLite 持久化
with SqliteSaver.from_conn_string("checkpoints.db") as saver:
    app = graph.compile(checkpointer=saver)

config = {"configurable": {"thread_id": "user_123"}}

# 第一轮对话
app.invoke({"messages": [HumanMessage(content="我叫小明")]}, config)

# 第二轮对话（记住我叫小明）
app.invoke({"messages": [HumanMessage(content="我叫什么？")]}, config)
# 输出：你叫小明

使用场景：

• 多轮对话机器人

• 长任务（如数据清洗）中断后恢复

• 调试：回溯到某个节点状态

调侃：没有 checkpointer 的 Agent 是金鱼，7 秒记忆。加上它，就是大象。

**interrupt_before** / **interrupt_after** —— 人机协同

函数签名：

app = graph.compile(
    checkpointer=saver,
    interrupt_before=["tools"],  # 进入 tools 节点前暂停
    interrupt_after=["send_email"]  # 执行完 send_email 后暂停
)

作用：在指定节点前后自动中断，等待人工干预后继续。

恢复执行：

# 检查状态
state = app.get_state(config)

# 人工确认后继续
app.invoke(None, config)

使用场景：

• 敏感操作审批（发送邮件、转账、删除数据）

• 人工纠错（AI 生成的代码需要人审核）

• 人机协作的客服系统

调侃：这就像给 AI 配了个“领导审批”环节，防止它胡来。

**get_state** / **update_state** —— 时间旅行

函数签名：

# 获取当前状态
state = app.get_state(config)

# 获取历史快照
history = list(app.get_state_history(config))

# 回滚到某个历史状态
app.update_state(config, history[-2].values)

作用：

• get_state：获取指定 thread 的当前状态

• get_state_history：获取所有历史快照（时间旅行）

• update_state：手动更新状态（可用于修正错误）

使用场景：

• 调试：回退到出错前，修改输入重试

• 人工修正：AI 判断错了，人手动改状态后继续

• 审计：追溯 Agent 的每一步决策

调侃：这是 LangGraph 的“后悔药”，吃了能回到过去。

第五章：可视化与辅助函数 —— 让你“看到”图

**get_graph** —— 画图

函数签名：

# 生成 Mermaid 格式
graph_representation = app.get_graph().draw_mermaid()

# 保存为 PNG（需要安装 pygraphviz）
with open("graph.png", "wb") as f:
    f.write(app.get_graph().draw_mermaid_png())

作用：生成图的可视化表示，支持 Mermaid 和 PNG。
使用场景：

• 给老板/客户展示你的 Agent 架构

• 调试：检查节点连接是否正确

• 文档：自动生成流程图

调侃：这比手画流程图快多了，还能随代码自动更新。

**START** / **END** —— 特殊节点

函数签名：

from langgraph.graph import START, END

graph.add_edge(START, "first_node")
graph.add_edge("last_node", END)

作用：

• START：虚拟的入口点，用来简化边的添加

• END：虚拟的结束点，表示图执行完毕

代码示例：

# 这样写更简洁
graph.add_edge(START, "node_a")
graph.add_edge("node_a", END)

使用场景：几乎所有图都会用到，尤其是用 add_edge 替代 set_entry_point 时 。

📋 速查表：一图看懂所有函数

函数	作用	一句话总结
StateGraph	创建图对象	买个毛坯房
add_node	加节点	砌墙
add_edge	加固定边	铺直路
add_conditional_edges	加条件边	铺岔路
set_entry_point	设入口	定大门
compile	编译图	装引擎
invoke	同步执行	一次跑完
stream	流式执行	一步步跑
MemorySaver	内存持久化	临时记忆
SqliteSaver	SQLite持久化	永久记忆
interrupt_before/after	中断点	设红绿灯
get_state	获取状态	看当前
update_state	更新状态	改历史
get_graph	可视化	画图纸

🎁 附赠：官方资源导航

• API 参考文档：https://langchain-ai.github.io/langgraph/reference/

• GitHub 示例库：https://github.com/langchain-ai/langgraph/tree/main/examples

• LangGraph Studio：https://langchain-ai.github.io/langgraph/studio/（可视化调试工具）

🏁 写在最后

这篇文章就像一本字典，不需要一次性读完。
当你写代码时，忘了某个函数怎么用，回来翻翻就行。

使用建议：

• 第一次接触：看“图构建家族”，跑通第一个例子

• 做多轮对话：看“持久化家族”

• 做人机协同：看 interrupt_before 和 update_state

• 调试复杂 Agent：看 get_graph 和 get_state_history

最后送大家一句话：

LangGraph 的 API 不难，难的是把业务流程抽象成图。
但只要画得出流程图，就能用 LangGraph 写出来。