第 17 期 | 主管模式 (Supervisor Agent)：打造统筹中心

更新于 2026/4/15

嘿，各位未来的 AI 架构师们！欢迎回到我们的《LangGraph 多智能体专家课》。我是你们的老朋友，那位对技术挑剔，对教学更挑剔的导师。今天，我们要给我们的“AI 万能内容创作机构”注入一个真正的“大脑”——一个不干具体活，但却能统筹全局、运筹帷幄的“主管模式 (Supervisor Agent)”。

想象一下，你的内容机构现在已经有了 Researcher（研究员）、Writer（写手）、Editor（编辑）等专家，他们个个身怀绝技。但问题来了：当客户抛来一个需求时，谁来决定这个需求是需要先做市场调研，还是直接可以开写，亦或是需要先进行资料搜集？总不能让 Writer 直接去问客户“我该写什么”吧？这不专业！

这就是我们今天要解决的核心痛点：一个智能的调度员，一个只负责分发任务、不亲自动手的主管。 它将是我们的中央指挥部，确保每一个用户请求都能精准地抵达最合适的专家手中。

🎯 本期学习目标

在本期课程中，你将不仅仅是敲几行代码，更是要理解多智能体协作的精髓，掌握构建智能调度系统的核心技能。具体来说，本期你将收获：

  1. 理解 Supervisor Agent 的核心价值与定位： 为什么我们需要一个“不干活”的调度员？它在多智能体系统中的战略意义是什么？
  2. 掌握 LangGraph 中 Supervisor 模式的实现技巧： 学习如何利用 LangGraph 的 StateGraph 和条件边（Conditional Edges）来构建一个灵活且强大的主管代理。
  3. 学会如何设计高效的路由策略： 掌握 Prompt Engineering 的精髓，让 Supervisor 能够根据用户需求，智能地将任务分发给 Researcher 或 Writer 等特定 Agent。
  4. 提升 AI 机构的整体智能与效率： 引入 Supervisor 后，我们的 AI 内容机构将拥有更强的适应性和自动化能力，减少人工干预，提高响应速度。

📖 原理解析

为什么需要 Supervisor Agent？

在我们的 AI Content Agency 项目中，随着 Agent 数量的增加和任务复杂度的提升，显式地定义任务流变得至关重要。最初，我们可能让一个 Planner Agent 直接分解任务，但这在某些场景下仍然不够灵活。例如：

  • 需求模糊性： 用户说“我需要一篇关于 AI 的文章”。这可能意味着需要先研究最新的 AI 进展，也可能只是要一篇关于 AI 基础知识的普及文。谁来判断？
  • 资源最优分配： 如果某个请求直接可以写，就不应该浪费时间去研究。反之，如果缺乏背景知识就盲目写作，质量会大打折扣。谁来做这个决策？
  • 扩展性挑战： 当我们未来增加更多的 Agent（比如 SEO 专家、翻译专家、图片生成专家）时，如何确保新 Agent 能够无缝集成到工作流中，并被正确地调用？

Supervisor Agent 的出现，就是为了解决这些问题。它是一个“决策节点”，一个“高阶大脑”，它的核心职责是：基于对用户输入和当前系统状态的理解，决定下一步应该由哪个专家 Agent 来处理任务。 它不执行具体的业务逻辑（比如写作或研究），只负责“指挥交通”。

Supervisor Agent 的工作原理

在 LangGraph 中实现 Supervisor Agent，其核心思想是利用图的条件边（Conditional Edges）。

  1. Supervisor 作为一个独立的 LLM 节点： 我们会专门配置一个 LLM，给它一个非常明确的指令：分析用户需求，并输出一个预定义的操作指令（例如，"RESEARCH"、"WRITE"、"FINISH"）。
  2. 状态（State）的传递： 整个工作流共享一个 AgentState，其中会包含用户原始请求、Supervisor 的决策等信息。
  3. 条件路由函数： 我们会定义一个路由函数，它会检查 AgentState 中 Supervisor 输出的决策。根据这个决策，路由函数会动态地决定图的下一个节点。
    • 如果 Supervisor 说 "RESEARCH"，就路由到 Researcher Agent。
    • 如果 Supervisor 说 "WRITE"，就路由到 Writer Agent。
    • 如果 Supervisor 说 "FINISH"，就结束当前流程。
    • 未来我们还可以扩展：如果 Supervisor 说 "EDIT"，就路由到 Editor Agent。

