第 03 期 | 业务节点 (Nodes)：Graph 的真正执行者 — LangGraph 多智能体专家课：从底层逻辑到复杂工作流的重构

🎯 本期学习目标

同学们好，我是你们的老朋友，那位在 AI 架构圈子里摸爬滚打十年、头发掉了不少但热情只增不减的老张。上期我们初识了 LangGraph 的世界观，知道了它如何用图结构来编排复杂的 AI 工作流。但光有图的骨架还不行，骨架里得有血有肉，有能干活的工人。这一期，咱们就来深入理解 LangGraph 的核心执行单元——业务节点 (Nodes)。

学完这一期，你将：

透彻理解 LangGraph 中 Node 的本质：它不仅仅是一个函数，更是我们 AI 机构里的“部门”或“专家”，负责执行具体的业务逻辑。
掌握如何编写基本的 Worker Nodes：学会封装大模型推理逻辑，让我们的 Planner (规划师) Agent 能真正动起来，开始思考。
精通节点内的数据修改与状态管理：理解 Node 如何与 Graph 的全局 State 交互，确保信息在不同 Agent 之间顺畅流转。
构建 Agency 项目的第一个核心节点：为我们的“AI 万能内容创作机构”搭建 Planner 节点，迈出构建复杂多智能体工作流的第一步。

📖 原理解析

同学们，想象一下，我们的“AI 万能内容创作机构”就像一家高科技公司。公司里有各种各样的部门：企划部、研发部、内容创作部、编辑部等等。每个部门都有其独特的职能，接收任务，处理信息，然后将结果传递给下一个部门。

在 LangGraph 的世界里，这些“部门”或者“专家”就是我们的 Nodes (业务节点)。

一个 Node，最核心的理解，它就是一个可调用的函数 (Callable)。这个函数接收当前的 Graph State（你可以理解为整个公司共享的“项目进展报告”或“任务看板”），执行自己的业务逻辑（比如调用大模型进行思考、查询数据库、调用外部工具），然后返回一个对 State 的更新。

核心思想：Node 接收 State，处理，返回 State 的“差值” (delta)。

为什么是“差值”？这是 LangGraph 乃至很多函数式编程范式的一个精妙之处：状态的不可变性 (Immutability)。每个节点都不是直接修改全局 State，而是生成一个新的 State 片段，然后 LangGraph 框架会负责将这些片段合并到全局 State 中。这大大简化了并发和调试的复杂度，你永远知道每个节点“做了什么改变”，而不是“把什么改了”。

我们的 AI Content Agency 中的 `Node`

在我们的“AI 万能内容创作机构”中，第一个要登场的重量级角色就是 Planner (规划师)。它的职责是接收客户的原始需求，然后将其分解成可执行的、具体的创作任务。

输入：初始的客户需求 (topic)。
处理：Planner 节点将调用一个大型语言模型 (LLM)，让它根据 topic 构思内容大纲、确定目标受众、提炼核心观点，并规划后续的创作步骤。
输出：更新 State，添加 plan_outline（内容大纲）、target_audience（目标受众）、research_tasks（研究任务列表）等信息。

这个过程，完美地体现了一个 Node 的生命周期。

Mermaid 图解：Planner 节点的引入

让我们通过一个 Mermaid 图来直观地看看 Planner 节点在我们的 Graph 中是如何运作的。它将是我们的第一个核心节点，为整个内容创作流程奠定基础。

graph TD
    A[用户输入: 初始创作需求] --> B{初始化 Graph State};
    B --> C(Planner Node: 规划师);
    
    subgraph Planner Node 内部逻辑
        C --> C1[接收当前 State];
        C1 --> C2[调用 LLM 进行内容规划];
        C2 --> C3[生成内容大纲、研究任务等];
        C3 --> C4[返回 State 更新];
    end
    
    C --> D{Graph State 更新};
    D --> E[准备进入下一阶段: 例如 Research Node];
    
    style A fill:#f9f,stroke:#333,stroke-width:2px;
    style B fill:#bbf,stroke:#333,stroke-width:2px;
    style C fill:#ccf,stroke:#333,stroke-width:2px;
    style C1 fill:#fcf,stroke:#333,stroke-width:1px;
    style C2 fill:#fcf,stroke:#333,stroke-width:1px;
    style C3 fill:#fcf,stroke:#333,stroke-width:1px;
    style C4 fill:#fcf,stroke:#333,stroke-width:1px;
    style D fill:#bbf,stroke:#333,stroke-width:2px;
    style E fill:#afa,stroke:#333,stroke-width:2px;

