Skip to main content

Documentation Index

Fetch the complete documentation index at: https://nvd-54.mintlify.app/llms.txt

Use this file to discover all available pages before exploring further.

​
概览

监督者模式是一种多智能体架构，其中一个中央监督者智能体协调专业的工作智能体。当任务需要不同类型的专业知识时，此方法表现出色。与其构建一个跨领域管理工具选择的单一智能体，不如创建专注的专家，由理解整体工作流的监督者来协调。 在本教程中，你将构建一个个人助手系统，通过一个现实的工作流来演示这些优势。系统将协调两个具有根本不同职责的专家：
  • 一个日历智能体，处理日程安排、可用性检查和事件管理。
  • 一个邮件智能体，管理通信、撰写消息和发送通知。
我们还将集成人机协作审查，允许用户根据需要批准、编辑和拒绝操作（例如外发邮件）。

​
为什么使用监督者？

多智能体架构允许你将工具分配到各个工作者中，每个工作者都有自己的提示词或指令。考虑一个直接访问所有日历和邮件 API 的智能体：它必须从许多类似的工具中选择，理解每个 API 的确切格式，并同时处理多个领域。如果性能下降，将相关工具和关联的提示词分离到逻辑组中可能会有所帮助（部分是为了管理迭代改进）。

​
概念

我们将涵盖以下概念：

​
设置

​
安装

本教程需要 langchain 包： 
pip install langchain
更多详情请参阅我们的安装指南。

​
LangSmith

设置 LangSmith 以检查智能体内部的运行情况。然后设置以下环境变量： 
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."

​
组件

我们需要从 LangChain 的集成套件中选择一个聊天模型：
👉 Read the OpenAI chat model integration docs
pip install -U "langchain[openai]"

import os
from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-..."

model = init_chat_model("gpt-5.4")

​
1. 定义工具

首先定义需要结构化输入的工具。在实际应用中，这些将调用真实的 API（Google Calendar、SendGrid 等）。本教程使用桩来演示模式。 
from langchain.tools import tool

@tool
def create_calendar_event(
    title: str,
    start_time: str,       # ISO 格式："2024-01-15T14:00:00"
    end_time: str,         # ISO 格式："2024-01-15T15:00:00"
    attendees: list[str],  # 邮箱地址
    location: str = ""
) -> str:
    """创建日历事件。需要精确的 ISO 日期时间格式。"""
    # 桩：实际中会调用 Google Calendar API、Outlook API 等。
    return f"Event created: {title} from {start_time} to {end_time} with {len(attendees)} attendees"


@tool
def send_email(
    to: list[str],  # 邮箱地址
    subject: str,
    body: str,
    cc: list[str] = []
) -> str:
    """通过邮件 API 发送邮件。需要格式正确的地址。"""
    # 桩：实际中会调用 SendGrid、Gmail API 等。
    return f"Email sent to {', '.join(to)} - Subject: {subject}"


@tool
def get_available_time_slots(
    attendees: list[str],
    date: str,  # ISO 格式："2024-01-15"
    duration_minutes: int
) -> list[str]:
    """检查给定参会者在特定日期的日历可用性。"""
    # 桩：实际中会查询日历 API
    return ["09:00", "14:00", "16:00"]

​
2. 创建专业子智能体

接下来，我们将创建处理每个领域的专业子智能体。

​
创建日历智能体

日历智能体理解自然语言的日程安排请求，并将它们转换为精确的 API 调用。它处理日期解析、可用性检查和事件创建。 
from langchain.agents import create_agent


CALENDAR_AGENT_PROMPT = (
    "你是一个日历日程安排助手。"
    "将自然语言日程安排请求（例如'下周二下午 2 点'）"
    "解析为正确的 ISO 日期时间格式。"
    "需要时使用 get_available_time_slots 检查可用性。"
    "如果没有合适的时间段，停止并在响应中确认不可用。"
    "使用 create_calendar_event 安排事件。"
    "始终在最终响应中确认安排了什么。"
)

calendar_agent = create_agent(
    model,
    tools=[create_calendar_event, get_available_time_slots],
    system_prompt=CALENDAR_AGENT_PROMPT,
)
测试日历智能体，看看它如何处理自然语言日程安排： 
query = "安排下周二下午 2 点的团队会议，时长 1 小时"

