1. 从零理解MCP与A2A:多智能体系统的两块基石
1.1 为什么单智能体不够用了
过去一年我陆续做了七八个基于大模型的自动化项目,从最简单的文档问答到稍微复杂点的代码生成流水线,踩过的坑基本能写一本小册子。最开始大家的思路都很朴素:一个模型、一段系统提示词、几个工具函数,能跑通就上线。但项目一旦稍微复杂一点,比如要同时处理数据抓取、格式清洗、逻辑校验、结果输出这四件事,单智能体就开始露怯了。
问题出在哪?我总结下来主要是三个:上下文污染、职责耦合、错误传播。你让一个模型既当调度员又当执行者,它会在长对话里逐渐忘记自己最初的角色设定;你把所有工具都塞给它,它选错工具的概率会随着工具数量线性上升;最要命的是,一旦中间某一步出错,后面所有步骤都会基于错误结果继续往下跑,最后给你一个看起来像模像样但完全错误的答案。
多智能体架构就是为了解决这三个问题而生的。核心思路很简单:把一个大任务拆成若干个子任务,每个子任务交给一个专门的智能体,智能体之间通过标准化的协议通信。这就像一家公司,你不会让一个人同时干销售、财务、法务和运维,而是分部门协作,部门之间有明确的交接流程。
但这里有个关键问题:智能体之间怎么通信?如果每个智能体都用自己的一套接口,那系统会变得极其脆弱,加一个新智能体就要改一堆代码。这就是MCP和A2A要解决的事情。
1.2 MCP到底是什么:给智能体装上标准化的手
MCP全称是Model Context Protocol,翻译过来叫模型上下文协议。你可以把它理解成智能体和外部工具之间的USB-C接口。在MCP出现之前,每个智能体要调用外部工具,都得自己写一套适配代码:调数据库要写一套,调搜索引擎要写一套,调文件系统又要写一套。工具一多,代码就变成了一团乱麻。
MCP的核心价值在于标准化。它定义了一套统一的通信规范,任何工具只要按照这个规范暴露自己的能力,任何智能体只要按照这个规范去调用,双方就能无缝对接。这就像你买了一个USB-C的硬盘,不管是插在笔记本上还是插在台式机上,都能直接用,不需要为每台电脑单独买一根线。
从技术实现上看,MCP通常采用客户端-服务端架构。智能体作为客户端,工具作为服务端,双方通过JSON-RPC格式的消息进行通信。一个典型的MCP服务端会暴露三类能力:资源(Resources)、工具(Tools)、提示模板(Prompts)。资源是只读的数据,比如文件内容、数据库记录;工具是可执行的操作,比如发送邮件、查询天气;提示模板是预定义的指令片段,方便智能体快速调用。
我实际用下来,MCP最爽的地方在于工具的热插拔。你写好一个MCP服务端,注册到智能体框架里,智能体就能立刻使用这个工具,不需要改任何核心代码。反过来,你想换一个工具实现,只要新实现符合MCP规范,直接替换就行,智能体那边完全无感知。
1.3 A2A又是什么:让智能体之间能对话
如果说MCP解决的是智能体和工具之间的通信问题,那A2A解决的就是智能体和智能体之间的通信问题。A2A全称是Agent-to-Agent Protocol,核心目标是让不同的智能体能够互相发现、互相调用、互相协作。
在没有A2A之前,多智能体系统通常采用硬编码的方式定义智能体之间的调用关系。比如调度智能体直接import执行智能体的函数,或者通过消息队列手动传递数据。这种方式在小规模场景下能用,但一旦智能体数量增多、部署位置分散,就会变得极其难以维护。
A2A的思路是把每个智能体都当成一个独立的服务。每个智能体对外暴露一个标准的接口描述,其他智能体通过这个描述来了解它能做什么、需要什么输入、会返回什么输出。这就像微服务架构中的服务发现机制,每个服务注册自己的能力,其他服务通过注册中心来查找和调用。
A2A协议通常包含几个核心概念:Agent Card(智能体名片),描述智能体的名称、能力、输入输出格式;Task(任务),表示一次具体的协作请求;Message(消息),任务执行过程中的通信单元。智能体之间通过交换Task和Message来完成协作。
我个人的体会是,A2A最大的价值在于解耦。调度智能体不需要知道执行智能体具体是怎么实现的,只需要知道它的Agent Card就行。执行智能体可以随时替换实现、升级版本、甚至换成完全不同的技术栈,只要Agent Card不变,整个系统就能继续运转。
1.4 MCP和A2A的关系:一个管纵向,一个管横向
很多人刚开始接触这两个概念时会混淆,觉得它们是不是竞争关系。其实不是,它们解决的是不同维度的问题。
MCP管的是纵向的通信,也就是智能体往下调用工具、数据源、外部服务。它让智能体能够方便地使用各种外部能力。
A2A管的是横向的通信,也就是智能体之间互相调用、协作、协商。它让多个智能体能够组成一个团队,共同完成复杂任务。
打个比方,MCP像是给每个员工配了一套标准化的办公设备接口,不管是用打印机、扫描仪还是投影仪,插上就能用。A2A像是给员工之间配了一套标准化的沟通流程,谁负责什么、怎么交接、怎么反馈,都有明确的规范。
在实际系统中,这两者通常是配合使用的。一个智能体既可以通过MCP调用外部工具,也可以通过A2A和其他智能体协作。两者结合,才能构建出真正灵活、可扩展的多智能体系统。
2. 多智能体系统的架构设计:从需求到方案
2.1 先想清楚要解决什么问题
在动手写代码之前,我习惯先花时间把需求理清楚。多智能体系统不是银弹,不是所有任务都适合用多智能体来解决。如果任务本身很简单,一个智能体加几个工具就能搞定,硬上多智能体只会增加复杂度和调试难度。
我一般用三个问题来判断是否需要多智能体架构:
第一,任务是否能自然拆分成多个子任务?比如一个报告生成任务,可以拆成数据收集、数据分析、报告撰写、格式校对四个子任务。如果拆不出来,那可能不适合多智能体。
第二,子任务之间是否需要不同的能力或知识?如果所有子任务都需要同样的能力,那一个智能体就够了。只有当子任务需要不同的工具、不同的提示词、不同的模型时,多智能体才有优势。
第三,子任务之间是否需要频繁交互?如果子任务之间几乎不需要交互,那用工作流引擎串起来就行,不一定需要A2A。只有当子任务之间需要动态协商、互相反馈时,A2A才真正发挥价值。
以我最近做的一个技术文档生成系统为例。需求是:给定一个开源项目的代码仓库,自动生成一份包含架构说明、API文档、使用示例的技术文档。这个任务可以拆成:代码结构分析、API接口提取、示例代码生成、文档整合四个子任务。每个子任务需要的工具和能力不同,而且API提取的结果会影响示例生成的内容,存在交互需求。所以这个场景适合用多智能体架构。
2.2 智能体角色的划分原则
确定要用多智能体之后,下一步是划分角色。我踩过的坑是:角色划分太细会导致通信开销爆炸,太粗又失去了多智能体的意义。
我的经验是遵循单一职责原则,但粒度控制在“一个智能体负责一个可独立验证的子任务”这个级别。什么叫可独立验证?就是这个子任务的输出可以单独拿出来检查对错,不需要依赖其他子任务的输出。
还是以技术文档生成系统为例。我最初把角色划分成:代码分析员、API提取员、示例生成员、文档撰写员、格式校对员五个角色。后来发现格式校对员和文档撰写员的职责高度重叠,合并成一个角色后系统反而更稳定。最终定下来四个角色:
- 代码分析智能体:负责解析代码仓库结构,识别模块划分、依赖关系、核心类和方法。
- API提取智能体:负责从代码中提取公开接口,生成结构化的API描述。
- 示例生成智能体:负责根据API描述生成可运行的使用示例。
- 文档整合智能体:负责把前面三个智能体的输出整合成最终文档,并做一致性检查。
每个角色的输入输出都明确定义,而且可以独立测试。比如我可以单独给API提取智能体喂一个代码文件,检查它提取的接口是否准确,不需要跑整个系统。
2.3 通信拓扑的选择:星型、网状还是混合
角色划分好之后,接下来要决定智能体之间怎么连接。常见的拓扑结构有三种:
星型拓扑:有一个中心调度智能体,所有其他智能体都只和调度智能体通信。这种结构最简单,调试最容易,但调度智能体容易成为瓶颈,而且单点故障风险高。
网状拓扑:智能体之间可以任意通信。这种结构最灵活,但通信路径复杂,调试困难,而且容易出现循环调用。
混合拓扑:结合前两者的特点,核心流程用星型,特定场景下允许智能体之间直接通信。
我实际项目中用得最多的是混合拓扑。还是以文档生成系统为例,整体流程是星型的:调度智能体依次调用代码分析、API提取、示例生成、文档整合四个智能体。但在示例生成阶段,示例生成智能体需要向API提取智能体查询具体的接口签名,这时候允许它们直接通信,不需要经过调度智能体转发。
这种设计的好处是:主流程清晰可控,同时保留了必要的灵活性。调度智能体负责整体进度和异常处理,具体智能体之间的细节交互可以自主完成。
2.4 状态管理:别让智能体变成无状态函数
多智能体系统的一个关键设计决策是:智能体是有状态的还是无状态的?
无状态智能体每次调用都是独立的,不保留任何历史信息。这种设计简单、可扩展,但无法处理需要上下文的场景。比如文档整合智能体需要知道前面三个智能体分别产出了什么,如果每次调用都重新获取,效率会很低。
有状态智能体保留自己的上下文,可以记住之前的交互。这种设计更自然,但会带来状态同步和持久化的问题。
我的做法是分层管理状态。每个智能体内部维护自己的局部状态,比如代码分析智能体记住已经分析过的文件列表。全局状态由一个专门的状态管理服务维护,所有智能体通过MCP访问这个服务来读写共享状态。
这样设计的好处是:智能体内部的状态变化不影响其他智能体,降低了耦合度;全局状态集中管理,避免了数据不一致的问题。而且状态管理服务本身也是一个MCP服务端,符合我们前面说的标准化原则。
3. 核心实现细节:MCP服务端与A2A通信的落地
3.1 用MCP封装一个工具服务端
先来看MCP服务端的具体实现。假设我们要封装一个“代码文件读取”工具,让智能体能够读取指定路径的代码文件内容。
MCP服务端的核心是定义一个符合协议规范的接口。以下是一个简化的Python实现示例:
from mcp.server import Server from mcp.types import Tool, TextContent import os app = Server("code-reader") @app.list_tools() async def list_tools(): return [ Tool( name="read_code_file", description="读取指定路径的代码文件内容", inputSchema={ "type": "object", "properties": { "path": {"type": "string", "description": "文件路径"}, "max_lines": {"type": "integer", "description": "最大读取行数", "default": 500} }, "required": ["path"] } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict): if name == "read_code_file": path = arguments["path"] max_lines = arguments.get("max_lines", 500) if not os.path.exists(path): return [TextContent(type="text", text=f"文件不存在: {path}")] with open(path, "r", encoding="utf-8") as f: lines = f.readlines()[:max_lines] content = "".join(lines) return [TextContent(type="text", text=content)]这个服务端定义了一个名为read_code_file的工具,接收路径和最大行数两个参数,返回文件内容。智能体通过MCP协议调用这个工具时,不需要知道文件读取的具体实现,只需要按照inputSchema传入参数即可。
这里有个细节值得注意:max_lines参数的设计。代码文件可能非常大,如果一次性全部读取,会消耗大量token。设置一个默认上限,既保证了智能体能获取足够的信息,又避免了资源浪费。这个上限具体设多少,要根据实际场景调整。我一般设500行,对于大多数代码文件来说足够覆盖核心逻辑。
3.2 A2A通信的实现:Agent Card与任务分发
A2A通信的核心是Agent Card。每个智能体启动时,会向注册中心注册自己的Agent Card,描述自己的能力和接口。
以下是一个Agent Card的示例:
{ "name": "api-extractor", "description": "从代码中提取公开API接口,生成结构化描述", "version": "1.0.0", "capabilities": { "input": { "type": "object", "properties": { "code_content": {"type": "string", "description": "代码文件内容"}, "language": {"type": "string", "description": "编程语言", "enum": ["python", "java", "go"]} }, "required": ["code_content", "language"] }, "output": { "type": "object", "properties": { "apis": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"}, "signature": {"type": "string"}, "description": {"type": "string"}, "parameters": {"type": "array"} } } } } } }, "endpoint": "http://localhost:8001/a2a" }调度智能体通过查询注册中心获取这个Agent Card,就知道api-extractor智能体需要代码内容和编程语言作为输入,会返回一个API列表。调度智能体不需要知道api-extractor内部是用正则表达式还是用AST解析,只需要按照Agent Card的约定发送请求即可。
任务分发的实现通常采用异步消息模式。调度智能体创建一个Task,发送给目标智能体,然后等待结果。目标智能体处理完成后,通过回调或者轮询的方式返回结果。
import httpx import asyncio async def dispatch_task(agent_endpoint: str, task_input: dict): async with httpx.AsyncClient() as client: response = await client.post( f"{agent_endpoint}/tasks", json={ "task_id": generate_task_id(), "input": task_input, "callback_url": "http://scheduler/callback" }, timeout=300.0 ) return response.json()这里有个实操心得:超时时间要设够。智能体处理任务的时间可能比预期长很多,特别是涉及大模型推理的场景。我最初设了60秒超时,结果经常出现任务还没处理完就超时的情况。后来改成300秒,并且加了重试机制,稳定性好了很多。
3.3 调度智能体的核心逻辑
调度智能体是整个系统的大脑,负责协调各个执行智能体。它的核心逻辑可以用一个状态机来描述:
class SchedulerAgent: def __init__(self): self.state = "idle" self.task_queue = [] self.results = {} async def execute_workflow(self, repo_path: str): # 第一步:代码分析 self.state = "analyzing" analysis_result = await self.call_agent( "code-analyzer", {"repo_path": repo_path} ) self.results["analysis"] = analysis_result # 第二步:API提取 self.state = "extracting" api_result = await self.call_agent( "api-extractor", { "code_content": analysis_result["core_files"], "language": analysis_result["language"] } ) self.results["apis"] = api_result # 第三步:示例生成 self.state = "generating" example_result = await self.call_agent( "example-generator", {"apis": api_result["apis"]} ) self.results["examples"] = example_result # 第四步:文档整合 self.state = "integrating" doc_result = await self.call_agent( "doc-integrator", { "analysis": analysis_result, "apis": api_result, "examples": example_result } ) self.results["document"] = doc_result self.state = "completed" return doc_result这个调度逻辑看起来简单,但实际实现时要考虑很多细节。比如错误处理:如果API提取失败了,是重试还是跳过?我的做法是区分错误类型,网络超时类的错误自动重试三次,逻辑错误类的错误直接上报给人工处理。
再比如进度追踪:用户需要知道当前执行到哪一步了。我在每个状态转换时都会更新一个进度对象,用户可以通过查询接口获取实时进度。
3.4 智能体之间的数据传递格式
智能体之间传递的数据格式直接影响系统的可维护性。我试过几种方案:
方案一:纯文本传递。上游智能体把结果拼成一段文本,下游智能体自己解析。这种方案最简单,但极其脆弱,上游改一个标点符号下游就可能解析失败。
方案二:JSON传递。上游智能体输出结构化JSON,下游智能体按照约定的schema解析。这种方案好很多,但需要严格维护schema版本。
方案三:JSON Schema + 版本号。在方案二的基础上,每个数据结构都带一个版本号,下游智能体根据版本号选择对应的解析逻辑。这种方案最稳健,但实现复杂度也最高。
我最终采用的是方案三的简化版:JSON + 必填字段校验。每个智能体的输出都定义一个必填字段列表,下游智能体在解析前先校验必填字段是否存在。如果缺失,直接报错并附带缺失字段信息,方便快速定位问题。
def validate_input(data: dict, required_fields: list): missing = [f for f in required_fields if f not in data] if missing: raise ValueError(f"输入数据缺少必填字段: {missing}") return True这个校验逻辑虽然简单,但在实际调试中帮我省了大量时间。很多时候问题不是出在逻辑上,而是出在数据格式上。有了这个校验,问题能第一时间暴露出来。
4. 实操中的常见问题与排查技巧
4.1 智能体调用超时怎么办
超时是多智能体系统中最常见的问题。我遇到过的超时场景主要有三类:
第一类:工具调用超时。智能体通过MCP调用外部工具时,工具响应太慢。比如调用一个远程API,对方服务器负载高,响应时间超过预期。
排查方法:在MCP服务端加日志,记录每个工具调用的开始时间和结束时间。如果发现某个工具频繁超时,考虑加缓存或者换一个更快的实现。
第二类:智能体推理超时。智能体本身处理任务的时间太长,通常是因为输入太大或者模型推理太慢。
排查方法:检查输入数据的大小。我遇到过一次,代码分析智能体接收了一个包含几万行代码的文件,导致推理时间超过5分钟。后来加了文件大小限制,超过阈值的文件先切分再处理。
第三类:通信超时。智能体之间的网络通信超时,通常是因为网络抖动或者目标智能体不可用。
排查方法:加健康检查机制,定期探测各个智能体的可用性。如果发现某个智能体不可用,调度智能体应该能够自动切换到备用智能体或者降级处理。
以下是我总结的超时问题速查表:
| 超时类型 | 典型原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 工具调用超时 | 外部服务响应慢 | 记录工具调用耗时 | 加缓存、换实现、设超时上限 |
| 智能体推理超时 | 输入过大、模型慢 | 检查输入大小和模型配置 | 切分输入、换更快的模型 |
| 通信超时 | 网络抖动、目标不可用 | 健康检查、网络监控 | 重试、切换备用、降级处理 |
4.2 智能体之间数据不一致怎么排查
数据不一致是另一个高频问题。典型表现是:上游智能体说处理成功了,下游智能体却拿不到数据。
我遇到过一次很典型的情况:API提取智能体输出了一个API列表,但文档整合智能体读取时发现列表是空的。排查了半天,最后发现是API提取智能体在输出时用了apis字段名,而文档整合智能体读取时用的是api_list字段名。两边字段名不一致,导致数据丢失。
这个问题的根本原因是接口约定没有严格执行。后来我加了一个强制校验:每个智能体的输出在发送前,必须通过一个schema校验器,确保字段名和类型符合约定。下游智能体在接收时也做一次校验,双重保险。
还有一个隐蔽的问题是时序问题。调度智能体同时调用两个智能体,期望它们都完成后才进入下一步。但如果其中一个先完成,另一个后完成,先完成的那个结果可能会被覆盖。解决方法是给每个结果加一个唯一标识,调度智能体根据标识来匹配结果,而不是根据完成顺序。
4.3 智能体陷入循环调用怎么破
循环调用是多智能体系统中最危险的问题之一。A调用B,B又调用A,无限循环,直到资源耗尽。
我遇到过一次:代码分析智能体发现某个文件需要API提取智能体处理,于是调用API提取智能体。API提取智能体在处理过程中发现需要更多代码上下文,又回头调用代码分析智能体。两个智能体互相调用,陷入了死循环。
解决循环调用的核心是调用链追踪。每次智能体调用时,在请求头里带上一个调用链ID和调用深度。如果调用深度超过阈值(比如5层),直接拒绝并返回错误。
MAX_CALL_DEPTH = 5 async def call_agent(endpoint: str, input_data: dict, call_chain: list = None): call_chain = call_chain or [] if len(call_chain) >= MAX_CALL_DEPTH: raise RecursionError(f"调用深度超过上限: {call_chain}") new_chain = call_chain + [endpoint] # 在请求中带上调用链信息 response = await http_post(endpoint, { "input": input_data, "call_chain": new_chain }) return response除了深度限制,还可以加时间窗口限制:同一个智能体在短时间内被多次调用时,触发告警。这能帮助发现潜在的循环调用。
4.4 性能优化的几个实操技巧
多智能体系统的性能优化,我总结下来主要有三个方向:
并行化。能并行执行的智能体调用尽量并行。比如代码分析和API提取如果不存在依赖关系,可以同时进行。我用asyncio.gather来实现并行调用,整体耗时从原来的串行30秒降到了并行15秒。
async def parallel_execute(tasks: list): results = await asyncio.gather(*tasks, return_exceptions=True) for i, result in enumerate(results): if isinstance(result, Exception): print(f"任务{i}失败: {result}") return results缓存。智能体的输出如果在一定时间内不会变化,可以缓存起来。比如代码分析的结果,只要代码仓库没变,就不需要重新分析。我用Redis做缓存,key是代码仓库的commit hash,value是分析结果。第二次请求同样的仓库时,直接从缓存读取,耗时从15秒降到了0.5秒。
流式输出。对于文档生成这种耗时较长的任务,采用流式输出可以让用户更早看到部分结果。文档整合智能体每生成一个章节就推送一次,用户不需要等整个文档生成完才能看到内容。
4.5 调试多智能体系统的实用工具
调试多智能体系统比调试单智能体复杂得多,因为涉及多个进程、多次通信。我常用的工具组合是:
日志聚合。所有智能体的日志统一收集到一个地方,按调用链ID分组。这样排查问题时可以看到一次完整调用涉及的所有智能体的日志。我用的是ELK栈,轻量级场景下用文件日志加grep也够用。
链路追踪。每个请求生成一个trace ID,在所有智能体之间传递。通过trace ID可以还原出完整的调用路径和耗时分布。OpenTelemetry是很好的选择,支持多种语言和框架。
可视化面板。我搭了一个简单的Web面板,实时显示各个智能体的状态、当前任务、历史成功率。这个面板在演示和排查问题时特别有用,一眼就能看出哪个智能体出了问题。
Mock智能体。在开发阶段,我会为每个智能体写一个Mock版本,返回固定的测试数据。这样可以在不依赖真实智能体的情况下测试调度逻辑,大大加快了开发速度。
5. 从单机到分布式:部署与扩展的实战经验
5.1 开发环境与生产环境的差异
开发阶段我通常把所有智能体跑在同一台机器上,通过localhost通信。这种配置简单、调试方便,但和生产环境差异很大。
生产环境需要考虑的问题多得多:网络延迟、服务发现、负载均衡、故障转移、安全认证。我踩过的最大的坑是网络延迟。开发时localhost通信延迟在毫秒级,生产环境跨机房通信延迟可能到几十毫秒。一个流程涉及十几次智能体调用,累积延迟就非常可观了。
应对策略是减少不必要的跨机房调用。把频繁交互的智能体部署在同一个机房,把独立性强的智能体分散部署。同时优化通信协议,用二进制格式代替JSON,减少传输数据量。
5.2 智能体的水平扩展
当单个智能体处理不过来时,需要水平扩展。多智能体系统的一个优势是扩展粒度细,可以只扩展瓶颈智能体,不需要整体扩容。
扩展的关键是无状态化。如果智能体是有状态的,扩展时需要考虑状态同步。我的做法是把状态外置到共享存储(如Redis),智能体本身变成无状态的。这样加机器就能直接分担负载,不需要复杂的状态迁移。
负载均衡策略我一般用最少连接数。每个智能体实例维护一个当前处理任务数,调度智能体选择任务数最少的实例来分发。这种策略比轮询更适应实际负载分布。
5.3 监控与告警体系的搭建
生产环境必须有完善的监控告警。我关注的指标主要有:
- 成功率:每个智能体的任务成功率,低于95%触发告警。
- 响应时间:P50、P95、P99响应时间,P99超过阈值触发告警。
- 队列长度:待处理任务数,持续增长说明处理能力不足。
- 资源使用:CPU、内存、网络IO,接近上限时提前扩容。
告警渠道我用的是企业微信机器人,简单直接。告警信息包含:哪个智能体、什么指标、当前值、阈值、建议操作。这样收到告警的人能快速判断严重程度并采取措施。
5.4 版本升级与灰度发布
多智能体系统的版本升级比单体应用复杂,因为涉及多个智能体的协同。我的做法是按依赖顺序升级:先升级被依赖的智能体,再升级依赖方。
比如文档生成系统中,API提取智能体被示例生成智能体依赖。升级时先升级API提取智能体,确保它的输出格式向后兼容,然后再升级示例生成智能体。
灰度发布策略是按流量比例。新版本智能体先接收10%的流量,观察一段时间没问题后逐步提高到50%、100%。如果发现问题,立即切回旧版本。
这里有个细节:Agent Card的版本管理。新版本智能体的Agent Card要保留旧版本的接口描述,确保旧版本的调度智能体还能正常调用。等所有调度智能体都升级完成后,再移除旧版本接口。
6. 实际项目复盘:技术文档生成系统的完整实现
6.1 项目背景与需求分析
这个项目的需求是:给定一个开源项目的代码仓库地址,自动生成一份完整的技术文档,包含项目概述、架构说明、API文档、使用示例四个部分。
项目启动前,我评估了几个关键指标:代码仓库大小(平均5000-20000行)、支持的编程语言(Python、Java、Go)、文档质量要求(需要人工审核通过)、生成时间要求(30分钟内完成)。
基于这些指标,我确定了技术方案:用四个智能体分别负责代码分析、API提取、示例生成、文档整合,通过MCP调用代码读取、文件写入等工具,通过A2A进行智能体间通信。
6.2 智能体实现细节
代码分析智能体的核心逻辑是遍历代码仓库,识别模块结构。我用了Python的ast模块来解析Python代码,用javalang解析Java代码,用go/parser解析Go代码。每个文件解析后,提取出类、方法、导入关系等信息,汇总成项目结构图。
这里有个坑:不同语言的解析器输出格式不一致。Python的ast输出是树形结构,Java的javalang输出是扁平列表,Go的parser输出又是另一种格式。我写了一个适配层,把三种格式统一转换成标准的项目结构JSON。
API提取智能体从代码分析结果中提取公开接口。判断一个接口是否公开,不同语言规则不同:Python看是否以_开头,Java看是否有public修饰符,Go看首字母是否大写。提取出的接口包含名称、签名、参数列表、返回值、文档字符串。
示例生成智能体根据API描述生成使用示例。这里用到了大模型的代码生成能力。我给每个API构造一个提示词,包含API签名、参数说明、返回值说明,让模型生成一段调用示例。生成后还会做一次语法检查,确保示例代码能通过编译。
文档整合智能体把前面三个智能体的输出整合成Markdown格式的文档。整合过程中会做一致性检查,比如API文档中提到的类是否在架构说明中出现过,示例代码中调用的方法是否在API文档中定义过。
6.3 性能数据与优化效果
系统上线后的性能数据:
| 指标 | 优化前 | 优化后 |
|---|---|---|
| 平均生成时间 | 45分钟 | 18分钟 |
| 成功率 | 78% | 94% |
| API提取准确率 | 85% | 96% |
| 示例代码可运行率 | 70% | 89% |
优化措施主要包括:并行化代码分析和API提取、缓存代码分析结果、优化提示词减少模型输出token数、增加重试机制。
6.4 踩过的坑与经验总结
这个项目做下来,最大的体会是:多智能体系统的难点不在智能体本身,而在智能体之间的协作。单个智能体的逻辑再复杂,也是可控的;但多个智能体之间的交互,会出现各种意想不到的情况。
我踩过的一个典型坑是智能体之间的数据竞争。代码分析智能体和API提取智能体都需要读取代码文件,如果同时读取同一个文件,可能会出现文件被锁的情况。后来加了一个文件访问队列,同一时间只允许一个智能体读取同一个文件。
另一个坑是提示词漂移。示例生成智能体的提示词在迭代过程中被修改了多次,每次修改都会影响生成结果的质量。后来我建立了提示词版本管理,每次修改都记录变更原因和效果对比,确保可追溯。
还有一个经验是:不要过度追求自动化。有些环节人工介入反而更高效。比如文档整合后的最终审核,我最初想用智能体自动审核,但效果不好。后来改成智能体生成审核建议,人工做最终确认,整体效率反而更高。
7. 后续扩展方向与个人建议
7.1 可以继续深挖的几个方向
这个系统目前只覆盖了技术文档生成这一个场景,但底层的MCP和A2A框架是通用的。我接下来打算往几个方向扩展:
多模态支持。目前的智能体只能处理文本,后续可以加入图像识别智能体,支持从架构图中提取信息,或者生成架构图。
交互式文档。生成的文档目前是静态的,后续可以加入交互式元素,比如点击API名称展开详细说明,或者在线运行示例代码。
多仓库支持。目前一次只能处理一个代码仓库,后续可以支持多个仓库的联合分析,生成跨项目的文档。
7.2 给刚接触多智能体的朋友几点建议
如果你刚开始接触多智能体系统,我的建议是从最简单的场景开始。不要一上来就搞复杂的多智能体协作,先用一个智能体加几个MCP工具跑通一个完整流程,然后再逐步拆分。
重视接口设计。智能体之间的接口一旦确定,后续修改成本很高。花时间把Agent Card设计好,把输入输出格式定义清楚,后面会省很多事。
日志和监控要早做。不要等到出问题了才想起来加日志。多智能体系统的调试难度比单智能体高一个数量级,没有完善的日志和监控,排查问题会非常痛苦。
保持简单。多智能体不是越复杂越好。每增加一个智能体,就增加一份通信开销和调试复杂度。能用三个智能体解决的问题,不要用五个。
7.3 关于MCP和A2A生态的观察
从我个人的使用体验来看,MCP和A2A目前还处于快速演进阶段。MCP的工具生态已经比较丰富了,常见的数据库、搜索引擎、文件系统都有现成的MCP服务端。A2A的生态还在建设中,智能体之间的协作模式还在探索。
我建议关注这两个协议的版本更新,及时跟进新特性。同时也要注意不要过度依赖某个特定实现。协议是标准,实现可以替换。把核心逻辑建立在协议层面,而不是某个具体的库或框架上,这样后续迁移成本会低很多。
最后分享一个我在实际项目中总结的小技巧:给每个智能体起一个有意义的名字。不要用agent1、agent2这种命名,而是用code-analyzer、api-extractor这种描述性的名字。这在日志排查和监控面板上会直观很多,一眼就能看出是哪个环节出了问题。这个习惯看似微不足道,但在长期维护中能节省大量时间。