这种模式的强大之处在于，它将决策逻辑与执行逻辑彻底分离。Supervisor 专注于“做什么”，而具体的 Agent 专注于“怎么做”。这使得系统更模块化、更易于维护和扩展。

Mermaid 图解核心概念

让我们通过一个 Mermaid 图来直观地理解 Supervisor Agent 的工作流：

graph LR
    A[用户请求] --> B(Supervisor Agent);

    subgraph Supervisor Agent Logic
        B -- 分析用户需求 --> C{决策: 哪个Agent最合适?};
        C -- "需要研究" --> D[Researcher Agent];
        C -- "可以直接写作" --> E[Writer Agent];
        C -- "无任务或完成" --> F[END];
    end

    D -- 研究完成，可能返回给Supervisor或直接结束 --> F;
    E -- 写作完成 --> F;

    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 D fill:#dfd,stroke:#333,stroke-width:2px;
    style E fill:#ffd,stroke:#333,stroke-width:2px;
    style F fill:#eee,stroke:#333,stroke-width:2px;

图解说明：

  1. 用户请求 (User Request): 一切的起点，客户向我们的 AI Content Agency 提交一个内容创作需求。
  2. Supervisor Agent (主管代理): 接收用户请求。它是本期设计的核心，一个独立的 LLM，不直接生成内容，而是分析请求，做出下一步的决策。
  3. 决策 (Decision): Supervisor Agent 的核心输出。它会根据预设的 Prompt，判断请求的性质，并输出一个明确的指令（例如“RESEARCH”或“WRITE”）。
  4. 条件路由 (Conditional Routing): 这是 LangGraph 的魔法所在。根据 Supervisor 的决策，工作流会被动态地路由到不同的专家 Agent。
    • Researcher Agent (研究员代理): 如果 Supervisor 判断需要先进行信息搜集或背景研究，任务就会被派发给 Researcher。
    • Writer Agent (写手代理): 如果 Supervisor 判断可以直接进行内容创作，任务就会被派发给 Writer。
    • END (结束): 在本期的简化模型中，Researcher 或 Writer 完成任务后，流程即告结束。或者，Supervisor 也可以直接判断为“无任务”或“已完成”，从而直接结束流程。

通过这个架构，我们的 AI Content Agency 变得更加智能和灵活。Supervisor 就像一个经验丰富的产品经理，在接收到用户需求后，能迅速判断出该由哪个部门的专家来具体执行，大大提高了整个机构的运作效率。

💻 实战代码演练 (Agency项目中的具体应用)

好了，原理讲得再天花乱坠，不如代码来得实在。是时候把我们的理论付诸实践，将 Supervisor Agent 融入到我们的 AI 内容机构中去了。

我们将使用 Python 和 langgraph 来构建这个系统。

import os
from typing import TypedDict, Annotated, List, Union
import operator

from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, END

# 请确保你已经设置了 OPENAI_API_KEY 环境变量
# You should have OPENAI_API_KEY environment variable set up
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 1. 定义我们的 AgentState
# 1. Define our AgentState
class AgentState(TypedDict):
    """
    一个AgentState代表了LangGraph中的共享状态。
    它将包含用户请求、主管的决策以及不同Agent的输出。
    
    AgentState represents the shared state in LangGraph.
    It will contain the user request, supervisor's decision, and outputs from different agents.
    """
    user_request: str
    supervisor_decision: Annotated[str, operator.setitem] # 主管的决策，用于路由
    research_output: Annotated[str, operator.setitem]     # 研究员的输出
    writing_output: Annotated[str, operator.setitem]      # 写手的输出
    messages: Annotated[List[BaseMessage], operator.add] # 消息历史，方便Agent查看上下文

