LangGraph Core

LangGraph Core 是 LangGraph 框架的核心库，用于构建可状态化、多参与者（multi-actor）的 LLM 应用。它将 LLM 应用中的步骤（steps）建模为图（graph）中的节点（nodes）和边（edges）。

LangGraph 的设计灵感来自 Pregel 和 Apache Beam。公共接口受 NetworkX 启发。

主要特性

• 状态管理 (State)：为你的应用维护和更新一个共享状态。
• 节点 (Nodes)：图的构建模块，可以是 Python 函数或可运行对象。
• 边 (Edges)：定义节点之间的控制流和数据流。
• 条件分支 (Conditional Branching)：基于当前状态在节点之间路由。
• 持久化 (Persistence)：内置支持跨节点执行检查点（checkpoint）和恢复状态。
• 流式处理 (Streaming)：支持实时输出节点结果。
• 人机交互 (Human-in-the-loop)：支持暂停执行并等待人工输入。
• 图编译/执行 (Graph Compilation/Execution)：将图定义编译为可执行对象。

安装

pip install langgraph-core

快速入门

1. 定义状态

状态是贯穿整个图结构的共享数据结构。

from typing import TypedDict

class GraphState(TypedDict):
    messages: list[str]
    next_agent: str

2. 定义节点

节点是执行工作的 Python 函数。每个函数接收当前状态并返回一个更新后的状态字典。

def agent_a(state: GraphState) -> dict:
    # 此处执行工作...
    return {"messages": state["messages"] + ["来自 Agent A 的消息"]}

def agent_b(state: GraphState) -> dict:
    return {"messages": state["messages"] + ["来自 Agent B 的消息"]}

def router(state: GraphState) -> dict:
    # 基于状态的路由逻辑
    return {"next_agent": "agent_b"}

3. 构建图

使用 StateGraph 类构建图，并通过 add_node 和 add_edge 方法连接节点。

from langgraph.graph import StateGraph, END

# 创建图构建器
builder = StateGraph(GraphState)

# 添加节点
builder.add_node("agent_a", agent_a)
builder.add_node("agent_b", agent_b)
builder.add_node("router", router)

# 设置入口点
builder.set_entry_point("agent_a")

# 添加边
builder.add_edge("agent_a", "router")
builder.add_conditional_edges(
    "router",
    lambda state: state["next_agent"],
    {
        "agent_b": "agent_b",
        "END": END
    }
)
builder.add_edge("agent_b", END)

4. 编译并执行

编译图后会生成一个可调用对象。

# 编译图
graph = builder.compile()

# 初始状态
initial_state = {"messages": [], "next_agent": ""}

# 执行图
for output in graph.stream(initial_state):
    # 流式输出结果
    for node_name, state_update in output.items():
        print(f"节点: {node_name}")
        print(f"更新: {state_update}")

高级用法

条件边 (Conditional Edges)

基于状态决定执行路径：

from langgraph.graph import StateGraph

def is_satisfied(state):
    return state.get("satisfied", False)

builder = StateGraph(MyState)
builder.add_conditional_edges(
    "check_node",
    is_satisfied,
    {
        True: "success_node",
        False: "retry_node"
    }
)

使用检查点进行持久化

启用检查和恢复：

from langgraph.checkpoint import MemorySaver

# 使用内存检查点保存器
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 执行时，状态会被自动保存检查点
config = {"configurable": {"thread_id": "1"}}
graph.invoke(initial_state, config)

# 从之前的检查点恢复执行
graph.invoke(None, config)  # 传入 None 表示从检查点恢复

人机交互

通过中断交互等待用户输入：

graph = builder.compile(interrupt_before=["human_input_node"])

# 执行直到中断
graph.invoke(initial_state)

# 用户提供输入并继续
graph.invoke({"human_input": "用户输入"})

API 参考

StateGraph

• __init__(state_schema): 使用状态模式初始化图。
• add_node(name, action): 添加节点。
• add_edge(source, target): 添加普通边。
• set_entry_point(name): 设置入口点。
• set_conditional_edge(source, condition, mapping): 设置条件边。
• compile(): 编译并返回可执行的图。
• invoke(input): 使用输入运行图。
• stream(input): 流式输出图执行结果。

MessageGraph

MessageGraph 是 StateGraph 的特化版本，其中的状态默认是一系列消息的列表。

from langgraph.graph import MessageGraph

builder = MessageGraph()
# 节点直接操作消息列表

示例

请参阅我们的 示例目录 了解更多使用案例。

贡献

欢迎贡献！请查看我们的 贡献指南 了解详情。

许可证

该项目使用 MIT 许可证 - 详情请参阅 LICENSE 文件。