第 5 章 Node 与函数式计算单元

5.1 Node 的本质

Node（节点）是图中的最小计算单元，本质上是一个可调用对象（Callable）：可以是普通 Python 函数、实现了 __call__ 的类实例，或任何符合 Runnable 协议的对象（如 LangChain 的 Chain、RunnableLambda）。节点接收当前 State（以及可选的运行时配置），返回一个对 State 的增量更新字典，运行时在超步末按 Channel 规则合并。

5.2 函数签名与类型

标准的节点函数签名：

from typing import Any
from langchain_core.runnables import RunnableConfig

def my_node(state: MyState, config: RunnableConfig | None = None) -> dict[str, Any]:
    # 读取 state，执行业务逻辑
    result = do_work(state["query"])
    # 返回增量字典，只包含需要更新的字段
    return {"messages": [AIMessage(content=result)], "attempts": state["attempts"] + 1}

• 第一个参数 state：当前超步开始时的只读状态快照，类型应与声明的 StateSchema 一致，编辑器据此提供自动补全；
• 第二个参数 config：可选，类型为 RunnableConfig，包含 configurable（用户在 invoke 时传入的自定义字典）、metadata、tags、recursion_limit、run_id 等运行时信息；
• 返回值：dict。返回 None 或空 dict 表示不更新任何字段；只返回需要变更的字段即可，不必回传完整 State。

节点支持同步函数（def）与异步函数（async def）。异步节点在 ainvoke/astream 执行时会被直接调度到 asyncio 事件循环，并发效率最高；同步节点会通过 run_in_executor 包装到线程池执行，避免阻塞事件循环。

5.3 节点分类

按功能常见的节点有五类：

• 普通函数节点：纯 Python 逻辑，如数据清洗、格式转换、规则判定；
• LLM 节点：内部调用大模型。推荐直接返回 {"messages": [resp]}，配合 add_messages reducer；
• 工具执行节点：执行 Tool/Function Calling 的节点，官方预构建了 ToolNode（langgraph-prebuilt），负责解析模型的 tool_calls、调用对应工具、回写 tool 消息；
• 路由节点（Supervisor）：本身不做重业务，只读取 State 并返回路由信号（或直接返回 Command，见第 6 章）；
• 子图节点：将一张编译后的子图作为节点加入父图，用于模块化复用（第 10 章详述）。

5.4 注册、命名与返回值约束

通过 StateGraph.add_node 注册节点：

builder.add_node("generate_joke", generate_joke)

命名约束：

• 节点名在图内必须唯一；
• 不能与保留字 "__start__"、"__end__"（即常量 START、END）重名；
• 建议使用 snake_case，便于在 trace 与 Studio 中阅读。

返回值约束：

• 必须是 dict（或第 6 章将介绍的 Command 对象）；
• 只返回需要更新的字段。返回完整 State 在功能上可行，但会让语义模糊——看代码无法区分"字段真的更新了"与"顺手带上了旧值"，同时增加序列化开销；
• 对 LastValue 字段，返回值即新值；对带 reducer 的字段，返回值作为 new 传入 reducer 与旧值合并；
• 节点不允许直接修改 state 对象内的可变字段（list/dict），这些修改不会被运行时感知。

5.5 运行时资源访问

在节点内部可通过以下方式访问运行时资源：

• config["configurable"]：调用方在 invoke(input, config={"configurable": {...}}) 传入的自定义字典，可承载用户 ID、会话 ID、业务上下文等；
• get_store()：获取跨线程共享的键值 Store，适合存长期记忆、知识库引用；
• get_writer() / get_stream_writer()：获取自定义流式写入器，节点可在执行过程中向客户端推送自定义事件（如进度条、中间思考片段），不必等到节点返回。

from langgraph.config import get_store, get_stream_writer