# 2. 初始化我们的 LLM
# 2. Initialize our LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# 3. 定义 Supervisor Agent
# 3. Define the Supervisor Agent
def create_supervisor_agent():
    """
    创建一个Supervisor Agent。它是一个LLM，根据用户请求决定下一步的动作。
    它不执行具体任务，只负责决策路由。
    
    Creates a Supervisor Agent. It's an LLM that decides the next action based on the user request.
    It doesn't perform specific tasks, only routes decisions.
    """
    system_prompt = """
    你是一个AI内容机构的主管调度员。你的任务是根据用户的请求，决定将任务分发给哪个专家Agent。
    你不能自己执行内容创作或研究。你的职责是分析请求，并输出一个明确的指令，指出下一步应该由谁来处理。
    
    可用的专家Agent有：
    - 'researcher': 当用户请求需要进行信息搜集、背景研究、事实核查或获取最新数据时。
    - 'writer': 当用户请求可以直接进行内容创作、文章撰写、文案生成或内容润色时。
    - 'FINISH': 当请求明确表示无需进一步处理，或者已经完成时。
    
    你的输出必须是上述三个字符串之一，不要包含其他任何内容。
    例如：
    用户请求: "我需要一篇关于 LangGraph Supervisor 模式的简介。"
    你的输出: "writer"
    
    用户请求: "我需要关于 LangChain 最新版本更新的资料，然后写一份总结报告。"
    你的输出: "researcher"
    
    用户请求: "谢谢，我已经拿到资料了。"
    你的输出: "FINISH"
    
    请严格按照这个格式输出，不要有额外的解释或寒暄。
    
    You are the supervisor dispatcher for an AI content agency. Your task is to determine which expert agent
    should handle the user's request. You do not perform content creation or research yourself.
    Your role is to analyze the request and output a clear instruction indicating who should handle the next step.
    
    Available expert agents are:
    - 'researcher': When the user request requires information gathering, background research, fact-checking, or obtaining the latest data.
    - 'writer': When the user request can directly proceed with content creation, article writing, copywriting, or content refinement.
    - 'FINISH': When the request clearly indicates no further processing is needed, or the task is complete.
    
    Your output must be one of the three strings above, without any additional content.
    Example:
    User request: "I need an introduction to the LangGraph Supervisor pattern."
    Your output: "writer"
    
    User request: "I need information about the latest LangChain updates, then write a summary report."
    Your output: "researcher"
    
    User request: "Thanks, I have the information."
    Your output: "FINISH"
    
    Please strictly adhere to this output format, without any extra explanations or greetings.
    """
    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        ("human", "{user_request}")
    ])
    return prompt | llm.bind(stop=["\n"]) # 使用 bind(stop=["\n"]) 确保只输出一行决策

supervisor_agent = create_supervisor_agent()

# 4. 定义 Researcher Agent 和 Writer Agent (简化版，仅用于演示路由)
# 4. Define Researcher Agent and Writer Agent (simplified for routing demonstration)

def researcher_node(state: AgentState):
    """
    研究员Agent节点。在这里，它只是模拟一个研究过程并返回结果。
    
    Researcher Agent node. Here, it simply simulates a research process and returns a result.
    """
    print(f"\n--- Researcher Agent 正在处理请求: {state['user_request']} ---")
    # 实际项目中，这里会调用搜索工具、数据库查询等
    # In a real project, this would involve calling search tools, database queries, etc.
    research_result = f"根据请求 '{state['user_request']}'，我搜集到了以下关键信息：\n" \
                      f"1. LangGraph Supervisor模式是用于多Agent系统中的任务调度和路由。\n" \
                      f"2. 它通过一个独立的LLM来决策，将任务分发给不同的执行Agent。\n" \
                      f"3. 核心是利用LangGraph的条件边实现动态工作流。"
    print("--- 研究完成 ---")
    return {"research_output": research_result, "messages": [AIMessage(content=f"Research completed: {research_result}")]}

