☰
多智能体系统实战:DeepAgents、MCP、A2A与Skills全流程落地
2026/9/26 18:22:07 网站建设 项目流程

1. 从单兵作战到团队协同:多智能体系统到底解决了什么问题

过去两年我一直在做AI应用落地,从最早的单一提示词工程,到后来的RAG检索增强,再到工作流编排,踩过的坑不算少。但真正让我觉得“路子变了”的,是第一次把一个大任务拆给多个Agent去协作完成的时候。那种感觉就像你一直一个人搬砖,突然发现可以组建一支施工队,每个人负责自己最擅长的环节,整体效率不是线性提升,而是指数级的。

DeepAgents、MCP、A2A、Skills这几个词,单独拎出来每一个都够写一篇长文,但它们组合在一起的时候,才真正构成了当前多智能体系统落地的完整拼图。我写这篇东西的初衷很简单:网上讲概念的多,讲实操的少;讲单个组件的多,讲怎么把它们串起来跑通全流程的少。所以这篇博文我会按照一个真实项目的推进节奏,从架构设计到代码落地,从踩坑记录到优化技巧,把这条链路完整走一遍。

这篇文章适合谁看?如果你已经用过LangChain或者类似的框架做过一些AI应用,想进一步了解多智能体怎么协作、MCP协议怎么接入外部工具、A2A怎么让Agent之间互相通信、Skills怎么让Agent具备可复用的专业能力,那这篇内容就是为你准备的。如果你是完全的新手,也没关系,我会在关键概念处做通俗解释,保证你能跟上节奏。

先说结论:多智能体系统的核心价值不在于“多”,而在于“分工”和“协作”。一个设计良好的多智能体系统,应该像一个高效的团队——每个成员有明确的职责边界,有标准化的沟通方式,有可复用的技能库,还有一个靠谱的协调者来分配任务和汇总结果。DeepAgents提供了协调框架,MCP解决了工具接入的标准化问题,A2A打通了Agent之间的通信壁垒,Skills则让每个Agent具备可积累的专业能力。四者缺一不可。

2. 核心组件深度拆解:每个角色到底在干什么

2.1 DeepAgents:多智能体的“项目管理层”

DeepAgents是LangChain生态里专门为多智能体协作设计的框架层。你可以把它理解成一个项目的管理层——它不直接干活,但负责定义组织架构、分配任务、管理上下文、处理异常。

我第一次接触DeepAgents的时候,最让我眼前一亮的是它的子智能体(SubAgents)机制。传统的做法是你写一个巨大的提示词,让一个Agent试图搞定所有事情。但DeepAgents允许你定义多个子智能体,每个子智能体有自己的系统提示词、自己的工具集、自己的职责范围。主智能体(Supervisor)负责理解用户意图,然后把任务拆解并分发给合适的子智能体。

这里有个关键的设计决策:子智能体之间是隔离的。什么意思?就是子智能体A的对话历史不会自动传递给子智能体B。这个设计初看有点反直觉,但仔细想想非常合理——它强制你做好任务边界划分,避免上下文污染。主智能体在分发任务时,需要把必要的上下文信息显式传递给子智能体,这反而让整个系统的行为更加可控和可预测。

在实际项目中,我通常会把子智能体按照职能划分,比如:

  • 研究智能体:负责信息检索和资料整理
  • 分析智能体:负责数据分析和逻辑推理
  • 写作智能体:负责内容生成和润色
  • 审核智能体:负责质量检查和事实核查

每个子智能体只需要关注自己的领域,提示词可以写得更精准,工具集也可以更精简。实测下来,这种分工方式比单一大模型处理复杂任务的成功率高出不少。

2.2 MCP:工具接入的“USB-C接口”

MCP全称是Model Context Protocol,你可以把它理解成AI世界的USB-C接口。在没有MCP之前,每接一个外部工具(比如数据库、API、文件系统),你都要写一套专门的适配代码。有了MCP之后,只要工具方提供了MCP Server,任何支持MCP的客户端都能直接调用。

MCP的架构其实很简单,核心就三个角色:

  • MCP Host:运行AI应用的环境,比如你的Agent框架
  • MCP Client:负责与MCP Server建立连接和通信
  • MCP Server:实际提供工具能力的一方

通信方式上,MCP支持stdio(标准输入输出)和SSE(Server-Sent Events)两种传输模式。stdio适合本地工具,比如文件操作、命令行执行;SSE适合远程服务,比如云端API、数据库查询。

