15 langgraph基本组件

好的，以下是 LangGraph 核心六大组件（1~6） 的详细、深入、结构化说明，剔除了子图与可视化等高级/辅助内容，专注于基础构建块的理解与使用。

1. State（状态）

1.1 定义与作用

State 是整个工作流的共享上下文，所有节点读取和更新它。它决定了“系统当前知道什么”以及“下一步能做什么”。所有节点函数接收 State 作为输入，并返回一个局部更新字典，框架自动将其合并到全局状态中。

1.2 类型定义方式

(1) 使用 `TypedDict`（推荐）


from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]   # 消息历史自动追加
    current_plan: str                         # 当前计划（覆盖式更新）
    tool_attempts: int                        # 工具调用次数

(2) 使用 Pydantic BaseModel（实验性支持）

注意：截至 LangGraph v0.1.x，官方更推荐 TypedDict + Annotated，因底层依赖对字典操作优化更好。

1.3 字段更新语义（Reducer）

LangGraph 通过 reducer 函数决定如何合并节点返回值到状态：

更新方式	写法	行为
覆盖（默认）	`field: str`	新值直接替换旧值
追加列表	`Annotated[list, operator.add]`	`state['field'] += new_value`
自定义 reducer	`Annotated[MyType, my_reducer]`	调用 `my_reducer(old, new)`

✅ 示例：messages 字段通常用 operator.add，确保多轮对话消息累积而非丢失。

1.4 最佳实践

状态应尽量扁平，避免深层嵌套（便于调试和 reducer 设计）。所有节点必须只通过返回 dict 修改状态，禁止直接修改传入的 state 对象（因可能被缓存或并行处理）。

2. Node（节点）

2.1 基本概念

节点是纯函数（或可调用对象），代表一个逻辑步骤。输入：当前完整 State输出：dict，表示对状态的增量更新

2.2 节点函数签名


def my_node(state: StateType) -> dict:
    # 读取 state
    last_msg = state["messages"][-1]

    # 执行逻辑（LLM、工具、计算等）
    result = some_action(last_msg)

    # 返回局部更新
    return {"messages": [result], "attempts": state.get("attempts", 0) + 1}

2.3 节点类型示例

节点类型	功能	返回示例
LLM Agent	调用大模型生成响应	`{"messages": [AIMessage(...)]}`
Tool Executor	执行工具并返回结果	`{"messages": [ToolMessage(content="...")]}`
Human-in-the-loop	等待人工输入（需外部注入）	通常由 stream 或 interrupt 机制处理
Validator	检查输出是否合规	`{"is_valid": True, "feedback": ""}`

2.4 注意事项

节点不应有副作用（如直接打印、写文件），除非明确需要（此时应封装在工具中）。节点不能直接跳转，跳转由边（Edge）控制。节点名称在图中必须唯一。

3. Edge（边）

3.1 边的本质

控制执行流的方向：从一个节点到另一个节点（或结束）。不包含业务逻辑，仅负责路由。

3.2 三种边类型详解

(1) 静态边 `add_edge(from, to)`

无条件跳转。常用于线性流程或循环回跳。


graph.add_edge("tool_executor", "agent")  # 工具执行完回到 agent

(2) 条件边 `add_conditional_edges(source, router, path_map)`

核心机制：实现动态决策。router 函数：接收 State，返回一个路由键（route key）。path_map：将路由键映射到目标节点名（或 END）。


def route_agent(state: AgentState) -> str:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "call_tool"
    return "end"

graph.add_conditional_edges(
    "agent",
    route_agent,
    {"call_tool": "tool_node", "end": END}
)

⚠️ router 返回值必须是 path_map 中的 key，否则会报错。

(3) 入口与出口

set_entry_point(node_name)：指定图启动时第一个执行的节点。END：特殊终止节点，流程在此结束，返回最终状态。

3.3 边的设计原则

避免环路无限循环：确保有退出条件（如最大尝试次数）。路由函数应轻量：不要在 router 中做耗时操作（如调用 LLM）。

4. StateGraph（状态图构建器）