def writer_node(state: AgentState):
    """
    写手Agent节点。它会根据用户请求和可能的研究结果来生成内容。
    
    Writer Agent node. It generates content based on the user request and potential research output.
    """
    print(f"\n--- Writer Agent 正在处理请求: {state['user_request']} ---")
    content_to_write = state['user_request']
    if state.get('research_output'):
        content_to_write += f"\n\n参考研究结果：\n{state['research_output']}"
    
    # 实际项目中，这里会调用LLM生成高质量内容
    # In a real project, this would involve calling an LLM to generate high-quality content
    writing_result = f"以下是根据您的请求生成的文章：\n\n" \
                     f"**标题：{state['user_request']}**\n\n" \
                     f"尊敬的客户，根据您的需求，我们为您精心撰写了以下内容。\n" \
                     f"内容核心旨在围绕 '{state['user_request']}' 展开，力求精准传达信息。\n" \
                     f"如果之前有研究结果，则会在内容中融入研究发现，确保内容的深度和广度。\n" \
                     f"（此处省略长篇内容，仅为演示）"
    print("--- 写作完成 ---")
    return {"writing_output": writing_result, "messages": [AIMessage(content=f"Writing completed: {writing_result}")]}

# 5. 定义路由函数
# 5. Define the router function
def route_supervisor_decision(state: AgentState):
    """
    根据Supervisor Agent的决策，决定下一个要执行的节点。
    
    Decides the next node to execute based on the Supervisor Agent's decision.
    """
    decision = state['supervisor_decision'].strip().lower() # 清理并转为小写，确保匹配
    print(f"\n--- Supervisor 决策: {decision} ---")
    if "researcher" in decision:
        return "researcher_node"
    elif "writer" in decision:
        return "writer_node"
    elif "finish" in decision:
        return "END"
    else:
        # 兜底机制，防止Supervisor输出意外内容
        # Fallback mechanism for unexpected supervisor output
        print(f"警告：Supervisor 输出了无法识别的决策: {decision}。默认路由到 Writer。")
        return "writer_node" # 默认路由，或者可以抛出错误

# 6. 构建 LangGraph
# 6. Construct the LangGraph
workflow = StateGraph(AgentState)

# 添加节点
# Add nodes
workflow.add_node("supervisor", lambda state: {"supervisor_decision": supervisor_agent.invoke({"user_request": state["user_request"]}).content, "messages": [AIMessage(content=f"Supervisor decided: {state['supervisor_decision']}")]})
workflow.add_node("researcher_node", researcher_node)
workflow.add_node("writer_node", writer_node)

# 设置入口点
# Set the entry point
workflow.set_entry_point("supervisor")

# 添加条件边：从 supervisor 到不同的 Agent
# Add conditional edges: from supervisor to different agents
workflow.add_conditional_edges(
    "supervisor",
    route_supervisor_decision,
    {
        "researcher_node": "researcher_node",
        "writer_node": "writer_node",
        "END": END
    }
)

# 添加普通边：从 Researcher 和 Writer 到 END (简化流程，本期不涉及复杂循环)
# Add normal edges: from Researcher and Writer to END (simplified flow, no complex loops in this episode)
workflow.add_edge("researcher_node", END)
workflow.add_edge("writer_node", END)

# 编译图
# Compile the graph
app = workflow.compile()

# 7. 运行演示
# 7. Run the demo
print("\n--- 演示开始 ---")

# 场景 1: 需要直接写作的请求
# Scenario 1: Request that can go directly to writing
print("\n=== 场景 1: 直接写作 ===")
inputs_1 = {"user_request": "请帮我写一篇关于 LangGraph Supervisor 模式的简介。", "messages": [HumanMessage(content="请帮我写一篇关于 LangGraph Supervisor 模式的简介。")]}
for s in app.stream(inputs_1):
    print(s)
    print("---")