我在项目里最常用的MCP Server包括:

  • 文件系统MCP:让Agent能读写本地文件
  • 数据库MCP:让Agent能查询和操作数据库
  • 浏览器MCP:让Agent能自动化操作网页
  • 代码执行MCP:让Agent能运行代码片段

注意:MCP Server的权限控制非常重要。我见过有人直接把生产数据库的MCP Server接给Agent,结果Agent一个误操作把表给删了。建议在MCP Server层面做好权限隔离,只暴露必要的操作接口。

2.3 A2A:Agent之间的“通信协议”

A2A全称是Agent-to-Agent Protocol,解决的是Agent之间怎么互相通信的问题。在DeepAgents框架内,子智能体之间的通信由框架本身管理。但当你需要跨框架、跨平台、甚至跨组织的Agent协作时,就需要A2A协议了。

A2A的核心概念包括:

  • Agent Card:每个Agent的能力描述文件,类似名片
  • Task:一个具体的任务单元,包含输入、输出和状态
  • Message:Agent之间传递的消息,支持文本、文件、结构化数据

A2A和MCP的区别经常有人搞混。简单说:MCP解决的是Agent怎么调用工具,A2A解决的是Agent怎么调用另一个Agent。一个是纵向的工具接入,一个是横向的协作通信。

在实际项目中,A2A的价值在分布式场景下特别明显。比如你的系统里有一个专门做图像生成的Agent部署在A服务器上,一个专门做文本分析的Agent部署在B服务器上,通过A2A协议,它们可以像调用本地函数一样互相协作,而不需要关心对方的具体实现。

2.4 Skills:Agent的“技能库”

Skills这个概念最近特别火,从Claude Code的Skills到各种Agent框架的Skills机制,本质上都是在解决同一个问题:怎么让Agent具备可复用、可组合、可积累的专业能力。

一个Skill通常包含:

  • 技能描述:这个技能是干什么的,什么时候该用
  • 执行逻辑:具体的步骤、提示词、工具调用序列
  • 输入输出规范:需要什么参数,返回什么结果
  • 示例:几个典型的使用案例

我自己的做法是把Skills按照领域分类管理,比如:

  • 文档处理类:PDF解析、Word生成、Markdown转换
  • 数据分析类:数据清洗、统计分析、可视化
  • 代码开发类:代码生成、代码审查、单元测试
  • 内容创作类:文案撰写、翻译、摘要

Skills最大的好处是可积累。你今天写了一个“竞品分析报告生成”的Skill,明天遇到类似任务时直接调用就行,不需要重新写提示词。而且Skills可以在团队内共享,新人来了直接调用现成的Skills,上手速度会快很多。

3. 全流程实战:从零搭建一个多智能体协作系统

3.1 环境准备与依赖安装

先说一下我的环境配置,这套配置在macOS和Ubuntu上都跑过,Windows的话建议用WSL2。

# 创建虚拟环境 python -m venv deepagents-env source deepagents-env/bin/activate # Windows用 deepagents-env\Scripts\activate # 安装核心依赖 pip install deepagents langchain langchain-openai mcp a2a-sdk # 安装常用MCP Server pip install mcp-server-filesystem mcp-server-sqlite mcp-server-fetch

这里有个坑要注意:deepagents和langchain的版本兼容性。我刚开始用的时候没锁版本,结果langchain更新了一个小版本,deepagents的某个API就变了。建议在requirements.txt里把版本锁死:

deepagents==0.1.x langchain==0.3.x langchain-openai==0.2.x mcp==1.0.x

API Key的配置我习惯用环境变量管理,不要硬编码在代码里:

export OPENAI_API_KEY="your-key-here" export ANTHROPIC_API_KEY="your-key-here" # 如果要用Claude

3.2 定义你的第一个子智能体

先从一个最简单的例子开始,定义一个研究智能体和一个写作智能体。

