1. 从单智能体到多智能体协作的演进
在LLM应用开发的早期阶段,开发者主要关注Prompt Engineering(提示词工程)和RAG(检索增强生成)技术。这些技术让单个智能体能够完成相对简单的任务,比如问答、文本生成等。但随着需求复杂度的提升,单个智能体的局限性逐渐显现:
- 上下文窗口限制:即使是当前最先进的LLM,其上下文窗口也无法容纳复杂项目的全部信息
- 技能单一性:单个智能体难以同时具备产品设计、代码编写、测试验证等多种专业技能
- 任务分解困难:面对"开发一个贪吃蛇游戏"这样的复合型任务,单个智能体缺乏有效的任务分解能力
这些限制催生了多智能体系统(MAS)的概念。在多智能体系统中,不同的智能体各司其职,通过协作完成复杂任务。这就引出了一个关键问题:智能体之间如何高效通信?
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. A2A协议的核心设计理念
A2A(Agent-to-Agent)协议不是某个具体的通信标准,而是一套设计模式和规范,它定义了智能体之间交互的基本原则。与传统的客户端-服务器模式不同,A2A协议中的每个智能体都具有双重身份:
- 服务提供者(Server):监听特定端口,响应其他智能体的请求
- 服务消费者(Client):在需要时主动寻找并调用其他智能体的能力
这种设计带来了几个显著优势:
- 去中心化:没有单点故障,系统更加健壮
- 动态发现:智能体可以随时加入或离开系统
- 能力导向:调用方只需关注"需要什么能力",而不必知道具体由哪个智能体提供
3. A2A协议的四个关键层级
3.1 传输层(Transport Layer)
传输层解决的是智能体之间如何建立连接的问题。常见的选择包括:
- HTTP/HTTPS:简单通用,适合大多数请求-响应式交互
- WebSocket:适合需要长连接和双向通信的场景
- SSE(Server-Sent Events):适用于服务端向客户端推送数据的场景
在我们的实现中,选择了HTTP作为基础传输协议,因为它:
- 被所有编程语言广泛支持
- 易于调试和监控
- 天然支持RESTful和RPC风格的交互
3.2 消息层(Messaging Layer)
消息层定义了智能体之间交换的数据格式。A2A协议推荐使用JSON-RPC 2.0,主要基于以下考虑:
- 语义明确:相比RESTful API,RPC风格的接口更能准确表达"执行某个动作"的意图
- 轻量级:JSON格式易于解析和生成
- 标准化:定义了清晰的请求、响应和错误格式
- 语言无关:任何支持JSON的语言都可以实现JSON-RPC客户端和服务端
一个典型的JSON-RPC 2.0请求如下:
json复制{
"jsonrpc": "2.0",
"method": "write_code",
"params": {
"requirement": "实现一个返回Hello World的HTTP服务"
},
"id": 123456
}
3.3 发现层(Discovery Layer)
发现层解决了"如何找到具备特定能力的智能体"的问题。A2A系统通常包含一个注册中心(Registry),智能体启动时会向注册中心注册自己的能力。当某个智能体需要特定服务时,它会查询注册中心,而不是直接连接具体的服务实例。
这种设计带来了几个好处:
- 动态扩展:新智能体可以随时加入系统
- 负载均衡:同一能力可以由多个智能体提供
- 故障隔离:当某个智能体不可用时,可以快速找到替代者
3.4 语义层(Semantic Layer)
语义层是A2A与传统微服务最大的不同之处。传统微服务传输的是结构化数据,而A2A中的智能体需要处理的是自然语言与结构化数据的混合体。这就要求:
- 自然语言理解:智能体需要能够解析模糊的指令
- 结构化输出:响应需要遵循约定的格式
- 上下文保持:在多轮交互中保持对话的连贯性
4. A2A与MCP协议的对比
MCP(Model Context Protocol)是Anthropic提出的另一种协议,主要用于标准化模型与工具/数据源的连接。它与A2A的主要区别如下:
| 特性 | MCP | A2A |
|---|---|---|
| 拓扑结构 | 星型(中心化) | 网状(去中心化) |
| 通信方向 | 单向(中心节点调用外围服务) | 双向(任意智能体间互相调用) |
| 适用场景 | 为单个模型扩展工具能力 | 多智能体协作完成复杂工作流 |
| 扩展性 | 受限于中心节点的处理能力 | 理论上可以无限扩展 |
MCP适合为单个智能体增加工具使用能力,而A2A更适合构建由多个专业智能体组成的协作系统。
5. 构建微型软件工厂的实践
5.1 系统架构设计
我们的"微型软件工厂"包含三个核心智能体:
-
产品经理(PM Agent):
- 职责:接收用户需求,生成产品需求文档(PRD)
- 能力标签:requirements_analysis
- 技术实现:使用LangChain.js处理自然语言需求
-
开发工程师(Dev Agent):
- 职责:根据PRD编写代码
- 能力标签:coding
- 技术实现:使用GPT模型生成代码片段
-
测试工程师(QA Agent):
- 职责:验证代码质量
- 能力标签:testing
- 技术实现:分析代码与需求的一致性
5.2 技术栈选择
- 运行时环境:Node.js(优秀的异步I/O处理能力)
- Web框架:Express(轻量级HTTP服务)
- LLM框架:LangChain.js(Prompt模板和模型调用)
- 通信协议:JSON-RPC 2.0(标准化的RPC协议)
5.3 核心代码实现
5.3.1 注册中心实现
注册中心是整个系统的协调者,负责记录各个智能体的能力和位置信息。
javascript复制const registryApp = express();
registryApp.use(bodyParser.json());
// 服务存储结构
const services = {};
// 注册接口
registryApp.post("/register", (req, res) => {
const { name, url, capabilities } = req.body;
services[name] = { url, capabilities };
res.json({ status: "ok" });
});
// 发现接口
registryApp.post("/discover", (req, res) => {
const { capability } = req.body;
const foundName = Object.keys(services).find(key =>
services[key].capabilities.includes(capability)
);
if (foundName) {
res.json({ url: services[foundName].url });
} else {
res.status(404).json({ error: "Service not found" });
}
});
registryApp.listen(3000, () => {
console.log("Registry running on port 3000");
});
5.3.2 智能体基类实现
我们抽象了一个AgentRuntime基类,封装了智能体的通用行为:
javascript复制class AgentRuntime {
constructor(name, port, capabilities) {
this.name = name;
this.port = port;
this.capabilities = capabilities;
this.app = express();
this.app.use(bodyParser.json());
this.model = new ChatOpenAI({ modelName: "gpt-3.5-turbo" });
this.app.post("/json-rpc", (req, res) => this.handleRpc(req, res));
}
async start() {
this.app.listen(this.port, async () => {
await this.registerSelf();
});
}
async registerSelf() {
await axios.post(`http://localhost:3000/register`, {
name: this.name,
url: `http://localhost:${this.port}/json-rpc`,
capabilities: this.capabilities
});
}
async handleRpc(req, res) {
const { jsonrpc, method, params, id } = req.body;
if (jsonrpc !== "2.0") {
return res.status(400).json({ error: "Invalid JSON-RPC version" });
}
try {
const result = await this[method](params);
res.json({ jsonrpc: "2.0", result, id });
} catch (error) {
res.status(500).json({ jsonrpc: "2.0", error, id });
}
}
async callOtherAgent(capability, method, params) {
const discoverRes = await axios.post(`http://localhost:3000/discover`, { capability });
const targetUrl = discoverRes.data.url;
const response = await axios.post(targetUrl, {
jsonrpc: "2.0",
method,
params,
id: Date.now()
});
return response.data.result;
}
}
5.3.3 具体智能体实现
基于AgentRuntime基类,我们可以轻松实现具体的业务智能体:
javascript复制// 产品经理智能体
class PMAgent extends AgentRuntime {
constructor() {
super("PM_Agent", 3001, ["requirements_analysis"]);
}
async start_project({ user_input }) {
const prompt = ChatPromptTemplate.fromTemplate(
"将以下需求转化为技术需求文档(PRD): {input}"
);
const chain = prompt.pipe(this.model).pipe(new StringOutputParser());
const prd = await chain.invoke({ input: user_input });
const codeResult = await this.callOtherAgent("coding", "write_code", { requirement: prd });
return {
prd: prd,
code: codeResult
};
}
}
// 开发工程师智能体
class DevAgent extends AgentRuntime {
constructor() {
super("Dev_Agent", 3002, ["coding"]);
}
async write_code({ requirement }) {
const prompt = ChatPromptTemplate.fromTemplate(
"根据以下需求编写代码:\n{req}\n只返回代码,不要解释"
);
const chain = prompt.pipe(this.model).pipe(new StringOutputParser());
const code = await chain.invoke({ req: requirement });
const testReport = await this.callOtherAgent("testing", "review_code", {
code,
requirement
});
return { code, testReport };
}
}
// 测试工程师智能体
class QAAgent extends AgentRuntime {
constructor() {
super("QA_Agent", 3003, ["testing"]);
}
async review_code({ code, requirement }) {
const prompt = ChatPromptTemplate.fromTemplate(
"验证以下代码是否满足需求:\n需求: {req}\n代码: {code}\n给出评估和建议"
);
const chain = prompt.pipe(this.model).pipe(new StringOutputParser());
return await chain.invoke({ req: requirement, code });
}
}
5.4 系统运行流程
-
初始化阶段:
- 启动注册中心(端口3000)
- 启动PM Agent(端口3001)、Dev Agent(端口3002)、QA Agent(端口3003)
- 各智能体向注册中心注册自己的能力
-
任务执行阶段:
- 用户向PM Agent发送需求(如"创建一个返回Hello World的HTTP服务")
- PM Agent生成PRD后,通过注册中心查找具备coding能力的智能体(Dev Agent)
- Dev Agent编写代码后,通过注册中心查找具备testing能力的智能体(QA Agent)
- QA Agent验证代码后,将结果返回给Dev Agent,Dev Agent再将完整结果返回给PM Agent
- PM Agent将最终结果返回给用户
-
扩展场景:
- 如果需要新增一个安全审计智能体,只需:
- 实现SecurityAgent类,注册security_audit能力
- 在Dev Agent的write_code方法中,增加对安全审计的调用
- 原有代码几乎不需要修改,体现了A2A架构的良好扩展性
- 如果需要新增一个安全审计智能体,只需:
6. 实践经验与优化建议
在实际实现和运行A2A系统时,我们总结出以下几点经验:
6.1 性能优化
-
连接池管理:
- 频繁创建和销毁HTTP连接会影响性能
- 建议使用axios的connection pooling或专门的RPC客户端库
-
异步处理:
- 智能体的方法应该是完全异步的
- 使用Promise.all()并行调用多个依赖服务
-
结果缓存:
- 对于相同参数的重复请求,可以考虑缓存结果
- 但要注意缓存失效条件,特别是当底层数据可能变化时
6.2 错误处理
-
重试机制:
- 网络请求可能会失败,需要实现指数退避重试
- 对于非幂等操作要特别小心
-
超时设置:
- 每个RPC调用都应该设置合理的超时时间
- 避免一个慢请求阻塞整个系统
-
熔断机制:
- 当某个智能体频繁失败时,应该暂时停止向其发送请求
- 可以使用类似Circuit Breaker的模式
6.3 监控与调试
-
日志记录:
- 记录每个RPC调用的请求和响应
- 包括调用方、被调用方、时间戳等元数据
-
分布式追踪:
- 为每个用户请求分配唯一的traceId
- 在系统间传递这个traceId,便于追踪完整的调用链
-
指标收集:
- 收集各智能体的调用次数、成功率、延迟等指标
- 使用Prometheus等工具进行可视化
7. 应用场景扩展
A2A协议不仅适用于软件开发场景,还可以应用于:
-
数据分析流水线:
- 数据采集 → 数据清洗 → 特征工程 → 模型训练 → 结果评估
- 每个环节由专门的智能体负责
-
客户服务系统:
- 意图识别 → 信息检索 → 回复生成 → 情感分析
- 根据客户问题动态组合不同的智能体
-
物联网控制:
- 传感器数据收集 → 异常检测 → 决策制定 → 执行器控制
- 分布式部署在不同设备上
8. 未来发展方向
-
协议标准化:
- 目前A2A还缺乏统一的标准
- 未来可能出现类似gRPC的专门为智能体通信设计的协议
-
安全机制:
- 智能体间的认证和授权
- 通信加密和完整性验证
-
服务质量管理:
- 基于SLA的智能体选择
- 动态负载均衡和弹性扩缩容
-
混合架构:
- A2A与MCP的结合
- 既有智能体间的水平协作,也有模型与工具的垂直集成
通过这个微型软件工厂的实现,我们展示了A2A协议在多智能体协作中的强大能力。相比传统的单体智能体架构,A2A提供了更好的灵活性、扩展性和可维护性。随着智能体技术的普及,A2A很可能成为构建复杂AI系统的标准范式之一。
