09 实战：用 Subagents 搭一个个人助理（主管 + 日历/邮件子智能体）

1
from langchain.tools import tool
2

3
@tool
4
def create_calendar_event(title: str, start_time: str,   # ISO："2024-01-15T14:00:00"
5
                          end_time: str,                 # ISO："2024-01-15T15:00:00"
6
                          attendees: list[str],          # 邮箱地址列表
7
                          location: str = "") -> str:
8
    """创建日历事件。要求严格的 ISO 时间格式。"""
9
    # 桩函数：真实项目里这里调 Google Calendar API / Outlook API
10
    return f"Event created: {title} from {start_time} to {end_time} with {len(attendees)} attendees"
11

12
@tool
13
def send_email(to: list[str], subject: str, body: str, cc: list[str] = []) -> str:
14
    """通过邮件 API 发送邮件。要求邮箱地址格式正确。"""
15
    return f"Email sent to {', '.join(to)} - Subject: {subject}"   # 桩函数：真实项目里调 SendGrid
16

17
@tool
18
def get_available_time_slots(attendees: list[str], date: str,   # ISO 格式："2024-01-15"
19
                             duration_minutes: int) -> list[str]:
20
    """查询指定日期内这些参与者的日历空闲情况。"""
21
    return ["09:00", "14:00", "16:00"]   # 桩函数：真实项目里查日历 API

1
@tool
2
def schedule_event(request: str) -> str:
3
    """用自然语言安排日历事件。用户想创建、修改或查询日历安排时用它；内部负责时间
4
    解析、空闲检查和事件创建。输入：自然语言的日程请求（例如"下周二下午 2 点开会"）。
5
    """
6
    result = calendar_agent.invoke({"messages": [{"role": "user", "content": request}]})
7
    return result["messages"][-1].text   # 只把最终答复交回主管
8

9
@tool
10
def manage_email(request: str) -> str:
11
    """用自然语言发送邮件。用户想发通知、提醒或任何邮件沟通时用它；内部负责提取收件
12
    人、生成标题和撰写正文。输入：自然语言的邮件请求（例如"给他们发个会议提醒"）。
13
    """
14
    result = email_agent.invoke({"messages": [{"role": "user", "content": request}]})
15
    return result["messages"][-1].text

1
SUPERVISOR_PROMPT = (
2
    "你是一个乐于助人的个人助理。你可以安排日历事件，也可以发送邮件。"
3
    "把用户的请求拆解成合适的工具调用，并协调这些结果。"
4
    "当一个请求涉及多个动作时，按需顺序或并行地使用多个工具。"
5
)
6

7
supervisor_agent = create_agent(model, tools=[schedule_event, manage_email],   # 只有两个"高层工具"
8
                                system_prompt=SUPERVISOR_PROMPT)

1
query = (
2
    "下周二下午 2 点和设计团队开 1 小时的会，"
3
    "并发封邮件提醒他们评审新的设计稿。"
4
)
5

6
stream = supervisor_agent.stream_events(
7
    {"messages": [{"role": "user", "content": query}]}, version="v3",
8
)
9
# 遍历方式同 §2；这一层的 tool_calls 里出现的，就是那两个"子智能体工具"

1
from langchain.agents.middleware import HumanInTheLoopMiddleware
2
from langgraph.checkpoint.memory import InMemorySaver
3

4
calendar_agent = create_agent(
5
    model,
6
    tools=[create_calendar_event, get_available_time_slots],
7
    system_prompt=CALENDAR_AGENT_PROMPT,
8
    middleware=[HumanInTheLoopMiddleware(
9
        interrupt_on={"create_calendar_event": True},   # 命中即暂停，等人工决定
10
        description_prefix="日历事件待审批",
11
    )],
12
)
13

14
# 邮件子智能体同理，只把拦截对象换成 send_email
15
email_agent = create_agent(
16
    model, tools=[send_email], system_prompt=EMAIL_AGENT_PROMPT,
17
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"send_email": True},
18
                                         description_prefix="外发邮件待审批")],
19
)

1
# 检查点只加在主管这一层：暂停与恢复用的都是主管的状态
2
supervisor_agent = create_agent(
3
    model,
4
    tools=[schedule_event, manage_email],
5
    system_prompt=SUPERVISOR_PROMPT,
6
    checkpointer=InMemorySaver(),
7
)

1
config = {"configurable": {"thread_id": "6"}}   # 检查点按 thread_id 归档
2

3
interrupts = []
4
stream = supervisor_agent.stream_events(
5
    {"messages": [{"role": "user", "content": query}]},
6
    config, version="v3",        # 带上 thread_id，检查点才有地方落
7
)
8
# 遍历同 §2；这次多了"有没有被拦住"的判断
9
if stream.interrupted:
10
    for interrupt_ in stream.interrupts:
11
        interrupts.append(interrupt_)
12
        print(f"\n已中断: {interrupt_.id}")

1
from langgraph.types import Command
2

3
resume = {}
4
for interrupt_ in interrupts:
5
    if interrupt_.id == "2b56f299be313ad8bc689eff02973f16":   # 邮件那条：改标题后放行
6
        edited_action = interrupt_.value["action_requests"][0].copy()
7
        edited_action["args"]["subject"] = "设计稿提醒"
8
        resume[interrupt_.id] = {"decisions": [{"type": "edit", "edited_action": edited_action}]}
9
    else:                                                     # 日历那条：直接批准
10
        resume[interrupt_.id] = {"decisions": [{"type": "approve"}]}
11

12
stream = supervisor_agent.stream_events(Command(resume=resume), config, version="v3")

1
from langchain.tools import tool, ToolRuntime
2

3
@tool
4
def schedule_event(request: str, runtime: ToolRuntime) -> str:
5
    """用自然语言安排日历事件。"""
6
    # 从主管的 state 里取出用户最初那一句原话
7
    original_user_message = next(
8
        m for m in runtime.state["messages"] if m.type == "human")
9
    prompt = (f"你正在协助处理下面这位用户的请求：\n\n{original_user_message.text}\n\n"
10
              f"你这次要完成的子任务：\n\n{request}")
11
    result = calendar_agent.invoke({"messages": [{"role": "user", "content": prompt}]})
12
    return result["messages"][-1].text

1
import json
2

3
# ...（前面几行同 §3：拿到 result = calendar_agent.invoke(...)）
4
# 方案一：只回一句确认文本（上下文最省）
5
return result["messages"][-1].text
6
# 方案二：回结构化数据，主管更容易做后续判断
7
# return json.dumps({"status": "success", "event_id": "evt_123",
8
#                    "summary": result["messages"][-1].text})