from deepagents import create_deep_agent from langchain_openai import ChatOpenAI # 初始化模型 model = ChatOpenAI(model="gpt-4o", temperature=0) # 定义研究子智能体 research_agent = { "name": "researcher", "description": "负责信息检索和资料整理,擅长从多个来源收集和筛选信息", "system_prompt": """你是一个专业的研究员。你的任务是: 1. 根据给定的主题,从可用工具中检索相关信息 2. 对信息进行筛选和整理,去除重复和低质量内容 3. 输出结构化的研究结果,包含来源和关键发现 注意:只输出事实性内容,不要加入个人观点""", "tools": [search_tool, fetch_tool] } # 定义写作子智能体 writer_agent = { "name": "writer", "description": "负责内容撰写和润色,擅长将研究结果转化为易读的文章", "system_prompt": """你是一个专业的内容创作者。你的任务是: 1. 根据研究结果,撰写结构清晰、逻辑通顺的文章 2. 使用通俗易懂的语言,避免过于专业的术语 3. 确保文章有明确的开头、主体和结尾 注意:不要编造研究结果中没有的信息""", "tools": [] } # 创建主智能体 supervisor = create_deep_agent( model=model, subagents=[research_agent, writer_agent], system_prompt="你是一个项目协调者,负责理解用户需求,将任务拆解并分配给合适的子智能体" )

这段代码看起来简单,但有几个关键点值得展开说。

第一,子智能体的description非常重要。主智能体就是靠这个description来判断该把任务分配给谁的。description写得越清晰,任务分配就越准确。我见过有人把description写成“一个有用的助手”,结果主智能体根本不知道该什么时候调用它。

第二,system_prompt要明确边界。每个子智能体的系统提示词里,我都会明确写“你的任务是”和“注意”两部分。前者定义职责,后者定义禁忌。特别是“不要编造信息”这条,在多智能体系统里尤其重要,因为子智能体之间信息传递时,一个环节的幻觉可能会被后续环节放大。

第三,工具集要精简。研究智能体只需要检索和抓取工具,写作智能体不需要工具。给子智能体分配过多工具反而会干扰它的判断。我的一般原则是:每个子智能体的工具不超过5个,超过的话就考虑再拆分。

3.3 接入MCP Server扩展工具能力

DeepAgents本身自带的工具比较有限,真正让它强大起来的是MCP生态。下面演示怎么接入一个文件系统MCP Server。

from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client # 配置MCP Server连接参数 server_params = StdioServerParameters( command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "/path/to/workspace"] ) # 建立连接并获取工具列表 async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools = await session.list_tools() # 将MCP工具转换为DeepAgents可用的格式 mcp_tools = [] for tool in tools.tools: mcp_tools.append({ "name": tool.name, "description": tool.description, "parameters": tool.inputSchema, "handler": lambda args, t=tool.name: session.call_tool(t, args) })

这段代码的关键在于工具格式转换。MCP Server返回的工具定义是JSON Schema格式,DeepAgents需要的是LangChain Tool格式,中间需要一个适配层。我一般会写一个通用的转换函数,避免每次接新MCP Server都重复写代码。

实际项目中,我通常会同时接入多个MCP Server:

MCP Server用途传输方式典型工具
filesystem文件读写stdioread_file, write_file, list_dir
sqlite数据库操作stdioquery, execute, list_tables
fetch网页抓取stdiofetch_url, extract_text
browser浏览器自动化SSEnavigate, click, screenshot
code-runner代码执行stdiorun_python, run_shell

提示:MCP Server的启动方式分两种。npx方式适合Node.js实现的Server,uvx方式适合Python实现的Server。如果网络环境不稳定,建议提前把Server包下载到本地,用本地路径启动。

3.4 用A2A协议实现跨Agent通信

当你的系统规模变大,子智能体可能部署在不同的服务上,这时候就需要A2A协议了。下面是一个简化的A2A通信示例。

from a2a import A2AClient, AgentCard, Task # 定义Agent Card,描述这个Agent的能力 card = AgentCard( name="image-generator", description="专业的图像生成Agent,支持文生图和图生图", capabilities=["text-to-image", "image-to-image"], endpoint="http://localhost:8001/a2a" ) # 创建A2A客户端 client = A2AClient(card.endpoint) # 发送任务 task = Task( input={"prompt": "一只在草地上奔跑的柴犬", "size": "1024x1024"}, output_format="image_url" ) result = await client.send_task(task) print(result.output) # 返回生成的图片URL

A2A协议最实用的地方在于Agent发现机制。你可以维护一个Agent注册中心,每个Agent启动时把自己的Agent Card注册上去,其他Agent需要某种能力时,先查询注册中心找到合适的Agent,再通过A2A协议发起调用。这套机制让系统具备了动态扩展的能力——新加一个Agent,不需要修改现有代码,只要注册上去就能被其他Agent发现和使用。

3.5 Skills的封装与调用

Skills的封装我推荐用“提示词模板+工具序列”的方式。下面是一个“竞品分析报告生成”Skill的示例。