4.1 核心职责

提供 API 用于声明式构建图结构。绑定状态类型、节点、边。

4.2 关键方法

方法	说明
`__init__(state_schema)`	初始化，传入 `TypedDict` 类型
`add_node(name: str, action)`	注册节点（name 必须唯一）
`add_edge(start, end)`	添加静态边
`add_conditional_edges(source, router, path_map)`	添加条件边
`set_entry_point(name)`	设置入口节点
`set_finish_point(name)`	（可选）设置显式结束节点（通常用 `END`）
`compile(checkpointer=None)`	编译为可执行应用

4.3 构建顺序建议

定义 State创建 StateGraph(State)add_node() 所有节点set_entry_point()添加边（先静态，后条件）compile()

❌ 错误：在 compile() 后再修改图结构 —— 编译后图是不可变的。

5. Compiled Graph（编译后的应用）

5.1 是什么？

StateGraph.compile() 返回的对象，是可执行的工作流引擎。封装了状态管理、执行调度、检查点等运行时逻辑。

5.2 执行接口

(1) `.invoke(input_state, config=None)`

同步执行，阻塞直到完成。返回最终状态。


final_state = app.invoke({"messages": [HumanMessage("你好")]})

(2) `.ainvoke(...)`

异步版本，适用于 async 环境。

(3) `.stream(input_state, config=None)`

流式执行：每完成一个节点就 yield 一次。返回格式：{"node_name": partial_update}


for chunk in app.stream({"messages": [...]}):
    print(chunk)
# 输出示例：
# {'agent': {'messages': [AIMessage(...)]}}
# {'tool_node': {'messages': [ToolMessage(...)]}}

✅ 适合前端实时渲染 Agent 思考过程。

(4) 支持中断与恢复（需 checkpointer）

可在特定节点设置中断点（通过自定义逻辑抛出 Interrupt）。恢复时传入相同 thread_id 即可继续。

5.3 配置参数 `config`

主要用于传递 checkpointer 上下文：


config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke(initial_state, config)

thread_id 是会话标识，不同 ID 对应独立状态历史。

6. Checkpointer（检查点机制）

虽然你要求只讲 1~6，但第 6 项“Compiled Graph”天然关联 Checkpointer，因其是运行时关键能力，故在此详述。

6.1 作用

持久化状态历史，实现：
多轮对话记忆中断后恢复会话回溯（审计、重试）并发会话隔离（通过 thread_id）

6.2 内置实现

类型	说明
`MemorySaver`	内存存储（默认，不持久）
`SqliteSaver`	本地 SQLite（开发测试）
`PostgresSaver`	PostgreSQL（生产推荐）
`RedisSaver`	Redis（高性能场景）

6.3 使用方式


from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string("checkpoints.sqlite")
app = workflow.compile(checkpointer=memory)

# 执行时绑定会话
config = {"configurable": {"thread_id": "session_abc"}}
app.invoke({"messages": [...]}, config)

6.4 状态版本控制

每次节点执行都会生成一个新版本的状态快照。可通过 checkpointer.get_tuple(config) 获取历史版本。支持时间旅行式调试。

6.5 注意事项

必须提供 thread_id，否则所有调用共享同一状态（通常不是你想要的）。检查点数据包含完整状态 + 元数据（时间戳、版本号等）。

组件协同工作流程总结


flowchart LR
    A[State Schema] --> B[StateGraph]
    B --> C[Add Nodes]
    C --> D[Add Edges]
    D --> E[Compile → App]
    E --> F[Invoke / Stream]
    F --> G[Checkpointer ←→ Persistent State]
    G --> H[Final or Intermediate State]

State 定义数据契约；Node 实现业务逻辑；Edge 控制执行路径；StateGraph 组装蓝图；Compiled App 执行流程；Checkpointer（可选但重要）提供状态持久化与会话管理。

这些组件共同构成了 LangGraph 的核心抽象，使得开发者能够以声明式、模块化、可测试的方式构建复杂的 AI 工作流。掌握它们的细节，是高效使用 LangGraph 的基础。