第 16 期 | 多智能体架构探秘：Flat vs Hierarchical — LangGraph 多智能体专家课：从底层逻辑到复杂工作流的重构

AI 创作机构到底是采取扁平讨论制好，还是上下级金字塔式汇报好？

欢迎回到《LangGraph 多智能体专家课》。我是你们的导师。

在过去的 15 期里，我们的「AI 万能内容创作机构 (AI Content Agency)」从零起步，已经拥有了 Planner（策划）、Researcher（研究员）、Writer（主笔）和 Editor（主编）四位得力干将。之前，我们让他们通过简单的线性流（Linear Flow）接力完成了一篇篇爆款文章。

但是，同学们，现实世界的业务哪有这么理想化？昨天，机构接到了一个大单：“请针对苹果最新发布的 Vision Pro 撰写一篇深度行业分析，同时输出一篇小红书种草文案，并配上对应的 Twitter 推广线索。”

如果还是用以前的“接力赛”模式，Researcher 查完资料丢给 Writer，Writer 瞬间懵逼：“我是先写深度分析，还是先写小红书？” 整个流程直接卡死。

当业务复杂度呈指数级上升时，单线流必死，多智能体架构的重构迫在眉睫。 今天，我们就来深度探讨并实操多智能体架构的两大核心流派：扁平化（Flat） 与 层级化（Hierarchical）。我们将为机构引入真正的“主理人”机制，让你的 Agent 团队从“草台班子”蜕变为“正规军”。

敲黑板，集中注意力，我们要开始烧脑了！

🎯 本期学习目标

认知升级（道）：深刻理解 Flat（扁平网络）与 Hierarchical（层级金字塔）架构的底层逻辑与适用场景。
架构重构（术）：掌握在 LangGraph 中引入 Supervisor（主管/路由）节点的标准设计模式。
实战落地（用）：使用 Python + LangGraph 重构我们的 Content Agency，实现由 Planner 统筹分发、各司其职的复杂工作流。
状态管理（深）：解决多 Agent 频繁交互导致的上下文爆炸（Context Bloat）问题。

📖 原理解析：草台班子 vs 现代企业

在多智能体系统（Multi-Agent System, MAS）中，Agent 之间的协作拓扑结构决定了系统的上限。我们来看看两种最经典的架构。

1. Flat Architecture（扁平化/点对点架构）

想象一下初创公司：老板、产品、研发坐在一起。遇到问题，大家在群里七嘴八舌地讨论。

机制：Agent 之间可以直接相互通信，或者通过一个共享的黑板（Blackboard）自由发言。
优点：沟通成本极低，灵活性极高。适合头脑风暴、剧本杀、辩论赛等需要发散性思维的场景。
缺点：失控。当任务复杂时，极容易陷入无限循环的“扯皮”中（比如 Writer 觉得资料不够打回给 Researcher，Researcher 觉得 Writer 不懂技术又打回去）。

2. Hierarchical Architecture（层级化/主管架构）

随着公司做大，必须引入“科层制”。我们需要一个包工头（Supervisor）。

机制：设立一个中心化的 Supervisor（在我们的项目中就是 Planner）。所有 Worker Agent（Researcher, Writer, Editor）不能直接互相交流。Worker 完成任务后，必须把结果汇报给 Supervisor，由 Supervisor 决定下一步是交给另一个 Worker，还是结束任务。
优点：秩序井然，逻辑严密，极度适合目标导向的复杂任务分解。
缺点：Supervisor 成了性能瓶颈（Bottleneck）。如果主管是个“傻子”（Prompt 没写好或 LLM 能力弱），整个团队就瘫痪了。

对于我们的 AI Content Agency 来说，面对多渠道、多格式的内容生成需求，Hierarchical（层级化）是唯一的解药。

下面这张图，直观地展示了我们今天要重构的架构走向：

