【Multi-Agent实战】如何用 LangChain 打造一个 Multi-Agent 实战项目

原创

用户11925555

修改于 2025-11-23 00:10:00

2850

【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？还是一个 Agent 带一堆工具？” “Multi-Agent 和平时的 Tools 调用有什么本质区别？”

🧠 一、先把概念掰开：单 Agent vs Multi-Agent

1.1 单 Agent：一个人又要算日期又要查天气还要写代码

你现在的 Demo 结构本质上是这样的：

用户问题
   ↓
一个 Agent（ReAct）
   ↓
在一堆 Tools 里选：
- Calculator
- Wikipedia
- terminal
- time
- weather(封装成 prompt 调用自己的 Agent)

特点：

所有事情都让一个 Agent 想：要不要查日期？要不要查天气？要不要写 Python？
逻辑复杂的时候，prompt 容易写爆，而且 debug 时很难定位到底是哪一步“想歪了”。

1.2 Multi-Agent：拆成“调度员 + 专家小组”

更工程一点的想法是：

既然人类解决问题是「项目经理 + 各种专家」的协作，那 LLM 也可以这么玩。

典型结构：

用户
  ↓
Supervisor（调度员 Agent）
  ├── 日期助手 DateAgent（只接 time 工具）
  ├── 天气助手 WeatherAgent（拿日期 + weather 数据）
  └── 生活顾问 LifeAdvisorAgent（综合上面结果，给出自然语言建议）

相比原来的单 Agent：

每个 Agent 的 Prompt 简单清晰（只负责一个角色）；
你可以明确说：

“我们是用 Supervisor 架构的 Multi-Agent 系统，而不是一个超大的 Monolithic Agent。”

这在博客 + 面试里都是「很容易讲清楚、又显得有设计感」的点。

📚 二、本文实战目标 & 架构图

2.1 我们要做的“小系统”

我们还是围绕你熟悉的天气 Demo，只是升级玩法：

用户一句： 「今天北京适合出门跑步吗？要不要带伞？」 由 AI 团队内部协作给出回答。

三类 Agent：

DateAgent：负责获取当前日期（只会用 time 工具）；
WeatherAgent：根据日期 + 城市，在一个“伪天气数据库”里查天气；
LifeAdvisorAgent：综合天气 + 用户问题，给出自然语言建议。

顶层还有一个：

SupervisorAgent：负责判断什么时候该找谁帮忙，串起整个流程。

2.2 架构图

在实现上，我们会用一个小技巧：

把“子 Agent”本身包装成一个 Tool，再挂给 Supervisor 使用。

这样既复用你对 tools + Agent 的理解，又自然过渡到 Multi-Agent 体系。

🔧 三、准备环境：LLM + 基础工具

3.1 LLM 客户端

from config import api_type, api_key, api_base, api_version, model_name
from langchain.chat_models import AzureChatOpenAI

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,
)

✅ 说明：我们后面所有 Agent 都共享这一个 llm； temperature 调低一点（0.3 左右）更适合“工具路由 + 规划”场景。

3.2 复用你的 `time` 工具

from langchain.agents import tool
from datetime import date

@tool
def time(text: str) -> str:
    """返回今天的日期字符串，用于和当前日期相关的问题。
    输入必须为空字符串；任何日期计算应在函数外进行。
    """
    return str(date.today())

和你原来的版本一样，我们只稍微把 docstring 写得更「提示词化」。

3.3 升级 `weather` 工具（不再套娃 Agent）

你的原始 weather 是直接在工具里面再调了一次 agent1，相当于「工具里套一个 Agent」，现在我们把它简化成：纯逻辑函数。

@tool
def 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}"

现在 weather 真正符合「(str) -> str 的工具函数」，方便挂在任意 Agent 上。

🤖 四、Step1：定义两个 Worker Agent（日期 & 天气）

4.1 DateAgent：只负责“日期相关”任务

思路：

这个 Agent 不直接面向用户；
它只用 time 这个工具；
Prompt 里告诉它：

“你是一个日期助手，只能回答日期 / YYYY-MM-DD 这种问题。”

from langchain.agents import initialize_agent, AgentType

date_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)

运行时的 ReAct 日志会很简单： Thought: 这是日期问题 → 调用 time； Observation: 返回 2025-11-22 → Final Answer。

4.2 WeatherAgent：拿日期 + 城市，查天气

WeatherAgent 的职责：

接收自然语言提问：

“今天北京的天气怎么样？” “明天上海是不是有风？”

利用：
time 来确定“今天是哪天”；
weather 在字典里查天气；
最终用自然语言回复用户。

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]) 工具查找天气。\n"
            "请最终用简洁的中文回答，比如：'今天北京下雨，出门建议带伞。'"
        )
    }
)

你现在就可以直接试：

res = weather_agent("今天北京的天气如何？")
print(res)

它会大致走这样几步：

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

关键技巧：

任何“可以接收字符串并返回字符串”的东西，都可以被包成一个 Tool。 AgentExecutor 本身也不例外。

5.1 把 DateAgent / WeatherAgent 包装成 Tool

from langchain.tools import Tool

date_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 上海刮风吗？'"
    ),
)

