1. 项目概述当多智能体学会“选择性倾听”在构建由多个大型语言模型LLM驱动的智能体系统时我们常常遇到一个核心困境智能体之间的对话很容易陷入信息过载或无效沟通的泥潭。想象一下在一个由策划、文案、设计、法务等多个AI角色组成的虚拟团队会议上每个智能体都在基于自己的知识库和当前任务背景发言。策划智能体抛出了一个长达千字的宏大方案其中包含了市场分析、用户画像、执行节奏等方方面面。对于文案智能体而言它真正需要关注的可能只是其中的“核心用户痛点描述”和“品牌调性关键词”而对于设计智能体重点则变成了“视觉风格参考”和“关键信息层级”。如果每个智能体都不得不“听”完所有冗长的上下文再费力地从中提取与自己相关的片段不仅会消耗巨大的计算资源直接体现为高昂的API调用成本和延迟更可能导致关键信息被淹没最终影响协同决策的质量和效率。这正是“通过上下文相关性的注意力引导来增强多智能体通信”这一课题要解决的核心问题。它不是一个简单的技术叠加而是一种通信范式的转变。其目标不是让智能体变得更“聪明”而是让它们变得更“专注”——学会在纷繁复杂的多轮对话中像经验丰富的团队成员一样主动将“注意力”聚焦在与自身角色和当前任务最相关的信息上。这里的“注意力”借鉴了Transformer架构中的核心机制但将其应用层面从模型内部的token关联提升到了智能体间消息传递的宏观调度。而“上下文相关性”则是实现这种引导的标尺它动态地评估历史对话中每一段信息对于接收方智能体的价值。通过这项技术我们期望达到的效果是降低通信冗余、提升决策效率、减少不必要的计算开销最终让多智能体系统在复杂任务上的协作更加流畅、精准且经济。2. 核心设计思路从全局对话到定向投喂传统的多智能体通信架构无论是简单的广播/订阅模式还是基于黑板Blackboard的共享内存模式本质上都是一种“推”的模式。发送方将完整的自身思考或行动结果打包成消息“推”给所有或指定的接收方。接收方则需要自行处理整个消息体。这种模式在智能体数量少、消息结构简单时尚可工作但随着智能体角色分化、任务复杂度提升其弊端暴露无遗。本项目的设计思路可以概括为将“广播”变为“窄播”将“完整推送”变为“摘要投喂”。其核心在于为每个智能体配备一个轻量级的“注意力引导器”。这个引导器不参与智能体自身的核心推理而是工作在通信层负责两件事1. 对发出的消息自动生成面向不同接收方的、定制化的“相关性摘要”2. 对接收到的消息快速评估其与自身任务的相关性并决定是深入处理、简要浏览还是直接忽略。2.1 注意力引导器的双工工作流这个引导器的工作是双工的贯穿通信的发送与接收两端。在发送端Sender Side Filtering 当一个智能体如策划Agent准备发言时其原始的、完整的思考内容我们称之为“原始消息体”会首先经过其自身的注意力引导器。这个引导器内置了所有协作智能体的角色画像Role Profile。角色画像不是简单的名称而是包含了该角色的核心职责、关注的关键信息类型如设计Agent关注“色彩”、“布局”、“风格”法务Agent关注“责任”、“条款”、“合规”等。引导器会基于接收方智能体的角色画像对原始消息体进行实时分析抽取出与之最相关的片段、关键词或总结性陈述生成一个“定向摘要”。这个摘要可能只是几个关键词、一两句话或者一个结构化的数据片段。最终发送出去的不再是冗长的全文而是这个轻量的“定向摘要”加上一个指向原始消息体的索引或链接如果需要深度查看。在接收端Receiver Side Scoring 当智能体如设计Agent收到来自其他智能体的消息可能是摘要也可能是完整消息时其自身的注意力引导器会立即启动。它会基于自己当前的任务状态例如正在设计首页Banner和角色画像计算接收到的消息的“上下文相关性分数”。这个分数综合考虑了语义相关性消息内容与自身职责关键词的匹配度。任务时序相关性消息是否与当前任务阶段强相关例如在需求确认阶段设计Agent应高度关注“需求变更”类消息在评审阶段则应关注“修改意见”。发送方权威性不同发送方的消息权重可能不同例如项目经理Agent的指令可能比另一个并行设计Agent的建议具有更高的默认相关性。基于这个分数引导器会决定消息的处理优先级高相关消息送入智能体的核心推理模块进行深度处理中等相关消息可能仅更新一下上下文记忆低相关消息则被标记或存档几乎不占用主要计算资源。2.2 上下文相关性的动态计算模型“上下文相关性”是本设计的灵魂其计算不能是静态的规则匹配而必须是一个动态、可学习的模型。一个基础但有效的实现可以分为三层表层关键词匹配层使用轻量级的嵌入模型如BGE-M3或text-embedding-3-small将消息内容和智能体的“关注词库”分别转化为向量计算余弦相似度。这一层速度快能快速过滤掉明显不相关的信息。任务状态感知层每个智能体维护一个简单的任务状态机如“等待需求”、“设计中”、“修改中”、“已完成”。系统预定义了一个“状态-信息类型”关联矩阵。例如设计Agent在“设计中”状态时对“新增需求”类消息的相关性权重会降低因为原则上不应中途加需求而对“同类参考案例”类消息的权重会升高。历史交互记忆层记录智能体间的历史交互。如果A智能体过去提供的某类信息频繁被B智能体采纳并产生了正反馈任务推进那么未来A向B发送的同类信息会获得更高的初始相关性分数。这相当于为智能体间建立了简单的“信任度”模型。这三层分数经过一个可配置的权重加权求和最终得出一个综合相关性分数。权重可以在系统初始化时设定也可以设计一个简单的元学习机制进行微调。注意相关性模型的设计切忌复杂化。初期完全可以使用基于规则和轻量嵌入的混合模型目标是“可用”和“高效”。引入复杂的神经网络来评估相关性可能会让引导器本身成为性能瓶颈违背了降低开销的初衷。3. 关键技术实现细节将上述设计思路落地需要解决几个关键的技术实现问题。这里我们以一个基于Python使用LangChain或AutoGen框架构建的多智能体系统为例拆解核心模块的实现。3.1 智能体角色画像的定义与嵌入角色画像是注意力引导的基础。我们需要为每个智能体定义一个结构化的画像。from pydantic import BaseModel from typing import List import numpy as np # 假设我们使用sentence-transformers进行轻量级嵌入 from sentence_transformers import SentenceTransformer class AgentProfile(BaseModel): agent_id: str role_name: str # 如 “Copywriter”, “Designer” core_responsibilities: List[str] # 核心职责描述列表 focus_keywords: List[str] # 关注的关键词列表 # 动态状态 current_task_phase: str “idle“ # 与任务状态机联动 # 嵌入向量缓存用于快速计算相似度 resp_embeddings: np.ndarray None keyword_embeddings: np.ndarray None def __init__(self, **data): super().__init__(**data) # 初始化时生成嵌入向量 embed_model SentenceTransformer(‘BAAI/bge-small-zh-v1.5’) # 以中文小模型为例 self.resp_embeddings embed_model.encode(‘ ‘.join(self.core_responsibilities)) self.keyword_embeddings embed_model.encode(‘ ‘.join(self.focus_keywords)) # 示例定义文案智能体画像 copywriter_profile AgentProfile( agent_id“agent_copy_01”, role_name“资深文案”, core_responsibilities[ “负责广告语、产品描述、社交媒体文案的撰写” “确保文案风格符合品牌调性” “提炼产品核心卖点并进行创意表达” ], focus_keywords[“品牌调性”, “用户痛点”, “核心卖点”, “广告语”, “ slogan”, “情感共鸣”, “简洁有力”] )3.2 消息的定向摘要生成在发送端我们需要一个摘要生成器。考虑到性能这里不采用完整的LLM进行长文本摘要而是采用“抽取概括”的两步法。class AttentionSteeringSender: def __init__(self, owner_profile: AgentProfile, all_profiles: Dict[str, AgentProfile]): self.owner owner_profile self.peer_profiles all_profiles # 所有协作智能体的画像 self.embed_model SentenceTransformer(‘BAAI/bge-small-zh-v1.5’) def generate_targeted_summary(self, raw_message: str, receiver_id: str) - dict: “”“为核心消息生成针对特定接收者的摘要”“” receiver_profile self.peer_profiles.get(receiver_id) if not receiver_profile: return {“summary”: raw_message, “is_filtered”: False} # 无法识别接收者回退到完整消息 # 1. 抽取计算原始消息每句话与接收者关注点的相似度 sentences self._split_into_sentences(raw_message) # 简单的分句 sentence_embeddings self.embed_model.encode(sentences) # 计算每句话与接收者关键词/职责的相似度 relevance_scores [] for sent_emb in sentence_embeddings: # 可以取与职责嵌入和关键词嵌入相似度的最大值或平均值 score_to_resp cosine_similarity(sent_emb, receiver_profile.resp_embeddings) score_to_keywords max([cosine_similarity(sent_emb, kw_emb) for kw_emb in receiver_profile.keyword_embeddings]) final_score max(score_to_resp, score_to_keywords) relevance_scores.append(final_score) # 2. 选择最相关的N句话例如top-3 top_n 3 top_indices np.argsort(relevance_scores)[-top_n:][::-1] top_sentences [sentences[i] for i in top_indices if relevance_scores[i] 0.3] # 设置阈值 # 3. 概括可选如果抽取的句子连贯性差可用一个超轻量模型或规则进行微调、连接 if len(top_sentences) 0: summary “[该消息与您的职责关联度较低如需查看请访问完整记录。]” elif len(top_sentences) 2: summary ‘ ‘.join(top_sentences) else: # 这里可以加入一个简单的连接词逻辑或者调用一个极简的文本生成API如只生成一句话总结 summary f“核心信息点{‘’.join(top_sentences)}” return { “summary”: summary, “is_filtered”: True, “raw_message_ref”: “msg_001”, # 指向原始消息的索引或ID “receiver_id”: receiver_id } def _split_into_sentences(self, text: str) - List[str]: # 简化的分句实际可使用更鲁棒的方法 import re sentences re.split(r‘[。!?]’, text) return [s.strip() for s in sentences if s.strip()]3.3 接收端的相关性评分与路由接收端的引导器更像一个路由器它根据评分决定消息的去向。class AttentionSteeringReceiver: def __init__(self, owner_profile: AgentProfile): self.owner owner_profile self.embed_model SentenceTransformer(‘BAAI/bge-small-zh-v1.5’) # 预定义的任务阶段-信息类型权重表 self.phase_weights { “designing”: {“requirement”: 0.2, “feedback”: 0.9, “reference”: 0.8}, “reviewing”: {“requirement”: 0.1, “feedback”: 1.0, “reference”: 0.5}, # ... 其他阶段 } def score_and_route(self, message_packet: dict) - str: “”“对接收到的消息包进行评分并返回路由决策”“” content message_packet.get(“summary”, message_packet.get(“full_content”, “”)) sender_id message_packet.get(“sender_id”) msg_type message_packet.get(“type”, “general”) # 消息类型由发送方标注 # 1. 计算语义相关性分数 content_embedding self.embed_model.encode(content) semantic_score max( cosine_similarity(content_embedding, self.owner.resp_embeddings), max([cosine_similarity(content_embedding, kw_emb) for kw_emb in self.owner.keyword_embeddings]) ) # 2. 计算任务阶段相关性分数 current_phase self.owner.current_task_phase phase_weight self.phase_weights.get(current_phase, {}).get(msg_type, 0.5) # 假设基础分是0.5根据权重调整 phase_score 0.5 * phase_weight # 3. 计算发送方权威性/信任度分数简化版固定权重或基于简单历史 trust_score self._get_sender_trust_score(sender_id) # 4. 综合评分加权平均权重可调 final_score 0.5 * semantic_score 0.3 * phase_score 0.2 * trust_score # 5. 路由决策 if final_score 0.7: return “HIGH_PRIORITY“ # 送入核心处理队列触发深度思考 elif final_score 0.4: return “MEDIUM_PRIORITY“ # 更新上下文可能进行快速回应 else: return “LOW_PRIORITY“ # 记录日志不触发主要计算 def _get_sender_trust_score(self, sender_id: str) - float: # 简化实现可维护一个sender_id - 历史交互正反馈率的字典 # 初始值设为0.5 return self.trust_history.get(sender_id, 0.5)4. 系统集成与通信流程实战有了发送器和接收器我们需要将它们集成到一个具体的多智能体框架中。这里以LangChain的AgentExecutor思路为例展示如何包裹原有的智能体调用流程。4.1 改造智能体调用链路假设我们原有的智能体调用是用户输入/上游消息-智能体LLM思考-输出行动/消息。改造后的链路变为接收阶段上游消息-AttentionSteeringReceiver.score_and_route()- 根据评分决定是放入核心队列还是仅更新背景。思考阶段智能体的LLM的prompt中不仅包含当前消息还包含由引导器维护的、高相关性的“精炼上下文”而不是完整的聊天历史。发送阶段智能体原始输出-AttentionSteeringSender.generate_targeted_summary()- 为每个接收者生成定制化摘要 -发送摘要包。class EnhancedMultiAgent: def __init__(self, base_agent, agent_profile, sender_module, receiver_module): self.base_agent base_agent # 原有的LangChain Agent或自定义Agent self.profile agent_profile self.sender sender_module self.receiver receiver_module self.refined_context [] # 存储高相关性精炼上下文 def run(self, incoming_message_packet: dict): # 步骤1接收并评分 priority self.receiver.score_and_route(incoming_message_packet) if priority “HIGH_PRIORITY“: # 将消息内容提炼后加入精炼上下文 core_content incoming_message_packet.get(“summary”) or incoming_message_packet.get(“content”) self.refined_context.append(f“From {incoming_message_packet[‘sender_id’]}: {core_content}”) # 限制精炼上下文长度避免膨胀 if len(self.refined_context) 5: self.refined_context.pop(0) # 步骤2组装Prompt使用精炼上下文而非完整历史 prompt self._assemble_prompt(self.refined_context, incoming_message_packet) # 步骤3调用基础智能体进行思考 raw_agent_output self.base_agent.invoke({“input”: prompt}) # 步骤4发送前处理 # 假设raw_agent_output中包含‘recipients’和‘content’ recipients raw_agent_output.get(“recipients”, [“all”]) raw_content raw_agent_output[“content”] messages_to_send [] for rec in recipients: if rec “all”: # 广播场景可以为每个已知角色生成摘要或回退到原始消息 for pid in self.sender.peer_profiles.keys(): if pid ! self.profile.agent_id: packet self.sender.generate_targeted_summary(raw_content, pid) packet[“sender_id”] self.profile.agent_id messages_to_send.append(packet) else: # 定向发送 packet self.sender.generate_targeted_summary(raw_content, rec) packet[“sender_id”] self.profile.agent_id messages_to_send.append(packet) return messages_to_send # 返回待发送的消息包列表 def _assemble_prompt(self, refined_ctx, current_msg): # 将精炼上下文和当前消息组装成给LLM的Prompt ctx_str “\n”.join(refined_ctx) if refined_ctx else “无近期高相关上下文。” current_content current_msg.get(“summary”) or current_msg.get(“content”, “”) prompt_template f“““ 你是一名{self.profile.role_name}你的核心职责是{‘’.join(self.profile.core_responsibilities)}。 以下是与你当前任务高度相关的近期对话背景 {ctx_str} 现在请处理以下最新消息 {current_content} 请根据你的角色和职责进行回应。 “““ return prompt_template4.2 通信总线的协调多个增强后的智能体需要一个协调者来管理消息的路由和传递。我们可以实现一个简单的中央通信总线Message Bus它负责维护所有智能体的注册信息包括其EnhancedMultiAgent实例。接收来自某个智能体的消息包列表。根据每个消息包中的receiver_id将其投递到对应智能体的接收队列中。可能还需要处理“广播”逻辑避免发送方为每个接收方重复生成摘要时造成的重复计算可以在总线层面优化让发送方生成一次总线负责复制和投递。class AttentionAwareMessageBus: def __init__(self): self.agents {} # agent_id - EnhancedMultiAgent instance def register_agent(self, agent_id: str, agent_instance): self.agents[agent_id] agent_instance def deliver_message(self, sender_id: str, message_packets: List[dict]): for packet in message_packets: receiver_id packet.get(“receiver_id”) if receiver_id in self.agents: # 将消息包放入接收方智能体的待处理队列这里简化为直接调用 # 实际应用中这里应该是异步的队列投递 self.agents[receiver_id].receive_queue.put(packet) elif receiver_id “all”: # 广播逻辑投递给除发送者外的所有智能体 for aid, agent in self.agents.items(): if aid ! sender_id: agent.receive_queue.put(packet)5. 性能优化与高级策略基础实现完成后我们需要关注性能和效果的优化。多智能体系统的通信开销是累积的一点微小的优化都能带来显著的规模收益。5.1 摘要生成的缓存与复用为每个(原始消息, 接收者角色类型)对缓存生成的摘要。因为角色类型如“设计师”、“文案”是有限的而同一原始消息对不同角色的摘要是相对静态的。当策划Agent向多个设计师广播同一方案时只需生成一次“面向设计师”的摘要即可复用。这可以大幅减少发送端的计算量。class CachedSummarySender(AttentionSteeringSender): def __init__(self, owner_profile, all_profiles): super().__init__(owner_profile, all_profiles) self.cache {} # key: (message_hash, receiver_role) - summary def generate_targeted_summary(self, raw_message: str, receiver_id: str): receiver_profile self.peer_profiles.get(receiver_id) if not receiver_profile: return {“summary”: raw_message, “is_filtered”: False} msg_hash hash(raw_message) # 或使用更健壮的哈希 cache_key (msg_hash, receiver_profile.role_name) if cache_key in self.cache: summary self.cache[cache_key] else: summary super().generate_targeted_summary(raw_message, receiver_id) self.cache[cache_key] summary return summary5.2 相关性模型的在线微调初始的权重如语义、任务阶段、信任度的权重0.5, 0.3, 0.2和任务阶段关联矩阵可能不是最优的。我们可以引入一个简单的反馈机制进行在线微调。例如当智能体基于一条低相关性评分但被标记为高优先级进行人工或规则干预的消息做出了关键正确决策后系统可以记录这次“误判”并自动调整该消息类型在当前任务阶段的权重或者调整发送方-接收方之间的信任度分数。这可以让系统在运行中自我进化越来越贴合实际协作模式。5.3 处理“信息孤岛”风险过度过滤可能导致“信息孤岛”——即某些智能体因为长期接收不到“低相关性”但潜在重要的跨界信息而变得视野狭窄。为了缓解这个问题可以引入“随机探索”或“重要度衰减”机制。随机探索以一个小概率如5%即使消息相关性评分很低也将其提升为MEDIUM_PRIORITY让智能体有机会接触到“圈外”信息。重要度衰减对于长期未被任何智能体高优先级处理的消息其“全局重要度”会随时间缓慢提升。当提升到一定阈值时通信总线可以将其作为“待办事项提醒”或“可能被忽略的重要信息”推送给项目经理角色智能体或所有智能体进行二次确认。6. 效果评估与常见问题排查部署这样一个系统后如何评估其效果又会遇到哪些典型问题6.1 核心评估指标通信量压缩率比较引入注意力引导前后单位时间内智能体间传递的消息总token数或字节数。目标是有显著下降例如30%-70%。任务完成时间在相同的复杂协作任务上比较系统完成任务所需的平均对话轮次或总耗时。目标是减少无效沟通缩短完成时间。LLM调用成本/延迟由于处理的消息更精炼每个智能体调用LLM进行深度思考的频次和输入长度可能下降直接降低API调用成本和响应延迟。决策质量通过人工评估或预设的客观指标如文案匹配度、设计稿通过率判断在信息过滤后智能体的输出质量是否下降、持平或提升。理想情况是质量持平或提升因为干扰信息减少了。相关性评分准确率抽样检查人工标注一批消息对对于特定智能体的真实相关性高/中/低与系统自动评分进行对比计算准确率、召回率。6.2 常见问题与解决方案问题1摘要丢失关键信息导致下游智能体决策失误。排查检查发送端摘要生成逻辑。是否过于激进地只保留了top-N句话相似度阈值是否设置过高解决引入“关键实体强制保留”机制。使用NER识别消息中的人名、项目名、时间、数字等实体确保它们出现在摘要中。采用“摘要关键标签”的形式。除了文本摘要额外生成一组与接收方相关的关键词标签供接收方快速抓取。在摘要末尾附加“如需了解关于[某个可能相关但被过滤的领域]的细节请查阅完整消息”的提示并提供便捷的完整消息访问接口。问题2智能体陷入局部对话对全局进展失去感知。排查检查“信息孤岛”风险缓解机制是否生效。检查任务阶段-信息类型关联矩阵是否定义得太窄导致阶段切换时无法接收必要信息。解决引入“全局公告板”机制。将涉及任务目标变更、重大里程碑达成等全局性信息强制以完整或高权重摘要形式广播给所有智能体绕过相关性过滤。定期如每5轮对话向所有智能体发送一次“任务状态简报”由系统自动生成汇总各角色进展和关键决策点。问题3相关性评分模型表现不稳定时好时坏。排查检查用于计算相似度的嵌入模型是否与领域匹配。通用的sentence-transformers模型在特定专业领域如法律、医疗可能表现不佳。解决使用领域内数据对嵌入模型进行微调如果数据充足。采用混合模型结合基于规则的关键词匹配确保核心术语不被遗漏和向量相似度取并集。实现一个简单的反馈回路让智能体可以对消息的相关性评分进行“点赞/点踩”用这些数据定期微调评分模型的权重。问题4系统初期由于缺乏历史交互数据信任度分数全是初始值引导效果差。解决设置一个“冷启动期”。在系统运行的前N轮对话或前X个小时内适当降低信任度分数的权重甚至暂时禁用主要依赖语义和任务阶段相关性。提供“种子信任关系”。系统管理员可以根据业务逻辑预先设置一些智能体间的信任权重例如产品经理对设计师的指令权重更高。在实际部署中从一个简单的版本开始设定明确的评估基线然后小步快跑根据上述指标和问题逐一迭代优化是确保项目成功的关键。这项技术的价值不在于一蹴而就的完美而在于它为多智能体系统引入了一种可度量、可优化的通信效率维度让AI协作从“能对话”走向“会对话”。 
长河君 · 2026/8/19 3:57:24 · 阅读 1.2k 多智能体通信优化：基于注意力引导的上下文相关性过滤技术
1. 项目概述当多智能体学会“选择性倾听”在构建由多个大型语言模型LLM驱动的智能体系统时我们常常遇到一个核心困境智能体之间的对话很容易陷入信息过载或无效沟通的泥潭。想象一下在一个由策划、文案、设计、法务等多个AI角色组成的虚拟团队会议上每个智能体都在基于自己的知识库和当前任务背景发言。策划智能体抛出了一个长达千字的宏大方案其中包含了市场分析、用户画像、执行节奏等方方面面。对于文案智能体而言它真正需要关注的可能只是其中的“核心用户痛点描述”和“品牌调性关键词”而对于设计智能体重点则变成了“视觉风格参考”和“关键信息层级”。如果每个智能体都不得不“听”完所有冗长的上下文再费力地从中提取与自己相关的片段不仅会消耗巨大的计算资源直接体现为高昂的API调用成本和延迟更可能导致关键信息被淹没最终影响协同决策的质量和效率。这正是“通过上下文相关性的注意力引导来增强多智能体通信”这一课题要解决的核心问题。它不是一个简单的技术叠加而是一种通信范式的转变。其目标不是让智能体变得更“聪明”而是让它们变得更“专注”——学会在纷繁复杂的多轮对话中像经验丰富的团队成员一样主动将“注意力”聚焦在与自身角色和当前任务最相关的信息上。这里的“注意力”借鉴了Transformer架构中的核心机制但将其应用层面从模型内部的token关联提升到了智能体间消息传递的宏观调度。而“上下文相关性”则是实现这种引导的标尺它动态地评估历史对话中每一段信息对于接收方智能体的价值。通过这项技术我们期望达到的效果是降低通信冗余、提升决策效率、减少不必要的计算开销最终让多智能体系统在复杂任务上的协作更加流畅、精准且经济。2. 核心设计思路从全局对话到定向投喂传统的多智能体通信架构无论是简单的广播/订阅模式还是基于黑板Blackboard的共享内存模式本质上都是一种“推”的模式。发送方将完整的自身思考或行动结果打包成消息“推”给所有或指定的接收方。接收方则需要自行处理整个消息体。这种模式在智能体数量少、消息结构简单时尚可工作但随着智能体角色分化、任务复杂度提升其弊端暴露无遗。本项目的设计思路可以概括为将“广播”变为“窄播”将“完整推送”变为“摘要投喂”。其核心在于为每个智能体配备一个轻量级的“注意力引导器”。这个引导器不参与智能体自身的核心推理而是工作在通信层负责两件事1. 对发出的消息自动生成面向不同接收方的、定制化的“相关性摘要”2. 对接收到的消息快速评估其与自身任务的相关性并决定是深入处理、简要浏览还是直接忽略。2.1 注意力引导器的双工工作流这个引导器的工作是双工的贯穿通信的发送与接收两端。在发送端Sender Side Filtering 当一个智能体如策划Agent准备发言时其原始的、完整的思考内容我们称之为“原始消息体”会首先经过其自身的注意力引导器。这个引导器内置了所有协作智能体的角色画像Role Profile。角色画像不是简单的名称而是包含了该角色的核心职责、关注的关键信息类型如设计Agent关注“色彩”、“布局”、“风格”法务Agent关注“责任”、“条款”、“合规”等。引导器会基于接收方智能体的角色画像对原始消息体进行实时分析抽取出与之最相关的片段、关键词或总结性陈述生成一个“定向摘要”。这个摘要可能只是几个关键词、一两句话或者一个结构化的数据片段。最终发送出去的不再是冗长的全文而是这个轻量的“定向摘要”加上一个指向原始消息体的索引或链接如果需要深度查看。在接收端Receiver Side Scoring 当智能体如设计Agent收到来自其他智能体的消息可能是摘要也可能是完整消息时其自身的注意力引导器会立即启动。它会基于自己当前的任务状态例如正在设计首页Banner和角色画像计算接收到的消息的“上下文相关性分数”。这个分数综合考虑了语义相关性消息内容与自身职责关键词的匹配度。任务时序相关性消息是否与当前任务阶段强相关例如在需求确认阶段设计Agent应高度关注“需求变更”类消息在评审阶段则应关注“修改意见”。发送方权威性不同发送方的消息权重可能不同例如项目经理Agent的指令可能比另一个并行设计Agent的建议具有更高的默认相关性。基于这个分数引导器会决定消息的处理优先级高相关消息送入智能体的核心推理模块进行深度处理中等相关消息可能仅更新一下上下文记忆低相关消息则被标记或存档几乎不占用主要计算资源。2.2 上下文相关性的动态计算模型“上下文相关性”是本设计的灵魂其计算不能是静态的规则匹配而必须是一个动态、可学习的模型。一个基础但有效的实现可以分为三层表层关键词匹配层使用轻量级的嵌入模型如BGE-M3或text-embedding-3-small将消息内容和智能体的“关注词库”分别转化为向量计算余弦相似度。这一层速度快能快速过滤掉明显不相关的信息。任务状态感知层每个智能体维护一个简单的任务状态机如“等待需求”、“设计中”、“修改中”、“已完成”。系统预定义了一个“状态-信息类型”关联矩阵。例如设计Agent在“设计中”状态时对“新增需求”类消息的相关性权重会降低因为原则上不应中途加需求而对“同类参考案例”类消息的权重会升高。历史交互记忆层记录智能体间的历史交互。如果A智能体过去提供的某类信息频繁被B智能体采纳并产生了正反馈任务推进那么未来A向B发送的同类信息会获得更高的初始相关性分数。这相当于为智能体间建立了简单的“信任度”模型。这三层分数经过一个可配置的权重加权求和最终得出一个综合相关性分数。权重可以在系统初始化时设定也可以设计一个简单的元学习机制进行微调。注意相关性模型的设计切忌复杂化。初期完全可以使用基于规则和轻量嵌入的混合模型目标是“可用”和“高效”。引入复杂的神经网络来评估相关性可能会让引导器本身成为性能瓶颈违背了降低开销的初衷。3. 关键技术实现细节将上述设计思路落地需要解决几个关键的技术实现问题。这里我们以一个基于Python使用LangChain或AutoGen框架构建的多智能体系统为例拆解核心模块的实现。3.1 智能体角色画像的定义与嵌入角色画像是注意力引导的基础。我们需要为每个智能体定义一个结构化的画像。from pydantic import BaseModel from typing import List import numpy as np # 假设我们使用sentence-transformers进行轻量级嵌入 from sentence_transformers import SentenceTransformer class AgentProfile(BaseModel): agent_id: str role_name: str # 如 “Copywriter”, “Designer” core_responsibilities: List[str] # 核心职责描述列表 focus_keywords: List[str] # 关注的关键词列表 # 动态状态 current_task_phase: str “idle“ # 与任务状态机联动 # 嵌入向量缓存用于快速计算相似度 resp_embeddings: np.ndarray None keyword_embeddings: np.ndarray None def __init__(self, **data): super().__init__(**data) # 初始化时生成嵌入向量 embed_model SentenceTransformer(‘BAAI/bge-small-zh-v1.5’) # 以中文小模型为例 self.resp_embeddings embed_model.encode(‘ ‘.join(self.core_responsibilities)) self.keyword_embeddings embed_model.encode(‘ ‘.join(self.focus_keywords)) # 示例定义文案智能体画像 copywriter_profile AgentProfile( agent_id“agent_copy_01”, role_name“资深文案”, core_responsibilities[ “负责广告语、产品描述、社交媒体文案的撰写” “确保文案风格符合品牌调性” “提炼产品核心卖点并进行创意表达” ], focus_keywords[“品牌调性”, “用户痛点”, “核心卖点”, “广告语”, “ slogan”, “情感共鸣”, “简洁有力”] )3.2 消息的定向摘要生成在发送端我们需要一个摘要生成器。考虑到性能这里不采用完整的LLM进行长文本摘要而是采用“抽取概括”的两步法。class AttentionSteeringSender: def __init__(self, owner_profile: AgentProfile, all_profiles: Dict[str, AgentProfile]): self.owner owner_profile self.peer_profiles all_profiles # 所有协作智能体的画像 self.embed_model SentenceTransformer(‘BAAI/bge-small-zh-v1.5’) def generate_targeted_summary(self, raw_message: str, receiver_id: str) - dict: “”“为核心消息生成针对特定接收者的摘要”“” receiver_profile self.peer_profiles.get(receiver_id) if not receiver_profile: return {“summary”: raw_message, “is_filtered”: False} # 无法识别接收者回退到完整消息 # 1. 抽取计算原始消息每句话与接收者关注点的相似度 sentences self._split_into_sentences(raw_message) # 简单的分句 sentence_embeddings self.embed_model.encode(sentences) # 计算每句话与接收者关键词/职责的相似度 relevance_scores [] for sent_emb in sentence_embeddings: # 可以取与职责嵌入和关键词嵌入相似度的最大值或平均值 score_to_resp cosine_similarity(sent_emb, receiver_profile.resp_embeddings) score_to_keywords max([cosine_similarity(sent_emb, kw_emb) for kw_emb in receiver_profile.keyword_embeddings]) final_score max(score_to_resp, score_to_keywords) relevance_scores.append(final_score) # 2. 选择最相关的N句话例如top-3 top_n 3 top_indices np.argsort(relevance_scores)[-top_n:][::-1] top_sentences [sentences[i] for i in top_indices if relevance_scores[i] 0.3] # 设置阈值 # 3. 概括可选如果抽取的句子连贯性差可用一个超轻量模型或规则进行微调、连接 if len(top_sentences) 0: summary “[该消息与您的职责关联度较低如需查看请访问完整记录。]” elif len(top_sentences) 2: summary ‘ ‘.join(top_sentences) else: # 这里可以加入一个简单的连接词逻辑或者调用一个极简的文本生成API如只生成一句话总结 summary f“核心信息点{‘’.join(top_sentences)}” return { “summary”: summary, “is_filtered”: True, “raw_message_ref”: “msg_001”, # 指向原始消息的索引或ID “receiver_id”: receiver_id } def _split_into_sentences(self, text: str) - List[str]: # 简化的分句实际可使用更鲁棒的方法 import re sentences re.split(r‘[。!?]’, text) return [s.strip() for s in sentences if s.strip()]3.3 接收端的相关性评分与路由接收端的引导器更像一个路由器它根据评分决定消息的去向。class AttentionSteeringReceiver: def __init__(self, owner_profile: AgentProfile): self.owner owner_profile self.embed_model SentenceTransformer(‘BAAI/bge-small-zh-v1.5’) # 预定义的任务阶段-信息类型权重表 self.phase_weights { “designing”: {“requirement”: 0.2, “feedback”: 0.9, “reference”: 0.8}, “reviewing”: {“requirement”: 0.1, “feedback”: 1.0, “reference”: 0.5}, # ... 其他阶段 } def score_and_route(self, message_packet: dict) - str: “”“对接收到的消息包进行评分并返回路由决策”“” content message_packet.get(“summary”, message_packet.get(“full_content”, “”)) sender_id message_packet.get(“sender_id”) msg_type message_packet.get(“type”, “general”) # 消息类型由发送方标注 # 1. 计算语义相关性分数 content_embedding self.embed_model.encode(content) semantic_score max( cosine_similarity(content_embedding, self.owner.resp_embeddings), max([cosine_similarity(content_embedding, kw_emb) for kw_emb in self.owner.keyword_embeddings]) ) # 2. 计算任务阶段相关性分数 current_phase self.owner.current_task_phase phase_weight self.phase_weights.get(current_phase, {}).get(msg_type, 0.5) # 假设基础分是0.5根据权重调整 phase_score 0.5 * phase_weight # 3. 计算发送方权威性/信任度分数简化版固定权重或基于简单历史 trust_score self._get_sender_trust_score(sender_id) # 4. 综合评分加权平均权重可调 final_score 0.5 * semantic_score 0.3 * phase_score 0.2 * trust_score # 5. 路由决策 if final_score 0.7: return “HIGH_PRIORITY“ # 送入核心处理队列触发深度思考 elif final_score 0.4: return “MEDIUM_PRIORITY“ # 更新上下文可能进行快速回应 else: return “LOW_PRIORITY“ # 记录日志不触发主要计算 def _get_sender_trust_score(self, sender_id: str) - float: # 简化实现可维护一个sender_id - 历史交互正反馈率的字典 # 初始值设为0.5 return self.trust_history.get(sender_id, 0.5)4. 系统集成与通信流程实战有了发送器和接收器我们需要将它们集成到一个具体的多智能体框架中。这里以LangChain的AgentExecutor思路为例展示如何包裹原有的智能体调用流程。4.1 改造智能体调用链路假设我们原有的智能体调用是用户输入/上游消息-智能体LLM思考-输出行动/消息。改造后的链路变为接收阶段上游消息-AttentionSteeringReceiver.score_and_route()- 根据评分决定是放入核心队列还是仅更新背景。思考阶段智能体的LLM的prompt中不仅包含当前消息还包含由引导器维护的、高相关性的“精炼上下文”而不是完整的聊天历史。发送阶段智能体原始输出-AttentionSteeringSender.generate_targeted_summary()- 为每个接收者生成定制化摘要 -发送摘要包。class EnhancedMultiAgent: def __init__(self, base_agent, agent_profile, sender_module, receiver_module): self.base_agent base_agent # 原有的LangChain Agent或自定义Agent self.profile agent_profile self.sender sender_module self.receiver receiver_module self.refined_context [] # 存储高相关性精炼上下文 def run(self, incoming_message_packet: dict): # 步骤1接收并评分 priority self.receiver.score_and_route(incoming_message_packet) if priority “HIGH_PRIORITY“: # 将消息内容提炼后加入精炼上下文 core_content incoming_message_packet.get(“summary”) or incoming_message_packet.get(“content”) self.refined_context.append(f“From {incoming_message_packet[‘sender_id’]}: {core_content}”) # 限制精炼上下文长度避免膨胀 if len(self.refined_context) 5: self.refined_context.pop(0) # 步骤2组装Prompt使用精炼上下文而非完整历史 prompt self._assemble_prompt(self.refined_context, incoming_message_packet) # 步骤3调用基础智能体进行思考 raw_agent_output self.base_agent.invoke({“input”: prompt}) # 步骤4发送前处理 # 假设raw_agent_output中包含‘recipients’和‘content’ recipients raw_agent_output.get(“recipients”, [“all”]) raw_content raw_agent_output[“content”] messages_to_send [] for rec in recipients: if rec “all”: # 广播场景可以为每个已知角色生成摘要或回退到原始消息 for pid in self.sender.peer_profiles.keys(): if pid ! self.profile.agent_id: packet self.sender.generate_targeted_summary(raw_content, pid) packet[“sender_id”] self.profile.agent_id messages_to_send.append(packet) else: # 定向发送 packet self.sender.generate_targeted_summary(raw_content, rec) packet[“sender_id”] self.profile.agent_id messages_to_send.append(packet) return messages_to_send # 返回待发送的消息包列表 def _assemble_prompt(self, refined_ctx, current_msg): # 将精炼上下文和当前消息组装成给LLM的Prompt ctx_str “\n”.join(refined_ctx) if refined_ctx else “无近期高相关上下文。” current_content current_msg.get(“summary”) or current_msg.get(“content”, “”) prompt_template f“““ 你是一名{self.profile.role_name}你的核心职责是{‘’.join(self.profile.core_responsibilities)}。 以下是与你当前任务高度相关的近期对话背景 {ctx_str} 现在请处理以下最新消息 {current_content} 请根据你的角色和职责进行回应。 “““ return prompt_template4.2 通信总线的协调多个增强后的智能体需要一个协调者来管理消息的路由和传递。我们可以实现一个简单的中央通信总线Message Bus它负责维护所有智能体的注册信息包括其EnhancedMultiAgent实例。接收来自某个智能体的消息包列表。根据每个消息包中的receiver_id将其投递到对应智能体的接收队列中。可能还需要处理“广播”逻辑避免发送方为每个接收方重复生成摘要时造成的重复计算可以在总线层面优化让发送方生成一次总线负责复制和投递。class AttentionAwareMessageBus: def __init__(self): self.agents {} # agent_id - EnhancedMultiAgent instance def register_agent(self, agent_id: str, agent_instance): self.agents[agent_id] agent_instance def deliver_message(self, sender_id: str, message_packets: List[dict]): for packet in message_packets: receiver_id packet.get(“receiver_id”) if receiver_id in self.agents: # 将消息包放入接收方智能体的待处理队列这里简化为直接调用 # 实际应用中这里应该是异步的队列投递 self.agents[receiver_id].receive_queue.put(packet) elif receiver_id “all”: # 广播逻辑投递给除发送者外的所有智能体 for aid, agent in self.agents.items(): if aid ! sender_id: agent.receive_queue.put(packet)5. 性能优化与高级策略基础实现完成后我们需要关注性能和效果的优化。多智能体系统的通信开销是累积的一点微小的优化都能带来显著的规模收益。5.1 摘要生成的缓存与复用为每个(原始消息, 接收者角色类型)对缓存生成的摘要。因为角色类型如“设计师”、“文案”是有限的而同一原始消息对不同角色的摘要是相对静态的。当策划Agent向多个设计师广播同一方案时只需生成一次“面向设计师”的摘要即可复用。这可以大幅减少发送端的计算量。class CachedSummarySender(AttentionSteeringSender): def __init__(self, owner_profile, all_profiles): super().__init__(owner_profile, all_profiles) self.cache {} # key: (message_hash, receiver_role) - summary def generate_targeted_summary(self, raw_message: str, receiver_id: str): receiver_profile self.peer_profiles.get(receiver_id) if not receiver_profile: return {“summary”: raw_message, “is_filtered”: False} msg_hash hash(raw_message) # 或使用更健壮的哈希 cache_key (msg_hash, receiver_profile.role_name) if cache_key in self.cache: summary self.cache[cache_key] else: summary super().generate_targeted_summary(raw_message, receiver_id) self.cache[cache_key] summary return summary5.2 相关性模型的在线微调初始的权重如语义、任务阶段、信任度的权重0.5, 0.3, 0.2和任务阶段关联矩阵可能不是最优的。我们可以引入一个简单的反馈机制进行在线微调。例如当智能体基于一条低相关性评分但被标记为高优先级进行人工或规则干预的消息做出了关键正确决策后系统可以记录这次“误判”并自动调整该消息类型在当前任务阶段的权重或者调整发送方-接收方之间的信任度分数。这可以让系统在运行中自我进化越来越贴合实际协作模式。5.3 处理“信息孤岛”风险过度过滤可能导致“信息孤岛”——即某些智能体因为长期接收不到“低相关性”但潜在重要的跨界信息而变得视野狭窄。为了缓解这个问题可以引入“随机探索”或“重要度衰减”机制。随机探索以一个小概率如5%即使消息相关性评分很低也将其提升为MEDIUM_PRIORITY让智能体有机会接触到“圈外”信息。重要度衰减对于长期未被任何智能体高优先级处理的消息其“全局重要度”会随时间缓慢提升。当提升到一定阈值时通信总线可以将其作为“待办事项提醒”或“可能被忽略的重要信息”推送给项目经理角色智能体或所有智能体进行二次确认。6. 效果评估与常见问题排查部署这样一个系统后如何评估其效果又会遇到哪些典型问题6.1 核心评估指标通信量压缩率比较引入注意力引导前后单位时间内智能体间传递的消息总token数或字节数。目标是有显著下降例如30%-70%。任务完成时间在相同的复杂协作任务上比较系统完成任务所需的平均对话轮次或总耗时。目标是减少无效沟通缩短完成时间。LLM调用成本/延迟由于处理的消息更精炼每个智能体调用LLM进行深度思考的频次和输入长度可能下降直接降低API调用成本和响应延迟。决策质量通过人工评估或预设的客观指标如文案匹配度、设计稿通过率判断在信息过滤后智能体的输出质量是否下降、持平或提升。理想情况是质量持平或提升因为干扰信息减少了。相关性评分准确率抽样检查人工标注一批消息对对于特定智能体的真实相关性高/中/低与系统自动评分进行对比计算准确率、召回率。6.2 常见问题与解决方案问题1摘要丢失关键信息导致下游智能体决策失误。排查检查发送端摘要生成逻辑。是否过于激进地只保留了top-N句话相似度阈值是否设置过高解决引入“关键实体强制保留”机制。使用NER识别消息中的人名、项目名、时间、数字等实体确保它们出现在摘要中。采用“摘要关键标签”的形式。除了文本摘要额外生成一组与接收方相关的关键词标签供接收方快速抓取。在摘要末尾附加“如需了解关于[某个可能相关但被过滤的领域]的细节请查阅完整消息”的提示并提供便捷的完整消息访问接口。问题2智能体陷入局部对话对全局进展失去感知。排查检查“信息孤岛”风险缓解机制是否生效。检查任务阶段-信息类型关联矩阵是否定义得太窄导致阶段切换时无法接收必要信息。解决引入“全局公告板”机制。将涉及任务目标变更、重大里程碑达成等全局性信息强制以完整或高权重摘要形式广播给所有智能体绕过相关性过滤。定期如每5轮对话向所有智能体发送一次“任务状态简报”由系统自动生成汇总各角色进展和关键决策点。问题3相关性评分模型表现不稳定时好时坏。排查检查用于计算相似度的嵌入模型是否与领域匹配。通用的sentence-transformers模型在特定专业领域如法律、医疗可能表现不佳。解决使用领域内数据对嵌入模型进行微调如果数据充足。采用混合模型结合基于规则的关键词匹配确保核心术语不被遗漏和向量相似度取并集。实现一个简单的反馈回路让智能体可以对消息的相关性评分进行“点赞/点踩”用这些数据定期微调评分模型的权重。问题4系统初期由于缺乏历史交互数据信任度分数全是初始值引导效果差。解决设置一个“冷启动期”。在系统运行的前N轮对话或前X个小时内适当降低信任度分数的权重甚至暂时禁用主要依赖语义和任务阶段相关性。提供“种子信任关系”。系统管理员可以根据业务逻辑预先设置一些智能体间的信任权重例如产品经理对设计师的指令权重更高。在实际部署中从一个简单的版本开始设定明确的评估基线然后小步快跑根据上述指标和问题逐一迭代优化是确保项目成功的关键。这项技术的价值不在于一蹴而就的完美而在于它为多智能体系统引入了一种可度量、可优化的通信效率维度让AI协作从“能对话”走向“会对话”。