在 LLM 应用开发的早期，我们的目光主要聚焦于 Prompt Engineering（提示词工程）和 RAG（检索增强生成）。随后，Agent（智能体）概念爆发，通过 Function Calling，LLM 拥有了使用计算器或搜索工具的“手”。

但当我们面对“开发一个贪吃蛇游戏”这样模糊且复杂的任务时，单个 Agent 往往显得力不从心——它的上下文窗口有限，它的“技能树”也不可能无限扩张。试图让一个 Prompt 处理所有逻辑，最终只会得到一个臃肿且不可维护的“上帝类”。

我们需要的是一个虚拟团队，即多智能体协作系统（MAS, Multi-Agent System）。在这个系统中，Agent 之间如何“说话”、如何“发现”彼此、如何“移交”任务，成为了关键问题。这就是 A2A (Agent-to-Agent) 协议诞生的背景。

什么是 A2A 协议？

A2A 协议并非指代某一个特定的硬性标准（如 TCP/IP），而是指代一种设计模式和通信规范。它定义了智能体之间如何进行点对点的交互。

在 A2A 的架构中，每个 Agent 不再仅仅是被动等待人类指令的工具箱，它们发生了身份的质变：

• 它是 Server：监听网络端口，随时准备响应其他 Agent 的请求。
• 它是 Client：在处理任务受阻时，主动寻找并调用其他 Agent 的能力。

A2A 协议通常定义了四个层面的标准：

• 传输层 (Transport Layer):
  • Agent 之间如何连接？最常见的是 HTTP/HTTPS（简单通用），但在需要实时流式传输（Streaming）的场景下，WebSocket 或 SSE (Server-Sent Events) 是更好的选择。
• 消息层 (Messaging Layer):
  • 数据包长什么样？A2A 强烈推荐 JSON-RPC 2.0。
  • 为什么不是 RESTful? REST 关注资源（Resource），而 Agent 关注行为（Action）。POST /generate_code 这种指令式的语义更符合 RPC（远程过程调用）的直觉。
• 发现层 (Discovery Layer):
  • 我怎么找到队友？A2A 系统中必须包含一个动态的注册中心 (Registry)。Agent 启动时汇报自己的“能力标签”（如 capability: image_generation），调用方通过标签查找 IP，而不是硬编码 URL。
• 语义层 (Semantic Layer):
  • 我们聊什么？这是 A2A 与传统微服务最大的不同。传统微服务传输的是确定的数据（Data），而 Agent 之间传输的是自然语言与结构化数据的混合体。A2A 协议要求 Agent 能够理解并处理模糊的指令，并返回结构化的结果。

它们遵循统一的接口规范（通常是 JSON-RPC），通过标准网络协议（如 HTTP/SSE/WebSocket）进行自主通信。

从 MCP 到 A2A：为什么要重新造轮子？

目前业界很火的是 MCP (Model Context Protocol)，由 Anthropic 提出。它非常优秀，但它解决的问题域与 A2A 有所不同。

MCP 的局限：中心化的星型拓扑

MCP 的核心是为了标准化“模型连接数据/工具”的方式。

• 架构：Host (如 Claude Desktop) <-> Server (工具/数据源)。
• 痛点：它本质上是星型结构。所有的决策压力都在 Host 身上。Server 之间无法直接通信（例如，Github Tool 无法直接告诉 Slack Tool 发送消息，必须经过 Host 中转）。这导致 Host 上下文容易爆炸，且难以实现真正的分布式自治。

A2A 的优势：去中心化的网状协作

A2A 旨在解决 Agent 之间的协作问题，模拟人类公司的运作模式。

• 对等网络：Agent 之间是平等的。PM Agent 可以直接调用 Dev Agent，无需经过用户中转。
• 服务发现：引入注册中心（Registry），解决“动态组队”的问题。Agent 上线后只需注册能力，无需硬编码 IP。
• 标准化交互：使用 JSON-RPC 2.0，明确了请求（Request）、响应（Response）和错误（Error）的格式，使得不同语言编写的 Agent 也能互相调用。

