第3章基本构件¶

学习目标¶

掌握 ToolNode、AgentNode 等预置构件的用法与适用场景
能够编写自定义节点并封装为可复用单元
理解节点间通信机制：状态字段、消息列表、reducer
学会把 LangChain 工具集成进 LangGraph 图
构建一个由预置节点与自定义节点混合组成的工作流

问题导入¶

第2章里我们手写了 agent 节点和 call_tools 节点。你会发现，这两个节点几乎每个 Agent 项目都要写一遍：调 LLM、解析工具调用、执行工具、回填结果。这种重复劳动既容易写错，也难以维护。LangGraph 在 langgraph.prebuilt 中提供了这些常用节点的预置实现，让你像搭积木一样组装 Agent。本章介绍这些积木，并教你打造自己的"定制积木"。

核心概念¶

1. 预置节点概览¶

langgraph.prebuilt 提供了几类开箱即用的构件：

ToolNode：接收消息列表，自动执行最后一条消息中的所有 tool_calls，把结果作为 ToolMessage 回写到状态。支持并行执行多个工具。
create_react_agent：一行代码生成完整的 ReAct Agent 图，内部已包含"LLM 节点 + ToolNode + 条件边循环"，是快速起步的首选。
AgentNode（概念性）：通常指封装了 LLM + 工具 + Prompt 的自定义节点，社区约定俗成的叫法。LangGraph 没有严格命名的 AgentNode 类，但官方示例常以此命名。

2. ToolNode 详解¶

ToolNode 接受一个工具列表，行为可概括为：

读取 state["messages"][-1] 的 tool_calls。
并行调用对应工具（默认并行，可用 executor 自定义）。
把每个结果包装成 ToolMessage，返回 {"messages": [tool_msg, ...]}。

它天然与 add_messages reducer 配合，结果会被追加到消息列表。若消息中没有 tool_calls，它会抛错——因此务必用条件边确保只有"确实要调工具"时才进入 ToolNode。

3. create_react_agent 详解¶

这是最常用的高层 API：

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(model, tools)

它内部构建的图等价于第2章我们手写的循环：START → agent → (条件边) → tools → agent → ... → END。支持的关键参数：

model：ChatModel 实例或绑定了工具的模型。
tools：工具列表。
state_schema：自定义状态类型（默认 MessagesState，含 messages 字段）。
checkpointer：注入检查点。
prompt：系统提示，支持字符串或返回消息的函数。

适合"标准 ReAct"场景；若需要多 Agent、复杂分支，仍应手写图。

4. 自定义节点¶

当预置构件不够用时，自定义节点遵循统一契约：

def my_node(state: State) -> dict:
    # 读 state、做事、返回要更新的字段
    return {"field": new_value}

设计自定义节点的要点：

单一职责：一个节点只做一件事（调一次 LLM、查一次库），便于复用与测试。
显式输入输出：节点依赖哪些字段、产出哪些字段，在 State 定义中一目了然。
无副作用优先：尽量返回纯数据，把 IO 边界（写库、发消息）集中在专门节点。

5. 节点间通信¶

节点之间不直接调用彼此，只通过 State 通信。三种常见模式：

消息列表模式：节点向 messages 追加消息，下游节点读最后一条。适合多轮对话。
字段传递模式：节点产出专用字段（如 retrieved_docs、draft），下游节点读取。适合流水线。
黑板模式：多个节点向一个共享字典写不同键，最后由汇总节点读取。适合并行 + 合并。

6. 工具集成¶

LangGraph 本身不重新发明工具，而是复用 LangChain 的工具生态：

@tool 装饰器：把普通函数变为带 schema 的工具。
StructuredTool.from_function：更灵活的封装。
BaseTool 子类：复杂工具逻辑。

工具可在 ToolNode 中被自动调用，也可在自定义节点中手动 tool.invoke(args) 调用。MCP 工具（见第7卷）也可适配为 LangChain Tool 后接入。

原理图¶

