LangGraph 第7章：Nodes 节点

Node（节点）是 LangGraph 图的基本执行单元。本章介绍节点的各种实现方式。

普通函数作为节点

最简单的节点就是一个普通 Python 函数：

from typing import TypedDict

class State(TypedDict):
    value: int
    result: str

def increment_node(state: State):
    """递增计数并返回结果"""
    new_value = state["value"] + 1
    return {
        "value": new_value,
        "result": f"当前值: {new_value}"
    }

基本规则：

• 函数签名：def node_name(state: StateType) -> dict:
• 参数：接收当前 State 对象
• 返回值：返回要更新的字段字典（只返回需要修改的字段）

LLM 调用节点

最常见的节点模式是调用 LLM：

from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage

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

def llm_call_node(state: State):
    """调用 LLM 生成回复"""
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

带系统提示词的 LLM 节点：

from langchain_core.messages import SystemMessage

SYSTEM_PROMPT = "你是一个专业的编程助手。"

def expert_llm_node(state: State):
    """带系统提示词的 LLM 调用"""
    messages = [SystemMessage(content=SYSTEM_PROMPT)]
    messages.extend(state["messages"])
    response = llm.invoke(messages)
    return {"messages": [response]}

异步节点（Async）

对于 IO 密集型操作，可以使用异步节点提升性能：

import asyncio
from typing import TypedDict

class AsyncState(TypedDict):
    data: str
    processed: str

async def fetch_data_node(state: AsyncState):
    """异步获取数据"""
    await asyncio.sleep(1)  # 模拟网络请求
    return {"data": "从远程获取的数据"}

async def process_node(state: AsyncState):
    """异步处理数据"""
    processed = await async_process(state["data"])
    return {"processed": processed}

# 异步执行
async def main():
    result = await app.ainvoke({"data": "", "processed": ""})

# 运行
import asyncio
asyncio.run(main())

异步节点和同步节点可以混用在同一个图中：

builder.add_node("fetch", fetch_data_node)    # 异步
builder.add_node("process", process_node)      # 异步
builder.add_node("format", format_node)        # 同步

builder.add_edge("fetch", "process")
builder.add_edge("process", "format")

类作为节点

除了函数，还可以使用可调用类（实现了 __call__ 方法的类）作为节点：

from typing import TypedDict, Any

class State(TypedDict):
    input: str
    counter: int
    output: str

class CounterNode:
    """计数器节点 - 类实现"""
    def __init__(self, increment: int = 1):
        self.increment = increment

    def __call__(self, state: State) -> dict:
        new_counter = state.get("counter", 0) + self.increment
        return {
            "counter": new_counter,
            "output": f"计数: {new_counter}"
        }

# 使用
builder.add_node("counter", CounterNode(increment=2))

类的优势：

• 可配置：通过构造函数参数配置节点行为
• 有状态：可以保存节点内部状态（注意与 State 区分）
• 可测试：易于注入依赖进行单元测试

带 LLM 的类节点：

class LLMNode:
    """可配置的 LLM 节点"""
    def __init__(self, model: str = "gpt-4", temperature: float = 0.7):
        self.llm = ChatOpenAI(model=model, temperature=temperature)

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

# 创建不同配置的 LLM 节点
creative_llm = LLMNode(temperature=0.9)
precise_llm = LLMNode(temperature=0.1)

builder.add_node("creative_writer", creative_llm)
builder.add_node("fact_checker", precise_llm)

节点返回值与状态更新

返回值格式

节点返回值必须是字典，键名对应 State 中定义的字段：

class State(TypedDict):
    text: str
    count: int
    items: Annotated[list, operator.add]
    metadata: dict

def example_node(state: State):
    # 返回部分字段更新
    return {
        "text": "新文本",
        "count": state["count"] + 1,
        "items": ["新项目"],
        "metadata": {"key": "value"}
    }

空返回值

如果节点不需要更新状态，可以返回空字典：

def logging_node(state: State):
    """仅记录日志，不修改状态"""
    print(f"当前状态: {state}")
    return {}  # 不更新任何字段

使用 Command 控制流程

除了返回字典，还可以使用 Command 同时更新状态和控制流程：

from langgraph.types import Command

def conditional_node(state: State):
    """根据条件分支并更新状态"""
    if state["count"] > 10:
        return Command(
            goto="end_node",
            update={"text": "计数超过10，结束"}
        )
    else:
        return Command(
            goto="process_node",
            update={"text": f"继续处理，当前计数: {state['count']}"}
        )

节点设计最佳实践

节点类型对比

下一章我们将学习 Edges 和条件路由，掌握如何让节点按照需要的方式流转。