特性	MCP (Model Context Protocol)	A2A (Agent-to-Agent)
核心目标	统一 AI 模型与数据/工具的连接	统一智能体之间的协作与通信
拓扑结构	星型 (Client-Host-Server)	网状 / 分布式微服务
通信流向	Host 主动拉取/调用	Agent 之间双向调用 (Peer-to-Peer)
适用场景	给 IDE、聊天机器人挂载工具	复杂工作流、虚拟软件公司

---

A2A 实践：构建微型软件工厂

为了演示这一概念，我们将构建一个微型软件工厂。在这个系统中，Agent 之间不再是同一个进程里的对象，而是独立的微服务。

角色定义

项目包含三个 Agent，它们通过标准化的 A2A 协议进行协作：

1. 产品经理 (PM Agent)： 接收用户模糊的输入，分析需求，输出标准化的《需求文档 (PRD)》。
2. 开发工程师 (Dev Agent)：接收 PRD，编写代码，输出《源代码》。
3. 测试工程师 (QA Agent)： 接收 PRD 和源代码，进行代码审查，输出《测试报告》。

技术栈选择

• Runtime: Node.js (利用其优秀的异步 I/O 处理并发请求)
• Web Framework: Express (模拟微服务 HTTP 接口)
• LLM Framework: LangChain.js (处理 Prompt Template 和 LLM 调用)
• Protocol: JSON-RPC 2.0 (为何选它？因为它无状态、轻量，且天然支持 method 和 params 的映射，非常适合映射 Agent 的 Function)

A2A 协作流程图

---

代码实现

为了方便演示，我们将所有服务写在一个文件中（index.js），通过不同的端口模拟分布式环境。

1. 基础设施搭建

首先，我们需要定义通信的“语言”。这里我们严格遵循 JSON-RPC 2.0 规范。

import express from "express";
import bodyParser from "body-parser";
import axios from "axios";
import { ChatOpenAI } from "@langchain/openai";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";
import dotenv from "dotenv";
import colors from "colors"; // 用于控制台彩色日志

dotenv.config();

// --- 协议层：JSON-RPC 2.0 封装 ---
class JsonRpcProtocol {
    static request(method, params, id = Date.now()) {
        return { jsonrpc: "2.0", method, params, id };
    }
    static success(result, id) {
        return { jsonrpc: "2.0", result, id };
    }
    static error(code, message, id) {
        return { jsonrpc: "2.0", error: { code, message }, id };
    }
}

2. 服务注册中心 (Service Registry)

这是 A2A 协议的大脑。Agent 不需要知道队友的 IP 地址，只需要知道队友具备什么“能力（Capability）”。这实现了位置解耦。

const REGISTRY_PORT = 3000;
const registryApp = express();
registryApp.use(bodyParser.json());

// 存储结构: { "AgentName": { url: "http://...", capabilities: ["coding"] } }
const services = {}; 

registryApp.post("/register", (req, res) => {
    const { name, url, capabilities } = req.body;
    services[name] = { url, capabilities };
    console.log(`[Registry] 🟢 服务上线: ${name} [${capabilities.join(", ")}]`.green);
    res.json({ status: "ok" });
});

registryApp.post("/discover", (req, res) => {
    const { capability } = req.body;
    // 简单的查找逻辑：寻找具备该能力的 Agent
    const foundName = Object.keys(services).find(key => 
        services[key].capabilities.includes(capability)
    );
    
    if (foundName) {
        const service = services[foundName];
        console.log(`[Registry] 🔍 发现请求: 寻找 "${capability}" -> 指向 ${service.url}`.cyan);
        res.json({ url: service.url });
    } else {
        console.log(`[Registry] ❌ 未找到具备 "${capability}" 的服务`.red);
        res.status(404).json({ error: "Service not found" });
    }
});

registryApp.listen(REGISTRY_PORT, () => {
    console.log(`[Registry] 注册中心启动于 http://localhost:${REGISTRY_PORT}`.bgBlue);
});

3. 通用 Agent 运行时 (Agent Runtime)

我们将 Agent 的通用行为（启动 HTTP 服务、处理 JSON-RPC、向注册中心汇报）抽象为一个基类。这让具体的业务 Agent 只需关注 Prompt 和逻辑。

