LangGraph

LangGraph 是一个用于构建多步骤、智能体化应用的库,它将执行流程建模为一个状态图, 而非一条简单的线性链。

什么是 LangGraph?

LangGraph 构建在与 LangChain 相同的组件之上(模型、提示词、工具), 但它不再用 | 进行线性组合,而是允许你将执行流程定义为一个由节点(Node)和边(Edge)构成的图—— 包括循环、条件分支,以及返回到之前的步骤。这正是真正实现 ReAct 模式所需要的。

为什么用图而不是线性链?

一条线性链(Chain)假设步骤 A 总是流向 B,再流向 C。但一个真实智能体的行为并非如此:调用某个工具之后, 结果可能不充分,智能体需要回到"决策"步骤(这是一个循环);或者根据用户请求的类型不同,可能走上完全不同的路径 (这是一个条件分支)。图能够自然地对这些模式建模。

核心概念

• State — 一个在所有节点之间传递并被更新的共享对象(相当于图的工作记忆)
• Node — 一个处理单元(一个 Python 函数或一条 LangChain 链),读取 State 并返回更新后的版本
• Edge — 两个节点之间的路径;可以是固定的,也可以是条件性的(根据 State 决定下一个节点)

一个示例图的示意图

代码示例

from langgraph.graph import StateGraph, END

def reason(state):
    # 模型决定:给出最终答案,还是调用工具
    return {"next": "tool" if state["needs_tool"] else END}

def call_tool(state):
    result = tools[state["tool_name"]](**state["tool_args"])
    return {"last_result": result, "needs_tool": False}

graph = StateGraph(dict)
graph.add_node("reason", reason)
graph.add_node("tool", call_tool)
graph.add_conditional_edges("reason", lambda s: s["next"])
graph.add_edge("tool", "reason")
graph.set_entry_point("reason")

app = graph.compile()
app.invoke({"needs_tool": True, "tool_name": "check_inventory", "tool_args": {"sku": "123"}})

持久化、检查点与人工介入

LangGraph 的一大特色能力,是在每个节点运行后自动保存 State(即检查点,Checkpoint)。这意味着一个图的执行 可以在中途暂停,之后——即使应用程序完全重启后——也能从完全相同的位置继续,因为整个 State 存储在数据库中 (例如 Postgres 或 Redis),而不仅仅存在于应用程序的内存里。

这一机制有两个重要用途:

• 人工介入(Human-in-the-loop) — 图可以在执行敏感操作(例如最终付款)之前有意暂停,等待人工批准,批准后再从完全相同的位置继续——不会丢失之前的任何步骤。
• 错误恢复 — 如果服务器在一次长时间运行的执行过程中崩溃,重新启动后,执行会从上一个成功的检查点继续,而不是从零开始。

LangGraph 与普通 LangChain 的对比

常见问题