for step in calendar_agent.stream(
    {"messages": [{"role": "user", "content": query}]}
):
    for update in step.values():
        for message in update.get("messages", []):
            message.pretty_print()

================================== Ai Message ==================================
Tool Calls:
  get_available_time_slots (call_EIeoeIi1hE2VmwZSfHStGmXp)
 Call ID: call_EIeoeIi1hE2VmwZSfHStGmXp
  Args:
    attendees: []
    date: 2024-06-18
    duration_minutes: 60
================================= Tool Message =================================
Name: get_available_time_slots

["09:00", "14:00", "16:00"]
================================== Ai Message ==================================
Tool Calls:
  create_calendar_event (call_zgx3iJA66Ut0W8S3NpT93kEB)
 Call ID: call_zgx3iJA66Ut0W8S3NpT93kEB
  Args:
    title: Team Meeting
    start_time: 2024-06-18T14:00:00
    end_time: 2024-06-18T15:00:00
    attendees: []
================================= Tool Message =================================
Name: create_calendar_event

Event created: Team Meeting from 2024-06-18T14:00:00 to 2024-06-18T15:00:00 with 0 attendees
================================== Ai Message ==================================

团队会议已安排在下周二 6 月 18 日下午 2:00，时长 1 小时。如果你需要添加参会者或地点，请告诉我！
智能体将”下周二下午 2 点”解析为 ISO 格式（“2024-01-16T14:00:00”），计算结束时间，调用 create_calendar_event，并返回自然语言确认。

​
创建邮件智能体

邮件智能体处理消息撰写和发送。它专注于提取收件人信息、制作适当的主题和正文，以及管理邮件通信。 
EMAIL_AGENT_PROMPT = (
    "你是一个邮件助手。"
    "根据自然语言请求撰写专业邮件。"
    "提取收件人信息并制作适当的主题和正文。"
    "使用 send_email 发送消息。"
    "始终在最终响应中确认发送了什么。"
)

email_agent = create_agent(
    model,
    tools=[send_email],
    system_prompt=EMAIL_AGENT_PROMPT,
)
测试邮件智能体的自然语言请求： 
query = "给设计团队发一封关于审查新设计稿的提醒"

for step in email_agent.stream(
    {"messages": [{"role": "user", "content": query}]}
):
    for update in step.values():
        for message in update.get("messages", []):
            message.pretty_print()

================================== Ai Message ==================================
Tool Calls:
  send_email (call_OMl51FziTVY6CRZvzYfjYOZr)
 Call ID: call_OMl51FziTVY6CRZvzYfjYOZr
  Args:
    to: ['design-team@example.com']
    subject: Reminder: Please Review the New Mockups
    body: Hi Design Team,

This is a friendly reminder to review the new mockups at your earliest convenience. Your feedback is important to ensure that we stay on track with our project timeline.

Please let me know if you have any questions or need additional information.

Thank you!

Best regards,
================================= Tool Message =================================
Name: send_email

Email sent to design-team@example.com - Subject: Reminder: Please Review the New Mockups
================================== Ai Message ==================================

我已向设计团队发送了一封提醒邮件，请他们审查新的设计稿。如果你需要就此话题进一步沟通，请告诉我！
智能体从非正式请求中推断出收件人，制作了专业的主题和正文，调用 send_email，并返回确认。每个子智能体都有窄焦点和领域特定的工具与提示词，使其能在特定任务上表现出色。

​
3. 将子智能体包装为工具

现在将每个子智能体包装为监督者可以调用的工具。这是创建分层系统的关键架构步骤。监督者将看到像”schedule_event”这样的高级工具，而不是像”create_calendar_event”这样的低级工具。 
@tool
def schedule_event(request: str) -> str:
    """使用自然语言安排日历事件。

    当用户想要创建、修改或检查日历约会时使用此工具。
    处理日期/时间解析、可用性检查和事件创建。

    输入：自然语言日程安排请求（例如'下周二下午 2 点与设计团队开会'）
    """
    result = calendar_agent.invoke({
        "messages": [{"role": "user", "content": request}]
    })
    return result["messages"][-1].text


@tool
def manage_email(request: str) -> str:
    """使用自然语言发送邮件。

    当用户想要发送通知、提醒或任何邮件通信时使用此工具。
    处理收件人提取、主题生成和邮件撰写。

    输入：自然语言邮件请求（例如'给他们发一封关于会议的提醒'）
    """
    result = email_agent.invoke({
        "messages": [{"role": "user", "content": request}]
    })
    return result["messages"][-1].text
