【multi-agent】一、如何用 langchain 打造一个 multi-agent 实战
✍ 前言
这篇就是给已经玩过 langchain tools + agent 的同学看的进阶篇：
time / weather 这种自定义工具；也看过 AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION 一路 Thought / Action / Observation 的 Debug Log；但一到 Multi-Agent / AI 团队协作，就会有点懵：? 一、先把概念掰开：单 Agent vs Multi-Agent
1.1 单 Agent：一个人又要算日期又要查天气还要写代码
你现在的 Demo 结构本质上是这样的：
<code class="python">用户问题 ↓一个 Agent（ReAct） ↓在一堆 Tools 里选：- Calculator- Wikipedia- terminal- time- weather(封装成 prompt 调用自己的 Agent)</code>
特点：
所有事情都让一个 Agent 想：要不要查日期？要不要查天气？要不要写 Python？逻辑复杂的时候，prompt 容易写爆，而且 debug 时很难定位到底是哪一步“想歪了”。1.2 Multi-Agent：拆成“调度员 + 专家小组”
更工程一点的想法是：
典型结构：
<code class="python">用户 ↓Supervisor（调度员 Agent） ├── 日期助手 DateAgent（只接 time 工具） ├── 天气助手 WeatherAgent（拿日期 + weather 数据） └── 生活顾问 LifeAdvisorAgent（综合上面结果，给出自然语言建议）</code>
相比原来的单 Agent：
每个 Agent 的 Prompt 简单清晰（只负责一个角色）；你可以明确说：这在博客 + 面试里都是「很容易讲清楚、又显得有设计感」的点。
? 二、本文实战目标 & 架构图
2.1 我们要做的“小系统”
我们还是围绕你熟悉的天气 Demo，只是升级玩法：
三类 Agent：
DateAgent：负责获取当前日期（只会用time 工具）；WeatherAgent：根据日期 + 城市，在一个“伪天气数据库”里查天气；LifeAdvisorAgent：综合天气 + 用户问题，给出自然语言建议。顶层还有一个：
SupervisorAgent：负责判断什么时候该找谁帮忙，串起整个流程。2.2 架构图
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 免费无限量使用 DeepSeek R1 模型☜☜☜
在实现上，我们会用一个小技巧：
这样既复用你对 tools + Agent 的理解，又自然过渡到 Multi-Agent 体系。
? 三、准备环境：LLM + 基础工具
3.1 LLM 客户端
<code class="python">from config import api_type, api_key, api_base, api_version, model_namefrom langchain.chat_models import AzureChatOpenAIllm = AzureChatOpenAI( openai_api_base=api_base, openai_api_version=api_version, deployment_name=model_name, openai_api_key=api_key, openai_api_type=api_type, temperature=0.3,)</code>
3.2 复用你的 time 工具
<code class="python">from langchain.agents import toolfrom datetime import date@tooldef time(text: str) -> str: """返回今天的日期字符串，用于和当前日期相关的问题。 输入必须为空字符串；任何日期计算应在函数外进行。 """ return str(date.today())</code>
3.3 升级 weather 工具（不再套娃 Agent）
你的原始 weather 是直接在工具里面再调了一次 agent1，相当于「工具里套一个 Agent」，现在我们把它简化成：纯逻辑函数。
<code class="python">@tooldef weather(query: str) -> str: """根据内置 weather_info 字典，返回给定城市在某一天的天气。 输入格式：(城市名[, 日期])，例如： - "北京"（日期缺省，则用今天） - "北京,2023-07-15" 返回格式："{date} {city} 的天气情况为：{weather}" """ # 1. 内置“伪天气数据库” weather_info = { "2023-07-14": { "北京": "sunny", "上海": "cloudy", }, "2023-07-15": { "北京": "rainy", "上海": "windy", }, } # 2. 解析输入 parts = [x.strip() for x in query.split(",") if x.strip()] if len(parts) == 1: city = parts[0] today = str(date.today()) date_str = today else: city, date_str = parts[0], parts[1] # 3. 查表 city_weather = weather_info.get(date_str, {}) w = city_weather.get(city, "未知") return f"{date_str} {city} 的天气情况为：{w}"</code>现在 weather 真正符合「(str) -> str 的工具函数」，方便挂在任意 Agent 上。
? 四、Step1：定义两个 Worker Agent（日期 & 天气）
4.1 DateAgent：只负责“日期相关”任务
思路：
这个 Agent 不直接面向用户；它只用time 这个工具；Prompt 里告诉它：<code class="python">from langchain.agents import initialize_agent, AgentTypedate_tools = [time]date_agent = initialize_agent( tools=date_tools, llm=llm, agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION, handle_parsing_errors=True, verbose=True,)# 简单测一下：test = date_agent("今天的日期？只返回 YYYY-MM-DD。")print("DateAgent 测试输出：", test)</code>4.2 WeatherAgent：拿日期 + 城市，查天气
WeatherAgent 的职责：
接收自然语言提问：利用：
time 来确定“今天是哪天”；weather 在字典里查天气；最终用自然语言回复用户。<code class="python">weather_tools = [time, weather]weather_agent = initialize_agent( tools=weather_tools, llm=llm, agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION, handle_parsing_errors=True, verbose=True, agent_kwargs={ "system_message": ( "你是一个天气查询助手，需要根据用户问题决定：" "1）是否需要获取今天日期(time 工具)，" "2）调用 weather(city[,date]) 工具查找天气。" "请最终用简洁的中文回答，比如：'今天北京下雨，出门建议带伞。'" ) })</code><code class="python">res = weather_agent("今天北京的天气如何？")print(res)</code>它会大致走这样几步：
Thought：这是天气问题 → Action: 调用time；Observation：2025-11-22；Thought：需要查北京这一天的天气 → Action: weather("北京,2025-11-22")；Observation：2025-11-22 北京 的天气情况为：未知（因为我们字典里没有这天）；Final Answer：用自然语言综合一下（比如说“在已有数据中没有查到”）。如果你把系统日期改成 2023-07-15，就正好对应你之前的 Demo ：）
? 五、Step2：把子 Agent 包装成 Tool，构建 Supervisor Multi-Agent
关键技巧：
5.1 把 DateAgent / WeatherAgent 包装成 Tool
<code class="python">from langchain.tools import Tooldate_agent_tool = Tool( name="DateAssistant", func=lambda q: date_agent(q)["output"], description=( "日期助手，擅长回答'今天几号'、'某天是星期几'这类问题。" "如果用户提问中包含日期计算，可以调用我。" ),)weather_agent_tool = Tool( name="WeatherAssistant", func=lambda q: weather_agent(q)["output"], description=( "天气助手，擅长回答某地某天的天气情况，比如：" "'今天北京的天气如何？'、'2023-07-15 上海刮风吗？'" ),)</code>
注意这里我用 lambda q: agent(q)["output"]，是为了兼容 initialize_agent 返回的 dict 结构。
5.2 定义 SupervisorAgent：只负责“谁干活”和“最后说话”
SupervisorAgent 不直接操作 time / weather，它只看这两个「子 Agent Tool」：
<code class="python">supervisor_tools = [date_agent_tool, weather_agent_tool]supervisor = initialize_agent( tools=supervisor_tools, llm=llm, agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION, handle_parsing_errors=True, verbose=True, agent_kwargs={ "system_message": ( "你是一个调度员（Supervisor Agent）。" "你自己不直接查日期或天气，而是根据用户问题，在以下两个助手之间做选择：" "1）DateAssistant：专门处理日期相关问题；" "2）WeatherAssistant：专门处理天气相关问题。" "如果问题里同时涉及日期和天气，比如'今天北京适合跑步吗，要不要带伞？'，" "你可以把原问题一个整体转交给 WeatherAssistant，让它做综合判断。" "最后，你用自然语言总结回答用户。" ) })</code>? 六、实战：问一句“今天北京适合跑步吗？”
现在我们可以像平时调用单 Agent 一样，直接问 Supervisor：
<code class="python">question = "今天北京适合出门跑步吗？要不要带伞？"result = supervisor(question)print(result)</code>
一个典型的 ReAct 日志（只保留关键部分）会是这样：
<code class="python">> Entering new AgentExecutor chain...Question: 今天北京适合出门跑步吗？要不要带伞？Thought: 这个问题与天气强相关，应该交给 WeatherAssistant。Action:{ "action": "WeatherAssistant", "action_input": "今天北京适合出门跑步吗？要不要带伞？"}Observation: 今天北京可能有小雨，建议出门跑步时避开降雨时段，并携带雨具。Thought: 已经从 WeatherAssistant 处拿到了结论，我只需要用自然语言复述并稍作组织。Final Answer: 今天北京有雨，虽然可以出门跑步，但建议选择雨势较小的时间段，并随身携带雨伞或雨衣，以免淋湿。> Finished chain.</code>你会发现三个明显的变化：
顶层日志里出现了WeatherAssistant 这个“工具名”；但我们知道它实际上不是一个普通 Tool，而是一个完整的 AgentExecutor；这就是最小可运行的 Multi-Agent：Agent 作为 Tool 被另一个 Agent 编排。? 七、再加一个 LifeAdvisorAgent（可选加戏）
如果你想架构再“戏剧化”一点，可以再加一个：
WeatherAgent 只负责事实（天气本身）；LifeAdvisorAgent 负责按生活场景给建议（跑步 / 上班 / 约会）；Supervisor 根据问题内容决定是直接走 WeatherAgent，还是先问 WeatherAgent 再让 LifeAdvisor 做决策。简单示意（伪代码，只给你思路）：
<code class="python"># 1. LifeAdvisorAgent：输入是“天气描述 + 用户原问题”，输出是建议life_advisor = initialize_agent( tools=[], llm=llm, agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION, handle_parsing_errors=True, verbose=True, agent_kwargs={ "system_message": ( "你是生活场景顾问。输入会包含：" "1）天气描述；" "2）用户原始需求，例如跑步/通勤/约会等；" "你只需要基于这两点给出具体建议。" ) })life_advisor_tool = Tool( name="LifeAdvisor", func=lambda q: life_advisor(q)["output"], description="根据天气描述和用户需求，给出生活建议。")</code>然后在 Supervisor 里：
先把问题转交给 WeatherAssistant，拿到纯天气描述；再把天气描述 + 用户原问题 交给 LifeAdvisor；最后由 Supervisor 输出最终结果。这样你就有了：
<code class="python">Supervisor → WeatherAgent → LifeAdvisorAgent</code>
一个完整的 三层多 Agent 协作链，在博客里画图 + 贴日志，会非常有“工程感”。
? 八、手撕代码
最后给你一个「收拢版」的完整脚本，适合贴在文章末尾当“手撕代码”块。
你可以根据自己的 Azure / OpenAI 配置改一改头部。
<code class="python">"""multi_agent_weather.py一个最小可运行的 Multi-Agent Demo：Supervisor + DateAgent + WeatherAgent"""from datetime import datefrom config import api_type, api_key, api_base, api_version, model_namefrom langchain.chat_models import AzureChatOpenAIfrom langchain.agents import ( tool, initialize_agent, AgentType,)from langchain.tools import Tool# ========= 1. LLM =========llm = AzureChatOpenAI( openai_api_base=api_base, openai_api_version=api_version, deployment_name=model_name, openai_api_key=api_key, openai_api_type=api_type, temperature=0.3,)# ========= 2. 基础工具 =========@tooldef time(text: str) -> str: """返回今天的日期字符串，用于和当前日期相关的问题。 输入必须为空字符串；任何日期计算应在函数外进行。 """ return str(date.today())@tooldef weather(query: str) -> str: """根据内置 weather_info 字典，返回给定城市在某一天的天气。 输入格式：(城市名[, 日期])，例如： - "北京" - "北京,2023-07-15" 返回格式："{date} {city} 的天气情况为：{weather}" """ weather_info = { "2023-07-14": { "北京": "sunny", "上海": "cloudy", }, "2023-07-15": { "北京": "rainy", "上海": "windy", }, } parts = [x.strip() for x in query.split(",") if x.strip()] if len(parts) == 1: city = parts[0] date_str = str(date.today()) else: city, date_str = parts[0], parts[1] city_weather = weather_info.get(date_str, {}) w = city_weather.get(city, "未知") return f"{date_str} {city} 的天气情况为：{w}"# ========= 3. 子 Agent：DateAgent =========date_agent = initialize_agent( tools=[time], llm=llm, agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION, handle_parsing_errors=True, verbose=True,)date_agent_tool = Tool( name="DateAssistant", func=lambda q: date_agent(q)["output"], description=( "日期助手，擅长回答'今天几号'、'某天是星期几'这类问题。" "如果用户提问中包含日期计算，可以调用我。" ),)# ========= 4. 子 Agent：WeatherAgent =========weather_agent = initialize_agent( tools=[time, weather], llm=llm, agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION, handle_parsing_errors=True, verbose=True, agent_kwargs={ "system_message": ( "你是一个天气查询助手，需要根据用户问题决定：" "1）是否需要获取今天日期(time 工具)，" "2）调用 weather(city[,date]) 工具查找天气。" "最终用简洁中文回答，比如：'今天北京下雨，出门建议带伞。'" ) },)weather_agent_tool = Tool( name="WeatherAssistant", func=lambda q: weather_agent(q)["output"], description=( "天气助手，擅长回答某地某天的天气情况，比如：" "'今天北京的天气如何？'、'2023-07-15 上海刮风吗？'" ),)# ========= 5. Supervisor Multi-Agent =========supervisor = initialize_agent( tools=[date_agent_tool, weather_agent_tool], llm=llm, agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION, handle_parsing_errors=True, verbose=True, agent_kwargs={ "system_message": ( "你是一个调度员（Supervisor Agent）。" "你自己不直接查日期或天气，而是根据用户问题，在以下助手之间做选择：" "1）DateAssistant：专门处理日期相关问题；" "2）WeatherAssistant：专门处理天气相关问题。" "如果问题里同时涉及日期和天气，比如" "'今天北京适合跑步吗，要不要带伞？'，" "你可以把原问题整体转交给 WeatherAssistant。" "最后，你用自然语言总结回答用户。" ) },)# ========= 6. 入口 =========if __name__ == "__main__": question = "今天北京适合出门跑步吗？要不要带伞？" result = supervisor(question) print("用户问题：", question) print("系统回答：", result)</code> 
