1. 为什么智能体之间需要一套自己的“语言”
过去两年,我参与过不少智能体相关的项目,从单体的问答机器人到多体协作的任务编排系统都摸过一遍。一个越来越明显的感受是:单个智能体做得再强,它的能力边界依然受限于自身接入的工具和数据源。真正能解决复杂问题的方案,几乎都走向了多智能体协作——让擅长检索的去做检索,擅长推理的去做推理,擅长调用外部系统的去做执行。
但问题也随之而来。当你把三五个甚至十几个智能体拉到一起干活时,第一道坎不是模型能力,而是它们之间怎么说话。我见过太多团队在这一步上重复造轮子:A 团队用自定义的 JSON 字段传递任务,B 团队用另一套字段描述能力,C 团队干脆把上下文整个塞进 prompt 里硬传。结果就是每接入一个新智能体,就要写一层适配代码,系统越做越脆,改一处崩三处。
这就是Agent Network Protocol(ANP)这类协议要解决的核心问题。它试图为智能体之间的发现、描述、通信、协作定义一套公共的“语法”和“礼仪”。你可以把它理解成智能体世界的 HTTP + DNS + 服务注册中心的合体:既要知道对方在哪、能干什么,也要知道怎么把请求发过去、怎么把结果拿回来。
这篇内容适合三类人看:一是正在做多智能体系统、被互操作问题折磨的开发者;二是想理解智能体架构演进方向的技术负责人;三是准备面试、需要把“智能体通信协议”这块讲清楚的同学。我会尽量把协议背后的设计动机、关键机制、落地时的坑都摊开讲,而不是停留在概念层面。
提示:本文讨论的是智能体通信协议的通用设计思路与工程实践,不涉及任何特定厂商的私有实现细节,重点在于帮你建立一套可迁移的认知框架。
2. 拆解 ANP 要解决的四个核心问题
在动手设计或选型之前,先把问题定义清楚。我复盘过多个多智能体项目,发现通信层面的痛点基本收敛到四个方向。ANP 这类协议的每一块设计,几乎都能对应到其中某一个。
2.1 身份与发现问题:怎么找到“对的那个智能体”
单体系统里,你调用一个函数是确定的,函数名写死就行。但多智能体系统里,智能体可能是动态上下线的,能力也可能随时扩展。今天需要一个“能查天气的智能体”,明天可能需要“能查天气且支持未来七天预报的智能体”。如果靠硬编码地址,系统根本没法扩展。
所以协议的第一件事是身份标识。每个智能体需要一个全局唯一、可解析的标识,类似域名之于网站。这个标识不能是易变的 IP 或端口,而应该是一个稳定的逻辑名称,背后再通过某种注册或发现机制映射到实际端点。
第二件事是能力描述。光有名字不够,还得让别的智能体知道“我能做什么、需要什么输入、会返回什么输出”。这就引出了能力声明的问题——是用自由文本描述,还是用结构化 schema?我的经验是,自由文本适合人类阅读,结构化 schema 才适合机器匹配。一个成熟的协议通常会两者兼备:一段自然语言简介用于快速理解,一份机器可解析的能力清单用于自动路由。
2.2 消息格式与语义:怎么保证“说的是一回事”
找到了智能体,接下来是发消息。这里最容易踩的坑是语义歧义。举个真实例子:A 智能体发了一个字段叫time,本意是“任务创建时间”;B 智能体收到后理解成“任务执行时间”,结果调度逻辑全乱了。字段名一样,语义完全不同。
ANP 这类协议通常会在消息层做几件事:定义统一的消息信封(envelope),把“谁发的、发给谁、什么时候发的、这条消息属于哪个会话”这些元信息和业务负载分离;对业务负载则推荐使用带类型约束的结构化格式,而不是裸 JSON。元信息与负载分离这个设计非常关键,它让协议层可以独立处理路由、重试、审计,而不用关心业务内容。
另一个常被忽视的点是版本兼容。协议一旦发布就会被各种实现采用,字段增删在所难免。好的协议会预留扩展字段,并规定“未知字段应被忽略而非报错”,这样新旧实现才能共存。
2.3 交互模式:一次请求够不够用
最简单的交互是请求-响应:我问一句,你答一句。但智能体协作远不止于此。我遇到过这些场景:
- 一个任务需要多个智能体接力完成,前一个的输出是后一个的输入,形成任务链。
- 一个智能体需要同时向多个智能体广播请求,然后汇总结果,这是并行协作。
- 长任务需要持续推送进度,这是流式交互。
- 两个智能体需要多轮协商才能达成一致,这是会话式交互。
如果协议只支持请求-响应,上面这些场景都得在应用层自己拼。所以 ANP 通常会把交互模式抽象出来,至少覆盖同步调用、异步任务、流式推送、会话保持这几种。选型时要特别关注协议对异步和流式的支持程度,因为这直接决定了你的系统能不能处理长耗时任务。
2.4 安全与信任:怎么防止“被冒充”和“被滥用”
智能体之间通信,安全不是可选项。我见过一个内部系统,因为没做调用方鉴权,一个测试智能体误调了生产环境的执行智能体,直接触发了一批真实操作。教训很深刻。
协议层面的安全通常包括:身份认证(确认对方真是它声称的那个智能体)、授权(确认它有权调用这个能力)、消息完整性(防止传输中被篡改)、审计留痕(出问题能追溯)。这几项里,认证和授权是基础,审计是兜底。很多团队前期为了快速跑通会跳过鉴权,后期补起来代价极大,因为要改所有调用方。我的建议是哪怕内部系统,也至少加上简单的令牌校验。
3. 一个可落地的 ANP 分层设计参考
理解了要解决的问题,接下来看协议本身怎么分层。我参考过几种不同的智能体通信方案,发现结构上大同小异,基本可以归纳成四层。这个分层不是标准答案,但它是我在实际项目里验证过、比较清晰的一种组织方式。
3.1 传输层:先解决“消息能不能到”
传输层只关心一件事:把字节从 A 送到 B。至于字节里是什么,它不管。这一层通常直接复用成熟的网络传输能力,比如基于 HTTP 的短连接、基于长连接的推送通道、或者消息队列。
这里有个选型上的实际考量。如果你的智能体交互以短任务为主,HTTP 请求-响应足够用,实现简单、调试方便。但如果涉及长任务进度推送、双向流式对话,就需要长连接或流式通道。我在一个实时协作项目里最初用轮询,延迟高、资源浪费大,后来换成流式推送,体验提升非常明显。
传输层还要考虑可靠性。网络抖动、对端重启都是常态。协议需要定义重试策略、超时时间、幂等性保证。幂等性尤其重要:如果一条“创建任务”的消息因为超时被重发,接收方不能创建两个任务。通常的做法是给每条消息带一个唯一 ID,接收方对相同 ID 的消息去重。
3.2 消息层:定义“消息长什么样”
消息层规定消息的结构。一个典型的消息信封包含这些字段:
| 字段 | 作用 | 是否必填 |
|---|---|---|
| message_id | 消息唯一标识,用于去重和追踪 | 是 |
| sender | 发送方智能体标识 | 是 |
| receiver | 接收方智能体标识 | 是 |
| session_id | 会话标识,关联同一任务的多条消息 | 视场景 |
| timestamp | 发送时间,用于排序和超时判断 | 是 |
| type | 消息类型,如 request/response/event | 是 |
| payload | 业务负载 | 是 |
| extensions | 扩展字段,供未来使用 | 否 |
这个结构看起来简单,但每个字段都有讲究。比如session_id,它的作用是把一次协作过程中的多条消息串起来。没有它,日志就是一盘散沙,排查问题时会非常痛苦。再比如type,它让接收方能快速判断该走哪条处理逻辑,而不用去猜 payload 的内容。
payload 的格式我强烈建议用带 schema 约束的结构化数据,而不是自由文本。自由文本虽然灵活,但机器解析成本高、容易出错。如果确实需要传自然语言,也应该把它放在一个明确的字段里,而不是整个 payload 都是文本。
3.3 能力层:描述“我能做什么”
能力层是 ANP 区别于普通消息协议的关键。它让智能体能够对外声明自己的能力,并让其他智能体自动发现和匹配。
一个能力描述通常包含:能力名称、功能简介、输入参数 schema、输出结果 schema、调用约束(如频率限制、权限要求)。我用过一个简化版的能力描述,大致长这样:
{ "capability": "weather.query", "description": "查询指定城市指定日期的天气情况", "input_schema": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名称"}, "date": {"type": "string", "format": "date", "description": "查询日期"} }, "required": ["city"] }, "output_schema": { "type": "object", "properties": { "temperature": {"type": "number"}, "condition": {"type": "string"} } } }有了这份描述,调度方就可以根据任务需求自动匹配合适的智能体,而不用人工配置路由规则。这是从“硬编码编排”走向“动态编排”的关键一步。
3.4 协作层:编排“多个智能体怎么配合”
协作层处理的是多智能体之间的配合逻辑。它定义了几种常见的协作模式,并规定了每种模式下消息怎么流转。
以任务链为例:智能体 A 完成任务后,需要把结果交给智能体 B。协议可以规定,A 在响应里带上“下一步建议接收方”,或者由一个独立的编排智能体来负责转发。两种方式各有优劣:前者去中心化、灵活,但链路长了容易失控;后者中心化、可控,但编排智能体可能成为瓶颈。实际项目里我倾向于混合模式:简单链路让智能体自己传递,复杂流程交给编排层统一管理。
协作层还要处理失败与补偿。任务链中某一环失败了怎么办?是整体回滚,还是跳过继续?协议需要提供明确的错误传播机制和补偿指令,否则系统在异常情况下会陷入不一致状态。
4. 从零搭建一个最小可用的智能体通信 Demo
光讲设计容易飘,我们动手搭一个最小可用的版本。目标很简单:两个智能体,一个负责接收任务并调用另一个,跑通“发现-调用-返回”的完整链路。我用 Python 写,因为生态成熟、上手快。
4.1 环境准备与依赖选择
先明确技术选型。传输层我用 HTTP,因为调试直观、依赖少。消息格式用 JSON。能力描述用 JSON Schema。整个 Demo 不引入重型框架,目的是让你看清协议层到底在做什么。
需要的基础依赖:
pip install fastapi uvicorn requests pydantic选 FastAPI 是因为它自带请求校验和文档生成,能省掉不少样板代码。Pydantic 用来做消息和能力描述的 schema 校验。这里有个经验:协议实现里一定要有强校验,不能靠“约定”。我见过太多因为字段类型不对导致下游崩溃的案例,加一层校验能挡掉大部分低级错误。
4.2 定义消息信封与能力描述
先写消息信封的模型。用 Pydantic 定义,好处是自动校验、自动序列化。
from pydantic import BaseModel, Field from typing import Any, Optional from datetime import datetime class MessageEnvelope(BaseModel): message_id: str = Field(..., description="消息唯一标识") sender: str = Field(..., description="发送方智能体标识") receiver: str = Field(..., description="接收方智能体标识") session_id: Optional[str] = Field(None, description="会话标识") timestamp: datetime = Field(default_factory=datetime.utcnow) type: str = Field(..., description="消息类型") payload: Any = Field(..., description="业务负载") extensions: Optional[dict] = Field(default_factory=dict)再定义能力描述模型:
class Capability(BaseModel): capability: str description: str input_schema: dict output_schema: dict这两个模型是整个协议的地基。注意extensions字段我给了默认空字典,这样旧实现遇到新字段不会报错,符合前面说的版本兼容原则。
4.3 实现智能体注册与发现
真实系统里,注册和发现通常由一个注册中心承担。Demo 里我们简化成一个内存字典,但接口设计要保留,方便以后替换成真正的注册中心。
class Registry: def __init__(self): self.agents = {} def register(self, agent_id: str, endpoint: str, capabilities: list): self.agents[agent_id] = { "endpoint": endpoint, "capabilities": capabilities } def discover(self, capability_name: str): results = [] for agent_id, info in self.agents.items(): for cap in info["capabilities"]: if cap.capability == capability_name: results.append({ "agent_id": agent_id, "endpoint": info["endpoint"], "capability": cap }) return resultsdiscover方法按能力名称匹配,返回所有能提供该能力的智能体。实际生产中这里会复杂得多,可能要按负载、按版本、按权限过滤,但核心逻辑就是这个匹配过程。
4.4 跑通一次跨智能体调用
现在写两个智能体。一个叫calculator,提供加法能力;一个叫orchestrator,接收用户请求后去发现并调用calculator。
先写 calculator:
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class AddRequest(BaseModel): a: float b: float @app.post("/invoke/add") def add(req: AddRequest): return { "result": req.a + req.b, "agent": "calculator" }再写 orchestrator,它启动时向注册中心注册自己,收到请求后去发现 calculator 并调用:
import requests from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() registry = Registry() # 注册 calculator registry.register( agent_id="calculator", endpoint="http://localhost:8001", capabilities=[Capability( capability="math.add", description="两数相加", input_schema={"type": "object", "properties": {"a": {"type": "number"}, "b": {"type": "number"}}}, output_schema={"type": "object", "properties": {"result": {"type": "number"}}} )] ) class TaskRequest(BaseModel): a: float b: float @app.post("/task") def handle_task(req: TaskRequest): candidates = registry.discover("math.add") if not candidates: return {"error": "no agent found"} target = candidates[0] resp = requests.post( f"{target['endpoint']}/invoke/add", json={"a": req.a, "b": req.b}, timeout=5 ) return resp.json()启动两个服务,向 orchestrator 发一个请求,就能看到它自动发现 calculator 并完成调用。这个 Demo 虽然简单,但已经包含了 ANP 的核心要素:注册、发现、能力匹配、跨智能体调用。
4.5 Demo 跑通后必须补的三件事
Demo 能跑不代表能用。我在实际项目里跑通最小链路后,通常会立刻补三件事,否则上线必出问题。
第一件是超时与重试。上面代码里timeout=5是硬编码的,真实场景要根据能力类型动态设置。查询类可以短一点,执行类要长一点。重试要配合幂等,否则会重复执行。
第二件是错误传播。calculator 挂了,orchestrator 要能感知并返回有意义的错误,而不是抛一个 500 让调用方一脸懵。协议层应该定义标准的错误码和错误结构。
第三件是日志与追踪。每条消息的message_id和session_id都要打进日志,这样一次跨智能体调用才能被完整还原。没有追踪能力的多智能体系统,排查问题基本靠猜。
5. 协议落地时最容易翻车的几个地方
设计和 Demo 都讲完了,接下来是我最想分享的部分——踩坑经验。这些东西文档里通常不会写,但每一个都能让你少熬几个通宵。
5.1 能力描述写得含糊,导致路由错乱
我遇到过一个典型案例:两个智能体都声明了“文本处理”能力,但一个擅长摘要,一个擅长翻译。调度方按能力名匹配,随机选了一个,结果把翻译任务发给了摘要智能体,输出完全不对。
根因是能力描述粒度太粗。解决办法是把能力拆细,用更精确的命名,比如text.summarize和text.translate,而不是笼统的text.process。同时,能力描述里的description要写清楚适用场景和限制条件,方便调度方做二次判断。
注意:能力命名建议采用“领域.动作”的两段式结构,既保证唯一性,又便于按前缀批量匹配。
5.2 消息体过大拖垮传输
有一次系统突然变慢,排查发现是某个智能体把整个知识库的检索结果塞进 payload 传给下游,单条消息几兆。传输层扛不住,超时频发。
这个坑的本质是没有约定消息体大小上限。协议应该规定单条消息的推荐大小和硬上限,超过上限的内容应该走引用机制——传一个资源 ID,让接收方按需拉取,而不是直接塞进消息里。大消息不仅慢,还会撑爆日志和审计系统,后患无穷。
5.3 会话状态管理失控
多轮协作场景下,会话状态放哪里是个大问题。放发送方,接收方每次都要等上下文;放接收方,发送方重启后状态就丢了;放第三方存储,又引入额外依赖和延迟。
我的经验是按会话生命周期分层处理:短会话(几轮内结束)状态放内存,随会话结束释放;长会话状态持久化,但要设置过期时间,避免无限增长。关键是协议要明确会话的创建、续期、销毁规则,不能让实现方各自为政。
5.4 鉴权缺失导致越权调用
前面提过这个坑,这里展开说。一个内部系统,所有智能体共用一个令牌,结果任何一个智能体被攻破,整个网络就沦陷了。正确做法是每个智能体有独立身份,调用时携带自己的凭证,接收方按能力粒度做授权。比如摘要智能体只能调用文本读取能力,不能调用数据删除能力。
授权模型建议用“主体-能力-资源”三元组来描述,清晰且易于审计。别嫌麻烦,安全这块省下的时间,出事时要加倍还回去。
5.5 版本升级引发连锁故障
协议字段一改,所有实现方都要跟着改,这是分布式系统的经典难题。我经历过一次字段重命名,因为没做兼容,导致一半智能体通信失败。
应对策略有三条:新增字段而非修改字段;未知字段一律忽略;重大变更通过版本号隔离,新旧版本并行运行一段时间。协议设计要有“向后兼容”的肌肉记忆,每次改动前先问一句:老实现收到这条新消息会怎样?
6. 智能体通信协议与常见技术概念的边界
聊到这里,有必要厘清几个容易混淆的概念。我在面试和评审中经常遇到把这些混为一谈的情况,理清楚对理解 ANP 的定位很有帮助。
6.1 和传统 RPC、消息队列的区别
有人会问:我用 gRPC 或消息队列不也能让智能体通信吗?能,但不够。传统 RPC 解决的是“调用一个已知服务”,它假设服务地址和接口是预先确定的。而智能体通信的核心诉求是动态发现和语义匹配——我不知道谁会处理这个任务,需要协议帮我找。消息队列解决的是解耦和削峰,它不关心消息语义,也不提供能力描述。ANP 是在这些传输能力之上,加了一层“智能体语义层”。
6.2 和模型上下文协议的关系
模型上下文协议主要解决的是智能体与工具、数据源之间的连接问题,偏“纵向”集成。ANP 解决的是智能体与智能体之间的协作问题,偏“横向”互联。两者不是替代关系,而是互补。一个完整的系统可能同时需要:用上下文协议接工具,用 ANP 接其他智能体。
6.3 和编排框架的边界
LangGraph、AutoGen 这类编排框架,关注的是“多个智能体怎么组织成一个工作流”。它们内部也会定义消息传递机制,但通常是框架私有的。ANP 的价值在于跨框架互操作——用 A 框架写的智能体,能和用 B 框架写的智能体通信。如果你的系统永远只用一种框架,可能感受不到 ANP 的必要性;但一旦涉及异构集成,它就是刚需。
7. 我对智能体通信协议演进的一点判断
最后聊点个人看法,不算总结,就是这几年做下来的一些直觉。
智能体通信协议目前还处在早期,标准未定,各家实现差异很大。但我判断有几个方向是确定的。一是能力描述会越来越标准化,就像 OpenAPI 之于 REST,未来可能出现智能体能力的通用描述规范。二是安全机制会从可选变成必选,随着智能体接入真实业务系统,鉴权和审计会成为准入门槛。三是协议会向轻量化演进,太重的东西落地成本高,开发者会用脚投票。
对开发者来说,现在不必急着押注某一个具体协议,但理解协议背后的设计思想是值得的。发现、描述、通信、协作、安全这几个维度,无论未来标准怎么变,都是绕不开的核心问题。把这些问题想清楚,换任何协议都能快速上手。
我在实际项目里的体会是,与其追求一套“完美协议”,不如先把最小可用的通信链路跑通,然后在真实场景里逐步补齐能力描述、鉴权、追踪这些模块。协议是长出来的,不是设计出来的。先跑起来,再谈优雅。