工具描述帮助监督者决定何时使用每个工具，所以要清晰具体。我们只返回子智能体的最终响应，因为监督者不需要看到中间推理或工具调用。

​
4. 创建监督者智能体

现在创建编排子智能体的监督者。监督者只看到高级工具，在领域级别而非单个 API 级别做出路由决策。 
SUPERVISOR_PROMPT = (
    "你是一个有用的个人助手。"
    "你可以安排日历事件和发送邮件。"
    "将用户请求分解为适当的工具调用并协调结果。"
    "当请求涉及多个操作时，按顺序使用多个工具。"
)

supervisor_agent = create_agent(
    model,
    tools=[schedule_event, manage_email],
    system_prompt=SUPERVISOR_PROMPT,
)

​
5. 使用监督者

现在用需要跨多个领域协调的复杂请求测试你的完整系统：

​
示例 1：简单的单领域请求

query = "安排明天上午 9 点的团队站会"

for step in supervisor_agent.stream(
    {"messages": [{"role": "user", "content": query}]}
):
    for update in step.values():
        for message in update.get("messages", []):
            message.pretty_print()

================================== Ai Message ==================================
Tool Calls:
  schedule_event (call_mXFJJDU8bKZadNUZPaag8Lct)
 Call ID: call_mXFJJDU8bKZadNUZPaag8Lct
  Args:
    request: Schedule a team standup for tomorrow at 9am with Alice and Bob.
================================= Tool Message =================================
Name: schedule_event

团队站会已安排在明天上午 9:00，Alice 和 Bob 参加。如果你需要做任何更改或添加更多详情，请告诉我！
================================== Ai Message ==================================

Alice 和 Bob 的团队站会已安排在明天上午 9:00。如果你需要进一步的安排或调整，请告诉我！
监督者将此识别为日历任务，调用 schedule_event，日历智能体处理日期解析和事件创建。
要完全透明地了解信息流，包括每次聊天模型调用的提示词和响应，请查看上述运行的 LangSmith 追踪。

​
示例 2：复杂的多领域请求

query = (
    "安排下周二下午 2 点与设计团队的 1 小时会议，"
    "并给他们发一封关于审查新设计稿的邮件提醒。"
)

for step in supervisor_agent.stream(
    {"messages": [{"role": "user", "content": query}]}
):
    for update in step.values():
        for message in update.get("messages", []):
            message.pretty_print()

================================== Ai Message ==================================
Tool Calls:
  schedule_event (call_YA68mqF0koZItCFPx0kGQfZi)
 Call ID: call_YA68mqF0koZItCFPx0kGQfZi
  Args:
    request: meeting with the design team next Tuesday at 2pm for 1 hour
  manage_email (call_XxqcJBvVIuKuRK794ZIzlLxx)
 Call ID: call_XxqcJBvVIuKuRK794ZIzlLxx
  Args:
    request: send the design team an email reminder about reviewing the new mockups
================================= Tool Message =================================
Name: schedule_event

你与设计团队的会议已安排在下周二 6 月 18 日下午 2:00 到 3:00。如果你需要添加更多详情或做更改，请告诉我！
================================= Tool Message =================================
Name: manage_email

我已向设计团队发送了一封邮件提醒，请他们审查新的设计稿。如果你需要包含更多信息或收件人，请告诉我！
================================== Ai Message ==================================

你与设计团队的会议已安排在下周二 6 月 18 日下午 2:00 到 3:00。

我还向设计团队发送了一封邮件提醒，请他们审查新的设计稿。

如果你想为会议添加更多细节或在邮件中包含额外信息，请告诉我！
监督者识别出这需要日历和邮件两个操作，调用 schedule_event 安排会议，然后调用 manage_email 发送提醒。每个子智能体完成其任务，监督者将两个结果综合为连贯的响应。
参阅 LangSmith 追踪 查看上述运行的详细信息流，包括每次聊天模型调用的提示词和响应。

​
完整可运行示例

以下是一个可运行脚本中的所有内容：

​
理解架构