图解说明：

用户输入：一切的起点，客户给我们的原始需求。
初始化 Graph State：LangGraph 会根据我们定义的 State 结构，创建一个初始状态。
Planner Node (规划师)：这是我们本期要重点实现的节点。它接收当前的 State。
Planner Node 内部逻辑：
- 接收当前 State：获取当前工作流的所有上下文信息。
- 调用 LLM 进行内容规划：这是核心业务逻辑，我们在这里会与大模型进行交互，让它发挥“规划师”的作用。
- 生成内容大纲、研究任务等：LLM 的输出会被结构化，成为我们后续 Agent 的输入。
- 返回 State 更新：节点将生成一个字典，包含需要添加到 State 中的新信息。
Graph State 更新：LangGraph 框架负责将 Planner Node 返回的更新合并到全局 State 中。
准备进入下一阶段：更新后的 State 将作为输入，传递给图中的下一个节点（例如，未来的 Researcher Node）。

理解了这个流程，你就能抓住 LangGraph 的精髓了：每个节点都是一个纯粹的函数，接收状态，返回状态的更新，而整个系统通过这些状态的流转和聚合，实现了复杂的工作流。

💻 实战代码演练

好了，理论说得再多，不如撸起袖子干一场。现在，我们就来为我们的“AI 万能内容创作机构”编写第一个核心的业务节点：Planner Node。

首先，我们需要定义我们的 Graph State。这个 State 将是所有 Agent 共享的信息中心。为了方便，我们使用 TypedDict 来定义它，它就像一个明确了字段和类型的数据结构。

# agency_state.py
from typing import TypedDict, List, Dict, Optional

class AgencyState(TypedDict):
    """
    AI 内容创作机构的共享状态。
    这个 TypedDict 定义了在整个 LangGraph 工作流中，所有 Agent 共享和操作的数据结构。
    """
    topic: str  # 初始内容创作主题
    plan_outline: Optional[str]  # Planner 生成的内容大纲
    target_audience: Optional[str] # Planner 确定的目标受众
    research_tasks: Optional[List[str]] # Planner 规划的研究任务列表
    research_results: Optional[Dict[str, str]] # Researcher 收集的研究结果
    draft_content: Optional[str] # Writer 撰写的内容初稿
    editor_feedback: Optional[str] # Editor 提供的修改意见
    final_content: Optional[str] # 最终定稿的内容

    # 还可以添加其他字段，例如：
    # current_agent_name: str # 当前正在执行的 Agent 名称，用于调试或日志
    # error_message: Optional[str] # 错误信息，用于错误处理

接下来，我们来编写 Planner Node 的代码。它将是一个 Python 函数，接收 AgencyState，然后返回一个字典，代表对 AgencyState 的更新。

我们将使用 ChatOpenAI 来模拟 LLM 的能力。如果你没有 OpenAI API Key，也可以用其他兼容 LangChain 的 LLM，或者直接返回硬编码的模拟数据。

# nodes.py
import os
from typing import Dict, Any, List
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

# 从 .env 文件加载环境变量，例如 OPENAI_API_KEY
load_dotenv()

# 导入我们定义的 AgencyState
from agency_state import AgencyState

