零基础入门- LangChain1.0 多智能体系统（人工中断）

前期准备

环境准备

我们需要安装相关的依赖：

pip install langchain langchain-community dashscope

另外我们还需要到百炼大模型平台获取通义千问大模型的 API_KEY。

代码准备

我们还是拿上节课官网的代码来进行使用：

"""
Personal Assistant Supervisor Example

This example demonstrates the tool calling pattern for multi-agent systems.
A supervisor agent coordinates specialized sub-agents (calendar and email)
that are wrapped as tools.
"""

from langchain_core.tools import tool
from langchain.agents import create_agent
from langchain_community.chat_models import ChatTongyi

# ============================================================================
# Step 1: Define low-level API tools (stubbed)
# ============================================================================

@tool
def create_calendar_event(
    title: str,
    start_time: str,  # ISO format: "2024-01-15T14:00:00"
    end_time: str,    # ISO format: "2024-01-15T15:00:00"
    attendees: list[str],  # email addresses
    location: str = ""
) -> str:
    """Create a calendar event. Requires exact ISO datetime format."""
    returnf"Event created: {title} from {start_time} to {end_time} with {len(attendees)} attendees"

@tool
def send_email(
    to: list[str],      # email addresses
    subject: str,
    body: str,
    cc: list[str] = []
) -> str:
    """Send an email via email API. Requires properly formatted addresses."""
    returnf"Email sent to {', '.join(to)} - Subject: {subject}"

@tool
def get_available_time_slots(
    attendees: list[str],
    date: str,  # ISO format: "2024-01-15"
    duration_minutes: int
) -> list[str]:
    """Check calendar availability for given attendees on a specific date."""
    return ["09:00", "14:00", "16:00"]

# ============================================================================
# Step 2: Create specialized sub-agents
# ============================================================================

model = ChatTongyi(model="qwen-max")  # for example

calendar_agent = create_agent(
    model,
    tools=[create_calendar_event, get_available_time_slots],
    system_prompt=(
        "You are a calendar scheduling assistant. "
        "Parse natural language scheduling requests (e.g., 'next Tuesday at 2pm') "
        "into proper ISO datetime formats. "
        "Use get_available_time_slots to check availability when needed. "
        "Use create_calendar_event to schedule events. "
        "Always confirm what was scheduled in your final response."
    )
)

email_agent = create_agent(
    model,
    tools=[send_email],
    system_prompt=(
        "You are an email assistant. "
        "Compose professional emails based on natural language requests. "
        "Extract recipient information and craft appropriate subject lines and body text. "
        "Use send_email to send the message. "
        "Always confirm what was sent in your final response."
    )
)

# ============================================================================
# Step 3: Wrap sub-agents as tools for the supervisor
# ============================================================================

@tool
def schedule_event(request: str) -> str:
    """Schedule calendar events using natural language.

    Use this when the user wants to create, modify, or check calendar appointments.
    Handles date/time parsing, availability checking, and event creation.

    Input: Natural language scheduling request (e.g., 'meeting with design team
    next Tuesday at 2pm')
    """
    result = calendar_agent.invoke({
        "messages": [{"role": "user", "content": request}]
    })
    return result["messages"][-1].text

@tool
def manage_email(request: str) -> str:
    """Send emails using natural language.

    Use this when the user wants to send notifications, reminders, or any email
    communication. Handles recipient extraction, subject generation, and email
    composition.

    Input: Natural language email request (e.g., 'send them a reminder about
    the meeting')
    """
    result = email_agent.invoke({
        "messages": [{"role": "user", "content": request}]
    })
    return result["messages"][-1].text

# ============================================================================
# Step 4: Create the supervisor agent
# ============================================================================

supervisor_agent = create_agent(
    model,
    tools=[schedule_event, manage_email],
    system_prompt=(
        "You are a helpful personal assistant. "
        "You can schedule calendar events and send emails. "
        "Break down user requests into appropriate tool calls and coordinate the results. "
        "When a request involves multiple actions, use multiple tools in sequence."
    )
)