flowchart TB
    subgraph 预置构件
        CRA[create_react_agent<br/>一键生成完整 ReAct 图]
        TN[ToolNode<br/>自动执行 tool_calls]
    end
    subgraph 自定义节点
        LLM_NODE[LLMNode<br/>调用模型生成]
        RETRIEVE[RetrieveNode<br/>检索知识]
        GUARD[GuardrailNode<br/>安全校验]
        ROUTER[RouterNode<br/>任务路由]
    end
    subgraph 通信层: 共享 State
        MSGS[(messages)]
        DOCS[(retrieved_docs)]
        DRAFT[(draft)]
        META[(metadata)]
    end

    CRA --> TN
    CRA --> LLM_NODE
    LLM_NODE <--> MSGS
    TN <--> MSGS
    RETRIEVE --> DOCS
    LLM_NODE --> DRAFT
    GUARD --> DRAFT
    ROUTER --> META

图：预置构件与自定义节点通过共享 State 的不同字段通信。messages 承载对话，retrieved_docs 承载检索结果，draft 承载中间产物。

示例代码¶

下面用 create_react_agent 快速起步，再演示如何混入自定义节点构建"检索增强 Agent"。

import os
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage

# ---------- 方式一：一行创建标准 ReAct Agent ----------
@tool
def search_api(query: str) -> str:
    """搜索引擎查询。"""
    return f"搜索结果：关于「{query}」的前 3 条链接。"

@tool
def calculator(expression: str) -> str:
    """数学计算器。"""
    try:
        return str(eval(expression))  # 演示用，生产环境请用安全沙箱
    except Exception as e:
        return f"计算失败：{e}"

llm = ChatOpenAI(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))

react_agent = create_react_agent(
    model=llm,
    tools=[search_api, calculator],
    prompt="你是一个严谨的研究助手，回答前先用 search_api 检索，涉及计算用 calculator。",
)

# 直接可用
# result = react_agent.invoke({"messages": [{"role": "user", "content": "1+2 等于几？"}]})


# ---------- 方式二：混入自定义节点，构建检索增强 Agent ----------
class RagState(TypedDict):
    messages: Annotated[list, add_messages]
    retrieved_docs: list   # 自定义字段：检索结果

def retrieve_node(state: RagState) -> dict:
    """自定义检索节点：从用户最后一条消息提取关键词并检索。"""
    last_user_msg = state["messages"][-1].content
    # 演示：实际接向量库
    docs = [f"知识片段-{i}: 与「{last_user_msg[:10]}」相关的内容" for i in range(2)]
    return {"retrieved_docs": docs}

def rag_generate_node(state: RagState) -> dict:
    """自定义生成节点：把检索结果拼进上下文再调用 LLM。"""
    docs_text = "\n".join(state["retrieved_docs"])
    sys = SystemMessage(content=f"仅依据以下知识回答，无依据时说不知道：\n{docs_text}")
    response = llm.invoke([sys] + state["messages"])
    return {"messages": [response]}

builder = StateGraph(RagState)
builder.add_node("retrieve", retrieve_node)
builder.add_node("generate", rag_generate_node)
builder.add_node("tools", ToolNode([search_api, calculator]))

# 定义一个绑工具的 LLM 节点
def agent_node(state: RagState) -> dict:
    llm_with_tools = llm.bind_tools([search_api, calculator])
    # 把检索结果作为系统消息注入
    docs_text = "\n".join(state.get("retrieved_docs", []))
    sys = SystemMessage(content=f"你是助手。参考知识：\n{docs_text}")
    return {"messages": [llm_with_tools.invoke([sys] + state["messages"])]}

builder.add_node("agent", agent_node)

builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "agent")
builder.add_conditional_edges("agent", lambda s: "tools" if s["messages"][-1].tool_calls else END,
                              {"tools": "tools", END: END})
builder.add_edge("tools", "agent")  # 工具执行后回到 agent

rag_agent = builder.compile()

if __name__ == "__main__":
    result = rag_agent.invoke({
        "messages": [{"role": "user", "content": "帮我查一下「LangGraph」并算算 2024 减 2023。"}],
        "retrieved_docs": [],
    })
    print("最终回复：", result["messages"][-1].content)
    print("检索片段数：", len(result["retrieved_docs"]))

方式一用 create_react_agent 一行起步，适合标准场景；方式二混入 retrieve 自定义节点和预置 ToolNode，构建出"先检索、再思考、可调工具"的复合 Agent。两者共享同一套工具与模型，差别只在图的拓扑。

企业案例¶

案例 1：可复用节点库¶