graph TD
    subgraph "❌ 过去的痛点：扁平/线性架构 (草台班子)"
        R1[Researcher] -->|资料| W1[Writer]
        W1 -->|初稿| E1[Editor]
        E1 -->|打回修改?| W1
        style R1 fill:#ffcccc,stroke:#333,stroke-width:2px
        style W1 fill:#ffcccc,stroke:#333,stroke-width:2px
        style E1 fill:#ffcccc,stroke:#333,stroke-width:2px
    end

    subgraph "✅ 今天的重构：Hierarchical 层级架构 (正规军)"
        User((客户需求)) --> S{Planner\n(Supervisor 主管)}
        
        S -->|1. 分配调研任务| R2[Researcher]
        R2 -.->|2. 提交报告| S
        
        S -->|3. 分配撰写任务| W2[Writer]
        W2 -.->|4. 提交初稿| S
        
        S -->|5. 分配审校任务| E2[Editor]
        E2 -.->|6. 提交终稿| S
        
        S -->|7. 任务全部完成| FINISH(((交付输出)))
        
        style S fill:#ff9900,stroke:#333,stroke-width:4px,color:#fff
        style R2 fill:#cce5ff,stroke:#333,stroke-width:2px
        style W2 fill:#cce5ff,stroke:#333,stroke-width:2px
        style E2 fill:#cce5ff,stroke:#333,stroke-width:2px
        style FINISH fill:#ccffcc,stroke:#333,stroke-width:2px
    end

看懂了吗？在新的架构中，Planner 变成了整个 Graph 的路由中枢（Routing Hub）。它不干脏活累活，它只做两件事：思考（Think）和派单（Delegate）。

💻 实战代码演练

废话不多说，我们直接上代码。我们将使用 LangGraph 的 StateGraph 来构建这个层级网络。为了确保稳定输出路由决策，我们将使用 OpenAI 的 Structured Output（结构化输出）特性来武装我们的 Supervisor。

1. 环境准备与状态定义

首先，我们需要定义整个机构的“共享账本”——AgentState。

import operator
from typing import Annotated, Sequence, TypedDict, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END

# 1. 定义系统状态 (The State of the Agency)
# 这里继承 TypedDict，messages 使用 operator.add 来实现消息的追加
class AgencyState(TypedDict):
    # 记录所有的历史对话和工作成果
    messages: Annotated[Sequence[BaseMessage], operator.add]
    # 记录当前应该由谁来接手下一个任务
    next_agent: str

2. 打造核心大脑：Supervisor (Planner)

这是本期的灵魂代码。我们要强迫大模型输出一个特定的 JSON 结构，告诉 LangGraph 下一步该往哪走。

# 2. 定义 Supervisor 的路由结构 (Structured Output)
# 我们机构目前有三个打工人
MEMBERS = ["Researcher", "Writer", "Editor"]

class RouterDecision(BaseModel):
    """Supervisor decides who should act next."""
    # 强制 LLM 只能从这几个选项里挑，或者输出 FINISH
    next_agent: str = Field(
        description="The next agent to act. Choose from 'Researcher', 'Writer', 'Editor', or 'FINISH' if the whole project is done."
    )
    reasoning: str = Field(
        description="Brief explanation of why this agent was chosen."
    )

# 初始化聪明的大脑 (建议用 gpt-4o，主管不能太笨)
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 使用 with_structured_output 绑定 Pydantic 模型
supervisor_chain = llm.with_structured_output(RouterDecision)

# 编写 Supervisor 节点的逻辑
def supervisor_node(state: AgencyState):
    """
    Supervisor Node: Analyzes the current state and routes to the next agent.
    主管节点：分析当前进度，并派发给下一个 Agent。
    """
    system_prompt = (
        "You are the Chief Planner of an elite AI Content Agency.\n"
        "Your team members are: {members}.\n"
        "Your job is to read the conversation/work history and decide who should act next.\n"
        "Rule 1: Always start with 'Researcher' to gather facts.\n"
        "Rule 2: Pass to 'Writer' to draft the content based on research.\n"
        "Rule 3: Pass to 'Editor' to review and refine the draft.\n"
        "Rule 4: If the Editor has approved the final content and all requirements are met, output 'FINISH'.\n"
        "DO NOT do the work yourself. Just route it."
    ).format(members=", ".join(MEMBERS))
    
    messages = [{"role": "system", "content": system_prompt}] + state["messages"]
    
    # 调用 LLM 进行决策
    print("🧠 [Planner] is thinking...")
    decision = supervisor_chain.invoke(messages)
    print(f"🎯 [Planner] Decision: Next up is -> {decision.next_agent}. Reason: {decision.reasoning}")
    
    # 关键点：返回更新后的 state，特别是 next_agent 字段
    return {"next_agent": decision.next_agent}