# 初始化大模型
# 这里使用 GPT-4o，你也可以根据需要选择其他模型
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# --- 1. Planner Node (规划师节点) ---
def planner_node(state: AgencyState) -> Dict[str, Any]:
    """
    规划师节点：根据初始主题，生成内容大纲、目标受众和研究任务。
    这个节点封装了 LLM 的推理逻辑，用于将高层次的需求转化为具体的执行计划。
    
    Args:
        state (AgencyState): 当前的全局状态。
                              预期 state 中包含 'topic' 字段。

    Returns:
        Dict[str, Any]: 包含对 state 更新的字典。
                        将添加 'plan_outline', 'target_audience', 'research_tasks'。
    """
    print("\n--- 执行 Planner Node ---")
    topic = state.get("topic")
    if not topic:
        raise ValueError("AgencyState 中缺少 'topic' 字段，Planner 无法工作。")

    # 1. 定义 LLM 的 Prompt Template
    # 我们希望 LLM 不仅给出大纲，还要思考受众和后续研究点
    planner_prompt = ChatPromptTemplate.from_messages(
        [
            ("system", "你是一位经验丰富的内容策略师和规划师。你的任务是根据给定的主题，制定一个详细的内容创作计划。"),
            ("human", """
            请为以下主题生成一个详细的内容创作计划。计划应包含：
            1.  **内容大纲 (Plan Outline)**：结构清晰，至少包含 3-5 个主要章节和每个章节的要点。
            2.  **目标受众 (Target Audience)**：详细描述这篇内容的理想读者是谁，他们的兴趣、痛点和知识水平。
            3.  **研究任务 (Research Tasks)**：列出至少 3-5 项具体的研究任务，以确保内容的深度和准确性。这些任务应是后续研究员 Agent 需要完成的。
            
            主题: {topic}
            
            请以 JSON 格式返回结果，例如：
            ```json
            {{
                "plan_outline": "...",
                "target_audience": "...",
                "research_tasks": ["...", "...", "..."]
            }}
            ```
            """)
        ]
    )

    # 2. 构建 LLM Chain
    # 使用 .with_structured_output(dict) 确保 LLM 输出为结构化的字典
    # 注意：with_structured_output 依赖于模型的能力，如果模型不支持，可能需要自定义解析器
    # 对于 GPT-4o 这样的模型，它通常能很好地遵循 JSON 格式指令
    planner_chain = planner_prompt | llm.with_structured_output(
        schema={
            "type": "object",
            "properties": {
                "plan_outline": {"type": "string", "description": "详细的内容大纲"},
                "target_audience": {"type": "string", "description": "目标受众描述"},
                "research_tasks": {"type": "array", "items": {"type": "string"}, "description": "需要进行的研究任务列表"},
            },
            "required": ["plan_outline", "target_audience", "research_tasks"],
        }
    )

    # 3. 执行 LLM 推理
    try:
        planning_output = planner_chain.invoke({"topic": topic})
        print("Planner 成功生成计划。")
        # 直接返回 LLM 的输出作为 state 的更新
        # with_structured_output 已经返回了字典，可以直接用
        return planning_output
    except Exception as e:
        print(f"Planner Node 执行失败: {e}")
        # 如果 LLM 调用失败，我们可以选择抛出异常，或者返回一个包含错误信息的更新
        # 这里我们选择返回一个错误信息，让 Graph 可以处理
        return {"error_message": f"Planner failed: {str(e)}"}

# --- 模拟运行 Planner Node ---
if __name__ == "__main__":
    print("--- 模拟运行 Planner Node ---")
    
    # 1. 模拟初始状态
    initial_state: AgencyState = {
        "topic": "探索 LangGraph 在构建多智能体应用中的核心优势与实践案例",
        "plan_outline": None,
        "target_audience": None,
        "research_tasks": None,
        "research_results": None,
        "draft_content": None,
        "editor_feedback": None,
        "final_content": None,
    }
    
    print(f"初始状态: {initial_state['topic']}")
    
    # 2. 调用 Planner Node
    updated_fields = planner_node(initial_state)
    
    # 3. 将更新合并到状态中 (LangGraph 框架会做这一步)
    # 这里我们手动模拟合并
    final_state = initial_state.copy()
    for key, value in updated_fields.items():
        # 如果是列表或字典，需要更复杂的合并逻辑，这里简化为直接覆盖
        final_state[key] = value

    print("\n--- Planner Node 执行后的状态更新 ---")
    print(f"内容大纲: \n{final_state.get('plan_outline')}")
    print(f"目标受众: {final_state.get('target_audience')}")
    print(f"研究任务: {final_state.get('research_tasks')}")
    print(f"完整最终状态: {final_state}")

    # 尝试另一个主题
    print("\n--- 再次模拟运行 Planner Node (新主题) ---")
    initial_state_2: AgencyState = {
        "topic": "人工智能伦理与监管：挑战与机遇",
        "plan_outline": None,
        "target_audience": None,
        "research_tasks": None,
        "research_results": None,
        "draft_content": None,
        "editor_feedback": None,
        "final_content": None,
    }
    updated_fields_2 = planner_node(initial_state_2)
    final_state_2 = initial_state_2.copy()
    for key, value in updated_fields_2.items():
        final_state_2[key] = value

    print("\n--- Planner Node 执行后的状态更新 (新主题) ---")
    print(f"内容大纲: \n{final_state_2.get('plan_outline')}")
    print(f"目标受众: {final_state_2.get('target_audience')}")
    print(f"研究任务: {final_state_2.get('research_tasks')}")

