节点 Node

上一篇讲了 State 是共享数据容器，这一篇讲真正"干活"的执行单元——节点（Node）。节点是 LangGraph 里最朴素的部件，但也是你写代码时打交道最多的地方。

一、节点的本质

节点就是普通 Python 函数。仅此而已。

def my_node(state: State) -> dict:
    # 读 state
    query = state["user_query"]
    # 干点啥
    answer = f"回答: {query}"
    # 返回要更新的字段
    return {"answer": answer}

签名约定：

• 输入：state: State，当前状态的快照。
• 输出：dict，要更新的字段子集。框架会用 reducer 把它合并进状态。
• 也可以返回完整 state：dict 里包含所有字段也行，行为和只返回子集一样（按字段用 reducer 合并）。

LangGraph 不要求节点继承某个类、不要求实现某个接口——任何 Callable[[State], dict] 都能当节点。这点和 LangGraph4j 不同，Java 那边要 implements AgentNode 或传 lambda。

二、节点的返回值如何影响 State

回顾 状态 State 的机制：节点返回的 dict 是"差量"。

class State(TypedDict):
    a: str
    b: Annotated[int, add]
    c: str   # 这个字段不被节点更新

def node(state: State) -> dict:
    return {"a": "新值", "b": 5}  # c 不出现

执行后：

• a 被覆盖为 "新值"（默认策略）
• b 加 5（reducer add）
• c 保持不变（节点没返回它）

返回 dict 里没出现的字段 = 不动，这是新手最常踩的坑之一。如果想清空某字段，显式返回 {"a": ""} 或 {"a": None}（要看 reducer 怎么处理 None）。

三、START 和 END 是特殊节点

LangGraph 内置两个特殊节点：

• START：图的入口。所有图的执行从 START 出发。你必须 add_edge(START, "某节点") 把入口接上。
• END：图的出口。某条边指向 END 表示"图结束"。

from langgraph.graph import StateGraph, START, END

g.add_edge(START, "first")   # 入口接到 first
g.add_edge("last", END)      # last 节点执行完结束

START 和 END 不是函数，是常量标识符。你不能 add_node(START, fn)——它们是图的"边界"。

四、节点里能做什么

节点的内部逻辑完全自由。常见任务：

1. 调 LLM：用 ChatOpenAI 之类生成回答。
2. 调工具：搜索、查数据库、调 API、发邮件。
3. 读写 IO：读文件、写日志、写数据库。
4. 纯计算：做数据处理、聚合、格式化。
5. 路由判断：基于 state 计算下一步该去哪（配合条件边）。
6. 调外部服务：调用第三方 SDK。
7. 人机交互：调用 interrupt() 暂停等用户输入。

def call_llm(state: State) -> dict:
    """节点：调 LLM 生成回答。"""
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(state["messages"])
    return {"messages": [response]}  # add_messages 会追加

def search_db(state: State) -> dict:
    """节点：查数据库。"""
    docs = my_db.query(state["user_query"])
    return {"retrieved_docs": docs}

def format_answer(state: State) -> dict:
    """节点：纯计算，格式化答案。"""
    docs = state["retrieved_docs"]
    answer = "\n".join(f"- {d}" for d in docs)
    return {"answer": f"找到 {len(docs)} 条结果：\n{answer}"}

五、单一职责原则

虽然节点能干任何事，但推荐每个节点只做一件事。原因：

• 可测试：单一职责的函数好写单测。
• 可复用：一个"格式化"节点能在多个图里复用。
• 可调试：出问题时一眼定位是哪一步。
• 可组合：小节点能像积木一样拼出复杂流程。

反例（不推荐）：

def do_everything(state: State) -> dict:
    # 一个节点又查库又调 LLM 又格式化又写日志——别这样
    docs = db.query(...)
    resp = llm.invoke(...)
    log.write(...)
    return {...}

正例（推荐）：

def retrieve(state): ...
def generate(state): ...
def log_result(state): ...

六、完整示例：多节点协作

下面写一个"查询-生成-润色"三节点协作图：

# nodes_demo.py
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI


class State(TypedDict):
    user_query: str
    retrieved: list[str]
    answer: str
    messages: Annotated[list, add_messages]


llm = ChatOpenAI(model="gpt-4o-mini")


def retrieve(state: State) -> dict:
    """节点1：模拟检索。"""
    query = state["user_query"]
    # 这里假装查到了两篇文档
    docs = [f"关于《{query}》的资料A", f"关于《{query}》的资料B"]
    return {"retrieved": docs}


def generate(state: State) -> dict:
    """节点2：基于检索结果生成回答。"""
    docs = state["retrieved"]
    query = state["user_query"]
    prompt = f"用户问题：{query}\n资料：{docs}\n请生成简短回答。"
    resp = llm.invoke([HumanMessage(content=prompt)])
    return {"answer": resp.content, "messages": [resp]}