你的系统有三个层次。底层包含需要精确格式的刚性 API 工具。中间层包含接受自然语言、将其转换为结构化 API 调用并返回自然语言确认的子智能体。顶层包含路由到高级能力并综合结果的监督者。 这种关注点分离提供了几个好处：每个层都有专注的职责，你可以在不影响现有层的情况下添加新领域，并且可以独立地测试和迭代每个层。

​
6. 添加人机协作审查

对敏感操作纳入人机协作审查是明智的做法。LangChain 包含内置中间件来审查工具调用，在此情况下是子智能体调用的工具。 让我们为两个子智能体添加人机协作审查：
  • 我们配置 create_calendar_event 和 send_email 工具为中断，允许所有响应类型（approve、edit、reject）
  • 我们仅向顶层智能体添加检查点器。这是暂停和恢复执行所必需的。
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware 
from langgraph.checkpoint.memory import InMemorySaver 


calendar_agent = create_agent(
    model,
    tools=[create_calendar_event, get_available_time_slots],
    system_prompt=CALENDAR_AGENT_PROMPT,
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={"create_calendar_event": True},
            description_prefix="日历事件待审批",
        ),
    ],
)

email_agent = create_agent(
    model,
    tools=[send_email],
    system_prompt=EMAIL_AGENT_PROMPT,
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={"send_email": True},
            description_prefix="外发邮件待审批",
        ),
    ],
)

supervisor_agent = create_agent(
    model,
    tools=[schedule_event, manage_email],
    system_prompt=SUPERVISOR_PROMPT,
    checkpointer=InMemorySaver(),
)
让我们重复查询。注意我们将中断事件收集到列表中以便下游访问： 
query = (
    "安排下周二下午 2 点与设计团队的 1 小时会议，"
    "并给他们发一封关于审查新设计稿的邮件提醒。"
)

config = {"configurable": {"thread_id": "6"}}

interrupts = []
for step in supervisor_agent.stream(
    {"messages": [{"role": "user", "content": query}]},
    config,
):
    for update in step.values():
        if isinstance(update, dict):
            for message in update.get("messages", []):
                message.pretty_print()
        else:
            interrupt_ = update[0]
            interrupts.append(interrupt_)
            print(f"\n已中断：{interrupt_.id}")

================================== Ai Message ==================================
Tool Calls:
  schedule_event (call_t4Wyn32ohaShpEZKuzZbl83z)
 Call ID: call_t4Wyn32ohaShpEZKuzZbl83z
  Args:
    request: Schedule a meeting with the design team next Tuesday at 2pm for 1 hour.
  manage_email (call_JWj4vDJ5VMnvkySymhCBm4IR)
 Call ID: call_JWj4vDJ5VMnvkySymhCBm4IR
  Args:
    request: Send an email reminder to the design team about reviewing the new mockups before our meeting next Tuesday at 2pm.

已中断：4f994c9721682a292af303ec1a46abb7

已中断：2b56f299be313ad8bc689eff02973f16
这次我们中断了执行。让我们检查中断事件： 
for interrupt_ in interrupts:
    for request in interrupt_.value["action_requests"]:
        print(f"已中断：{interrupt_.id}")
        print(f"{request['description']}\n")

已中断：4f994c9721682a292af303ec1a46abb7
日历事件待审批

Tool: create_calendar_event
Args: {'title': 'Meeting with the Design Team', 'start_time': '2024-06-18T14:00:00', 'end_time': '2024-06-18T15:00:00', 'attendees': ['design team']}

已中断：2b56f299be313ad8bc689eff02973f16
外发邮件待审批

Tool: send_email
Args: {'to': ['designteam@example.com'], 'subject': 'Reminder: Review New Mockups Before Meeting Next Tuesday at 2pm', 'body': "Hello Team,\n\nThis is a reminder to review the new mockups ahead of our meeting scheduled for next Tuesday at 2pm. Your feedback and insights will be valuable for our discussion and next steps.\n\nPlease ensure you've gone through the designs and are ready to share your thoughts during the meeting.\n\nThank you!\n\nBest regards,\n[Your Name]"}
我们可以使用 Command 通过引用 ID 来为每个中断指定决策。有关更多详情，请参阅人机协作指南。为了演示目的，这里我们将接受日历事件，但编辑外发邮件的主题： 
from langgraph.types import Command 