代码解析：

AgencyState 定义：我们首先定义了 AgencyState 这个 TypedDict。这是整个机构的“共享内存”，所有 Agent 都通过它来读写信息。清晰定义状态结构是构建稳健多智能体系统的基石。
planner_node 函数：
- 输入：它接收一个 state: AgencyState 参数。这是 LangGraph 传入的当前全局状态。
- 业务逻辑封装：
  - 我们从 state 中提取 topic。
  - 构建了一个 ChatPromptTemplate，这是与 LLM 交互的核心。我们精心设计了 Prompt，要求 LLM 扮演“内容策略师”，并以结构化的 JSON 格式返回结果，包含大纲、受众和研究任务。
  - 使用 llm.with_structured_output 是一个非常强大的技巧，它指导 LLM 直接输出符合我们定义的 JSON Schema 的数据，大大简化了后续的解析工作。这就像给 LLM 戴上了一副“结构化输出”的眼镜，让它看得更清楚，吐字更清晰。
  - 调用 planner_chain.invoke() 执行 LLM 推理。
- 输出：函数返回一个字典。这个字典的键值对将用于更新 AgencyState。注意，我们只返回了需要新增或修改的字段，而不是整个 AgencyState。LangGraph 会自动将这些更新合并到当前的全局状态中。
- 错误处理：加入了 try-except 块，以应对 LLM 调用可能出现的错误，提供更好的健壮性。
模拟运行 (if __name__ == "__main__":)：
- 这部分代码展示了如何在没有完整 LangGraph 框架的情况下，单独测试我们的 planner_node。
- 我们创建了一个 initial_state，然后调用 planner_node。
- 最后，手动模拟了 LangGraph 将节点返回的更新合并到 initial_state 的过程，打印出最终状态，以验证节点的功能。

通过这个 Planner Node，我们已经成功地将一个复杂的“内容规划”任务，封装成了一个可执行、可复用、可测试的 LangGraph 节点。它接收输入，利用大模型的能力进行复杂推理，然后以结构化的方式输出结果，更新了我们机构的共享项目状态。这正是 Node 作为 Graph 真正执行者的体现！

坑与避坑指南

作为一名资深 AI 架构师，我见过太多的坑。现在，我把一些最常见的、最容易踩的“雷”分享给你，让你少走弯路。

状态管理混乱：Node 返回值的误区
- 坑：新手常犯的错误是，在 Node 中直接返回一个完整的 AgencyState 对象，或者返回与 AgencyState 结构不符的字典。
- 后果：这会导致状态更新不正确，轻则数据丢失，重则整个 Graph 逻辑崩溃。LangGraph 期望你返回的是一个字典，其中的键值对是需要添加到或覆盖当前 State 中相应字段的。
- 避坑指南：
  - 只返回 delta (差值)：你的 Node 函数应该只返回一个字典，包含你希望更新或添加的 State 字段。例如，planner_node 只返回 {"plan_outline": "...", "target_audience": "...", "research_tasks": [...]}。
  - 类型匹配：确保返回字典中字段的类型与 AgencyState 中定义的类型一致。TypedDict 或 Pydantic Model 的严格类型定义能帮你规避很多问题。
  - 理解合并逻辑：LangGraph 默认的 StateGraph 会对字典进行浅层合并。这意味着如果你返回 {"research_results": {"new_key": "value"}}，它会直接替换掉旧的 research_results 整个字典，而不是合并内部的键。如果需要深层合并，你需要自定义 StateGraph 的 reducer 方法。