3. 定义打工人节点 (Worker Nodes)

为了演示清晰，我们用简单的 Prompt 来模拟这三个专家的工作。在真实的业务代码中，你可以为他们挂载各自的 Tools（比如 Researcher 挂载 Tavily 搜索，Writer 挂载 Markdown 格式化工具）。

# 3. 定义 Worker Nodes
# 辅助函数：将 Worker 的输出包装成 AIMessage，并标明身份
def worker_node_factory(role_name: str, system_instruction: str):
    def node(state: AgencyState):
        print(f"🛠️ [{role_name}] is working on the task...")
        worker_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
        messages = [{"role": "system", "content": system_instruction}] + state["messages"]
        response = worker_llm.invoke(messages)
        
        # 必须在消息前缀加上名字，让 Supervisor 知道是谁干的
        final_message = AIMessage(
            content=response.content, 
            name=role_name # 标记消息来源
        )
        return {"messages": [final_message]}
    return node

# 实例化三个打工人
researcher_node = worker_node_factory(
    "Researcher", 
    "You are a meticulous Researcher. Read the request, provide detailed bullet points of facts and data. Do not write the final article."
)

writer_node = worker_node_factory(
    "Writer", 
    "You are an expert Writer. Take the research provided by the Researcher and draft a compelling piece of content based on the user's request."
)

editor_node = worker_node_factory(
    "Editor", 
    "You are a strict Editor. Review the Writer's draft. Fix any typos, improve the tone, and output the FINAL polished version. Explicitly state 'FINAL VERSION APPROVED' at the end."
)

4. 组装图网络：构建金字塔

现在，我们要把主管和打工人用 LangGraph 的边（Edges）连接起来。这是 Hierarchical 架构的核心体现。

# 4. 构建 LangGraph
workflow = StateGraph(AgencyState)

# 添加所有节点
workflow.add_node("Supervisor", supervisor_node)
workflow.add_node("Researcher", researcher_node)
workflow.add_node("Writer", writer_node)
workflow.add_node("Editor", editor_node)

# 所有的 Worker 工作完成后，必须无条件汇报给 Supervisor
for member in MEMBERS:
    workflow.add_edge(member, "Supervisor")

# 定义条件路由逻辑
def router(state: AgencyState):
    # 根据 Supervisor 设置的 next_agent 来决定走向
    next_node = state["next_agent"]
    if next_node == "FINISH":
        return END
    return next_node

# Supervisor 的下一步是条件边（Conditional Edge）
workflow.add_conditional_edges(
    "Supervisor", # 起点
    router,       # 路由函数
    {
        "Researcher": "Researcher",
        "Writer": "Writer",
        "Editor": "Editor",
        END: END
    }
)

# 设定图的入口：任务进来，先找主管 Planner
workflow.add_edge(START, "Supervisor")

# 编译图
agency_app = workflow.compile()

5. 模拟运行 Demo

让我们来看看这个“正规军”是如何处理复杂任务的。

# 5. 运行测试
if __name__ == "__main__":
    task_prompt = "Write a short, engaging tweet about the latest breakthrough in Quantum Computing. Make it accessible to the public."
    
    print(f"🚀 User Request: {task_prompt}\n")
    print("-" * 50)
    
    initial_state = {
        "messages": [HumanMessage(content=task_prompt)],
        "next_agent": ""
    }
    
    # 设定递归上限，防止死循环
    config = {"recursion_limit": 15}
    
    for chunk in agency_app.stream(initial_state, config=config):
        # 打印当前执行完毕的节点名称
        if "__end__" not in chunk:
            node_name = list(chunk.keys())[0]
            print(f"✅ [{node_name}] finished its turn.\n")
            print("-" * 50)
            
    print("🎉 Project Completed Successfully!")