resume = {}
for interrupt_ in interrupts:
    if interrupt_.id == "2b56f299be313ad8bc689eff02973f16":
        # 编辑邮件
        edited_action = interrupt_.value["action_requests"][0].copy()
        edited_action["args"]["subject"] = "设计稿审查提醒"
        resume[interrupt_.id] = {
            "decisions": [{"type": "edit", "edited_action": edited_action}]
        }
    else:
        resume[interrupt_.id] = {"decisions": [{"type": "approve"}]}

interrupts = []
for step in supervisor_agent.stream(
    Command(resume=resume),
    config,
):
    for update in step.values():
        if isinstance(update, dict):
            for message in update.get("messages", []):
                message.pretty_print()
        else:
            interrupt_ = update[0]
            interrupts.append(interrupt_)
            print(f"\n已中断：{interrupt_.id}")

================================= Tool Message =================================
Name: schedule_event

你与设计团队的会议已安排在下周二 6 月 18 日下午 2:00 到 3:00。
================================= Tool Message =================================
Name: manage_email

你给设计团队的邮件提醒已发送。以下是发送的内容：

- 收件人：designteam@example.com
- 主题：设计稿审查提醒
- 正文：在下周二下午 2 点会议前审查新设计稿的提醒，请准备好反馈和讨论。

如果你需要进一步帮助，请告诉我！
================================== Ai Message ==================================

- 你与设计团队的会议已安排在下周二 6 月 18 日下午 2:00 到 3:00。
- 已向设计团队发送关于在会议前审查新设计稿的邮件提醒。

如果你需要进一步帮助，请告诉我！
运行按照我们的输入继续执行。

​
7. 进阶：控制信息流

默认情况下，子智能体只接收来自监督者的请求字符串。你可能希望传递额外的上下文，如对话历史或用户偏好。

​
向子智能体传递额外的对话上下文

from langchain.tools import tool, ToolRuntime

@tool
def schedule_event(
    request: str,
    runtime: ToolRuntime
) -> str:
    """使用自然语言安排日历事件。"""
    # 自定义子智能体接收的上下文
    original_user_message = next(
        message for message in runtime.state["messages"]
        if message.type == "human"
    )
    prompt = (
        "你正在协助以下用户咨询：\n\n"
        f"{original_user_message.text}\n\n"
        "你的任务是处理以下子请求：\n\n"
        f"{request}"
    )
    result = calendar_agent.invoke({
        "messages": [{"role": "user", "content": prompt}],
    })
    return result["messages"][-1].text
这允许子智能体看到完整的对话上下文，这对于解决诸如”安排在明天同一时间”（引用之前的对话）等歧义很有用。
你可以在 LangSmith 追踪的聊天模型调用中查看子智能体接收到的完整上下文。

​
控制监督者接收的内容

你也可以自定义返回给监督者的信息： 
import json

@tool
def schedule_event(request: str) -> str:
    """使用自然语言安排日历事件。"""
    result = calendar_agent.invoke({
        "messages": [{"role": "user", "content": request}]
    })

    # 选项 1：仅返回确认消息
    return result["messages"][-1].text

    # 选项 2：返回结构化数据
    # return json.dumps({
    #     "status": "success",
    #     "event_id": "evt_123",
    #     "summary": result["messages"][-1].text
    # })
重要提示： 确保子智能体的提示词强调其最终消息应包含所有相关信息。一个常见的失败模式是子智能体执行工具调用但不在最终响应中包含结果。

​
8. 关键要点

监督者模式创建了抽象层，每层都有清晰的职责。设计监督者系统时，从清晰的领域边界开始，为每个子智能体提供专注的工具和提示词。为监督者编写清晰的工具描述，在集成之前独立测试每个层，并根据你的特定需求控制信息流。
何时使用监督者模式当你有多个不同的领域（日历、邮件、CRM、数据库）、每个领域有多个工具或复杂逻辑、你想要集中的工作流控制，以及子智能体不需要直接与用户对话时，使用监督者模式。对于只有少量工具的简单情况，使用单个智能体。当智能体需要与用户对话时，使用交接。对于智能体间的对等协作，考虑其他多智能体模式。

​
下一步

了解交接的智能体间对话，探索上下文工程来微调信息流，阅读多智能体概览比较不同模式，并使用 LangSmith 调试和监控你的多智能体系统。
通过 MCP 连接这些文档到 Claude、VSCode 等，获取实时答案。
在 GitHub 上编辑此页面 或 提交 issue。