资讯动态

智能体通信协议ANP设计解析:从发现到协作的工程实践

发布时间:2026/10/14 16:02:25 来源:尧图企业网站定制
1. 为什么智能体之间需要一套自己的“语言”过去两年我参与过不少智能体相关的项目从单体的问答机器人到多体协作的任务编排系统都摸过一遍。一个越来越明显的感受是单个智能体做得再强它的能力边界依然受限于自身接入的工具和数据源。真正能解决复杂问题的方案几乎都走向了多智能体协作——让擅长检索的去做检索擅长推理的去做推理擅长调用外部系统的去做执行。但问题也随之而来。当你把三五个甚至十几个智能体拉到一起干活时第一道坎不是模型能力而是它们之间怎么说话。我见过太多团队在这一步上重复造轮子A 团队用自定义的 JSON 字段传递任务B 团队用另一套字段描述能力C 团队干脆把上下文整个塞进 prompt 里硬传。结果就是每接入一个新智能体就要写一层适配代码系统越做越脆改一处崩三处。这就是Agent Network ProtocolANP这类协议要解决的核心问题。它试图为智能体之间的发现、描述、通信、协作定义一套公共的“语法”和“礼仪”。你可以把它理解成智能体世界的 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_factorydatetime.utcnow) type: str Field(..., description消息类型) payload: Any Field(..., description业务负载) extensions: Optional[dict] Field(default_factorydict)再定义能力描述模型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。先写 calculatorfrom 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_idcalculator, endpointhttp://localhost:8001, capabilities[Capability( capabilitymath.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}, timeout5 ) return resp.json()启动两个服务向 orchestrator 发一个请求就能看到它自动发现 calculator 并完成调用。这个 Demo 虽然简单但已经包含了 ANP 的核心要素注册、发现、能力匹配、跨智能体调用。4.5 Demo 跑通后必须补的三件事Demo 能跑不代表能用。我在实际项目里跑通最小链路后通常会立刻补三件事否则上线必出问题。第一件是超时与重试。上面代码里timeout5是硬编码的真实场景要根据能力类型动态设置。查询类可以短一点执行类要长一点。重试要配合幂等否则会重复执行。第二件是错误传播。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未来可能出现智能体能力的通用描述规范。二是安全机制会从可选变成必选随着智能体接入真实业务系统鉴权和审计会成为准入门槛。三是协议会向轻量化演进太重的东西落地成本高开发者会用脚投票。对开发者来说现在不必急着押注某一个具体协议但理解协议背后的设计思想是值得的。发现、描述、通信、协作、安全这几个维度无论未来标准怎么变都是绕不开的核心问题。把这些问题想清楚换任何协议都能快速上手。我在实际项目里的体会是与其追求一套“完美协议”不如先把最小可用的通信链路跑通然后在真实场景里逐步补齐能力描述、鉴权、追踪这些模块。协议是长出来的不是设计出来的。先跑起来再谈优雅。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