预期终端输出模拟：

🚀 User Request: Write a short, engaging tweet about the latest breakthrough in Quantum Computing...

🧠 [Planner] is thinking... 🎯 [Planner] Decision: Next up is -> Researcher. Reason: Need to gather the latest facts on quantum computing breakthroughs first. ✅ [Supervisor] finished its turn.

🛠️ [Researcher] is working on the task... ✅ [Researcher] finished its turn.

🧠 [Planner] is thinking... 🎯 [Planner] Decision: Next up is -> Writer. Reason: Research is complete, now we need to draft the tweet. ✅ [Supervisor] finished its turn.

... (直到 FINISH) 🎉 Project Completed Successfully!

💣 坑与避坑指南 (高阶视角的排错经验)

作为有着 10 年经验的老兵，我必须告诉你们，跑通上面的 Demo 只是第一步。在真实的生产环境中，Hierarchical 架构有几个致命的坑，踩进去就是万劫不复。

坑一：Supervisor 陷入“死循环” (Infinite Loop)

现象：Editor 觉得文章不行，打回给 Writer；Writer 改了又给 Editor，Editor 还是不满意。两人通过 Supervisor 互相踢皮球，你的 API Token 余额瞬间清零。避坑：

硬性阻断：在 workflow.compile() 时必须设置 recursion_limit（如代码中的 15）。
Prompt 约束：在 Supervisor 的 Prompt 中明确规定：“最多允许修改 2 次，超过次数强制 FINISH”。
状态注入：在 AgentState 中加入一个 revision_count: int 字段，每次打回加 1，达到阈值直接路由到 END。

坑二：上下文爆炸 (Context Bloat)

现象：由于所有的 Worker 都要把结果汇报给 Supervisor，messages 列表会越来越长。到了第 10 轮交互，每次调用 LLM 都要带上几万 Token 的废话。避坑：不要无脑使用 operator.add 累加所有消息。引入**状态压缩（State Summarization）**机制。或者在 State 中区分 scratchpad（打草稿用，定期清空）和 final_artifacts（最终成果，只保留最新版）。

坑三：杀鸡用牛刀 (Over-engineering)

现象：客户只是要求“把这句话翻译成英文”，你却启动了 Planner -> Researcher -> Writer -> Editor 完整流程，耗时 30 秒，花费 0.5 元。避坑：架构没有绝对的好坏，只有适不适合。如果你的业务 80% 是简单任务，请在 Supervisor 前面加一个 Triage Node（分诊台节点）。如果是简单任务，直接调用轻量级模型单次输出；复杂任务，再进入层级化网络。

📝 本期小结

今天，我们完成了一次史诗级的重构。

我们探讨了 Flat 架构的灵活与混乱，以及 Hierarchical 架构的严谨与可控。通过引入 LangGraph 的 add_conditional_edges 和大模型的结构化输出能力，我们成功打造了一个拥有独立思考和派单能力的 Supervisor (Planner)。

你的 AI Content Agency 现在不再是几个自顾自干活的散兵游勇，而是一个由大脑统帅、各司其职的现代化 AI 生产流水线。

思考题留给大家：在今天的代码中，Supervisor 每次只能派发给一个 Agent。如果客户要求“同时去查百度和谷歌”，我们能让两个 Researcher 并行（Parallel） 工作，然后再汇总给 Writer 吗？

这涉及到 LangGraph 的高级分支与合并（Fan-out / Fan-in）机制。别着急，我们下期《第 17 期 | 并行处理与 Map-Reduce：让你的团队拥有分身术》再见！

下课！同学们记得把代码跑起来，别光看不练！