LLM 推理逻辑与业务逻辑耦合过紧
- 坑：把所有的 LLM 调用、Prompt 工程、输出解析、数据处理等逻辑都塞到一个巨大的 Node 函数里。
- 后果：代码难以阅读、难以维护、难以测试。一旦 Prompt 需要调整，或者 LLM 模型需要切换，整个 Node 都可能需要重写。
- 避坑指南：
  - 单一职责原则 (SRP)：一个 Node 应该只负责一个核心的业务功能。例如，Planner Node 只负责规划，不负责研究或写作。
  - 模块化封装：将 LLM 相关的逻辑（Prompt 定义、Chain 构建、解析器）封装成独立的函数或类。就像我们代码中 planner_prompt | llm.with_structured_output(...) 这样，链式调用让逻辑更清晰。
  - Prompt 外部化：考虑将复杂的 Prompt 存储在外部文件或配置中，方便管理和迭代。
缺乏健壮性：错误处理不到位
- 坑：不处理 LLM 调用失败、API 超时、数据解析错误等异常情况。
- 后果：生产环境下一旦出现网络问题或 LLM 返回非预期格式，整个工作流就会中断，用户体验极差。
- 避坑指南：
  - try-except 是你的朋友：像我们 planner_node 中那样，总是用 try-except 包裹 LLM 调用和其他可能失败的操作。
  - 优雅降级或错误状态：当发生错误时，不要直接让程序崩溃。可以返回一个特殊的 State 更新（例如 {"error_message": "..."}），让后续的 Node 或者 Graph 的 conditional_edge 来判断并处理这些错误，例如跳转到错误处理流程，或者重试。
  - 日志记录：详细的日志能帮助你快速定位问题。在每个 Node 的开始和结束，以及关键步骤都打印日志。
环境配置与依赖管理
- 坑：硬编码 API 密钥，或者没有正确加载环境变量。
- 后果：代码安全性差，部署困难，不同环境下的行为不一致。
- 避坑指南：
  - 使用环境变量：所有敏感信息（如 API 密钥）都应该通过环境变量管理。我们使用了 python-dotenv 来加载 .env 文件，这是一个很好的实践。
  - requirements.txt：始终维护一个清晰的 requirements.txt 文件，确保所有依赖都能被正确安装。

记住，构建多智能体系统就像搭建一座高楼，每个 Node 都是一个砖块。只有每个砖块都坚固、每个连接都牢靠，这栋大楼才能屹立不倒。

📝 本期小结

各位未来的 AI 架构师们，恭喜你们！这一期，我们不仅深入理解了 LangGraph Node 的核心概念——作为 Graph 的真正执行者，它如何接收状态、执行业务逻辑并返回状态更新——更重要的是，我们亲手为我们的“AI 万能内容创作机构”打造了第一个至关重要的部门：Planner Node。

我们学习了如何：

定义清晰的共享状态 AgencyState，它是我们所有 Agent 协作的基础。
编写一个功能完备的 Node 函数，封装了 LLM 的推理逻辑，通过精心设计的 Prompt 让它扮演“内容策略师”的角色。
利用 llm.with_structured_output 确保 LLM 输出的结构化和可靠性。
理解 Node 返回状态更新的机制，而非直接修改全局状态。

通过 Planner Node 的实战，你已经迈出了构建复杂多智能体工作流的第一步。它将用户的高层次需求，转化为具体、可执行的创作计划，为后续的 Researcher、Writer 等 Agent 铺平了道路。

当然，我们还探讨了在构建 Node 时常见的“坑”以及如何优雅地避开它们，这些都是你未来在大型项目中能够独当一面的宝贵经验。

下一期，我们将继续深入，把我们这个 Planner Node 真正地接入到 LangGraph 的 StateGraph 中，并开始定义图的边 (Edges)，让我们的 Planner 能够将它生成的工作计划，无缝地传递给下一个智能体。敬请期待！

第 03 期 | 业务节点 (Nodes)：Graph 的真正执行者