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

开篇为什么我们需要多智能体在传统的 LLM 开发中我们往往倾向于给一个 Agent 堆叠无数个 Tool。但随着业务逻辑变复杂单一 Agent 会面临“选择困难症”工具回调错误和“上下文迷失”的问题。LangChain 1.0 推出的多智能体架构核心思想就是模块化。本系列文章将通过一个“个人助理”案例带大家掌握最主流的Supervisor集中式架构。本篇上集重点我们先不急着写“主管”而是先把基础打牢——编写两个专业的执行者Calendar Agent和Email Agent。代码先放两个work agent的完整代码calander_agent.pyfrom langchain_openai import ChatOpenAI kimi_model ChatOpenAI( modelkimi-k2.5, api_keysk-uQ***, base_urlhttps://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 fEvent created:{title} from {start_time} to {end_time} with {len(attendees)} attendees. calander_agent create_agent( modelkimi_model, system_promptCALENDAR_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_modevalues ): event[messages][-1].pretty_print() query Schedule a team meeting [aaaabc.com] on 2026-03-30 at 2pm, 1hour for event in calander_agent.stream( {messages:{role:user,content:query}}, stream_modevalues ): event[messages][-1].pretty_print() # test_calander_agent()email_agent.pyfrom langchain_openai import ChatOpenAI kimi_model ChatOpenAI( modelkimi-k2.5, api_keysk-uQp***, base_urlhttps://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 fEmail sent to {,.join(to)}-Subject:{subject} email_agent create_agent( modelkimi_model, system_promptEMAIL_SYSTEM_PROMPT, tools[send_email] ) def test_email_agent(): query hello for event in email_agent.stream( {messages:{role:user,content:query}}, stream_modevalues ): event[messages][-1].pretty_print() query Schedule a team meeting [aaaabc.com] on 2026-03-30 at 2pm, 1hour for event in email_agent.stream( {messages:{role:user,content:query}}, stream_modevalues ): event[messages][-1].pretty_print() # test_email_agent()详解1.初始化大模型这个在之前的课程中已经说过无数次了不再做过多赘述。from langchain_openai import ChatOpenAI # 配置 Kimi 模型 kimi_model ChatOpenAI( modelkimi-k2.5, api_keyYOUR_API_KEY, base_urlhttps://api.moonshot.cn/v1, extra_body{thinking: {type: disabled}} # 优化 API 响应速度 )2.构建agent的提示词和tools这里就和单Agent时一样设置好PROMPT提示词。定义本智能体需要的toolscalander_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 fEvent 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 fEmail sent to {,.join(to)}-Subject:{subject}3.创建agent这里和单智能体的创建一样以前教过不做过多赘述calander_agent create_agent( modelkimi_model, system_promptCALENDAR_SYSTEM_PROMPT, tools[get_available_time_slots,create_calendar_event] )email_agent create_agent( modelkimi_model, system_promptEMAIL_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_modevalues ): event[messages][-1].pretty_print() query Schedule a team meeting [aaaabc.com] on 2026-03-30 at 2pm, 1hour for event in calander_agent.stream( {messages:{role:user,content:query}}, stream_modevalues ): 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_modevalues ): event[messages][-1].pretty_print() query Schedule a team meeting [aaaabc.com] on 2026-03-30 at 2pm, 1hour for event in email_agent.stream( {messages:{role:user,content:query}}, stream_modevalues ): event[messages][-1].pretty_print() # test_email_agent()代码运行结果展示我们先发送了一个hello之后发送了指令。第一个截图中AI得到hello的消息后介绍了自己的功能得到第二个指令后检查并使用了自己的【create_calendar_event】Tool最后返回了Successfully和Event Detailscalander_agent.pyemail_agent.py第二个截图中AI得到hello的消息后介绍了自己的功能得到第二个指令后检查并使用了自己的【send_email】Tool最后返回了Successfully sent the .....小结通过上面的代码我们完成了两个具备独立作战能力的执行者。它懂业务理解自然语言中的时间。会操作能够准确调用 API。有边界只处理日历事务或者操作邮件。这种高内聚、低耦合的设计为我们下集引入 Supervisor Agent 奠定了基础。在多智能体系统中只有每个 Worker 足够专业整体的协作才会有意义。