print("\n=== 场景 1 结束 ===\n")

# 场景 2: 需要先研究的请求
# Scenario 2: Request that requires research first
print("\n=== 场景 2: 需要研究 ===")
inputs_2 = {"user_request": "我需要关于 LangChain 最新版本更新的资料，然后写一份总结报告。", "messages": [HumanMessage(content="我需要关于 LangChain 最新版本更新的资料，然后写一份总结报告。")]}
for s in app.stream(inputs_2):
    print(s)
    print("---")
print("\n=== 场景 2 结束 ===\n")

# 场景 3: 明确表示结束的请求
# Scenario 3: Request explicitly indicating completion
print("\n=== 场景 3: 明确结束 ===")
inputs_3 = {"user_request": "谢谢，我已经拿到资料了，不需要其他内容了。", "messages": [HumanMessage(content="谢谢，我已经拿到资料了，不需要其他内容了。")]}
for s in app.stream(inputs_3):
    print(s)
    print("---")
print("\n=== 场景 3 结束 ===\n")

print("\n--- 演示结束 ---")

代码解析：

  1. AgentState 定义： 我们定义了一个 TypedDict 作为我们 LangGraph 的共享状态。它包含了 user_request（用户原始请求）、supervisor_decision（主管的决策，这是路由的关键）、research_output 和 writing_output（各 Agent 的结果），以及 messages（用于 Agent 之间传递上下文）。
  2. create_supervisor_agent()： 这是本期的核心。我们构建了一个 ChatOpenAI 实例，并为其设计了一个极其精细的 system_prompt。这个 Prompt 明确告诉 Supervisor 它的角色（调度员）、它能做什么（决策路由）、不能做什么（不干活），以及它应该输出什么格式（"researcher"、"writer" 或 "FINISH"）。llm.bind(stop=["\n"]) 是一个小技巧，确保 LLM 在生成第一个换行符后停止，这有助于保持输出的简洁和格式化。
  3. researcher_node 和 writer_node： 为了简化，这两个 Agent 节点在这里只是模拟了各自的工作，并返回一个字符串作为结果。在实际项目中，它们会集成更复杂的逻辑，例如调用外部工具（搜索、数据库等）或更强大的 LLM 来执行任务。
  4. route_supervisor_decision()： 这是 LangGraph 条件边所需的路由函数。它接收当前的 AgentState，解析 supervisor_decision 字段，并返回下一个节点的名称（字符串）。注意我们加入了 strip().lower() 来增强鲁棒性，以及一个简单的兜底机制，以防 Supervisor 输出意外内容。
  5. 图的构建：
    • 我们使用 StateGraph(AgentState) 初始化图。
    • add_node() 注册了 Supervisor、Researcher 和 Writer 节点。
      • supervisor 节点是一个 lambda 函数，它调用 supervisor_agent 来获取决策，并更新 supervisor_decision 字段。
    • set_entry_point("supervisor") 明确了流程从 Supervisor 开始。
    • add_conditional_edges() 是这里的关键。它告诉 LangGraph：从 supervisor 节点出来后，调用 route_supervisor_decision 函数来决定下一步去哪里。这个函数返回的字符串（例如 "researcher_node"）会匹配到 add_conditional_edges 的第三个参数字典中的键，从而导向相应的节点。
    • add_edge("researcher_node", END) 和 add_edge("writer_node", END)：在本期中，为了突出 Supervisor 的路由功能，我们让 Researcher 和 Writer 完成任务后直接结束流程。在更复杂的场景中，它们可能会返回给 Supervisor 进行二次决策，或者进入其他 Agent（如 Editor）。
  6. 演示运行： 我们通过三个不同的用户请求来演示 Supervisor Agent 的效果：一个直接去 Writer，一个先去 Researcher，一个直接结束。观察控制台输出，你会清楚地看到 Supervisor 的决策和任务的流向。

通过这个实战演练，你现在已经掌握了如何在 LangGraph 中构建一个强大的 Supervisor Agent，作为你多智能体系统的中央调度员。