def polish(state: State) -> dict:
    """节点3：润色回答。"""
    text = state["answer"]
    polished = f"💡 {text}"
    return {"answer": polished, "messages": [AIMessage(content=polished)]}


g = StateGraph(State)
g.add_node("retrieve", retrieve)
g.add_node("generate", generate)
g.add_node("polish", polish)
g.add_edge(START, "retrieve")
g.add_edge("retrieve", "generate")
g.add_edge("generate", "polish")
g.add_edge("polish", END)
app = g.compile()

result = app.invoke({
    "user_query": "什么是 LangGraph",
    "retrieved": [],
    "answer": "",
    "messages": [],
})

print("最终回答：", result["answer"])
print("消息历史：", [m.content if hasattr(m, "content") else m for m in result["messages"]])

执行顺序（由边决定，与 add_node 顺序无关）：

graph LR
    S([START]) --> R[retrieve 检索]
    R --> G[generate 生成]
    G --> P[polish 润色]
    P --> E([END])

观察：

• 每个节点只读自己需要的字段，写自己负责的字段。
• messages 字段被多个节点追加，最终保留完整历史。
• 节点之间不互相调用——它们都只跟 state 打交道。

七、节点的副作用与可测试性

节点分两类：

• 纯函数节点：只读 state + 返回 dict，没有副作用。最好测试——传 state 进去断言返回值即可。
• 副作用节点：调 LLM、查数据库、读写文件。测试时需要 mock 或测试替身。

建议把"纯计算"和"副作用"分离：

# 纯计算（好测）
def compose_prompt(state) -> dict:
    return {"prompt": f"基于 {state['docs']} 回答 {state['query']}"}

# 副作用（要 mock）
def call_llm(state) -> dict:
    resp = llm.invoke(state["prompt"])
    return {"answer": resp.content}

八、节点的多种形态

普通函数

最常见：

def my_node(state: State) -> dict:
    return {...}

Lambda

简单逻辑可用 lambda：

g.add_node("echo", lambda state: {"echo": state["input"]})

类方法

需要保持状态（注意：图执行时框架会调一次）：

class MyProcessor:
    def __init__(self, llm):
        self.llm = llm

    def __call__(self, state: State) -> dict:
        resp = self.llm.invoke(state["messages"])
        return {"messages": [resp]}

g.add_node("proc", MyProcessor(llm))

接收额外配置

通过 config 传运行时参数（详见 编译与运行）：

def my_node(state: State, config: RunnableConfig) -> dict:
    user_id = config["configurable"]["user_id"]
    ...

九、与 LangGraph4j 对比

维度	Python	Java (LangGraph4j)
节点载体	普通函数 / lambda / __call__ 对象	实现 AgentNode 接口 / lambda
签名	(state: State) -> dict	apply(state: State) -> Map<String,Object>
类型推断	鸭子类型，灵活	静态类型，严格
配置注入	第二参数 config	第二参数 config
闭包	原生支持	需 lambda 捕获

Java 因类型严格，节点通常是命名类或 lambda；Python 更灵活，函数即节点。

十、常见踩坑

1. 返回字段不在 State 里被忽略：节点返回 {"foo": "bar"}，但 State 没定义 foo——静默丢失。先核对 State 定义。
2. 节点返回 None：应该返回 dict（哪怕空 {}）。返回 None 在某些版本被当"无更新"，新手最好显式 return \{\}。
3. 节点里直接改入参 state：state["a"] = "新" 不会生效。节点收到的是快照，必须通过返回值更新。
4. 节点执行顺序误解：以为 add_node 调用顺序决定执行顺序——错，由边决定。
5. 节点既读又写同一字段造成循环依赖：如节点 A 写 x，节点 B 读 x 写 x，又回到 A——会形成无限循环。配 recursion_limit 防御。
6. 节点内开了全局资源不释放：如每次都新建数据库连接不关闭。建议用上下文管理器或预先初始化复用。
7. 副作用节点没法测试：把副作用和纯计算分开，让纯计算部分可单测。
8. 节点太"重"：一个节点干 5 件事，难调试。拆成 5 个小节点。
9. 节点返回了非 JSON 可序列化对象：如返回 datetime 对象、自定义类实例。checkpointer 持久化时会出错。尽量返回基本类型，或用 Pydantic 模型。
10. 依赖外部状态未传入：节点函数里用了模块级全局变量（如 llm = ChatOpenAI()），测试时难替换。建议显式传参或用工厂函数构造节点。

十一、小结

• 节点的本质是普通 Python 函数 (state: State) -> dict。
• 返回的 dict 是"差量"，按 reducer 合并进状态；没出现的字段不变。
• START 和 END 是特殊节点，表示图的边界。
• 节点内可做任何事：调 LLM、调工具、读写 IO、纯计算。
• 单一职责：一个节点干一件事，便于测试、复用、调试。
• 与 LangGraph4j 概念对应，Python 因函数一等公民更灵活。

下一篇讲 边 Edge——把节点连起来。