class CompetitorAnalysisSkill: name = "competitor_analysis" description = "根据给定的公司名称,生成竞品分析报告" input_schema = { "company_name": {"type": "string", "description": "目标公司名称"}, "competitors": {"type": "array", "items": {"type": "string"}, "description": "竞品公司列表"} } steps = [ { "step": 1, "action": "research", "prompt": "检索{company_name}和{competitors}的最新产品信息、定价策略和市场表现" }, { "step": 2, "action": "analyze", "prompt": "对比分析各公司的优劣势,输出SWOT分析" }, { "step": 3, "action": "write", "prompt": "基于分析结果,撰写一份结构化的竞品分析报告" } ] async def execute(self, agent, params): context = {} for step in self.steps: prompt = step["prompt"].format(**params) result = await agent.run(prompt, context=context) context[f"step_{step['step']}"] = result return context["step_3"]

这个Skill的设计思路是把复杂任务拆成有序的步骤,每一步的输出作为下一步的输入。这样做的好处是每个步骤的提示词可以写得很精准,而且中间结果可以单独检查和调试。

Skills的管理我建议用注册表模式:

class SkillRegistry: def __init__(self): self.skills = {} def register(self, skill): self.skills[skill.name] = skill def get(self, name): return self.skills.get(name) def list_all(self): return [ {"name": s.name, "description": s.description} for s in self.skills.values() ] registry = SkillRegistry() registry.register(CompetitorAnalysisSkill()) # 注册更多Skills...

主智能体在规划任务时,可以先查询SkillRegistry,看看有没有现成的Skill可以用。有的话直接调用,没有的话再走通用的子智能体协作流程。这种“Skill优先”的策略能大幅提升常见任务的处理效率。

4. 踩坑实录与性能优化

4.1 常见问题速查表

问题现象可能原因排查方法解决方案
子智能体不被调用description不清晰打印主智能体的决策日志重写description,增加使用场景说明
任务分配错误子智能体职责重叠检查各子智能体的system_prompt明确边界,合并或拆分职责
MCP工具调用失败连接超时或权限不足单独测试MCP Server检查网络、路径权限、Server日志
上下文丢失子智能体间未传递必要信息检查任务分发时的context参数显式传递上下文,或使用共享内存
输出格式不稳定提示词约束不够收集失败案例增加输出格式示例,使用结构化输出
响应速度慢串行执行过多分析各步骤耗时并行化独立步骤,使用流式输出
幻觉放大子智能体编造信息对比各环节输出增加审核智能体,要求标注来源

4.2 三个让我印象深刻的坑

第一个坑:子智能体之间的“信息孤岛”问题。

前面说过DeepAgents的子智能体之间是隔离的,这个设计本身没问题,但实际用的时候很容易忘掉这一点。我做过一个项目,研究智能体检索了一堆资料,写作智能体写出来的文章却完全没用到这些资料。排查了半天才发现,主智能体在分发任务给写作智能体时,没有把研究结果传过去。

解决方案有两种:一是主智能体在分发任务时显式传递上下文,二是在子智能体之间建立共享的“工作记忆”。我一般用第一种,因为更可控。具体做法是在主智能体的system_prompt里明确写:“在调用写作智能体时,必须把研究智能体的输出作为context传入。”

第二个坑:MCP Server的并发问题。

有一次我同时启动了多个子智能体,每个都需要调用文件系统MCP Server。结果发现文件读写经常出错,日志显示“resource busy”。原因是MCP Server默认是单线程处理请求的,多个客户端同时调用会冲突。

解决方案是给每个子智能体分配独立的MCP Server实例,或者使用支持并发的MCP Server实现。我后来改用了一个支持连接池的MCP Server,问题就解决了。这个坑让我意识到,MCP虽然标准化了工具接入,但并发控制还是需要自己处理。

第三个坑:Skills的版本管理。

Skills用多了之后,版本管理就成了问题。同一个Skill,今天改一版,明天改一版,过段时间自己都忘了哪个版本是稳定的。更麻烦的是,不同项目可能依赖同一个Skill的不同版本。

我的解决方案是给每个Skill加版本号,用语义化版本管理。Skill的注册表里记录版本信息,调用时可以指定版本。同时维护一个CHANGELOG,记录每次修改的内容和原因。这套做法借鉴了软件包管理的思路,虽然有点重,但对于长期维护的项目来说非常值得。

4.3 性能优化的几个实用技巧

技巧一:并行化独立任务。如果两个子智能体的任务之间没有依赖关系,就让它们并行执行。DeepAgents支持异步调用,用asyncio.gather就能实现。