某电商平台把常见能力封装成节点库并发布到内部 PyPI：ProductSearchNode、OrderQueryNode、RefundNode、ComplaintClassifyNode、GuardrailNode。各业务线像搭积木一样组合：售前 Agent = Classify → ProductSearch → Generate；售后 Agent = Classify → OrderQuery → (Refund | Escalate)。节点库统一了工具调用、错误处理与日志埋点，新 Agent 上线周期从两周缩短到两天。

案例 2：把 MCP 工具接入 LangGraph¶

研发团队将第7卷的 MCP Client 适配为 LangChain Tool，再用 ToolNode 接入 LangGraph。这样 Agent 既能调用本地文件系统工具，也能调用远程 MCP Server 暴露的数据库工具，且全部由 ToolNode 统一调度、并行执行。关键是 MCP 工具的 schema 自动转换为 LangChain 工具的 args_schema，无需手写。

常见问题¶

Q1：create_react_agent 能否自定义状态字段？ 能。传入 state_schema=YourState，并把额外字段在节点中维护。但若状态逻辑复杂（多分支、自定义 reducer），建议手写图更可控。

Q2：ToolNode 与自己写工具执行节点有何区别？ ToolNode 内置了并行执行、错误包装、tool_call_id 回填等逻辑，且与条件边约定一致（无 tool_calls 会报错）。自己写更灵活但易漏细节。生产环境推荐 ToolNode。

Q3：一个节点能调用多个 LLM 吗？ 能。节点内部可以多次 llm.invoke，或并行调用多个不同模型（如一个生成、一个审核）。但要权衡延迟与成本，复杂逻辑可拆为多个节点。

Q4：自定义节点如何复用？ 把节点定义为独立函数或类，放在 nodes/ 包中，配合 State 的标准字段约定。团队可约定"输入消息列表 + 输出消息列表"的通用接口，实现跨项目复用。

Q5：节点内抛异常会怎样？ 异常会向上抛出终止本次 invoke。若已配置 checkpointer，已执行节点的状态已持久化，可从断点恢复。建议在节点内捕获可预期异常并写入状态字段（如 error），由条件边决定是重试还是兜底。

面试题¶

1. ToolNode 解决了什么问题？ 答：自动执行消息中的 tool_calls，并行调度、回填 ToolMessage、维护 tool_call_id 关联，免去手写工具执行循环，且与 add_messages reducer 无缝配合。

2. create_react_agent 内部构建了什么样的图？ 答：START → agent(LLM) → 条件边 → tools(ToolNode) → agent → ... → END，即标准 ReAct 循环图。条件边根据是否有 tool_calls 决定走 tools 还是 END。

3. 节点间通信有哪些模式？ 答：消息列表模式（追加消息，适合对话）、字段传递模式（专用字段，适合流水线）、黑板模式（多节点写共享字典，适合并行合并）。

4. 何时该用 create_react_agent，何时手写图？ 答：标准 ReAct、单 Agent 用工具，用 create_react_agent 快速起步；涉及多 Agent、复杂分支、自定义状态字段与 reducer、人在回路时，手写图更可控。

5. 如何把外部工具（如 MCP 工具）接入 LangGraph？ 答：适配为 LangChain Tool（实现 args_schema 与 _run），再用 ToolNode 或在自定义节点中 tool.invoke 调用。MCP 工具的 schema 可自动转换。

练习题¶

用 create_react_agent 搭建一个"代码助手"，工具包括 read_file 和 run_python，并设计一个系统提示约束它只回答 Python 相关问题。
实现一个 GuardrailNode：检查 LLM 输出是否包含敏感词，包含则返回固定兜底话术。把它插入示例二的 generate 与 END 之间。
把示例二的 retrieve_node 改造为并行检索两个数据源（向量库 + 关键词库），用 fan-out / fan-in 合并结果。

小结¶

本章介绍了 LangGraph 的"积木"：预置的 ToolNode、create_react_agent 让你快速起步，自定义节点让你灵活扩展。节点间通过共享 State 通信，三种模式覆盖对话、流水线、并行合并等场景。工具复用 LangChain 生态，MCP 等外部工具也可适配接入。下一章我们进入状态与记忆，讲解如何用检查点让图"记住过去、暂停当下、恢复未来"。

第3章 基本构件¶