# ============================================================================
# Step 5: Use the supervisor
# ============================================================================

if __name__ == "__main__":
    # Example: User request requiring both calendar and email coordination
    user_request = (
        "Schedule a meeting with the design team next Tuesday at 2pm for 1 hour, "
        "and send them an email reminder about reviewing the new mockups."
    )

    print("User Request:", user_request)
    print("
" + "="*80 + "
")

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

加入 HITL（Human-in-the-Loop）人类审核机制

概念

在LangChain1.0提到了一个内置中间件 HumanInTheLoopMiddleware，这个中间件允许我们能够在“子代理调用工具之前”插入人工判断点。

列如当智能体执行关键操作（如发邮件、创建日程）时，先中断流程，由人类手动“审批 / 编辑 / 拒绝”，再继续执行。

那在我们的系统中，我们也可以添加进该机制，从而实现：

确保发出去的邮件不会出错
日程安排前想确认时间和人选
涉及敏感操作（下单、删库、通知大群）能够提前预判

给 Sub-agent 加上中断中间件

第一我们可以给两个智能体加上中间件：

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware 

calendar_agent = create_agent(
    model,
    tools=[create_calendar_event, get_available_time_slots],
    system_prompt=(
        "You are a calendar scheduling assistant. "
        "Parse natural language scheduling requests (e.g., 'next Tuesday at 2pm') "
        "into proper ISO datetime formats. "
        "Use get_available_time_slots to check availability when needed. "
        "Use create_calendar_event to schedule events. "
        "Always confirm what was scheduled in your final response."
    ),
    middleware=[ 
        HumanInTheLoopMiddleware( 
            interrupt_on={"create_calendar_event": True}, 
            description_prefix="Calendar event pending approval", 
        ), 
    ], 
)

email_agent = create_agent(
    model,
    tools=[send_email],
    system_prompt=(
        "You are an email assistant. "
        "Compose professional emails based on natural language requests. "
        "Extract recipient information and craft appropriate subject lines and body text. "
        "Use send_email to send the message. "
        "Always confirm what was sent in your final response."
    ),
    middleware=[ 
        HumanInTheLoopMiddleware( 
            interrupt_on={"send_email": True}, 
            description_prefix="Outbound email pending approval", 
        ), 
    ], 
)

对于第一个 calendar_agent 而言，并不是两个工具都需要标记为需要人工审核，而是只有 create_calendar_event 这个工具需要进行审核。如果只是单纯使用工具 get_available_time_slots 查询可用时间的话是不需要审核的。

而对于第二个 email_agent 而言，send_email 这个工具也是需要进行审核。

给 Supervisor 加上记忆 checkpointer

虽然对于 Supervisor Agent 而言，由于 sub_agent 里已经设置了人工审核，所以这里就不需要再进行设置了。但是 HumanInTheLoopMiddleware 里有一个硬性要求，就是需要添加上记忆，因此我们可以为 supervisor_agent 添加 checkpointer 参数：

from langgraph.checkpoint.memory import InMemorySaver

supervisor_agent = create_agent(
    model=model,
    tools=[schedule_event, manage_email],
    system_prompt=(
        "You are a helpful personal assistant. "
        "You can schedule calendar events and send emails. "
        "Break down user requests into appropriate tool calls and coordinate the results. "
        "When a request involves multiple actions, use multiple tools in sequence."
    ),
    checkpointer=InMemorySaver()  #  激活状态保存
)

这样子，在运行过程中的信息就能够保存下来了，那么在传入人工的指令后也可以基于之前的记忆继续恢复了。

对于子代理而言，由于其本身就是为了完成任务而设置的，除非是一些需要基于上下文完成任务的智能体工具，不然是不需要添加记忆进去的。添加了反而可能导致状态不一致或冗余。

智能体调用

当然和前面一样，我们一样可以提出问题，然后调用 supervisor_agent 进行回复：

from langchain_core.messages import HumanMessage
from langgraph.types import Command
import json

query = (
    "Schedule a meeting with the design team next Tuesday at 2pm for 1 hour, "
    "and send them an email reminder about reviewing the new mockups."
)

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

#  执行 + 中断检测
response = supervisor_agent.invoke(
    {"messages": [HumanMessage(content=query)]},
    config=config
)

但是由于存在着需要人工介入的情况，因此我们可以参考前面讲解 HumanInTheLoopMiddleware 时使用的方法来实现处理：

#  循环处理多个中断（逐个审批直到流程结束）
while"__interrupt__"in response:
    print("
 检测到中断，进入人工审核环节...")

    #  打印用户原始提问
    user_msg = next((m.content for m in response["messages"] if m.type == "human"), "无")
    print(f"
 用户提问：{user_msg}")

    # 遍历中断对象（可能包含多个 action request）
    for interrupt in response["__interrupt__"]:
        for i, req in enumerate(interrupt.value["action_requests"]):
            print(f" 描述：{req.get('description', '无')}")
            print(f"✅ 可选操作：{interrupt.value['review_configs'][i]['allowed_decisions']}")

        #  人工输入决策
        decision_type = input("
请输入你的决定（approve / edit / reject）：").strip().lower()
        if decision_type notin ["approve", "edit", "reject"]:
            print("⚠️ 无效输入，默认设置为 reject")
            decision_type = "reject"

        decision_payload = {"type": decision_type}

        # ✍️ 如果选择 edit：展示原请求并让用户直接输入完整 JSON
        if decision_type == "edit":
            print("
️ 当前模型原始调用指令如下：")
            print(json.dumps(req, indent=2, ensure_ascii=False))
            print("
✍️ 请粘贴你想执行的完整新调用 JSON（例如：）")
            print('{"name": "send_email", "args": {"to": ["x@example.com"], "subject": "测试", "body": "内容"}}')

            try:
                new_raw = input("
请输入新的调用 JSON：").strip()
                decision_payload["edited_action"] = json.loads(new_raw)
            except Exception as e:
                print(f"❌ JSON 格式错误：{e}")
                print("⚠️ 自动拒绝此调用")
                decision_payload = {"type": "reject", "message": "编辑失败，输入格式错误"}

        # ❌ 如果拒绝，填写理由
        elif decision_type == "reject":
            reason = input("请输入拒绝理由：")
            decision_payload["message"] = reason

        # ✅ 恢复执行（Resume）
        try:
            response = supervisor_agent.invoke(
                Command(resume={interrupt.id: {"decisions": [decision_payload]}}),
                config=config
            )
        except Exception as e:
            print(f"❌ 执行恢复时发生错误：{e}")
            breakr/>
# ✅ 最终结果输出
print("
 流程结束，模型最终回复：")
print(response["messages"][-1].content)

当需要调用 create_calendar_event 或 send_email 这两个工具时，就会进入到中断的界面，列如我们直接运行代码，会出现：

 检测到中断，进入人工审核环节...

 用户提问：Schedule a meeting with the design team next Tuesday at 2pm for 1 hour, and send them an email reminder about reviewing the new mockups.
 描述：Outbound email pending approval

Tool: send_email
Args: {'to': ['design.team@example.com'], 'subject': 'Reminder: Review New Mockups Before Meeting', 'body': "Hello Design Team, 

This is a friendly reminder to review the new mockups before our scheduled meeting. Your feedback and insights are crucial for the project's progress.

Best regards,
[Your Name]"}
✅ 可选操作：['approve', 'edit', 'reject']
请输入你的决定（approve / edit / reject）：

（1）approve

这里我们就可以输入我们的决定，如果输入的是 approve 的话，那就会继续运行下去，直到遇到下一个需要决定的工具或模型调用结束返回结果：

 检测到中断，进入人工审核环节...

 用户提问：Schedule a meeting with the design team next Tuesday at 2pm for1 hour, and send them an email reminder about reviewing the new mockups.
 描述：Outbound email pending approval

Tool: send_email
Args: {'to': ['design.team@example.com'], 'subject': 'Reminder: Review New Mockups Before Meeting', 'body': "Hello Design Team, 

This is a friendly reminder to review the new mockups before our scheduled meeting. Your feedback and insights are crucial for the project's progress.

Best regards,
[Your Name]"}
✅ 可选操作：['approve', 'edit', 'reject']

请输入你的决定（approve / edit / reject）：approve
 描述：Calendar event pending approval

Tool: create_calendar_event
Args: {'title': 'Meeting with Design Team', 'start_time': '2023-10-10T14:00:00', 'end_time': '2023-10-10T15:00:00', 'attendees': ['design_team@example.com'], 'location': ''}
✅ 可选操作：['approve', 'edit', 'reject']

请输入你的决定（approve / edit / reject）：approve

 流程结束，模型最终回复：
The meeting with the design team has been scheduled for next Tuesday, October 10th, from2 PM to 3 PM. Additionally, an email reminder has been sent to the design team, reminding them to review the new mockups before the meeting.

（2）reject

如果我们不同意返回拒绝后，我们还需要输入拒绝的理由。这个时候流程并不会就此终止，由于我们拒绝的不过只是 sub-agent 而已，其会将我们反馈后的信息给到 Supervisor agent 那，然后 Supervisor agent 会根据我们拒绝的意见进行调整及修改：

 检测到中断，进入人工审核环节...

 用户提问：Schedule a meeting with the design team next Tuesday at 2pm for1 hour, and send them an email reminder about reviewing the new mockups.
 描述：Outbound email pending approval

Tool: send_email
Args: {'to': ['design.team@example.com'], 'subject': 'Reminder: Review New Mockups Before Meeting', 'body': 'Hello Design Team,

I hope this message finds you well. This is a friendly reminder to review the new mockups before our scheduled meeting. Your feedback is crucial, and we want to ensure that we make the most out of our discussion.

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

Best regards,
[Your Name]'}
✅ 可选操作：['approve', 'edit', 'reject']

请输入你的决定（approve / edit / reject）：reject
请输入拒绝理由：语气不够诚恳
 描述：Calendar event pending approval

Tool: create_calendar_event
Args: {'title': 'Meeting with Design Team', 'start_time': '2023-10-10T14:00:00', 'end_time': '2023-10-10T15:00:00', 'attendees': ['design_team@example.com'], 'location': ''}
✅ 可选操作：['approve', 'edit', 'reject']

请输入你的决定（approve / edit / reject）：approve

 检测到中断，进入人工审核环节...

 用户提问：Schedule a meeting with the design team next Tuesday at 2pm for1 hour, and send them an email reminder about reviewing the new mockups.
 描述：Outbound email pending approval

Tool: send_email
Args: {'to': ['design.team@example.com'], 'subject': 'Friendly Reminder: New Mockups Review Before Our Meeting', 'body': "Hello Design Team,

I hope this message finds you well. I'm reaching out with a gentle reminder to take some time to review the new mockups ahead of our upcoming meeting. Your insights are incredibly valuable, and I believe they will be instrumental in making our session as productive as possible.

If you have any questions or if there's anything else you need from me, please don't hesitate to reach out.

Looking forward to our discussion.

Warm regards,
[Your Name]"}
✅ 可选操作：['approve', 'edit', 'reject']

请输入你的决定（approve / edit / reject）：approve

 流程结束，模型最终回复：
The meeting with the design team has been scheduled for next Tuesday, October 10, from2:00 PM to 3:00 PM. Additionally, an email reminder has been sent to the design team. The email kindly asks them to review the new mockups before our meeting. Here are the details of the email that was sent:

- **To:** design.team@example.com
- **Subject:** Friendly Reminder: New Mockups Review Before Our Meeting
- **Body:**

  Hello Design Team,

  I hope this message finds you well. I'm reaching out with a gentle reminder to take some time to review the new mockups ahead of our upcoming meeting. Your insights are incredibly valuable, and I believe they will be instrumental in making our session as productive as possible.        

  If you have any questions or if there's anything else you need from me, please don't hesitate to reach out.

  Looking forward to our discussion.

  Warm regards,
  [Your Name]
  

If you need any further assistance or another email, feel free to let me know!

列如这里一开始说想要去发邮件，我拒绝了说要正式一点。然后后面其正常调用了一下时间后，再更新了一版更正式的邮箱信息给我了。所以可以看到，通过人工的介入能够让模型更符合我们的想法去完成任务。

（3）edit

那最后一种策略就是我们自己来调整。这里我们就需要修改完整的调用命令，然后回传过去，让其执行新的操作：

 检测到中断，进入人工审核环节...

 用户提问：Schedule a meeting with the design team next Tuesday at 2pm for1 hour, and send them an email reminder about reviewing the new mockups.
 描述：Outbound email pending approval

Tool: send_email
Args: {'to': ['design.team@company.com'], 'subject': 'Reminder: Review New Mockups', 'body': 'Dear Design Team, 

This is a friendly reminder to review the new mockups that have been shared with you. Your feedback is essential for us to move forward with the project. Please take some time to go through them and share your thoughts by the end of this week.

Best regards,
[Your Name]'}
✅ 可选操作：['approve', 'edit', 'reject']

请输入你的决定（approve / edit / reject）：edit

️ 当前模型原始调用指令如下：
{
"name": "send_email",
"args": {
    "to": [
      "design.team@company.com"
    ],
    "subject": "Reminder: Review New Mockups",
    "body": "Dear Design Team, 

This is a friendly reminder to review the new mockups that have been shared with you. Your feedback is essential for us to move forward with the project. Please take some time to go through them and share your thoughts by the end of this week.

Best regards,
[Your Name]"
  },
"description": "Outbound email pending approval

Tool: send_email
Args: {'to': ['design.team@company.com'], 'subject': 'Reminder: Review New Mockups', 'body': 'Dear Design Team, \n\nThis is a friendly reminder to review the new mockups that have been shared with you. Your feedback is essential for us to move forward with the project. Please take some time to go through them and share your thoughts by the end of this week.\n\nBest regards,\n[Your Name]'}"
}

✍️ 请粘贴你想执行的完整新调用 JSON（例如：）
{"name": "send_email", "args": {"to": ["x@example.com"], "subject": "测试", "body": "内容"}}

请输入新的调用 JSON：{"name": "send_email", "args": {"to": ["x@example.com"], "subject": "测试", "body": "内容"}}
 描述：Calendar event pending approval

Tool: create_calendar_event
Args: {'title': 'Meeting with Design Team', 'start_time': '2023-10-10T14:00:00', 'end_time': '2023-10-10T15:00:00', 'attendees': ['design_team@example.com'], 'location': ''}
✅ 可选操作：['approve', 'edit', 'reject']

请输入你的决定（approve / edit / reject）：approve

 流程结束，模型最终回复：
The meeting with the design team has been scheduled for next Tuesday, October 10th, from2 PM to 3 PM. Additionally, an email reminder has been sent to the design team at design.team@company.com. The subject of the email is"Reminder: Review New Mockups"and it kindly asks them to review the new mockups and provide their feedback by the end of the week.

列如这里我们就把本来传给 design.team@company.com 的内容改成了传给 x@example.com ，并且把里面内容都改变了。但是 Supervisor agent 还是觉得自己传给的是正确的人。所以如果我们觉得模型出的问题比较小，我们自己来改也可以的话，那就可以使用 edit 方法。但是如果问题比较大想让大模型自己来改的话，那就使用 reject，并输入出问题的地方，让大模型自己来进行修改。

控制信息流

所谓的控制信息流，实则就是掌握 不同层级的智能体看到什么、返回什么。

如果子代理获取到的信息太少，那么可能就会执行错误。如果看到的信息太多，又容易造成提示干扰或暴露隐私。类似的，要是子代理给模型的信息太多或者太少都可能会造成影响。

因此我们可以根据任务定制信息获取的方式。

向子代理注入额外上下文

在我们当前的代码里，实则 sub_agent 能看到的信息只是 Supervisor_agent 发送的查询请求：

@tool
def schedule_event(request: str) -> str:
    """Schedule calendar events using natural language.

    Use this when the user wants to create, modify, or check calendar appointments.
    Handles date/time parsing, availability checking, and event creation.

    Input: Natural language scheduling request (e.g., 'meeting with design team
    next Tuesday at 2pm')
    """
    result = calendar_agent.invoke({
        "messages": [{"role": "user", "content": request}]
    })
    return result["messages"][-1].text

但如果我们希望让子代理看到“上层全部对话历史”或其它状态信息，我们可以通过 ToolRuntime 注入状态：

from langchain.tools import tool, ToolRuntime

@tool
def schedule_event(
    request: str,
    runtime: ToolRuntime
) -> str:
    """Schedule calendar events using natural language.

    Use this when the user wants to create, modify, or check calendar appointments.
    Handles date/time parsing, availability checking, and event creation.

    Input: Natural language scheduling request (e.g., 'meeting with design team
    next Tuesday at 2pm')
    """
    # Customize context received by sub-agent
    original_user_message = next(
        message for message in runtime.state["messages"]
        if message.type == "human"
    )
    prompt = (
        "You are assisting with the following user inquiry:

"
        f"{original_user_message.text}

"
        "You are tasked with the following sub-request:

"
        f"{request}"
    )
    result = calendar_agent.invoke({
        "messages": [{"role": "user", "content": prompt}],
    })
    return result["messages"][-1].text

这里面我们获取到的是在 runtime 中的 state[“messages”] （也就是完整历史对话信息）里的 HumanMessage 信息。这里的目的就是提取用户原始消息（即当前轮输入）给到 sub_agent。并且重新构造一个新的 prompt ，然后再进行调用。

所以这个时候 sub_agent 不再只能接收到调用的信息，还能知道部分当前任务的信息并智能化的做出调用的决定。列如更新后的 prompt 可能长下面这样：

You are assisting with the following user inquiry:

Schedule a meeting with the design team next Tuesday at 2pm for 1 hour, and send them an email reminder about reviewing the new mockups.        

You are tasked with the following sub-request:

Schedule a meeting with the design team next Tuesday at 2pm for 1 hour

那这里我额外补充一下，这个 ToolRuntime 里包含的信息有：

ToolRuntime(
  state: dict | BaseModel,    #  AgentState 当前智能体状态（短期记忆）
  context: dict | None,     #  Runtime Context 执行上下文（运行时传入）
  config: RunnableConfig,    # ⚙️ 当前执行配置，包含 run_id、metadata、tags 等
  stream_writer: StreamWriter,  #  实时流式输出对象
  tool_call_id: str | None,   #  当前工具调用唯一标识符
  store: BaseStore | None    #  Store 持久化存储接口（长期记忆）
)

这样我们就可以在工具时将需要的信息传入进来了。

让子代理返回更多信息

除了向子代理里注入更多的信息，类似的我们可以让子代理返回更多的信息，列如在 return 的时候并不是仅仅把 result[“messages”][-1].text ，而是返回一些状态信息：

import json

@tool
def schedule_event(request: str) -> str:
    """Schedule calendar events using natural language."""
    result = calendar_agent.invoke({
        "messages": [{"role": "user", "content": request}]
    })

    # Option 1: Return just the confirmation message
    return result["messages"][-1].text

    # Option 2: Return structured data
    # return json.dumps({
    #     "status": "success",
    #     "event_id": "evt_123",
    #     "summary": result["messages"][-1].text
    # })

通过 json.dumps ，我们可以把特定格式的 json 内容转为字符串的形式然后再返回给大模型。列如上面的内容会得到：

"{"status": "success", "event_id": "evt_123", "summary": "The event has been scheduled for next Tuesday at 2 PM."}"

这样Supervisor_agent 就会知道成功运行且运行的 id 是 evt_123 了。

本章完整的代码如下所示：

"""
Personal Assistant Supervisor Example

This example demonstrates the tool calling pattern for multi-agent systems.
A supervisor agent coordinates specialized sub-agents (calendar and email)
that are wrapped as tools.
"""

from langchain_core.tools import tool
from langchain.agents import create_agent
from langchain_community.chat_models import ChatTongyi

# ============================================================================
# Step 1: Define low-level API tools (stubbed)
# ============================================================================

@tool
def create_calendar_event(
    title: str,
    start_time: str,  # ISO format: "2024-01-15T14:00:00"
    end_time: str,    # ISO format: "2024-01-15T15:00:00"
    attendees: list[str],  # email addresses
    location: str = ""
) -> str:
    """Create a calendar event. Requires exact ISO datetime format."""
    returnf"Event created: {title} from {start_time} to {end_time} with {len(attendees)} attendees"

@tool
def send_email(
    to: list[str],      # email addresses
    subject: str,
    body: str,
    cc: list[str] = []
) -> str:
    """Send an email via email API. Requires properly formatted addresses."""
    returnf"Email sent to {', '.join(to)} - Subject: {subject}"

@tool
def get_available_time_slots(
    attendees: list[str],
    date: str,  # ISO format: "2024-01-15"
    duration_minutes: int
) -> list[str]:
    """Check calendar availability for given attendees on a specific date."""
    return ["09:00", "14:00", "16:00"]

# ============================================================================
# Step 2: Create specialized sub-agents
# ============================================================================

model = ChatTongyi(model="qwen-max")  # for example

from langchain.agents.middleware import HumanInTheLoopMiddleware

calendar_agent = create_agent(
    model,
    tools=[create_calendar_event, get_available_time_slots],
    system_prompt=(
        "You are a calendar scheduling assistant. "
        "Parse natural language scheduling requests (e.g., 'next Tuesday at 2pm') "
        "into proper ISO datetime formats. "
        "Use get_available_time_slots to check availability when needed. "
        "Use create_calendar_event to schedule events. "
        "Always confirm what was scheduled in your final response."
    ),
    middleware=[ 
        HumanInTheLoopMiddleware( 
            interrupt_on={"create_calendar_event": True}, 
            description_prefix="Calendar event pending approval", 
        ), 
    ], 
)

email_agent = create_agent(
    model,
    tools=[send_email],
    system_prompt=(
        "You are an email assistant. "
        "Compose professional emails based on natural language requests. "
        "Extract recipient information and craft appropriate subject lines and body text. "
        "Use send_email to send the message. "
        "Always confirm what was sent in your final response."
    ),
    middleware=[ 
        HumanInTheLoopMiddleware( 
            interrupt_on={"send_email": True}, 
            description_prefix="Outbound email pending approval", 
        ), 
    ], 
)

# ============================================================================
# Step 3: Wrap sub-agents as tools for the supervisor
# ============================================================================

from langchain.tools import tool, ToolRuntime

@tool
def schedule_event(
    request: str,
    runtime: ToolRuntime
) -> str:
    """Schedule calendar events using natural language.

    Use this when the user wants to create, modify, or check calendar appointments.
    Handles date/time parsing, availability checking, and event creation.

    Input: Natural language scheduling request (e.g., 'meeting with design team
    next Tuesday at 2pm')
    """
    # Customize context received by sub-agent
    original_user_message = next(
        message for message in runtime.state["messages"]
        if message.type == "human"
    )
    prompt = (
        "You are assisting with the following user inquiry:

"
        f"{original_user_message.text}

"
        "You are tasked with the following sub-request:

"
        f"{request}"
    )
    result = calendar_agent.invoke({
        "messages": [{"role": "user", "content": prompt}],
    })
    return json.dumps({
        "status": "success",
        "event_id": "evt_123",
        "summary": result["messages"][-1].text
    })

@tool
def manage_email(request: str) -> str:
    """Send emails using natural language.

    Use this when the user wants to send notifications, reminders, or any email
    communication. Handles recipient extraction, subject generation, and email
    composition.

    Input: Natural language email request (e.g., 'send them a reminder about
    the meeting')
    """
    result = email_agent.invoke({
        "messages": [{"role": "user", "content": request}]
    })
    return result["messages"][-1].text

# ============================================================================
# Step 4: Create the supervisor agent
# ============================================================================

from langgraph.checkpoint.memory import InMemorySaver

supervisor_agent = create_agent(
    model=model,
    tools=[schedule_event, manage_email],
    system_prompt=(
        "You are a helpful personal assistant. "
        "You can schedule calendar events and send emails. "
        "Break down user requests into appropriate tool calls and coordinate the results. "
        "When a request involves multiple actions, use multiple tools in sequence."
    ),
    checkpointer=InMemorySaver()  #  激活状态保存
)<br/># ============================================================================<br/># Step 5: Use the supervisor<br/># ============================================================================
<br/># =====================  启动调用 + 人工审核逻辑 =====================

from langchain_core.messages import HumanMessage
from langgraph.types import Command
import json

query = (
    "Schedule a meeting with the design team next Tuesday at 2pm for 1 hour, "
    "and send them an email reminder about reviewing the new mockups."
)

config = {"configurable": {"thread_id": "6"}}r/>
#  执行 + 中断检测
response = supervisor_agent.invoke(
    {"messages": [HumanMessage(content=query)]},
    config=config
)r/>
#  循环处理多个中断（逐个审批直到流程结束）
while"__interrupt__"in response:
    print("
 检测到中断，进入人工审核环节...")

    #  打印用户原始提问
    user_msg = next((m.content for m in response["messages"] if m.type == "human"), "无")
    print(f"
 用户提问：{user_msg}")
<br/>    # 遍历中断对象（可能包含多个 action request）
    for interrupt in response["__interrupt__"]:
        for i, req in enumerate(interrupt.value["action_requests"]):
            print(f" 描述：{req.get('description', '无')}")
            print(f"✅ 可选操作：{interrupt.value['review_configs'][i]['allowed_decisions']}")

        #  人工输入决策
        decision_type = input("
请输入你的决定（approve / edit / reject）：").strip().lower()
        if decision_type notin ["approve", "edit", "reject"]:
            print("⚠️ 无效输入，默认设置为 reject")
            decision_type = "reject"

        decision_payload = {"type": decision_type}

        # ✍️ 如果选择 edit：展示原请求并让用户直接输入完整 JSON
        if decision_type == "edit":
            print("
️ 当前模型原始调用指令如下：")
            print(json.dumps(req, indent=2, ensure_ascii=False))
            print("
✍️ 请粘贴你想执行的完整新调用 JSON（例如：）")
            print('{"name": "send_email", "args": {"to": ["x@example.com"], "subject": "测试", "body": "内容"}}')

            try:
                new_raw = input("
请输入新的调用 JSON：").strip()
                decision_payload["edited_action"] = json.loads(new_raw)
            except Exception as e:
                print(f"❌ JSON 格式错误：{e}")
                print("⚠️ 自动拒绝此调用")
                decision_payload = {"type": "reject", "message": "编辑失败，输入格式错误"}

        # ❌ 如果拒绝，填写理由
        elif decision_type == "reject":
            reason = input("请输入拒绝理由：")
            decision_payload["message"] = reason

        # ✅ 恢复执行（Resume）
        try:
            response = supervisor_agent.invoke(
                Command(resume={interrupt.id: {"decisions": [decision_payload]}}),
                config=config
            )
        except Exception as e:
            print(f"❌ 执行恢复时发生错误：{e}")
            breakr/>
# ✅ 最终结果输出
print("
 流程结束，模型最终回复：")
print(response["messages"][-1].content)

总结

总的来说，本文通过一个“智能个人助理”示例，完整展示了如何构建一个可控、可协作、可审查的多智能体系统。系统由 Supervisor 智能体协调多个子智能体（如日程助手、邮件助手），并引入 Human-in-the-Loop 中间件，实现对关键操作（如发邮件、建日程）的人工干预与审核。同时，结合上下文注入与状态记忆机制，确保任务在智能化执行的同时具备良好的可恢复性与信息安全性。整体流程清晰，架构合理，为构建安全可控的智能体系统提供了实用模板与设计思路。