当前位置：首页 > news >正文

【由浅入深探究langchain】第二十一集-多智能体Supervisor Agent（上）

news 2026/6/18 19:55:36

开篇

为什么我们需要多智能体？

在传统的 LLM 开发中，我们往往倾向于给一个 Agent 堆叠无数个 Tool。但随着业务逻辑变复杂，单一 Agent 会面临“选择困难症”（工具回调错误）和“上下文迷失”的问题。

LangChain 1.0 推出的多智能体架构，核心思想就是模块化。本系列文章将通过一个“个人助理”案例，带大家掌握最主流的Supervisor（集中式）架构。

本篇（上集）重点：我们先不急着写“主管”，而是先把基础打牢——编写两个专业的执行者：Calendar Agent和Email Agent。

代码

先放两个work agent的完整代码

calander_agent.py

from langchain_openai import ChatOpenAI kimi_model = ChatOpenAI( model="kimi-k2.5", api_key="sk-uQ***", base_url="https://api.moonshot.cn/v1", # 重点：这里严格对应 Kimi 的 API 结构 extra_body={ "thinking": {"type": "disabled"} } ) from langchain_core.tools import tool from langchain.agents import create_agent CALENDAR_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." ) @tool def get_available_time_slots( attendees:list[str], date:str, duration_minutes:int ) ->list[str]: """ Check calendar availability for given attendees on a specific date. """ print("get_available_time_slots工具被调用") return ["09:00","14:00","16:00"] @tool def create_calendar_event( title:str, start_time:str, end_time:str, attendees:list[str], location:str = "" )->str: """ Create a calendar event.Requires exact ISO datatime format. """ print("create_calendar_event工具被调用") return f"Event created:{title} from {start_time} to {end_time} with {len(attendees)} attendees." calander_agent = create_agent( model=kimi_model, system_prompt=CALENDAR_SYSTEM_PROMPT, tools=[get_available_time_slots,create_calendar_event] ) def test_calander_agent(): query = "hello" for event in calander_agent.stream( {"messages":{"role":"user","content":query}}, stream_mode="values" ): event["messages"][-1].pretty_print() query = "Schedule a team meeting ['aaa@abc.com'] on 2026-03-30 at 2pm, 1hour" for event in calander_agent.stream( {"messages":{"role":"user","content":query}}, stream_mode="values" ): event["messages"][-1].pretty_print() # test_calander_agent()

email_agent.py

from langchain_openai import ChatOpenAI kimi_model = ChatOpenAI( model="kimi-k2.5", api_key="sk-uQp***", base_url="https://api.moonshot.cn/v1", # 重点：这里严格对应 Kimi 的 API 结构 extra_body={ "thinking": {"type": "disabled"} } ) from langchain_core.tools import tool from langchain.agents import create_agent 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" ) @tool def send_email( to:list[str], subject:str, body:str, cc:list[str] = [] )-> str: """ Send an email via email API,Requires properly formatted affresses. """ print("send_email工具被调用") return f"Email sent to {','.join(to)}-Subject:{subject}" email_agent = create_agent( model=kimi_model, system_prompt=EMAIL_SYSTEM_PROMPT, tools=[send_email] ) def test_email_agent(): query = "hello" for event in email_agent.stream( {"messages":{"role":"user","content":query}}, stream_mode="values" ): event["messages"][-1].pretty_print() query = "Schedule a team meeting ['aaa@abc.com'] on 2026-03-30 at 2pm, 1hour" for event in email_agent.stream( {"messages":{"role":"user","content":query}}, stream_mode="values" ): event["messages"][-1].pretty_print() # test_email_agent()

详解

1.初始化大模型

这个在之前的课程中已经说过无数次了，不再做过多赘述。

from langchain_openai import ChatOpenAI # 配置 Kimi 模型 kimi_model = ChatOpenAI( model="kimi-k2.5", api_key="YOUR_API_KEY", base_url="https://api.moonshot.cn/v1", extra_body={"thinking": {"type": "disabled"}} # 优化 API 响应速度 )

2.构建agent的提示词和tools

这里就和单Agent时一样，设置好PROMPT提示词。

定义本智能体需要的tools

calander_agent.py中定义了两个工具，一个叫get_available_time_slots，一个叫create_calendar_event，正如他们里面的描述，一个检查特定日期给定参会者的日历可用性，一个创建日历事件。需使用精确的ISO日期时间格式。

email_agent.py中定义了一个工具，叫做send_email，通过电子邮件API发送邮件，需提供格式正确的地址。当然我这里是写死的返回，实际开发中可以调用API完成功能。

CALENDAR_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." ) @tool def get_available_time_slots( attendees:list[str], date:str, duration_minutes:int ) ->list[str]: """ Check calendar availability for given attendees on a specific date. """ print("get_available_time_slots工具被调用") return ["09:00","14:00","16:00"] @tool def create_calendar_event( title:str, start_time:str, end_time:str, attendees:list[str], location:str = "" )->str: """ Create a calendar event.Requires exact ISO datatime format. """ print("create_calendar_event工具被调用") return f"Event created:{title} from {start_time} to {end_time} with {len(attendees)} attendees."

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" ) @tool def send_email( to:list[str], subject:str, body:str, cc:list[str] = [] )-> str: """ Send an email via email API,Requires properly formatted affresses. """ print("send_email工具被调用") return f"Email sent to {','.join(to)}-Subject:{subject}"

3.创建agent

这里和单智能体的创建一样，以前教过，不做过多赘述

calander_agent = create_agent( model=kimi_model, system_prompt=CALENDAR_SYSTEM_PROMPT, tools=[get_available_time_slots,create_calendar_event] )

email_agent = create_agent( model=kimi_model, system_prompt=EMAIL_SYSTEM_PROMPT, tools=[send_email] )

4.测试方法

分别测试两个agent是否正常使用，测试完后记得#注释掉测试方法，免得之后重复调用

def test_calander_agent(): query = "hello" for event in calander_agent.stream( {"messages":{"role":"user","content":query}}, stream_mode="values" ): event["messages"][-1].pretty_print() query = "Schedule a team meeting ['aaa@abc.com'] on 2026-03-30 at 2pm, 1hour" for event in calander_agent.stream( {"messages":{"role":"user","content":query}}, stream_mode="values" ): event["messages"][-1].pretty_print() # test_calander_agent()

def test_email_agent(): query = "hello" for event in email_agent.stream( {"messages":{"role":"user","content":query}}, stream_mode="values" ): event["messages"][-1].pretty_print() query = "Schedule a team meeting ['aaa@abc.com'] on 2026-03-30 at 2pm, 1hour" for event in email_agent.stream( {"messages":{"role":"user","content":query}}, stream_mode="values" ): event["messages"][-1].pretty_print() # test_email_agent()