简介围绕多智能体系统协作研究这份 PDF 文献聚焦 Internet of AgentsIoA框架的设计与实验验证面向关注异构智能体集成、分布式计算与自动化团队形成的研究者和开发者。它针对现有框架生态隔离、单设备仿真、通信协调僵化等问题提出灵活的多智能体集成协议、类即时消息架构以及动态团队组建与对话流控制机制并在通用助理、具身智能和检索增强生成等任务上验证协作效果。压缩包仅含 1 个 PDF 文件大小约 4.21MB便于集中阅读与归档。已有 249 人学习下载。读者可获取 IoA 的体系结构、关键机制与实验对比结论理解动态任务分配和异构智能体协作的实现思路也可结合论文公开的代码线索继续复现与拓展研究其中对第三方智能体集成和分布式协作场景的讨论适合作为方案设计与论文研读参考。1. 从单机编排到跨设备组网IoA 想解决的三个真问题同一个开放域任务AutoGPT 和 Open Interpreter 各自单跑与把它们接进同一个群聊协作后再跑后者的胜率是 66% 到 76%。这里没换更强的模型变的只是协作结构。IoAInternet of Agents由清华、北大、北邮和腾讯合作提出抓的就是这件反直觉的事多智能体系统的瓶颈常常不在单个 agent 的智力而在它们之间那根线怎么接。现有 agent 框架的毛病集中在三处。生态隔离agent 只能在自家生态里定义第三方能力接不进来单设备仿真几乎所有框架都在一台机器上假装分布式跟真实的多机多端差得远通信管道硬编码谁跟谁说话、何时从讨论切到执行全写在代码里任务一变就得改代码。IoA 的应对是三条一套 agent 集成协议、一套类即时消息的架构、一套动态的组队与对话流控制机制。适合人群也明确手里已经有 ReAct、AutoGPT 这类跑得起来的 agent、想试着把它们组网协作的开发者以及需要在多台机器上做分布式智能体实验的研究者。2. IoA 的服务端与客户端分层以及 Agent 注册发现协议理解 IoA 的第一步不是读它的实验数据而是看清 server 和 client 各自被切成了三层。这个切法决定了后面所有加一个 agent改一次路由的成本。2.1 交互层、数据层、基础层各自负责什么服务端是中心枢纽承担 agent 注册、发现和消息路由客户端是个体 agent 的 wrapper负责把第三方 agent 接进这套协议。两侧结构对称都是三层。层服务端区块客户端区块职责Interaction LayerAgent Query / Group Setup / Message RoutingTeam Formation / Communication找 agent、建群、消息扇出组队决策与消息处理Data LayerAgent Registry / Session ManagementAgent Contact / Group Info / Task Management注册表、活跃连接联系人、群信息、任务状态Foundation LayerData Infra / Network Infra / SecurityAgent Integration / Data Infra / Network Infra持久化、网络通信、鉴权第三方 agent 适配接口选中心化 hub 而不是全互联 P2P是为了让发现成本可控任何 agent 只要问一次服务端就能拿到候选名单不用维护一张全网路由表。代价是 server 成为单点所以 Foundation Layer 里的 Session Management 和 Security 必须做实否则重启一次服务端就等于全网掉线。分层带来的直接好处是扩展性——新增一种 agent 类型只需要改客户端的 Agent Integration Block消息路由逻辑一行都不用动。2.2 注册报文该写什么能力描述字段设计注册是发现的前提。一个 agent 加入 IoA 时它的 wrapper 要向服务端提交一份能力自述这份自述后面会被别的 agent 用自然语言检索到。{ agent_id: react_retriever_01, name: RetrieverAgent, endpoint: ws://10.0.12.7:8765/agent, capabilities: [web_search, vector_retrieval, citation], description: 擅长从本地向量库与公开网页检索事实输出带出处的片段, model: gpt-3.5-turbo, max_concurrency: 2, status: idle }capabilities是硬过滤条件查询里指定了就必须命中description是软匹配文本供语义检索用endpoint决定服务端把群消息投到哪个连接max_concurrency是自我保护防止一个 agent 被同时塞进十个群聊。description写得太泛比如我能做很多事情是召回噪声的主要来源经验做法是把领域、输入形态、输出形态三件事写清楚上面这条就属于合格写法领域是检索输入是查询串输出是带出处的片段。2.3 一次完整的注册、发现、建群链路客户端 wrapper 的核心逻辑不复杂但顺序不能乱。import asyncio, json, websockets class IoAClient: def __init__(self, server_url, profile): self.server_url server_url # 例如 ws://127.0.0.1:8765/ws self.profile profile # 2.2 里的注册报文 self.ws None async def connect(self): self.ws await websockets.connect(self.server_url) await self.ws.send(json.dumps({type: register, **self.profile})) ack json.loads(await self.ws.recv()) # 必须等 ack此时服务端才把 agent_id 与这条连接绑定 assert ack[type] register_ack, ack print(registered as, ack[agent_id]) async def query_agents(self, text, top_k3): await self.ws.send(json.dumps({ type: agent_query, query: text, # 自然语言描述需要什么能力 top_k: top_k # 返回候选数量 })) return json.loads(await self.ws.recv())[agents] async def heartbeat(self, interval20): while True: await asyncio.sleep(interval) await self.ws.send(json.dumps({ type: ping, agent_id: self.profile[agent_id] })) async def listen(self): async for raw in self.ws: msg json.loads(raw) if msg[type] group_message: print(msg[group_id], msg[sender], msg[content])register之后如果不先等register_ack就发别的消息服务端的 Session Management 还没完成绑定消息会被直接丢弃——这是接入时最容易撞的一堵墙。心跳必须放在独立协程里interval要小于服务端的session_timeout一般取它的三分之一。agent_query返回的是候选列表而不是直接建群建群这一步交给客户端的 Team Formation 决定这是 IoA 和硬编码 pipeline 的分水岭。top_k调大能让决策上下文更全但每个候选的 description 都要进 prompttoken 成本线性上涨实验阶段 3 到 5 比较划算。2.4 注册与发现的四个常见坑agent_id重名。多台机器用同一份默认配置启动注册表里后注册的把先注册的覆盖掉表现为队友时有时无。常见做法是把主机名或容器序号拼进 id。能力声明与实现不符。声明了code_execution却没有沙箱任务派过去直接失败而失败信息只回到群里排查要翻半天。注册报文里的capabilities应当由代码里的能力注册表自动生成而不是手写。心跳与长任务冲突。一个 agent 跑五分钟的检索任务时阻塞了事件循环心跳发不出去被判离线群里的任务随之超时。把执行体丢进线程池或子进程是标准解法。重连后没有重新注册。断线重连只是换了一条 WebSocket 连接Agent Registry 里记的还是旧 endpoint消息投过去石沉大海。客户端的重连回调里必须重发一次register。3. 自动化团队形成与对话流控制的有限状态机组队和对话流是 IoA 里最不像框架、最像机制的部分。它借的是即时消息的心智模型先查通讯录再拉群群内再分工。3.1 组队为什么不能写死硬编码的 pipeline 隐含一个假设——任务形态固定参与者固定轮次固定。真实任务里主 agent 得先看有谁能干活再决定拉谁进群进了群还得判断此刻是在讨论还是在执行。把这三件事写进代码任务一变就得改代码交给 agent 自己决定则需要一个受约束的决策空间否则群聊会发散。IoA 的做法是给决策空间加上状态边界。3.2 从 Agent Query 到 Group Setup 的执行链一次典型的组队过程分五步主 agent 把任务拆成能力需求描述例如需要一个能跑 Python 并读文件的执行者。发agent_query拿到候选列表。按capabilities做硬过滤再按description做一轮语义筛选通常留下 2 到 4 个。发group_setup服务端分配group_idMessage Routing 开始按群扇出。群内进入会话状态机由状态决定当前该说什么话。第 3 步值得单独说。候选太多会拖长决策上下文太少又容易漏掉合适的人。我一般会把硬过滤放在前面因为capabilities是确定性的而description的语义匹配有噪声先砍规模再精排的顺序更稳。3.3 会话状态的抽象与转移条件这一层借鉴了言语行为理论说话本身就是行动。群里的每条消息都带一个意图标签状态根据意图和上下文推进。状态允许的发言意图进入下一状态的条件建议轮次上限DISCUSSIONquestion / info有成员提出可执行方案6PROPOSALproposal / agree / object同意比例达标或主 agent 拍板3ASSIGNassign / accept所有子任务都有 owner2EXECUTIONprogress / result全部子任务回报完成视任务REVIEWverdict / reject校验通过或打回次数用尽2CLOSED—终态—from enum import Enum class State(str, Enum): DISCUSSION discussion PROPOSAL proposal ASSIGN assign EXECUTION execution REVIEW review CLOSED closed # 每个状态允许的意图越界的消息不参与转移判断只记录 ALLOWED { State.DISCUSSION: {question, info}, State.PROPOSAL: {proposal, agree, object}, State.ASSIGN: {assign, accept}, State.EXECUTION: {progress, result}, State.REVIEW: {verdict, reject}, } def next_state(state, intent, ctx, cfg): if intent not in ALLOWED[state]: return state, intent_not_allowed if state is State.DISCUSSION and intent proposal: return State.PROPOSAL, proposal_raised if state is State.PROPOSAL and ctx[agree_ratio] cfg[agree_ratio]: return State.ASSIGN, consensus if state is State.ASSIGN and ctx[unassigned] 0: return State.EXECUTION, all_assigned if state is State.EXECUTION and ctx[pending] 0: return State.REVIEW, all_reported if state is State.REVIEW and ctx[rejected] cfg[max_reject]: return State.CLOSED, accepted if ctx[turns_in_state] cfg[turn_limit][state]: return State.CLOSED, timeout # 兜底别让群聊无限循环 return state, continueintent由成员 agent 发消息时标注或用一个轻量分类器打标成本很低。turns_in_state是防死循环的关键闸门灵活不等于无界没有这个计数器两个 agent 互相 object 能把 token 烧光。3.4 状态机参数怎么调agree_ratio调太低一个 agent 提方案就全员执行质量塌得很快调太高PROPOSAL 阶段频繁卡死。经验值是成员数不超过 3 时取 0.55 人以上取 0.6 到 0.7因为人多了意见本来就更难统一。max_reject建议设 2第三次仍不通过就带着警示标注返回比无限重试省得多。turn_limit的 DISCUSSION 项最需要调它直接决定组队阶段花掉多少 token一般按单次任务预算的 20% 到 30% 倒推。4. 异构 Agent 接入与多设备分布式部署IoA 相对其它 agent 框架在集成上的关键差异是不要求第三方 agent 用同一套工具抽象只要能被包装成字符串进、字符串出就接得进来。4.1 Wrapper 适配层把第三方 agent 包成 IoA 客户端class ReActWrapper: 把任意 callable 形式的 agent 适配成 IoA 可调用的执行体。 def __init__(self, agent_id, fn, capabilities): self.agent_id agent_id self.fn fn # 签名固定为 fn(task: str) - str self.capabilities capabilities async def handle(self, msg, loop): task msg[content] # 阻塞型 agent 必须丢线程池否则心跳协程被卡住会被判离线 result await loop.run_in_executor(None, self.fn, task) return { type: group_message, group_id: msg[group_id], sender: self.agent_id, intent: result, # 供状态机判断 content: result, reply_to: msg[msg_id], # 便于做幂等与串线排查 }fn的签名约束是整套适配的核心AutoGPT、Open Interpreter、ReAct 循环、甚至一段纯 Python 函数只要满足这个签名就能接进来。run_in_executor不是为了性能是为了让事件循环空出来发心跳。reply_to带上原消息 id重连重发时按msg_id去重就能避免群里出现重复结论。4.2 分布式部署的配置项与端口规划# server.yaml server: host: 0.0.0.0 ws_port: 8765 session_timeout: 60 # 秒超时未心跳则踢掉会话 registry: backend: sqlite # 小规模实验够用多机压测换 postgres persist_path: ./data/registry.db routing: fanout_mode: group # group 按群扇出direct 点对点 security: token_ttl: 3600参数含义单机实验多机部署建议host监听地址127.0.0.10.0.0.0否则跨机连不上ws_portWebSocket 端口8765固定防火墙放行session_timeout心跳超时60不小于心跳间隔的 3 倍backend注册表存储sqlitepostgres避免多进程写冲突fanout_mode路由模式group群成员跨网段时仍用 grouptoken_ttl鉴权有效期3600与实验时长匹配避免中途过期4.3 启动顺序与联通性自检# 1) 起服务端先确认端口在听 python -m ioa.server --config server.yaml ss -lntp | grep 8765 # 2) 起两个能力不同的 agent可分别跑在不同机器上 python -m ioa.client --agent-id retriever_01 --endpoint ws://10.0.12.1:8765/ws python -m ioa.client --agent-id coder_01 --endpoint ws://10.0.12.1:8765/ws # 3) 查注册表确认两个 agent 都可被发现 python -m ioa.cli query --server ws://10.0.12.1:8765/ws \ --text 能写并执行 python 代码的 agent第 3 步最容易被跳过但注册成功不等于能被发现。query返回空通常意味着description与查询文本的语义距离太远或者该 agent 的status不是idle。定位这类问题时先用一段几乎照抄description的文本查一次如果查得到就说明是描述写法的问题而不是链路问题。4.4 故障对照表现象可能原因排查动作连上但注册失败token 过期或 ack 超时看服务端 Security 日志队友列表为空capabilities 过滤过严换纯文本 query 再试一次群消息重复重连后没做去重客户端按 msg_id 幂等agent 频繁被判离线长任务阻塞心跳执行移到线程池或子进程跨机器不通endpoint 填了 127.0.0.1改 bind 地址与注册 endpoint群建了但没人说话group_setup 成员列表为空检查建群前的候选筛选逻辑5. 验证实验设计与 RAG 场景下的两个调优技巧5.1 胜率评估怎么设计才站得住开放域任务上的对比实验结论强度取决于控制变量的干净程度。几个必须做的动作同一批任务分别跑单 agent 和 IoA 组队judge 用固定的模型且对两边匿名任务顺序随机化以抵消上下文缓存带来的偏差每一组至少重复 3 次取均值。IoA 在 GAIA 上只用几个基础 ReAct agent 就超过了此前的工作在 RAG 问答上 GPT-3.5 版本的实现接近甚至超过 GPT-4这两个结果的说服力恰恰来自任务与 judge 的一致性而不是任务数量堆得多。评估脚本里建议把每次运行的group_id、状态转移序列和最终答案一起落盘出问题时才能回放整场对话。5.2 让 GPT-3.5 逼近 GPT-4 的两个协作技巧第一个技巧是拆职责而不是拆步骤。把检索和回答分给两个 agentretriever 只负责返回带出处的片段answer agent 只负责在给定片段上组织答案。同一个模型同时干这两件事时上下文里既有噪声片段又有写作指令注意力会被分散拆开之后两个 agent 的 prompt 都更短更聚焦这是收益最大的改动。第二个技巧是把自检挪出原上下文。在 REVIEW 状态放一个独立的校验 agent只做一件事逐句判断答案是否被片段支撑。def review(answer, passages, judge): prompt ( 只依据给定片段逐句检查答案。 输出 JSON: {\unsupported\: [句子], \verdict\: \pass\|\reject\} ) payload f片段:\n{passages}\n\n答案:\n{answer} out judge(prompt, payload) # judge 用与答题不同的系统提示 return out[verdict] pass关键在judge用的是另一套系统提示、另一次调用而不是让答题的 agent 自己回看。同一上下文里的自检几乎必然通过因为模型倾向于认可自己刚才的输出。max_reject建议设 2第三次仍不通过就直接返回带警示标注的答案比无限重试省 token也更接近真实系统的行为。本文还有配套的精品资源点击获取