import asyncio async def parallel_tasks(): task1 = research_agent.arun("检索A主题") task2 = research_agent.arun("检索B主题") results = await asyncio.gather(task1, task2) return results

技巧二:缓存常用结果。有些MCP工具的调用结果变化不频繁,比如文件列表、数据库表结构,可以加一层缓存。我用的是内存缓存加TTL,简单有效。

技巧三:流式输出。对于长文本生成任务,开启流式输出能大幅提升用户体验。DeepAgents支持token级别的流式返回,配合前端的打字机效果,感知速度会快很多。

技巧四:模型分级。不是所有子智能体都需要用最贵的模型。研究智能体和审核智能体可以用强模型,格式转换和简单分类用轻量模型就行。我一般会配置一个模型路由,根据任务复杂度自动选择。

5. 从能跑到好用:多智能体系统的工程化思考

5.1 可观测性建设

多智能体系统最让人头疼的就是调试。一个任务经过多个子智能体,中间还穿插MCP工具调用,出了问题很难定位。我的做法是全链路日志追踪。

每个任务分配一个唯一的trace_id,所有子智能体的输入输出、MCP工具调用记录、耗时统计都带上这个trace_id。然后用一个简单的Web界面展示整个调用链路,哪个环节慢了、哪个环节出错了,一目了然。

import uuid import logging class TraceLogger: def __init__(self): self.traces = {} def start_trace(self): trace_id = str(uuid.uuid4()) self.traces[trace_id] = {"events": [], "start_time": time.time()} return trace_id def log_event(self, trace_id, agent_name, event_type, data): self.traces[trace_id]["events"].append({ "timestamp": time.time(), "agent": agent_name, "type": event_type, "data": data })

这套东西看起来简单,但实际排查问题时能省下大量时间。我建议在项目初期就把可观测性做好,不要等到出问题了再补。

5.2 错误处理与降级策略

多智能体系统里,任何一个环节出错都可能导致整个任务失败。我的策略是分级降级:

  • 一级降级:子智能体调用失败,重试2次
  • 二级降级:重试仍失败,切换到备用子智能体
  • 三级降级:备用也失败,返回部分结果并标注失败环节
  • 四级降级:完全失败,返回错误信息和排查建议

关键是不要让整个系统因为一个环节失败就完全崩溃。用户宁愿拿到一个不完整的结果,也不愿意看到“系统错误”四个字。

5.3 安全边界设计

多智能体系统因为涉及多个Agent和外部工具,安全边界的设计尤为重要。我的做法是:

  • 工具权限最小化:每个子智能体只分配完成其职责所必需的工具
  • 操作审计:所有MCP工具调用都记录日志,敏感操作需要二次确认
  • 输入输出过滤:对用户输入和Agent输出都做内容安全检查
  • 资源限制:限制单个任务的执行时间、Token消耗和工具调用次数

注意:如果你的系统接入了代码执行或文件写入类的MCP工具,一定要做好沙箱隔离。我见过Agent误删文件的案例,血的教训。

5.4 持续迭代的方法论

多智能体系统不是一次搭好就完事的,需要持续迭代。我的迭代节奏是:

第一周:跑通主流程。不追求完美,先让整个链路能跑起来,哪怕每个环节都很粗糙。

第二周:优化提示词。收集失败案例,针对性优化各子智能体的system_prompt。

第三周:补充Skills。把重复出现的任务模式封装成Skills,减少重复劳动。

第四周:性能调优。分析耗时瓶颈,做并行化和缓存优化。

之后就是持续收集反馈、持续迭代。我一般会维护一个“问题池”,用户反馈的问题、自己发现的问题都扔进去,每周挑几个优先级高的解决。

这套方法论听起来很朴素,但实际执行下来效果很好。多智能体系统的建设是一个长期过程,不要指望一步到位。

最后分享一个我个人的体会:多智能体系统的核心挑战不在于技术,而在于任务拆解和边界定义。技术框架会不断更新,MCP协议会迭代,A2A标准会完善,但“怎么把一个复杂任务拆成合理的子任务,怎么定义每个子智能体的职责边界”这个问题的答案,更多来自于对业务的理解,而不是对技术的掌握。我见过太多技术很牛但任务拆解一塌糊涂的系统,也见过技术一般但任务拆解非常合理的系统跑得风生水起。所以如果你刚开始做多智能体,建议先花时间想清楚任务怎么拆,再动手写代码。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询