class AgentRuntime {
    constructor(name, port, rolePrompt, capabilities) {
        this.name = name;
        this.port = port;
        this.capabilities = capabilities;
        
        this.app = express();
        this.app.use(bodyParser.json());
        // 使用轻量模型以加快演示速度，生产环境建议使用 GPT-4
        this.model = new ChatOpenAI({ modelName: "gpt-4o-mini", temperature: 0.7 });

        // 绑定 JSON-RPC 入口
        this.app.post("/json-rpc", (req, res) => this.handleRpc(req, res));
    }

    async start() {
        return new Promise((resolve) => {
            this.app.listen(this.port, async () => {
                console.log(`[${this.name}] 🚀 启动于 Port ${this.port}`.yellow);
                await this.registerSelf();
                resolve();
            });
        });
    }

    async registerSelf() {
        try {
            await axios.post(`http://localhost:${REGISTRY_PORT}/register`, {
                name: this.name,
                url: `http://localhost:${this.port}/json-rpc`,
                capabilities: this.capabilities
            });
        } catch (e) {
            console.error(`[${this.name}] 注册失败: registry offline?`);
        }
    }

    // --- 核心：处理 JSON-RPC 请求 ---
    async handleRpc(req, res) {
        const { jsonrpc, method, params, id } = req.body;
        // 1. 协议校验
        if (jsonrpc !== "2.0") {
            return res.status(400).json(JsonRpcProtocol.error(-32600, "Invalid Request", id));
        }

        console.log(`[${this.name}] 📩 收到任务: ${method}`.white);

        try {
            // 2. 方法路由：将 RPC method 映射为 Class method
            if (this[method] && typeof this[method] === 'function') {
                const result = await this[method](params);
                res.json(JsonRpcProtocol.success(result, id));
            } else {
                res.status(404).json(JsonRpcProtocol.error(-32601, "Method not found", id));
            }
        } catch (error) {
            console.error(error);
            res.status(500).json(JsonRpcProtocol.error(-32000, error.message, id));
        }
    }

    // --- 核心：A2A 调用能力 (Client 模式) ---
    async callOtherAgent(capability, method, params) {
        console.log(`[${this.name}] 📞 寻找具备 "${capability}" 能力的队友...`.gray);
        
        // 1. 发现 (Discovery)
        const discoverRes = await axios.post(`http://localhost:${REGISTRY_PORT}/discover`, { capability });
        const targetUrl = discoverRes.data.url;

        // 2. 通信 (Communication)
        console.log(`[${this.name}] 📤 发送任务到 ${targetUrl}: ${method}`.gray);
        const rpcPayload = JsonRpcProtocol.request(method, params);
        const response = await axios.post(targetUrl, rpcPayload);
        
        if (response.data.error) throw new Error(response.data.error.message);
        return response.data.result;
    }
}

4. 业务 Agent 实现

有了基类，定义具体的 Agent 就像写普通的业务类一样简单。LangChain 在这里负责处理非结构化的自然语言逻辑。

// --- 产品经理 (PM) ---
class PMAgent extends AgentRuntime {
    constructor() {
        super("PM_Agent", 3001, "你是产品经理", ["requirements_analysis"]);
    }

    // RPC Method: 入口方法
    async start_project({ user_input }) {
        console.log(`[PM] 正在分析用户需求: ${user_input}`);
        
        // 1. 利用 LLM 生成 PRD
        const chain = ChatPromptTemplate.fromTemplate(
            "将此需求转化为简短的技术需求文档(PRD)，包含功能点和技术栈建议: {input}"
        ).pipe(this.model).pipe(new StringOutputParser());
        const prd = await chain.invoke({ input: user_input });
        console.log(`[PM] ✅ PRD 生成完毕`.green);

        // 2. [A2A] 呼叫 Dev
        // PM 不知道 Dev 的 IP，只知道需要 "coding" 能力
        const codeResult = await this.callOtherAgent("coding", "write_code", { requirement: prd });

        return {
            status: "finished",
            prd_summary: prd.slice(0, 50) + "...",
            final_output: codeResult
        };
    }
}