注意这里我用 lambda q: agent(q)["output"]，是为了兼容 initialize_agent 返回的 dict 结构。

5.2 定义 SupervisorAgent：只负责“谁干活”和“最后说话”

SupervisorAgent 不直接操作 time / weather，它只看这两个「子 Agent Tool」：

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）。"
            "你自己不直接查日期或天气，而是根据用户问题，在以下两个助手之间做选择：\n"
            "1）DateAssistant：专门处理日期相关问题；\n"
            "2）WeatherAssistant：专门处理天气相关问题。\n"
            "如果问题里同时涉及日期和天气，比如'今天北京适合跑步吗，要不要带伞？'，"
            "你可以把原问题一个整体转交给 WeatherAssistant，让它做综合判断。\n"
            "最后，你用自然语言总结回答用户。"
        )
    }
)

🚀 六、实战：问一句“今天北京适合跑步吗？”

现在我们可以像平时调用单 Agent 一样，直接问 Supervisor：

question = "今天北京适合出门跑步吗？要不要带伞？"
result = supervisor(question)
print(result)

一个典型的 ReAct 日志（只保留关键部分）会是这样：

> Entering new AgentExecutor chain...
Question: 今天北京适合出门跑步吗？要不要带伞？
Thought: 这个问题与天气强相关，应该交给 WeatherAssistant。
Action:
{
  "action": "WeatherAssistant",
  "action_input": "今天北京适合出门跑步吗？要不要带伞？"
}

Observation: 今天北京可能有小雨，建议出门跑步时避开降雨时段，并携带雨具。

Thought: 已经从 WeatherAssistant 处拿到了结论，我只需要用自然语言复述并稍作组织。
Final Answer: 今天北京有雨，虽然可以出门跑步，但建议选择雨势较小的时间段，并随身携带雨伞或雨衣，以免淋湿。
> Finished chain.

你会发现三个明显的变化：

顶层日志里出现了 WeatherAssistant 这个“工具名”；
但我们知道它实际上不是一个普通 Tool，而是一个完整的 AgentExecutor；
这就是最小可运行的 Multi-Agent：Agent 作为 Tool 被另一个 Agent 编排。

🧪 七、再加一个 LifeAdvisorAgent（可选加戏）

如果你想架构再“戏剧化”一点，可以再加一个：

WeatherAgent 只负责事实（天气本身）；
LifeAdvisorAgent 负责按生活场景给建议（跑步 / 上班 / 约会）；
Supervisor 根据问题内容决定是直接走 WeatherAgent，还是先问 WeatherAgent 再让 LifeAdvisor 做决策。

简单示意（伪代码，只给你思路）：

# 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": (
            "你是生活场景顾问。输入会包含：\n"
            "1）天气描述；\n"
            "2）用户原始需求，例如跑步/通勤/约会等；\n"
            "你只需要基于这两点给出具体建议。"
        )
    }
)

life_advisor_tool = Tool(
    name="LifeAdvisor",
    func=lambda q: life_advisor(q)["output"],
    description="根据天气描述和用户需求，给出生活建议。"
)

然后在 Supervisor 里：

先把问题转交给 WeatherAssistant，拿到纯天气描述；
再把 天气描述 + 用户原问题 交给 LifeAdvisor；
最后由 Supervisor 输出最终结果。

这样你就有了：

Supervisor → WeatherAgent → LifeAdvisorAgent

一个完整的 三层多 Agent 协作链，在博客里画图 + 贴日志，会非常有“工程感”。

💼 八、手撕代码

最后给你一个「收拢版」的完整脚本，适合贴在文章末尾当“手撕代码”块。

你可以根据自己的 Azure / OpenAI 配置改一改头部。

"""
multi_agent_weather.py
一个最小可运行的 Multi-Agent Demo：
Supervisor + DateAgent + WeatherAgent
"""

from datetime import date

from config import api_type, api_key, api_base, api_version, model_name
from langchain.chat_models import AzureChatOpenAI
from 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. 基础工具 =========
@tool
def time(text: str) -> str:
    """返回今天的日期字符串，用于和当前日期相关的问题。
    输入必须为空字符串；任何日期计算应在函数外进行。
    """
    return str(date.today())


@tool
def 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]) 工具查找天气。\n"
            "最终用简洁中文回答，比如：'今天北京下雨，出门建议带伞。'"
        )
    },
)

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）。"
            "你自己不直接查日期或天气，而是根据用户问题，在以下助手之间做选择：\n"
            "1）DateAssistant：专门处理日期相关问题；\n"
            "2）WeatherAssistant：专门处理天气相关问题。\n"
            "如果问题里同时涉及日期和天气，比如"
            "'今天北京适合跑步吗，要不要带伞？'，"
            "你可以把原问题整体转交给 WeatherAssistant。\n"
            "最后，你用自然语言总结回答用户。"
        )
    },
)


# ========= 6. 入口 =========
if __name__ == "__main__":
    question = "今天北京适合出门跑步吗？要不要带伞？"
    result = supervisor(question)
    print("用户问题：", question)
    print("系统回答：", result)

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

腾讯技术创作特训营S16

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

腾讯技术创作特训营S16

登录后参与评论

0 条评论

热度