构建AI智能体：四十七、Agent2Agent多智能体系统：基础通信与任务协作实现 - 指南
<style>pre { white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important; display: block !important; font-family: "Consolas", "Monaco", "Courier New", monospace !important; font-size: 14px !important; line-height: 1.6 !important; padding: 16px !important; margin: 16px 0 !important; background-color: rgba(248, 248, 248, 1) !important; border: 1px solid rgba(225, 228, 232, 1) !important; border-radius: 6px !important; tab-size: 4 !important; -moz-tab-size: 4 !important; max-width: 100% !important; box-sizing: border-box !important }code { font-family: "Consolas", "Monaco", "Courier New", monospace !important; font-size: 14px !important; white-space: pre !important; word-wrap: normal !important; word-break: normal !important; overflow-wrap: normal !important; display: inline !important; background: rgba(0, 0, 0, 0) !important; border: none !important; padding: 0 !important; margin: 0 !important; line-height: inherit !important }
pre code { background: rgba(0, 0, 0, 0) !important; border: 0 !important; border-radius: 0 !important; display: block !important; line-height: 1.6 !important; margin: 0 !important; max-width: none !important; overflow: visible !important; padding: 0 !important; white-space: pre !important; word-wrap: normal !important; word-break: normal !important; color: inherit !important }
.token.comment, .token.prolog, .token.doctype, .token.cdata { color: rgba(112, 128, 144, 1) !important; font-style: italic !important }
.token.punctuation { color: rgba(153, 153, 153, 1) !important }
.token.atrule, .token.attr-value, .token.keyword { color: rgba(0, 119, 170, 1) !important; font-weight: bold !important }
.token.function, .token.class-name { color: rgba(221, 74, 104, 1) !important; font-weight: bold !important }
.token.selector, .token.attr-name, .token.string, .token.char, .token.builtin, .token.inserted { color: rgba(102, 153, 0, 1) !important }
.token.property, .token.tag, .token.boolean, .token.number, .token.constant, .token.symbol, .token.deleted { color: rgba(153, 0, 85, 1) !important }
.cnblogs-markdown pre, .cnblogs-post-body pre { white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important; background-color: rgba(248, 248, 248, 1) !important; border: 1px solid rgba(225, 228, 232, 1) !important; border-radius: 6px !important; padding: 16px !important; margin: 16px 0 !important }
pre, pre, pre { white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important }</style>
<div class="htmledit_views atom-one-light" id="content_views"><h2>一、基本概述</h2><p> Agent2Agent（A2A）是一个专注于智能体间通信、协作与协调的框架，旨在促进多个智能体（Agent）之间的通信与协作。在人工智能领域，智能体是自主的软件实体，能够感知环境、执行动作、与其他智能体通信，单个智能体的能力有限，而多个智能体通过有效的交互可以解决更复杂的问题。A2A框架提供了标准化的通信协议、消息传递机制和协作模式，使智能体能够像人类团队一样协同工作，解决单个智能体难以处理的复杂问题。</p><p><img alt="" height="1350" src="https://i-blog.csdnimg.cn/direct/4a2005a8db7143d19b052ad346f9bd80.jpeg" width="2400"></p><h2>二、智能体的特征和类型</h2><h3>1. 智能体的特征</h3><p>智能体是A2A框架中的基本单位。每个智能体具有以下特性：</p><ul><li><strong>自主性：</strong>智能体能够在没有直接外部干预的情况下运作，控制自身内部状态和行为。自主性程度可以从完全自主到半自主不等，取决于系统设计需求。</li><li><strong>社交能力：</strong>智能体能够通过某种通信语言与其他智能体（或人类）进行交互。这种交互不仅仅是数据交换，还包括意图传达、协商协作等复杂社交行为。</li><li><strong>反应性：</strong>智能体能够感知所处环境（包括其他智能体），并对环境变化做出及时、适当的响应。这种响应不是简单的刺激-反应，而是基于内部状态的决策。</li><li><strong>主动性：</strong>智能体不仅能够对环境变化做出反应，还能够主动追求目标，表现出目标导向的行为。它们能够主动发起行动，而不仅仅是被动响应。</li><li><strong>知识库与推理能力：</strong>智能体拥有内部知识表示和推理机制，能够基于已有知识进行决策。知识库可以包括领域知识、交互历史、信任模型等。</li><li><strong>学习与适应能力：</strong>高级智能体具备从经验中学习的能力，能够根据历史交互结果调整行为策略，适应环境变化。</li></ul><h3>2. 智能体的类型</h3><ul><li><strong>反应型智能体：</strong>基于条件-动作规则运作，不维护复杂内部状态。它们对环境刺激做出直接反应，决策速度快但灵活性有限。适用于环境相对稳定、决策逻辑简单的场景。</li><li><strong>慎思型智能体：</strong>基于符号人工智能原理，拥有复杂的内部状态模型（信念、愿望、意图，即BDI架构）。它们能够进行规划推理，适合处理复杂、需要深思熟虑的决策任务。</li><li><strong>混合型智能体：</strong>结合反应型和慎思型的优点，在不同情境下采用不同的决策模式。简单情况使用快速反应机制，复杂情况启用深度推理能力。</li><li><strong>移动智能体：</strong>能够在网络中的不同节点间迁移，携带代码和数据在远程主机上执行。适用于分布式信息检索、网络管理等场景</li></ul><p> 在A2A框架中，智能体通过消息进行通信。每个智能体有一个唯一的标识符（Agent ID），并可以注册多种能力（Capabilities），这些能力描述了智能体能够执行的任务类型。</p><h2>三. 智能体的协作机制</h2><h3>1. 协作模式</h3><ul><li><strong>主从协调模式：</strong>一个主智能体（协调者）负责任务分解、分配和结果整合，多个从智能体（工作者）执行具体子任务。适用于任务结构清晰、需要集中控制的场景。</li><li><strong>对等协作模式：</strong>所有智能体地位平等，通过协商和共识机制协调行动。没有中心控制节点，系统更具鲁棒性但协调复杂度高。</li><li><strong>市场协作模式：</strong>基于经济学原理，智能体通过虚拟货币、竞价等市场机制进行资源分配和任务协调。适合资源稀缺、需要优化分配的场景。</li><li><strong>团队协作模式：</strong>智能体形成具有共同目标的团队，通过共享计划、联合承诺和相互意识实现深度协作。团队成员对团队目标有共同承诺。</li></ul><h3>2. 协作策略</h3><ul><li><strong>利他策略：</strong>智能体总是优先考虑其他智能体或整体系统的利益，即使可能损害自身短期利益。促进系统整体效能但可能被自私智能体利用。</li><li><strong>自私策略：</strong>智能体只考虑自身利益最大化，可能拒绝合作或提供虚假信息。在竞争性环境中常见，但可能降低系统整体效率。</li><li><strong>互惠策略：</strong>基于"以牙还牙"原则，智能体根据历史交互记录决定合作程度。对方过去合作则积极回应，对方背叛则相应报复。</li><li><strong>信任基础策略：</strong>基于信任模型评估合作伙伴的可靠性，根据信任等级调整合作深度。信任模型考虑历史表现、声誉推荐等多维度信息。</li></ul><h3>3. 冲突解决机制</h3><ul><li><strong>冲突检测</strong><ul><li>资源冲突：多个智能体竞争有限资源</li><li>目标冲突：智能体目标相互矛盾或竞争</li><li>计划冲突：行动序列在时间或空间上冲突</li></ul></li><li><strong>冲突解决策略</strong><ul><li>协商：各方通过多轮提议-反提议寻求共赢方案</li><li>调解：第三方中立智能体协助冲突各方达成协议</li><li>仲裁：权威智能体强制制定解决方案</li><li>投票：多数决定原则解决分歧</li></ul></li></ul><h2>四、智能体的通信协议</h2><p><strong>1. 消息（Message）</strong></p><p>消息是智能体之间通信的基本单元。A2A框架定义了标准的消息格式，包括：</p><ul><li>消息ID：唯一标识一条消息。</li><li>发送者ID：发送消息的智能体标识符。</li><li>接收者ID：接收消息的智能体标识符。</li><li>消息类型：描述消息的目的，例如任务请求、任务响应、协商等。</li><li>内容：消息的具体内容，格式由消息类型和具体协议决定。</li><li>时间戳：消息发送的时间。</li><li>会话ID：可选字段，用于将多条消息关联到一个会话中。</li><li>优先级：消息的紧急程度。</li></ul><p><strong>2. 传输层（Transport）</strong></p><p> 传输层负责消息的实际传递。它可以是本地的（在同一进程内）或分布式的（通过网络）。传输层的主要职责是：</p><ul><li>连接管理：建立和维护智能体之间的连接。</li><li>消息路由：将消息从发送者传递到正确的接收者。</li><li>可靠性：确保消息的可靠传递（例如，通过重试机制）。</li></ul><p> A2A框架可以支持多种传输实现，例如本地内存传输、基于消息队列（如RabbitMQ）的分布式传输等。</p><p><strong>3. 任务分配器</strong></p><p> 任务分配器负责将任务分配给合适的智能体。它根据智能体的能力和任务的需求进行匹配。任务分配算法可以是简单的轮询，也可以是基于能力匹配的复杂算法。</p><p><strong>4. 协商协议</strong></p><p> 协商协议定义了智能体之间如何通过多轮交互达成一致。常见的协商协议包括合同网协议和拍卖协议。</p><h2>五. 组织结构与协调机制</h2><h3>1. 智能体组织结构</h3><ul><li><strong>层次结构：</strong>明确的上下级关系，决策权集中在上层智能体。协调效率高但单点故障风险大，适合任务结构固定的场景。</li><li><strong>扁平结构：</strong>智能体间地位平等，通过对等协商协调行动。系统鲁棒性强但协调成本高，适合动态变化的环境。</li><li><strong>混合结构：</strong>结合层次和平扁结构的优点，在局部采用层次协调，在全局采用对等协商。平衡效率与灵活性。</li><li><strong>市场结构：</strong>基于供求关系和价格机制进行协调，智能体通过买卖服务实现资源分配。适合开放、动态的环境。</li></ul><h3>2. 基于角色的协调</h3><p><strong>2.1 角色定义</strong></p><p>角色是一组职责、权限和期望行为的封装。智能体通过扮演不同角色参与协作：</p><ul><li>角色职责: 该角色需要完成的任务和承担的责任</li><li>角色权限: 角色在组织中的决策权力和资源访问权限</li><li>角色关系: 不同角色间的协作关系和交互协议</li></ul><p><strong>2.2 动态角色分配</strong></p><p> 根据任务需求和组织状态动态分配角色，使系统能够适应环境变化。考虑智能体能力、负载状态等因素。</p><h2 style="background-color: rgba(0, 0, 0, 0)">六、智能体的协作流程</h2><h3 style="background-color: rgba(0, 0, 0, 0)">1. 整体协作流程图</h3><p class="img-center"><img alt="" height="1460" src="https://i-blog.csdnimg.cn/direct/0bba251435a44b84adcb4c6c2e7874ea.png" width="590"></p><p><strong>流程说明：</strong></p><p><strong>阶段1：协作准备阶段</strong></p><ul><li>任务识别与分解：分析任务需求，识别任务边界，将复杂任务分解为可管理的子任务。考虑任务间的依赖关系和执行顺序。</li><li>智能体发现与能力匹配：在智能体网络中寻找具备所需能力的参与者，评估各智能体的当前负载、历史表现和信誉度。</li><li>协作模式选择：根据任务特性、参与智能体关系和环境条件选择最合适的协作模式。</li></ul><p><strong>阶段2：通信与协商阶段</strong></p><ul><li>任务请求与分配：明确任务要求、质量标准和验收条件，建立任务分配协议。</li><li>能力协商与匹配：确认各智能体的具体能力参数，协商服务等级协议(SLA)。</li><li>资源协商与分配：分配计算、存储、网络等资源，建立资源使用配额和优先级。</li><li>计划协调与同步：制定详细执行计划，建立里程碑和检查点，同步各参与方的时间表。</li></ul><p><strong>阶段3：任务执行与监控阶段</strong></p><ul><li>并行任务执行：按照预定计划执行任务，处理任务间的依赖和同步。</li><li>进度监控与报告：实时跟踪执行状态，收集性能指标，生成进度报告。</li><li>异常检测与处理：监控系统健康状态，检测异常情况，启动应急预案。</li><li>动态调整与重分配：根据执行情况动态优化资源分配，调整任务优先级。</li></ul><p><strong>阶段4：结果整合与评估阶段</strong></p><ul><li>子结果收集：汇总各智能体的输出结果，验证数据的完整性和一致性。</li><li>冲突解决与融合：处理结果间的冲突和不一致，采用合适的融合策略。</li><li>质量评估与验证：评估最终结果的质量，验证是否满足原始需求。</li><li>经验学习与更新：总结协作经验，更新智能体的知识库和策略。</li></ul><p><strong>阶段5：协作终止与总结</strong></p><ul><li>释放所有分配的资源</li><li>生成协作总结报告</li><li>更新智能体信誉和性能记录</li><li>归档协作过程和结果数据</li></ul><h3 style="background-color: rgba(0, 0, 0, 0)">2. 协作模式选择决策树</h3><p class="img-center"><img alt="" height="678" src="https://i-blog.csdnimg.cn/direct/58e34eda433249dfadcac8f394f6ad70.png" width="750"></p><p style="background-color: rgba(0, 0, 0, 0)"><strong>详细说明：</strong></p><p style="background-color: rgba(0, 0, 0, 0)"><strong>决策因素分析：</strong></p><ul><li style="background-color: rgba(0, 0, 0, 0)">任务复杂度评估：
<ul><li style="background-color: rgba(0, 0, 0, 0)">简单任务：目标明确、步骤清晰、结果可预测</li><li style="background-color: rgba(0, 0, 0, 0)">复杂任务：多目标、不确定性高、需要动态调整</li></ul></li><li style="background-color: rgba(0, 0, 0, 0)">执行者数量考虑：
<ul><li style="background-color: rgba(0, 0, 0, 0)">单个执行者：任务规模小、专业要求集中</li><li style="background-color: rgba(0, 0, 0, 0)">多个执行者：需要分工协作、资源共享</li></ul></li><li style="background-color: rgba(0, 0, 0, 0)">任务依赖性分析：
<ul><li style="background-color: rgba(0, 0, 0, 0)">强依赖：严格的执行顺序、中间结果依赖</li><li style="background-color: rgba(0, 0, 0, 0)">弱依赖：可并行执行、独立性较强</li></ul></li><li style="background-color: rgba(0, 0, 0, 0)">资源竞争程度：
<ul><li style="background-color: rgba(0, 0, 0, 0)">高竞争：稀缺资源、需要优化分配</li><li style="background-color: rgba(0, 0, 0, 0)">低竞争：资源充足、分配简单</li></ul></li></ul><p style="background-color: rgba(0, 0, 0, 0)"><strong>协作模式特性：</strong></p><ul><li style="background-color: rgba(0, 0, 0, 0)">直接分配模式：
<ul><li style="background-color: rgba(0, 0, 0, 0)">适用场景：简单任务、单个专业执行者</li><li style="background-color: rgba(0, 0, 0, 0)">优势：决策简单、执行高效</li><li style="background-color: rgba(0, 0, 0, 0)">劣势：单点故障风险、缺乏冗余</li></ul></li><li style="background-color: rgba(0, 0, 0, 0)">主从协作模式：
<ul><li style="background-color: rgba(0, 0, 0, 0)">适用场景：需要集中协调的多参与者任务</li><li style="background-color: rgba(0, 0, 0, 0)">优势：职责明确、协调高效</li><li style="background-color: rgba(0, 0, 0, 0)">劣势：中心节点压力大、灵活性较低</li></ul></li><li style="background-color: rgba(0, 0, 0, 0)">流水线协作模式：
<ul><li style="background-color: rgba(0, 0, 0, 0)">适用场景：强依赖的序列化任务</li><li style="background-color: rgba(0, 0, 0, 0)">优势：专业化分工、质量控制容易</li><li style="background-color: rgba(0, 0, 0, 0)">劣势：瓶颈效应、整体效率受限于最慢环节</li></ul></li><li style="background-color: rgba(0, 0, 0, 0)">市场竞标模式：
<ul><li style="background-color: rgba(0, 0, 0, 0)">适用场景：资源竞争激烈的环境</li><li style="background-color: rgba(0, 0, 0, 0)">优势：资源优化配置、激励相容</li><li style="background-color: rgba(0, 0, 0, 0)">劣势：协商成本高、可能产生投机行为</li></ul></li><li style="background-color: rgba(0, 0, 0, 0)">对等协作模式：
<ul><li style="background-color: rgba(0, 0, 0, 0)">适用场景：平等参与者间的弱依赖协作</li><li style="background-color: rgba(0, 0, 0, 0)">优势：鲁棒性强、灵活性高</li><li style="background-color: rgba(0, 0, 0, 0)">劣势：协调复杂、决策效率较低</li></ul></li></ul><h3 style="background-color: rgba(0, 0, 0, 0)">3. 主从协作模式流程图</h3><p class="img-center"><img alt="" height="1013" src="https://i-blog.csdnimg.cn/direct/9870916977e44476b8f32d36e4121033.png" width="750"></p><p><br><strong>流程说明：</strong></p><p><strong>任务分解过程：</strong></p><ul><li>主智能体分析任务的复杂性和结构特征</li><li>识别子任务间的数据依赖和执行顺序约束</li><li>选择基于功能、数据或流程的分解策略</li><li>制定详细的子任务规格说明书</li></ul><p><strong>任务分配机制：</strong></p><ul><li>基于从智能体的能力专长进行任务匹配</li><li>考虑当前负载均衡和资源可用性</li><li>建立任务接受确认机制</li><li>设定任务执行的约束条件和质量标准</li></ul><p><strong>并行执行管理：</strong></p><ul><li>各从智能体独立执行分配的子任务</li><li>处理任务执行过程中的资源申请和释放</li><li>监控子任务执行进度和资源消耗</li><li>管理子任务间的数据交换和同步</li></ul><p><strong>结果收集与整合：</strong></p><ul><li>主智能体按预定协议收集子任务结果</li><li>验证子结果的完整性和一致性</li><li>处理可能的结果冲突和不匹配</li><li>采用合适的算法融合各子结果</li><li>生成最终的协作输出</li></ul><p><strong>反馈与学习：</strong></p><ul><li>向所有参与者反馈最终结果</li><li>记录协作过程中的关键指标</li><li>更新各智能体的性能评价</li><li>总结经验用于未来协作优化</li></ul><h3 style="background-color: rgba(0, 0, 0, 0)">4. 任务分解与分配流程</h3><p class="img-center"><img alt="" height="1200" src="https://i-blog.csdnimg.cn/direct/ebae14ec0c33469eaf67902e440c03a2.png" width="658"></p><p><strong>流程说明：</strong></p><p><strong>任务接收与分析：</strong></p><ul><li>详细理解任务需求和目标</li><li>识别任务的技术约束和业务约束</li><li>评估任务的规模、复杂性和紧急程度</li><li>确定任务的优先级和质量要求</li></ul><p><strong>分解决策过程：</strong></p><ul><li>基于任务特性决定是否需要分解</li><li>考虑团队的执行能力和资源状况</li><li>评估分解带来的管理和协调成本</li><li>权衡分解粒度与执行效率的平衡</li></ul><p><strong>分解策略选择：</strong></p><ul><li><strong>功能分解：</strong><ul><li>按不同的功能模块划分任务</li><li>每个模块具有明确的功能边界</li><li>定义清晰的模块接口和交互协议</li><li>适合功能独立性强的复杂系统</li></ul></li><li><strong>数据分解：</strong><ul><li>将大数据集划分为可并行处理的子集</li><li>各子任务处理不同的数据分区</li><li>需要设计有效的结果合并算法</li><li>适合数据密集型计算任务</li></ul></li><li><strong>时间分解：</strong><ul><li>按时间顺序将任务划分为多个阶段</li><li>每个阶段产生可验证的中间结果</li><li>建立阶段间的交付和验收机制</li><li>适合具有明显阶段性特征的项目</li></ul></li></ul><p><strong>子任务规格制定：</strong></p><ul><li>明确每个子任务的目标和交付物</li><li>定义输入输出格式和数据标准</li><li>设定质量验收标准和完成条件</li><li>建立进度监控和报告机制</li></ul><p><strong>智能体能力匹配：</strong></p><ul><li>评估各智能体的技术能力和专业领域</li><li>考虑智能体的当前工作负载和可用性</li><li>分析智能体的历史表现和信誉评级</li><li>优化整体系统的负载均衡</li></ul><p><strong>任务分配与依赖管理：</strong></p><ul><li>基于能力匹配结果分配子任务</li><li>建立任务间的依赖关系图</li><li>识别关键路径和瓶颈环节</li><li>制定风险缓解和应急计划</li></ul><h3 style="background-color: rgba(0, 0, 0, 0)">5. 智能体能力匹配矩阵</h3><p><img alt="" height="904" src="https://i-blog.csdnimg.cn/direct/856ece927c904c82918d442f0b2e9743.png" width="3226"><br><strong>详细说明：</strong></p><p><strong>任务需求维度分析：</strong></p><ul><li><strong>计算能力：</strong><ul><li>处理复杂算法的能力</li><li>并行计算和分布式处理支持</li><li>实时计算性能要求</li><li>计算精度和准确性标准</li></ul></li><li><strong>存储能力：</strong><ul><li>数据存储容量需求</li><li>存储访问速度和IO性能</li><li>数据持久化和备份要求</li><li>存储安全性和隐私保护</li></ul></li><li><strong>专业知识：</strong><ul><li>领域特定的知识库</li><li>专业算法和模型掌握</li><li>行业标准和规范理解</li><li>问题诊断和解决经验</li></ul></li><li><strong>响应速度：</strong><ul><li>任务响应的实时性要求</li><li>处理延迟的容忍度</li><li>并发请求处理能力</li><li>系统吞吐量性能</li></ul></li><li><strong>可靠性：</strong><ul><li>系统可用性和稳定性</li><li>故障恢复和容错能力</li><li>服务质量保证级别</li><li>安全性和可信度评级</li></ul></li></ul><p><strong>智能体能力评估：</strong></p><ul><li><strong>智能体A：</strong><ul><li>优势：强大的计算能力、深厚的专业知识、高可靠性</li><li>适用场景：复杂计算任务、专业问题解决、关键业务处理</li><li>限制：可能成本较高、资源占用大</li></ul></li><li><strong>智能体B：</strong><ul><li>优势：大容量存储能力、专业领域知识</li><li>适用场景：数据密集型任务、知识检索和处理</li><li>限制：计算能力相对有限、响应速度中等</li></ul></li><li><strong>智能体C：</strong><ul><li>优势：高性能计算、快速响应能力</li><li>适用场景：实时计算、高吞吐量处理</li><li>限制：专业知识相对缺乏、可靠性中等</li></ul></li><li><strong>智能体D：</strong><ul><li>优势：均衡的存储和响应能力、高可靠性</li><li>适用场景：综合型任务、需要稳定服务的场景</li><li>限制：专业深度可能不足、计算能力有限</li></ul></li></ul><p><strong>匹配策略：</strong></p><ul><li>基于任务需求的优先级进行加权匹配</li><li>考虑智能体的当前负载和可用性</li><li>评估历史合作表现和信誉评级</li><li>优化整体系统的资源利用效率</li></ul><h2>七、Agent2Agent 框架通信示例</h2><p> 示例实现一个简单的Agent2Agent通信框架，展示了多智能体系统的基本工作原理。下面我将逐部分分解代码并说明其设计意图。</p><p><strong>1. 基础定义</strong></p><p><strong>1.1 消息类型枚举</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">class MessageType(Enum):
TASK_REQUEST = "task_request" # 任务请求
TASK_RESPONSE = "task_response" # 任务响应
GREETING = "greeting" # 问候消息</code></pre>
<ul><li>定义标准化的消息类型，确保智能体间通信的一致性</li><li>便于消息路由和处理逻辑的分发</li><li>支持系统扩展，可以轻松添加新的消息类型</li></ul><p><strong>1.2 消息数据结构</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">@dataclass
class A2AMessage:
message_id: str # 消息唯一ID
sender_id: str # 发送者ID
receiver_id: str # 接收者ID
message_type: MessageType # 消息类型
content: Dict # 消息内容
timestamp: float # 时间戳</code></pre>
<ul><li>提供统一的消息格式，包含必要的元数据</li><li>支持消息追踪和调试（message_id, timestamp）</li><li>灵活的内容结构，适应不同类型的通信需求</li><li>便于序列化和网络传输</li></ul><p><strong>2. 通信传输层：SimpleTransport类</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">class SimpleTransport:
def __init__(self):
self.agents: Dict = {} # 智能体注册表</code></pre>
<ul><li>模拟网络通信层，解耦智能体间的直接依赖</li><li>提供消息路由功能，根据接收者ID分发消息</li><li>支持异步消息传递，模拟真实网络环境</li><li>作为智能体注册中心，管理参与协作的智能体</li></ul><p><strong>2.1 注册智能体：</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">def register_agent(self, agent_id: str, message_handler: callable):
self.agents = message_handler</code></pre>
<ul><li>建立智能体ID到消息处理函数的映射</li><li>支持动态加入和退出协作系统</li></ul><p><strong>2.2 发送消息：</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">async def send_message(self, message: A2AMessage):
# 检查接收者是否存在
# 模拟网络延迟
# 调用接收者的消息处理函数</code></pre>
<ul><li>提供可靠的消息传递机制</li><li>包含基本的错误检查（接收者存在性验证）</li><li>模拟真实网络环境（延迟、异步处理）</li></ul><p><strong>3. 智能体基类：SimpleAgent类</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">class SimpleAgent:
def __init__(self, agent_id: str, name: str, transport: SimpleTransport):
self.agent_id = agent_id
self.name = name
self.transport = transport
self.conversation_history: List = []</code></pre>
<ul><li>提供智能体的共同属性和行为</li><li>封装消息发送和接收的基础功能</li><li>维护对话历史，支持调试和分析</li><li>实现模板方法模式，子类可以重写特定行为</li></ul><p><strong>3.1 消息发送：</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">async def send_message(self, receiver_id: str, message_type: MessageType, content: Dict):
# 创建标准化消息
# 记录到对话历史
# 通过传输层发送</code></pre>
<ul><li>封装消息创建过程，确保格式一致性</li><li>自动维护对话历史</li><li>提供简单的发送接口，隐藏底层复杂性</li></ul><p><strong>3.2 消息接收：</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">async def receive_message(self, message: A2AMessage):
# 记录接收到的消息
# 提供基础处理（子类可扩展）</code></pre>
<ul><li>作为消息处理的入口点</li><li>提供基础的日志记录功能</li><li>采用模板方法模式，支持子类定制化处理</li></ul><p><strong>4. 专用智能体实现：TaskManagerAgent任务管理器</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">class TaskManagerAgent(SimpleAgent):
def __init__(self, agent_id: str, name: str, transport: SimpleTransport):
super().__init__(agent_id, name, transport)
self.pending_tasks: Dict = {} # 待处理任务
self.completed_tasks: Dict = {} # 已完成任务</code></pre>
<ul><li>实现任务协调者的角色</li><li>负责任务生命周期管理（创建、分配、跟踪）</li><li>维护任务状态，支持进度监控</li><li>处理工作者的响应和结果收集</li></ul><p><strong>4.1 任务创建与分配：</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">async def create_task(self, task_description: str, worker_id: str):
# 生成唯一任务ID
# 保存任务信息到待处理列表
# 发送任务请求给指定工作者</code></pre>
<ul><li>封装任务分配逻辑</li><li>自动生成任务标识，便于跟踪</li><li>维护任务状态，确保不丢失</li></ul><p><strong>4.2 消息处理扩展：</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">async def receive_message(self, message: A2AMessage):
await super().receive_message(message) # 基础处理
if message.message_type == MessageType.TASK_RESPONSE:
await self.handle_task_response(message)
elif message.message_type == MessageType.GREETING:
await self.handle_greeting(message)</code></pre>
<ul><li>继承基础消息处理功能</li><li>根据消息类型分发到专门的处理方法</li><li>展示多态性的应用</li></ul><p><strong>5. WorkerAgent类：工作者智能体</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">class WorkerAgent(SimpleAgent):
def __init__(self, agent_id: str, name: str, transport: SimpleTransport, skills: List):
super().__init__(agent_id, name, transport)
self.skills = skills
self.tasks_completed = 0</code></pre>
<ul><li>实现任务执行者的角色</li><li>封装特定领域的能力和技能</li><li>处理任务请求并生成结果</li><li>维护工作统计和性能指标</li></ul><p><strong>5.1 任务处理：</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">async def handle_task_request(self, message: A2AMessage):
# 解析任务信息
# 模拟任务执行（异步等待）
# 生成任务结果
# 发送任务完成响应</code></pre>
<ul><li>展示异步任务执行模式</li><li>模拟真实的工作负载和处理时间</li><li>封装结果生成逻辑</li></ul><p><strong>5.2 技能匹配处理：</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">def process_task(self, description: str) -> str:
if "计算" in description or "数学" in description:
return f"计算完成: 42 (使用 {self.skills} 技能)"
# 其他技能分支...</code></pre>
<ul><li>根据任务描述匹配相应的处理逻辑</li><li>展示基于内容的决策机制</li><li>集成智能体的技能信息到结果中</li></ul><p><strong>6. 程序启动运行 </strong></p><p><strong>6.1 基础演示</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">async def demo_agent_communication():
# 创建传输层和智能体实例
# 演示问候消息交换
# 演示任务分配和执行流程
# 展示统计信息</code></pre>
<ul><li>展示最基本的A2A通信场景</li><li>验证消息传递的正确性</li><li>演示任务生命周期的完整流程</li><li>提供可视化的执行跟踪</li></ul><p><strong>6.2 高级演示</strong></p>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">async def advanced_demo():
# 创建多个专业智能体
# 并行分配不同类型任务
# 展示跨领域协作</code></pre>
<ul><li>展示多智能体协作的复杂性</li><li>演示基于技能的智能体专业化</li><li>验证系统的可扩展性</li><li>展示并行任务处理能力</li></ul><p><strong>运行结果：</strong></p><p><span style="color: rgba(123, 127, 130, 1)">==================================================<br>
Agent2Agent 通信演示开始<br>
==================================================<br>
传输层: 智能体 manager_001 已注册<br>
智能体初始化: 任务管理器 (ID: manager_001)<br>
传输层: 智能体 worker_001 已注册<br>
智能体初始化: AI工作者 (ID: worker_001)</span></p><p><span style="color: rgba(123, 127, 130, 1)">============================== 场景1: 简单问候 ==============================<br>
传输层: manager_001 -> worker_001 <br>
AI工作者 收到来自 manager_001 的消息: greeting</span></p><p><span style="color: rgba(123, 127, 130, 1)">============================== 场景2: 任务分配 ==============================</span></p><p><span style="color: rgba(123, 127, 130, 1)">--- 任务 1 ---<br>
任务管理器 创建任务: 计算复杂的数学公式<br>
传输层: manager_001 -> worker_001 <br>
AI工作者 收到来自 manager_001 的消息: task_request<br>
AI工作者 开始处理任务: 计算复杂的数学公式<br>
传输层: worker_001 -> manager_001 <br>
任务管理器 收到来自 worker_001 的消息: task_response<br>
任务管理器 收到任务完成通知: task_1759064250<br>
结果: 计算完成: 42 (使用 ['Python编程', '数据分析', '机器学习'] 技能)<br>
AI工作者 完成任务: 计算复杂的数学公式</span></p><p><span style="color: rgba(123, 127, 130, 1)">--- 任务 2 ---<br>
任务管理器 创建任务: 分析销售数据趋势<br>
传输层: manager_001 -> worker_001 <br>
AI工作者 收到来自 manager_001 的消息: task_request<br>
AI工作者 开始处理任务: 分析销售数据趋势<br>
传输层: worker_001 -> manager_001 <br>
任务管理器 收到来自 worker_001 的消息: task_response<br>
任务管理器 收到任务完成通知: task_1759064253<br>
结果: 分析报告: 数据趋势良好<br>
AI工作者 完成任务: 分析销售数据趋势</span></p><p><span style="color: rgba(123, 127, 130, 1)">--- 任务 3 ---<br>
任务管理器 创建任务: 编写Python数据处理代码 <br>
传输层: manager_001 -> worker_001 <br>
AI工作者 收到来自 manager_001 的消息: task_request<br>
AI工作者 开始处理任务: 编写Python数据处理代码<br>
传输层: worker_001 -> manager_001 <br>
任务管理器 收到来自 worker_001 的消息: task_response<br>
任务管理器 收到任务完成通知: task_1759064256<br>
结果: 代码已编写: def solution(): return 'Hello World'<br>
AI工作者 完成任务: 编写Python数据处理代码</span></p><p><span style="color: rgba(123, 127, 130, 1)">--- 任务 4 ---<br>
任务管理器 创建任务: 生成月度报告摘要<br>
传输层: manager_001 -> worker_001 <br>
AI工作者 收到来自 manager_001 的消息: task_request<br>
AI工作者 开始处理任务: 生成月度报告摘要<br>
传输层: worker_001 -> manager_001 <br>
任务管理器 收到来自 worker_001 的消息: task_response<br>
任务管理器 收到任务完成通知: task_1759064259<br>
结果: 任务 '生成月度报告摘要' 已处理完成<br>
AI工作者 完成任务: 生成月度报告摘要</span></p><p><span style="color: rgba(123, 127, 130, 1)">============================== 结果统计 ==============================<br>
任务管理器统计:<br>
总消息数: 9<br>
完成任务: 4<br>
待处理任务: 0</span></p><p><span style="color: rgba(123, 127, 130, 1)"> 工作者统计:<br>
总消息数: 9<br>
完成任务: 4</span></p><p><span style="color: rgba(123, 127, 130, 1)">==================================================<br>
Agent2Agent 通信演示完成!<br>
==================================================</span></p><p><span style="color: rgba(123, 127, 130, 1)">******************** 高级演示 ********************<br>
传输层: 智能体 manager_1 已注册<br>
智能体初始化: 高级管理器 (ID: manager_1)<br>
传输层: 智能体 worker_1 已注册<br>
智能体初始化: 开发专家 (ID: worker_1)<br>
传输层: 智能体 worker_2 已注册<br>
智能体初始化: 数据分析师 (ID: worker_2)<br>
传输层: 智能体 worker_3 已注册<br>
智能体初始化: 技术作家 (ID: worker_3)<br>
高级管理器 创建任务: 开发新的数据处理模块<br>
传输层: manager_1 -> worker_1 <br>
高级管理器 创建任务: 分析用户行为数据<br>
传输层: manager_1 -> worker_2 <br>
高级管理器 创建任务: 编写项目文档<br>
传输层: manager_1 -> worker_3 <br>
高级管理器 创建任务: 优化算法性能<br>
传输层: manager_1 -> worker_1 <br>
开发专家 收到来自 manager_1 的消息: task_request<br>
开发专家 开始处理任务: 开发新的数据处理模块<br>
技术作家 收到来自 manager_1 的消息: task_request<br>
传输层: manager_1 -> worker_1 <br>
开发专家 收到来自 manager_1 的消息: task_request<br>
开发专家 开始处理任务: 开发新的数据处理模块<br>
技术作家 收到来自 manager_1 的消息: task_request<br>
开发专家 收到来自 manager_1 的消息: task_request<br>
开发专家 开始处理任务: 开发新的数据处理模块<br>
技术作家 收到来自 manager_1 的消息: task_request<br>
技术作家 收到来自 manager_1 的消息: task_request<br>
技术作家 开始处理任务: 编写项目文档<br>
数据分析师 收到来自 manager_1 的消息: task_request<br>
数据分析师 收到来自 manager_1 的消息: task_request<br>
数据分析师 开始处理任务: 分析用户行为数据<br>
数据分析师 开始处理任务: 分析用户行为数据<br>
开发专家 收到来自 manager_1 的消息: task_request<br>
开发专家 开始处理任务: 优化算法性能<br>
传输层: worker_1 -> manager_1 <br>
传输层: worker_2 -> manager_1 <br>
传输层: worker_3 -> manager_1 <br>
传输层: worker_1 -> manager_1 <br>
高级管理器 收到来自 worker_2 的消息: task_response<br>
高级管理器 收到任务完成通知: task_1759064261<br>
结果: 分析报告: 数据趋势良好<br>
数据分析师 完成任务: 分析用户行为数据<br>
高级管理器 收到来自 worker_3 的消息: task_response<br>
技术作家 完成任务: 编写项目文档<br>
高级管理器 收到来自 worker_1 的消息: task_response<br>
开发专家 完成任务: 开发新的数据处理模块<br>
高级管理器 收到来自 worker_1 的消息: task_response<br>
开发专家 完成任务: 优化算法性能</span></p><p><span style="color: rgba(123, 127, 130, 1)"> 高级演示完成!<br>
管理器处理了 8 条消息<br>
完成了 1 个跨领域任务</span></p><h2><span style="color: rgba(13, 0, 22, 1)">八、应用场景</span></h2><ul><li><span style="color: rgba(13, 0, 22, 1)">复杂问题求解：多个专业智能体协作解决单一智能体难以处理的复杂问题</span></li><li><span style="color: rgba(13, 0, 22, 1)">分布式系统管理：智能体协同管理分布式资源和服务</span></li><li><span style="color: rgba(13, 0, 22, 1)">多模态AI系统：不同模态的AI智能体（文本、图像、语音）协同工作</span></li><li><span style="color: rgba(13, 0, 22, 1)">自动化工作流：智能体协作完成端到端的自动化流程</span></li></ul><h2><span style="color: rgba(13, 0, 22, 1)">九、总结</span></h2><p><span style="color: rgba(13, 0, 22, 1)"> Agent2Agent框架为构建智能体间通信与协作系统提供了坚实的基础。通过标准化的消息格式、灵活的传输层和丰富的协作模式，开发者可以快速构建能够解决复杂问题的多智能体系统。随着人工智能技术的不断发展，A2A框架将在构建更智能、更自治的AI系统中发挥重要作用。</span></p><p><span style="color: rgba(13, 0, 22, 1)"> 该框架具有良好的扩展性，可以根据具体应用场景添加新的通信协议、协作算法和容错机制，满足不同领域的需求。</span></p><h2><span style="color: rgba(13, 0, 22, 1)">附录：示例完整代码</span></h2>
<pre style="white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important"><code class="language-python">import asyncio
import time
import uuid
from dataclasses import dataclass
from typing import Dict, Any, List, Optional
from enum import Enum
# ==================== 基础定义 ====================
class MessageType(Enum):
"""消息类型枚举"""
TASK_REQUEST = "task_request" # 任务请求
TASK_RESPONSE = "task_response" # 任务响应
GREETING = "greeting" # 问候消息
@dataclass
class A2AMessage:
"""
A2A消息类
定义智能体间通信的标准消息格式
"""
message_id: str # 消息唯一ID
sender_id: str # 发送者ID
receiver_id: str # 接收者ID
message_type: MessageType # 消息类型
content: Dict # 消息内容
timestamp: float # 时间戳
def to_dict(self) -> Dict:
"""将消息转换为字典格式，便于传输"""
return {
"message_id": self.message_id,
"sender_id": self.sender_id,
"receiver_id": self.receiver_id,
"message_type": self.message_type.value,
"content": self.content,
"timestamp": self.timestamp
}
class SimpleTransport:
"""
简单传输层
负责智能体间的消息传递（模拟网络通信）
"""
def __init__(self):
# 存储注册的智能体和它们的消息处理函数
self.agents: Dict = {}
def register_agent(self, agent_id: str, message_handler: callable):
"""
注册智能体到传输层
Args:
agent_id: 智能体ID
message_handler: 消息处理函数
"""
self.agents = message_handler
print(f" 传输层: 智能体 {agent_id} 已注册")
async def send_message(self, message: A2AMessage):
"""
发送消息到目标智能体
Args:
message: 要发送的消息
"""
print(f" 传输层: {message.sender_id} -> {message.receiver_id} "
f"[{message.message_type.value}]")
# 检查接收者是否存在
if message.receiver_id not in self.agents:
print(f" 错误: 智能体 {message.receiver_id} 未注册")
return
# 获取接收者的消息处理函数
receiver_handler = self.agents
# 模拟网络延迟
await asyncio.sleep(0.1)
# 将消息传递给接收者
await receiver_handler(message)
# ==================== 智能体定义 ====================
class SimpleAgent:
"""
简单智能体基类
所有智能体的共同属性和行为
"""
def __init__(self, agent_id: str, name: str, transport: SimpleTransport):
"""
初始化智能体
Args:
agent_id: 智能体唯一标识
name: 智能体名称
transport: 传输层实例
"""
self.agent_id = agent_id
self.name = name
self.transport = transport
self.conversation_history: List = []
# 向传输层注册自己
self.transport.register_agent(self.agent_id, self.receive_message)
print(f" 智能体初始化: {self.name} (ID: {self.agent_id})")
async def send_message(self, receiver_id: str, message_type: MessageType, content: Dict):
"""
发送消息给其他智能体
Args:
receiver_id: 接收者ID
message_type: 消息类型
content: 消息内容
"""
# 创建消息对象
message = A2AMessage(
message_id=str(uuid.uuid4()),# 生成唯一ID
sender_id=self.agent_id,
receiver_id=receiver_id,
message_type=message_type,
content=content,
timestamp=time.time()
)
# 保存到对话历史
self.conversation_history.append(message)
# 通过传输层发送消息
await self.transport.send_message(message)
async def receive_message(self, message: A2AMessage):
"""
接收消息的通用处理（子类可以重写）
Args:
message: 接收到的消息
"""
print(f" {self.name} 收到来自 {message.sender_id} 的消息: {message.message_type.value}")
# 保存到对话历史
self.conversation_history.append(message)
def get_conversation_summary(self):
"""获取对话摘要"""
return f"{self.name} 进行了 {len(self.conversation_history)} 次消息交换"
class TaskManagerAgent(SimpleAgent):
"""
任务管理器智能体
负责创建和分配任务
"""
def __init__(self, agent_id: str, name: str, transport: SimpleTransport):
super().__init__(agent_id, name, transport)
self.pending_tasks: Dict = {}# 待处理任务
self.completed_tasks: Dict = {} # 已完成任务
async def create_task(self, task_description: str, worker_id: str):
"""
创建新任务并分配给工作者
Args:
task_description: 任务描述
worker_id: 工作者智能体ID
"""
task_id = f"task_{int(time.time())}"
print(f" {self.name} 创建任务: {task_description}")
# 保存任务信息
self.pending_tasks = {
"description": task_description,
"worker": worker_id,
"status": "assigned",
"created_time": time.time()
}
# 发送任务请求
await self.send_message(
receiver_id=worker_id,
message_type=MessageType.TASK_REQUEST,
content={
"task_id": task_id,
"description": task_description,
"priority": "normal"
}
)
async def receive_message(self, message: A2AMessage):
"""
处理接收到的消息（重写父类方法）
"""
await super().receive_message(message)# 调用父类基础处理
if message.message_type == MessageType.TASK_RESPONSE:
# 处理任务响应
await self.handle_task_response(message)
elif message.message_type == MessageType.GREETING:
# 处理问候消息
await self.handle_greeting(message)
async def handle_task_response(self, message: A2AMessage):
"""
处理任务完成响应
Args:
message: 任务响应消息
"""
task_id = message.content.get("task_id")
result = message.content.get("result")
status = message.content.get("status")
if task_id in self.pending_tasks:
if status == "completed":
# 任务完成
self.completed_tasks = {
**self.pending_tasks,
"result": result,
"completion_time": time.time()
}
del self.pending_tasks
print(f" {self.name} 收到任务完成通知: {task_id}")
print(f" 结果: {result}")
else:
print(f" 任务 {task_id} 失败: {result}")
async def handle_greeting(self, message: A2AMessage):
"""
处理问候消息
Args:
message: 问候消息
"""
greeting_text = message.content.get("text", "")
print(f" {self.name} 收到问候: {greeting_text}")
class WorkerAgent(SimpleAgent):
"""
工作者智能体
负责执行具体任务
"""
def __init__(self, agent_id: str, name: str, transport: SimpleTransport, skills: List):
"""
初始化工作者智能体
Args:
skills: 技能列表
"""
super().__init__(agent_id, name, transport)
self.skills = skills
self.tasks_completed = 0
async def receive_message(self, message: A2AMessage):
"""
处理接收到的消息
"""
await super().receive_message(message)
if message.message_type == MessageType.TASK_REQUEST:
# 处理任务请求
await self.handle_task_request(message)
async def handle_task_request(self, message: A2AMessage):
"""
处理任务请求并执行任务
Args:
message: 任务请求消息
"""
task_id = message.content.get("task_id")
description = message.content.get("description")
print(f" {self.name} 开始处理任务: {description}")
# 模拟任务执行时间
await asyncio.sleep(1)
# 生成任务结果（根据任务描述和技能）
result = self.process_task(description)
# 发送任务完成响应
await self.send_message(
receiver_id=message.sender_id,
message_type=MessageType.TASK_RESPONSE,
content={
"task_id": task_id,
"status": "completed",
"result": result,
"worker_skills_used": self.skills
}
)
self.tasks_completed += 1
print(f" {self.name} 完成任务: {description}")
def process_task(self, description: str) -> str:
"""
处理具体任务（根据描述生成结果）
Args:
description: 任务描述
Returns:
任务结果
"""
if "计算" in description or "数学" in description:
return f"计算完成: 42 (使用 {self.skills} 技能)"
elif "分析" in description:
return f"分析报告: 数据趋势良好"
elif "编写" in description or "代码" in description:
return f"代码已编写: def solution(): return 'Hello World'"
else:
return f"任务 '{description}' 已处理完成"
# ==================== 演示程序 ====================
async def demo_agent_communication():
"""
演示智能体间通信的主函数
"""
print("=" * 50)
print(" Agent2Agent 通信演示开始")
print("=" * 50)
# 创建传输层实例
transport = SimpleTransport()
# 创建智能体实例
task_manager = TaskManagerAgent(
agent_id="manager_001",
name="任务管理器",
transport=transport
)
worker_agent = WorkerAgent(
agent_id="worker_001",
name="AI工作者",
transport=transport,
skills=["Python编程", "数据分析", "机器学习"]
)
# 等待一下让注册完成
await asyncio.sleep(0.1)
print("\n" + "="*30 + " 场景1: 简单问候 " + "="*30)
# 场景1: 简单问候
await task_manager.send_message(
receiver_id=worker_agent.agent_id,
message_type=MessageType.GREETING,
content={"text": "你好，我是任务管理器！"}
)
await asyncio.sleep(0.5)# 等待消息处理
print("\n" + "="*30 + " 场景2: 任务分配 " + "="*30)
# 场景2: 任务分配和执行
tasks = [
"计算复杂的数学公式",
"分析销售数据趋势",
"编写Python数据处理代码",
"生成月度报告摘要"
]
for i, task_desc in enumerate(tasks, 1):
print(f"\n--- 任务 {i} ---")
await task_manager.create_task(task_desc, worker_agent.agent_id)
await asyncio.sleep(1.5)# 等待任务完成
print("\n" + "="*30 + " 结果统计 " + "="*30)
# 显示结果统计
print(f" 任务管理器统计:")
print(f" 总消息数: {len(task_manager.conversation_history)}")
print(f" 完成任务: {len(task_manager.completed_tasks)}")
print(f" 待处理任务: {len(task_manager.pending_tasks)}")
print(f"\n 工作者统计:")
print(f" 总消息数: {len(worker_agent.conversation_history)}")
print(f" 完成任务: {worker_agent.tasks_completed}")
print("\n" + "="*50)
print(" Agent2Agent 通信演示完成!")
print("=" * 50)
async def advanced_demo():
"""
高级演示：多个智能体协作
"""
print("\n" + "*"*20 + " 高级演示 " + "*"*20)
transport = SimpleTransport()
# 创建多个智能体
manager = TaskManagerAgent("manager_1", "高级管理器", transport)
# 创建具有不同技能的多个工作者
developer = WorkerAgent("worker_1", "开发专家", transport,
["Python", "Java", "系统设计"])
analyst = WorkerAgent("worker_2", "数据分析师", transport,
["统计分析", "数据可视化", "机器学习"])
writer = WorkerAgent("worker_3", "技术作家", transport,
["文档编写", "技术说明", "API文档"])
await asyncio.sleep(0.1)
# 并行分配任务给不同的工作者
tasks = [
("开发新的数据处理模块", developer.agent_id),
("分析用户行为数据", analyst.agent_id),
("编写项目文档", writer.agent_id),
("优化算法性能", developer.agent_id),
]
# 同时启动所有任务
task_coroutines = [
manager.create_task(desc, worker_id)
for desc, worker_id in tasks
]
await asyncio.gather(*task_coroutines)
# 等待所有任务完成
await asyncio.sleep(3)
print(f"\n 高级演示完成!")
print(f" 管理器处理了 {len(manager.conversation_history)} 条消息")
print(f" 完成了 {len(manager.completed_tasks)} 个跨领域任务")
if __name__ == "__main__":
# 运行基础演示
asyncio.run(demo_agent_communication())
# 运行高级演示
asyncio.run(advanced_demo())</code></pre>
</div><br><br>
来源：https://www.cnblogs.com/tlnshuju/p/19122406
頁:
[1]