坑与避坑指南

作为一名经验丰富的 AI 架构师，我见过无数因为“想当然”而掉进的坑。Supervisor 模式虽强大，但也并非万能药，需要你精心打磨。

  1. Prompt Engineering 是王道，也是深渊：

    • 坑： 认为只要给 LLM 一个大致的指令它就能完美工作。Supervisor 的输出不符合预期，导致路由错误或系统崩溃。
    • 避坑： Supervisor Agent 的 Prompt 是整个模式的生命线。你必须极其精确、严谨地定义其职责、输出格式和可选的输出值。在 Prompt 中给出清晰的正反例（few-shot examples）是提升其稳定性的不二法门。例如，明确告诉它“只输出一个单词，不要解释”。
    • 我个人经验： 很多时候，一个看似简单的路由问题，其根源在于 Supervisor LLM 对用户意图的理解偏差或输出格式不规范。多花时间打磨 Prompt，远比事后调试复杂逻辑来得高效。

  2. 决策粒度与 Agent 职责的平衡：

    • 坑： Supervisor 的决策粒度过细，导致它需要做出太多微观决策，增加了其负担和出错概率；或者决策粒度过粗，导致具体 Agent 接收到的指令不够明确。
    • 避坑： 确保 Supervisor 专注于高层次的路由决策（即“谁来做”），而不是具体的执行细节（即“怎么做”）。具体 Agent 应该有足够的智能和工具来处理其领域的复杂性。例如，Supervisor 决定“写作”，Writer Agent 内部可以有自己的子图或工具来完成写作、润色、排版等。

  3. 异常处理与兜底机制：

    • 坑： Supervisor 有时会“犯傻”，输出不在你预期列表中的内容，导致路由函数无法识别，程序直接报错。
    • 避坑： 在路由函数中，必须包含一个健壮的异常处理和兜底机制。例如，如果 Supervisor 输出无法识别，可以默认路由到一个“错误处理 Agent”，或者重新回到 Supervisor 节点并附带错误信息，让它重新决策。我在代码中提供了一个简单的默认路由到 writer_node 的例子，但在生产环境中，你可能需要更复杂的策略。

  4. 状态管理与信息传递：

    • 坑： Supervisor 做了决策，但相关信息没有正确传递给下一个 Agent，导致 Agent 无法有效工作。
    • 避坑： 仔细设计 AgentState，确保所有 Agent 需要的信息都能在 StateGraph 中正确存储和更新。例如，user_request 必须从一开始就贯穿始终，而 research_output 只有 Researcher 节点会更新，但 Writer 节点需要读取它。明确每个节点对状态的读写权限。

  5. 成本考量：

    • 坑： 引入 Supervisor 会增加一次额外的 LLM 调用，这会带来额外的延迟和成本。
    • 避坑： 权衡利弊。对于简单的、确定性的任务，可能直接路由更高效。但对于复杂、模糊或需要动态决策的任务，Supervisor 的价值远超其成本。选择合适的 LLM 模型也很重要，Supervisor 可以使用更轻量级、响应速度更快的模型，因为它只做决策，不生成长内容。

记住，构建多智能体系统就像组建一支特种部队。Supervisor 是你的指挥官，它不需要亲上战场，但它的每一个指令都至关重要。

📝 本期小结

各位硬核玩家，恭喜你！在本期课程中，我们成功地为我们的“AI 万能内容创作机构”引入了主管模式 (Supervisor Agent)。

我们深入探讨了 Supervisor Agent 的战略意义：它是一个不干具体活，但却能统筹全局、精准分发任务的智能调度中心。通过 LangGraph 的 StateGraph 和强大的条件边，我们实现了一个灵活的工作流，让 Supervisor 能够根据用户请求，智能地将任务路由给 Researcher 或 Writer 等专业 Agent。

这个模式的引入，标志着我们的 AI 内容机构从简单的“一锤子买卖”迈向了更智能、