def my_node(state: MyState, config: RunnableConfig) -> dict:
    store = get_store()          # BaseStore 实例
    writer = get_stream_writer() # 自定义事件流
    # 示例：推送进度
    writer({"event": "progress", "step": "retrieving"})
    profile = store.get(("profiles",), key=state["user_id"])
    return {"profile": profile.value if profile else None}

高级注入：编写供 LLM 调用的工具函数时，可使用 InjectedState、InjectedStore 注解让 LangGraph 自动注入 State/Store，工具函数本身只暴露业务参数给模型：

from langgraph.prebuilt import InjectedState, InjectedStore
from langchain_core.tools import tool
from typing import Annotated

@tool
def search_profile(
    user_id: str,
    state: Annotated[MyState, InjectedState],
    store: Annotated[BaseStore, InjectedStore],
) -> str:
    ...

5.6 重试机制

节点支持声明式重试，通过 retry 参数传入 RetryPolicy：

from langgraph.errors import RetryPolicy
from datetime import timedelta

builder.add_node(
    "call_external_api",
    call_external_api,
    retry=RetryPolicy(
        max_attempts=3,
        initial_interval=timedelta(seconds=1),
        jitter=True,
        retry_on=(ConnectionError, TimeoutError),
    ),
)

重试在节点粒度生效：每次重试会以相同的入参 state 重新执行节点函数（同属一个超步，不产生新快照），节点内部产生的副作用（如已发出的 HTTP 请求）需要调用方自己保证幂等。

本章小结

Node 是接收 State、返回增量 dict 的可调用对象，支持同步/异步两种形态，分为普通函数、LLM、工具、路由、子图五类。节点通过 add_node 注册并命名，命名需唯一且避开保留字；节点内通过 config、get_store、get_stream_writer 访问运行时资源；通过 RetryPolicy 可声明节点级重试；节点严禁 in-place 修改 state，异步节点内严禁使用阻塞 IO。

常见错误

1. 在节点内直接修改 state。现象：list/dict 字段的变更不生效或偶发生效。原因：state 是只读快照，in-place 修改不经过 reducer。修复：将修改结果作为返回值。
2. 返回整个 State 覆盖其他字段。现象：节点返回完整 State 后，其他字段被意外覆盖。原因：对 LastValue 字段，返回值会直接替换旧值；带 reducer 字段若 reducer 不是幂等合并也可能异常。修复：只返回需要更新的字段。
3. 异步节点内使用同步阻塞 IO。现象：图整体并发度低，慢节点拖垮整个超步。原因：requests.get、time.sleep 等同步调用会阻塞事件循环。修复：改用 httpx.AsyncClient、asyncio.sleep 等异步原生 API。
4. 在节点中启动新的事件循环。现象：RuntimeError: This event loop is already running 或嵌套 loop 异常。原因：在已有事件循环中调用 asyncio.run 或 loop.run_until_complete。修复：节点直接 await 异步函数，避免手动管理 loop。
5. 节点抛出未捕获异常。现象：整图中断，后续节点不再执行。原因：节点内未做业务异常兜底。修复：对可恢复异常在节点内捕获并返回错误字段，对不可恢复异常才抛出，由运行时终止本次执行。

生产建议

• 节点保持单一职责：业务节点与路由节点分离，便于单测与替换；
• LLM 节点统一返回 {"messages": [AIMessage(...)]} 格式，配合 MessagesState，全图消息路径一致；
• 所有外部 IO（模型、HTTP、数据库）优先使用 async 实现，并为关键节点配置合理的重试与退避策略；
• 节点函数编写时严格遵守"只读 state、只返回增量"，CI 中使用 lint 规则禁止 in-place 修改（如禁止对 state 字段调用 .append/.update）；
• 自定义流式事件通过 get_stream_writer 推送，避免在节点内持有全局 socket/queue；
• 对会产生副作用的节点（扣库存、发通知）显式设计幂等键（基于 thread_id + superstep 版本号），避免重试导致重复执行。

---