// --- 程序员 (Dev) ---
class DevAgent extends AgentRuntime {
    constructor() {
        super("Dev_Agent", 3002, "你是全栈工程师", ["coding"]);
    }

    async write_code({ requirement }) {
        console.log(`[Dev] 收到 PRD，开始写代码...`.blue);
        
        const chain = ChatPromptTemplate.fromTemplate(
            "根据需求写一段 JavaScript 代码。只返回代码块，不要Markdown: {req}"
        ).pipe(this.model).pipe(new StringOutputParser());
        const code = await chain.invoke({ req: requirement });
        console.log(`[Dev] ✅ 代码编写完毕`.blue);

        // 3. [A2A] 呼叫 QA
        const testReport = await this.callOtherAgent("testing", "review_code", { code, requirement });

        return { code, qa_report: testReport };
    }
}

// --- 测试 (QA) ---
class QAAgent extends AgentRuntime {
    constructor() {
        super("QA_Agent", 3003, "你是测试工程师", ["testing"]);
    }

    async review_code({ code, requirement }) {
        console.log(`[QA] 收到代码，运行静态分析...`.magenta);
        
        const chain = ChatPromptTemplate.fromTemplate(
            "需求: {req}\n代码: {code}\n请给出简短评测结论(Pass/Fail)和改进建议:"
        ).pipe(this.model).pipe(new StringOutputParser());
        const report = await chain.invoke({ req: requirement, code });
        
        console.log(`[QA] ✅ 测试完毕`.magenta);
        return report;
    }
}

5. 启动与测试

最后，我们启动所有微服务，并模拟用户发起一个请求。

async function main() {
    // 启动所有微服务
    const pm = new PMAgent();
    const dev = new DevAgent();
    const qa = new QAAgent();

    await Promise.all([pm.start(), dev.start(), qa.start()]);

    // 等待注册完成
    await new Promise(r => setTimeout(r, 1000));

    console.log("\n==================================================");
    console.log("   A2A 分布式网络已建立 (HTTP/JSON-RPC)");
    console.log("==================================================\n");

    // 模拟外部用户调用
    const userPrompt = "帮我写一个简单的 Node.js HTTP Server，返回 Hello World";
    
    try {
        console.log(`[User] 🌐 提交需求给 PM...`.white);
        // 用户直接调用 PM 的 RPC 接口
        const response = await axios.post("http://localhost:3001/json-rpc", 
            JsonRpcProtocol.request("start_project", { user_input: userPrompt })
        );

        console.log("\n============= 最终交付成果 =============");
        console.log(JSON.stringify(response.data.result, null, 2).green);
    } catch (e) {
        console.error("调用失败:", e.message);
    }
}

main();

运行结果解析

当你运行这段代码时，你会看到控制台打印出类似微服务日志的流向：

1. Registry 启动，随后 PM, Dev, QA 分别启动并向 Registry 注册自己的能力。
2. User 向 PM 发送 HTTP POST 请求。
3. PM 生成 PRD 后，向 Registry 询问“谁懂 coding？”。Registry 返回 Dev 的 URL。
4. PM 直接向 Dev 发起 JSON-RPC 调用。
5. Dev 生成代码后，向 Registry 询问“谁懂 testing？”。Registry 返回 QA 的 URL。
6. Dev 将代码交给 QA。
7. QA 完成测试，结果一路回传。

总结

通过这个案例，我们展示了 A2A 协议的核心魅力：

1. 解耦 (Decoupling)：PM 根本不需要知道 QA 的存在，也不需要知道 Dev 的 IP 地址，它只关心“能力”。这意味着你可以随时替换掉 Dev Agent 的实现（比如从 GPT-3.5 换成 Claude 3.5 Sonnet），而无需修改 PM 的代码。
2. 可扩展性 (Scalability)：如果你想增加一个 Security Agent，只需启动它并注册 security_audit 能力，然后在流程中插入调用即可，无需重构现有代码。
3. 异构性 (Heterogeneity)：由于使用了标准的 JSON-RPC over HTTP，你的 Dev Agent 可以是用 Python 写的，QA Agent 可以是用 Rust 写的，只要遵循协议，它们就能无缝协作。