金融行业 Multi-Agent 落地实践：风险控制与客户服务的双重赋能

作者简介：15年金融科技架构经验，曾主导3家头部股份制银行/消费金融公司的 AI 风控、客服系统升级，主导设计的 Agent 协作系统日均处理交易超1000万笔，客户服务满意度提升32%。坚信“复杂金融逻辑 = 模块化Agent + 安全可信协作机制”。

核心概念与前置知识

核心概念

什么是 Multi-Agent System (MAS)？

Multi-Agent System（多智能体系统）是一种由多个自主、交互、协作/竞争的智能体（Agent）组成的分布式人工智能系统。每个智能体具有以下核心属性：

1. 自主性：无需外部直接干预即可根据环境和内部状态做出决策
2. 反应性：能感知环境（包括其他Agent的输出）并及时响应
3. 主动性：能主动设定并追求目标（非仅被动响应）
4. 社会性：能通过协议、消息与其他Agent进行协作、竞争或协商

金融级 Multi-Agent System 的特殊性

金融行业的 MAS 与通用领域（如游戏、自动驾驶）的 MAS 有本质区别，必须满足监管合规、安全可信、低延迟、高可用、可解释性五大核心约束：

双重赋能的核心定位

本文聚焦于金融行业的两大刚需场景落地：

1. 风险控制（反欺诈、信用评估、反洗钱、合规审计）：通过Agent协作覆盖风险识别、评估、决策、处置的全链路，解决传统风控“单一模型能力有限、规则引擎僵化、响应慢、可解释性弱”的痛点
2. 客户服务（智能客服、财富管理顾问、贷款咨询）：通过Agent协作覆盖咨询、需求分析、方案推荐、后续跟进的全旅程，解决传统客服“专业度不足、个性化程度低、无法处理复杂多轮/跨场景问题”的痛点

问题背景与行业痛点

金融行业数字化转型面临的“最后一公里”

过去10年，金融行业的数字化转型经历了**线上化（移动APP、网上银行）→ 智能化（单一AI模型，如人脸识别、OCR识别、风控评分卡）→ 平台化（大数据平台、AI中台）**三个阶段，但仍存在以下“最后一公里”的痛点：

痛点1：风控场景的“能力孤岛”与“决策滞后”

传统风控系统通常由规则引擎 + 单一/少数孤立模型组成，存在以下问题：

1. 能力孤岛严重：反欺诈模型、信用评估模型、反洗钱模型、合规审计模型由不同团队开发，部署在不同的系统中，数据无法实时共享，模型之间无法协作
2. 规则引擎僵化：规则由风控专家手工编写，更新周期长达数周甚至数月，无法应对新型欺诈手段的快速迭代
3. 决策维度单一：仅依赖结构化数据（如征信报告、交易流水），无法有效利用非结构化数据（如客户语音、聊天记录、社交媒体信息）
4. 决策全链路不可控：黑箱模型的输出无法提供全链路的解释，不符合监管要求；规则引擎与模型的输出冲突时，无法自动协商最优解
5. 处置效率低：风险识别后，需要人工介入进行处置，响应时间长达数小时甚至数天，容易造成损失扩大

痛点2：客户服务场景的“专业缺失”与“体验割裂”

传统客户服务系统通常由IVR（交互式语音应答） + 人工客服 + 单一FAQ（常见问题解答）模型组成，存在以下问题：

1. 专业度不足：FAQ模型仅能回答简单、标准化的问题，无法处理复杂的多轮/跨场景问题（如“我想申请一笔30万的装修贷款，同时把我的股票账户里的50万转到货币基金里，还要预约明天上午10点的理财经理线下咨询”）
2. 个性化程度低：所有客户收到的回复/方案都是标准化的，无法根据客户的历史行为、风险偏好、财务状况提供个性化的服务
3. 体验割裂：IVR、人工客服、手机APP、网上银行之间的数据无法实时共享，客户需要重复输入信息，体验极差
4. 人工成本高：复杂问题必须由人工客服处理，导致人工成本居高不下；同时，人工客服的专业水平参差不齐，服务质量难以保证
5. 全旅程覆盖不足：仅能覆盖“咨询-解答”的单一环节，无法覆盖“需求挖掘-方案推荐-交易协助-后续跟进-风险预警”的全客户旅程

政策驱动与技术成熟：Multi-Agent 落地的黄金时代

政策驱动

近年来，国家出台了一系列政策推动金融科技的发展，尤其是AI技术在金融领域的应用：

1. 《金融科技发展规划（2022-2025年）》：明确提出要“深化人工智能技术在金融领域的应用，构建智能风控、智能客服、智能投顾等系统，提升金融服务的效率和质量”
2. 《关于加快推进人工智能与实体经济深度融合的指导意见》：明确提出要“在金融领域推广应用多智能体系统、联邦学习等技术，提升金融风险防控能力和客户服务水平”
3. 《个人信息保护法》《金融数据安全管理办法》：对金融数据的隐私保护和安全管理提出了严格的要求，为金融级 MAS 的数据共享机制提供了法律依据

技术成熟

近年来，以下技术的成熟为金融级 MAS 的落地提供了技术支撑：

1. 大语言模型（LLM）：如 GPT-4、Claude 3、文心一言、通义千问等，具备强大的自然语言理解、生成、推理能力，是金融级 MAS 的“大脑”
2. 向量数据库：如 Pinecone、Weaviate、Milvus、Chroma 等，能高效存储和检索非结构化数据的向量表示，是金融级 MAS 的“记忆库”
3. 联邦学习：如 FedAvg、FedProx、FedBN 等，能在不共享原始数据的情况下实现跨机构/跨系统的模型协作，是金融级 MAS 的“数据共享桥梁”
4. 区块链：如 Hyperledger Fabric、FISCO BCOS 等，能实现不可篡改的Agent协作日志和决策记录，是金融级 MAS 的“审计账本”
5. 云原生技术：如 Docker、Kubernetes、Istio 等，能实现金融级 MAS 的高可用、低延迟、弹性伸缩，是金融级 MAS 的“基础设施”

金融级 Multi-Agent System 的核心架构设计

核心设计原则

金融级 MAS 的架构设计必须严格遵循以下8大核心原则：

1. 模块化设计：每个Agent的职责单一、明确，便于开发、测试、部署、维护和升级
2. 安全可信设计：内置数据隐私保护、身份认证、权限控制、审计日志、异常检测等机制
3. 低延迟高可用设计：采用分布式架构、微服务架构、缓存机制、负载均衡等技术
4. 可解释性设计：每个Agent的决策过程都必须可追溯、可解释，决策全链路必须有审计日志
5. 协作机制设计：采用明确的协作协议、消息格式、任务分配机制和冲突解决机制
6. 可扩展性设计：能方便地添加新的Agent、新的协作模式、新的数据源和新的模型
7. 兼容性设计：能与现有的金融系统（如核心银行系统、CRM系统、风控系统、客服系统）无缝集成
8. 监管合规设计：所有功能都必须符合监管要求，如《个人信息保护法》《金融数据安全管理办法》《反洗钱法》等

分层架构设计

基于以上核心设计原则，本文提出了一种金融级 Multi-Agent 分层协作架构，共分为5层：

第1层：基础设施层（Infrastructure Layer）

基础设施层是金融级 MAS 的“底座”，提供以下核心功能：

1. 计算资源：云服务器、GPU服务器、边缘计算设备
2. 存储资源：关系型数据库（如 MySQL、PostgreSQL、Oracle）、非关系型数据库（如 MongoDB、Redis）、向量数据库（如 Milvus、Weaviate）、区块链账本（如 Hyperledger Fabric、FISCO BCOS）
3. 网络资源：负载均衡、CDN、VPN、加密通信（如 TLS 1.3）
4. 容器编排：Docker、Kubernetes、Istio
5. 监控告警：Prometheus、Grafana、ELK Stack（Elasticsearch、Logstash、Kibana）、Jaeger

第2层：数据与知识层（Data & Knowledge Layer）

数据与知识层是金融级 MAS 的“记忆库”和“知识库”，提供以下核心功能：

1. 数据采集与清洗：从现有的金融系统（如核心银行系统、CRM系统、风控系统、客服系统）、外部数据源（如央行征信、百行征信、芝麻信用、社交媒体、新闻资讯）采集数据，并进行清洗、去重、标准化
2. 数据存储与管理：根据数据的类型（结构化、半结构化、非结构化）和敏感度（公开、内部、机密、绝密）进行分级存储和管理，内置数据加密、数据脱敏、数据访问控制等机制
3. 知识图谱构建：基于采集到的数据，构建金融领域的知识图谱（如客户知识图谱、产品知识图谱、风险知识图谱、市场知识图谱）
4. 向量嵌入与检索：使用大语言模型（如 GPT-4、通义千问）或专用的嵌入模型（如 sentence-transformers、text2vec）将非结构化数据（如客户语音、聊天记录、新闻资讯）转换为向量表示，并存储在向量数据库中，提供高效的相似度检索功能
5. 联邦学习平台：提供跨机构/跨系统的模型协作功能，能在不共享原始数据的情况下实现模型的联合训练和推理

第3层：Agent 层（Agent Layer）

Agent 层是金融级 MAS 的“核心执行单元”，由多个自主、交互、协作的专业Agent组成。根据职责的不同，Agent 可以分为以下4大类：

3.1 感知类 Agent（Perception Agents）

感知类 Agent 的职责是感知环境（包括外部环境和内部环境）的变化，并将感知到的信息转换为标准格式的消息，发送给决策类 Agent 或协作协调类 Agent。常见的感知类 Agent 包括：

1. 客户交互感知 Agent：负责感知客户的交互行为，如手机APP的点击、滑动、搜索，聊天记录的文本、语音、图片、视频，IVR的语音输入等
2. 交易感知 Agent：负责感知交易的发生，如转账、汇款、支付、贷款申请、理财购买等
3. 市场感知 Agent：负责感知市场的变化，如股票价格、基金净值、汇率、利率、新闻资讯、政策法规等
4. 风险感知 Agent：负责感知风险的发生，如客户账户的异常登录、异常交易、客户的信用评分下降、欺诈事件的发生等

3.2 专业能力类 Agent（Professional Capability Agents）

专业能力类 Agent 的职责是利用自身的专业知识和模型，完成特定的任务，并将任务结果转换为标准格式的消息，发送给决策类 Agent 或协作协调类 Agent。根据应用场景的不同，专业能力类 Agent 可以分为风控类 Agent和客户服务类 Agent两大类：

3.2.1 风控类 Agent（Risk Control Agents）

常见的风控类 Agent 包括：

1. 身份核验 Agent：负责核验客户的身份，如人脸识别、OCR识别、身份证实名认证、银行卡实名认证、手机号实名认证等
2. 反欺诈 Agent：负责识别欺诈行为，如交易欺诈、申请欺诈、营销欺诈、电信诈骗等
3. 信用评估 Agent：负责评估客户的信用状况，如信用评分、信用等级、信用额度等
4. 反洗钱 Agent：负责识别洗钱行为，如大额交易、可疑交易、跨境交易等
5. 合规审计 Agent：负责审计交易的合规性，如是否符合监管要求、是否符合银行内部规定等
6. 风险定价 Agent：负责根据客户的风险状况和市场情况，制定合理的产品价格，如贷款利率、理财收益率、保险费率等
7. 风险处置 Agent：负责根据风险的等级和类型，制定合理的风险处置方案，如冻结账户、拒绝交易、降额、封卡、报警等

3.2.2 客户服务类 Agent（Customer Service Agents）

常见的客户服务类 Agent 包括：

1. 意图识别 Agent：负责识别客户的意图，如查询余额、转账汇款、申请贷款、购买理财、投诉建议等
2. FAQ 解答 Agent：负责回答客户的简单、标准化的问题
3. 需求挖掘 Agent：负责挖掘客户的潜在需求，如根据客户的历史行为、风险偏好、财务状况，挖掘客户的贷款需求、理财需求、保险需求等
4. 方案推荐 Agent：负责根据客户的需求和风险偏好，推荐个性化的金融产品和服务方案
5. 交易协助 Agent：负责协助客户完成金融交易，如贷款申请、理财购买、保险投保等
6. 投诉处理 Agent：负责处理客户的投诉建议，如记录投诉内容、分析投诉原因、制定投诉处理方案、跟踪投诉处理进度、反馈投诉处理结果等
7. 后续跟进 Agent：负责对客户进行后续跟进，如提醒客户还款、提醒客户理财到期、向客户推荐新的金融产品和服务等

3.3 决策类 Agent（Decision-Making Agents）

决策类 Agent 的职责是根据感知类 Agent 发送的感知信息和专业能力类 Agent 发送的任务结果，做出最终的决策，并将决策结果转换为标准格式的消息，发送给执行类 Agent 或协作协调类 Agent。常见的决策类 Agent 包括：

1. 风控决策 Agent：负责根据身份核验 Agent、反欺诈 Agent、信用评估 Agent、反洗钱 Agent、合规审计 Agent、风险定价 Agent 发送的结果，做出最终的风控决策，如通过贷款申请、拒绝贷款申请、冻结账户、拒绝交易等
2. 客服决策 Agent：负责根据意图识别 Agent、FAQ 解答 Agent、需求挖掘 Agent、方案推荐 Agent 发送的结果，做出最终的客服决策，如直接回答客户的问题、转接专业能力类 Agent、转接人工客服、推荐金融产品和服务等

3.4 协作协调类 Agent（Collaboration & Coordination Agents）

协作协调类 Agent 的职责是协调多个 Agent 之间的协作，包括任务分配、消息路由、冲突解决、状态管理等。协作协调类 Agent 是金融级 MAS 的“大脑中枢”，常见的协作协调类 Agent 包括：

1. 任务分配 Agent：负责根据任务的类型、优先级、Agent 的负载情况和专业能力，将任务分配给最合适的 Agent
2. 消息路由 Agent：负责根据消息的类型、发送者、接收者，将消息路由到最合适的 Agent
3. 冲突解决 Agent：负责解决多个 Agent 之间的冲突，如多个风控类 Agent 的决策结果冲突、多个客户服务类 Agent 的推荐方案冲突等
4. 状态管理 Agent：负责管理所有 Agent 的状态和任务的状态，如 Agent 是否在线、任务是否完成、任务是否失败等
5. 对话管理 Agent：仅在客户服务场景中使用，负责管理客户与 Agent 之间的多轮对话，如上下文理解、对话历史记录、对话状态跟踪等

第4层：协作机制层（Collaboration Mechanism Layer）

协作机制层是金融级 MAS 的“游戏规则”，定义了 Agent 之间的协作协议、消息格式、任务分配机制和冲突解决机制。常见的协作机制包括：

1. 协作协议：如 Contract Net Protocol（合同网协议）、Blackboard Architecture（黑板架构）、Market-Oriented Programming（面向市场的编程）等
2. 消息格式：采用 JSON、Protobuf 或 XML 等标准格式，内置消息 ID、发送者 ID、接收者 ID、消息类型、消息内容、时间戳、签名等字段
3. 任务分配机制：如基于负载均衡的任务分配机制、基于专业能力的任务分配机制、基于优先级的任务分配机制、基于拍卖的任务分配机制等
4. 冲突解决机制：如基于投票的冲突解决机制、基于优先级的冲突解决机制、基于协商的冲突解决机制、基于规则的冲突解决机制等

第5层：应用层（Application Layer）

应用层是金融级 MAS 的“门面”，直接面向用户（包括客户、风控专员、客服专员、理财经理、银行管理层等），提供以下核心应用：

1. 智能风控平台：面向风控专员和银行管理层，提供风险识别、风险评估、风险决策、风险处置、风险监控、风险报告等功能
2. 智能客服平台：面向客户、客服专员和银行管理层，提供在线客服、语音客服、视频客服、投诉处理、客户反馈等功能
3. 智能财富管理平台：面向客户和理财经理，提供需求分析、方案推荐、交易协助、资产配置、风险预警、后续跟进等功能
4. 监管审计平台：面向监管机构和银行内部审计部门，提供决策全链路的可追溯性、可解释性、审计日志等功能

架构图

以下是本文提出的金融级 Multi-Agent 分层协作架构的 Mermaid 架构图：

核心算法原理与具体操作步骤

核心算法1：基于 Contract Net Protocol（合同网协议）的任务分配算法

问题背景

在金融级 MAS 中，任务分配是协作协调类 Agent 的核心职责之一。如何在保证低延迟、高可用、公平性的前提下，将任务分配给最合适的 Agent，是一个需要解决的核心问题。

问题描述

给定：

1. 一个任务集合 T={t1,t2,...,tn}T = \{t_1, t_2, ..., t_n\}T={t1​,t2​,...,tn​}，每个任务 $t_i$ 具有以下属性：
   • 任务 ID：$i{d}_i$
   • 任务类型：typei∈{身份核验,反欺诈,信用评估,...,意图识别,需求挖掘,...}type_i \in \{身份核验, 反欺诈, 信用评估, ..., 意图识别, 需求挖掘, ...\}typei​∈{身份核验,反欺诈,信用评估,...,意图识别,需求挖掘,...}
   • 任务优先级：priorityi∈{1,2,3,4,5}priority_i \in \{1, 2, 3, 4, 5\}priorityi​∈{1,2,3,4,5}（1 为最低优先级，5 为最高优先级）
   • 任务截止时间：$deadlin{e}_i$
   • 任务复杂度：$complexit{y}_i\in [0,1]$（0 为最简单，1 为最复杂）
2. 一个 Agent 集合 A={a1,a2,...,am}A = \{a_1, a_2, ..., a_m\}A={a1​,a2​,...,am​}，每个 Agent $a_j$ 具有以下属性：
   • Agent ID：$i{d}_j$
   • Agent 专业能力集合：skillsj⊆{身份核验,反欺诈,信用评估,...,意图识别,需求挖掘,...}skills_j \subseteq \{身份核验, 反欺诈, 信用评估, ..., 意图识别, 需求挖掘, ...\}skillsj​⊆{身份核验,反欺诈,信用评估,...,意图识别,需求挖掘,...}
   • Agent 当前负载：$loa{d}_j\in [0,1]$（0 为空闲，1 为满负载）
   • Agent 处理任务的平均成功率：$succes{s}_{r}at{e}_j\in [0,1]$
   • Agent 处理任务的平均延迟：$av{g}_{l}atenc{y}_j\in [0,+\infty )$（单位：毫秒）
   • Agent 是否在线：onlinej∈{True,False}online_j \in \{True, False\}onlinej​∈{True,False}

目标：
将任务集合 $T$ 中的每个任务 $t_i$ 分配给 Agent 集合 $A$ 中的一个在线的、具有相应专业能力的 Agent $a_j$，使得：

1. 任务的平均成功率最高
2. 任务的平均延迟最低
3. 高优先级的任务优先被处理
4. 任务在截止时间之前完成
5. Agent 的负载尽可能均衡

算法原理

Contract Net Protocol（合同网协议）是一种经典的分布式任务分配算法，模拟了人类社会的“招标-投标-中标”机制。本文对传统的 Contract Net Protocol 进行了改进，增加了优先级、截止时间、复杂度、负载均衡等约束条件，使其更适合金融级 MAS 的场景。

改进后的 Contract Net Protocol 的核心流程如下：

1. 招标阶段：任务分配 Agent 接收到任务 $t_i$ 后，根据任务的类型 $typ{e}_i$ 筛选出所有在线的、具有相应专业能力的 Agent（称为“合格投标者”），并向这些合格投标者发送“招标消息”。招标消息包含任务的所有属性（ID、类型、优先级、截止时间、复杂度）。
2. 投标阶段：合格投标者接收到招标消息后，根据自身的当前负载 $loa{d}_j$、处理任务的平均成功率 $succes{s}_{r}at{e}_j$、处理任务的平均延迟 $av{g}_{l}atenc{y}_j$、任务的优先级 $priorit{y}_i$、任务的截止时间 $deadlin{e}_i$、任务的复杂度 $complexit{y}_i$，计算出自身的“投标得分”。如果投标得分高于预设的“投标阈值”，则向任务分配 Agent 发送“投标消息”。投标消息包含 Agent 的 ID、投标得分、预计完成时间。
3. 中标阶段：任务分配 Agent 接收到所有合格投标者的投标消息后，根据投标得分从高到低对合格投标者进行排序，选择投标得分最高的合格投标者作为“中标者”，并向中标者发送“中标消息”，向其他合格投标者发送“未中标消息”。中标消息包含任务的所有属性。
4. 执行阶段：中标者接收到中标消息后，开始执行任务。如果任务执行成功，则向任务分配 Agent 发送“成功消息”；如果任务执行失败或预计无法在截止时间之前完成，则向任务分配 Agent 发送“失败消息”，任务分配 Agent 重新启动招标流程。
5. 反馈阶段：任务分配 Agent 接收到成功消息或失败消息后，更新中标者的当前负载 $loa{d}_j$、处理任务的平均成功率 $succes{s}_{r}at{e}_j$、处理任务的平均延迟 $av{g}_{l}atenc{y}_j$。

投标得分的计算

投标得分的计算是改进后的 Contract Net Protocol 的核心。本文提出了一种多目标加权求和的投标得分计算方法，公式如下：

$$scor{e}_{i,j}=w_1\times S_{success}(i,j)+w_2\times S_{latency}(i,j)+w_3\times S_{priority}(i,j)+w_4\times S_{deadline}(i,j)+w_5\times S_{load}(i,j)$$

其中：

1. $scor{e}_{i,j}$ 表示 Agent $a_j$ 对任务 $t_i$ 的投标得分，取值范围为 $[0,1]$，得分越高越好
2. $w_1,w_2,w_3,w_4,w_5$ 表示权重，满足 $w_1+w_2+w_3+w_4+w_5=1$，可以根据实际场景进行调整（例如，在风控场景中，$w_1$ 和 $w_3$ 可以设置得较大；在客服场景中，$w_2$ 和 $w_3$ 可以设置得较大）
3. $S_{success}(i,j)$ 表示成功率得分，计算公式如下：
   $$S_{success}(i,j)=succes{s}_{r}at{e}_j\times (1-complexit{y}_i)+complexit{y}_i\times success\_rate\_comple{x}_j$$
   其中，$success\_rate\_comple{x}_j$ 表示 Agent $a_j$ 处理复杂任务（复杂度 $\ge 0.7$）的平均成功率
4. $S_{latency}(i,j)$ 表示延迟得分，计算公式如下：
   $$S_{latency}(i,j)=1-\frac{avg\_latenc{y}_j\times (1+complexit{y}_i)}{max\_latency}$$
   其中，$max\_latency$ 表示预设的最大允许延迟（单位：毫秒），如果 $\frac{avg\_latenc{y}_j\times (1+complexit{y}_i)}{max\_latency}\ge 1$，则 $S_{latency}(i,j)=0$
5. $S_{priority}(i,j)$ 表示优先级得分，计算公式如下：
   $$S_{priority}(i,j)=\frac{priorit{y}_i}{5}$$
6. $S_{deadline}(i,j)$ 表示截止时间得分，计算公式如下：
   $$S_{deadline}(i,j)=1-\frac{current\_time+avg\_latenc{y}_j\times (1+complexit{y}_i)-deadlin{e}_i}{deadline\_buffer}$$
   其中，$current\_time$ 表示当前时间，$deadline\_buffer$ 表示预设的截止时间缓冲（单位：毫秒），如果 $current\_time+avg\_latenc{y}_j\times (1+complexit{y}_i)>deadlin{e}_i$，则 $S_{deadline}(i,j)=0$
7. $S_{load}(i,j)$ 表示负载得分，计算公式如下：
   $$S_{load}(i,j)=1-loa{d}_j$$

算法流程图

以下是改进后的 Contract Net Protocol 的 Mermaid 流程图：

算法源代码（Python）

以下是改进后的 Contract Net Protocol 的 Python 源代码实现：

import time
import random
from typing import List, Dict, Optional, Tuple

# 定义任务类
class Task:
    def __init__(self, id: str, type: str, priority: int, deadline: float, complexity: float):
        """
        初始化任务
        :param id: 任务 ID
        :param type: 任务类型
        :param priority: 任务优先级（1-5）
        :param deadline: 任务截止时间（时间戳，单位：秒）
        :param complexity: 任务复杂度（0-1）
        """
        self.id = id
        self.type = type
        self.priority = priority
        self.deadline = deadline
        self.complexity = complexity

    def __repr__(self):
        return f"Task(id={self.id}, type={self.type}, priority={self.priority}, deadline={self.deadline}, complexity={self.complexity})"

# 定义 Agent 类
class Agent:
    def __init__(self, id: str, skills: List[str], load: float = 0.0, success_rate: float = 0.9, 
                 success_rate_complex: float = 0.8, avg_latency: float = 100.0, online: bool = True):
        """
        初始化 Agent
        :param id: Agent ID
        :param skills: Agent 专业能力集合
        :param load: Agent 当前负载（0-1）
        :param success_rate: Agent 处理任务的平均成功率（0-1）
        :param success_rate_complex: Agent 处理复杂任务的平均成功率（0-1）
        :param avg_latency: Agent 处理任务的平均延迟（单位：毫秒）
        :param online: Agent 是否在线
        """
        self.id = id
        self.skills = skills
        self.load = load
        self.success_rate = success_rate
        self.success_rate_complex = success_rate_complex
        self.avg_latency = avg_latency
        self.online = online
        # 统计信息：总任务数、成功任务数、总延迟
        self.total_tasks = 0
        self.successful_tasks = 0
        self.total_latency = 0.0

    def __repr__(self):
        return f"Agent(id={self.id}, skills={self.skills}, load={self.load:.2f}, success_rate={self.success_rate:.2f}, avg_latency={self.avg_latency:.2f}, online={self.online})"

    def is_qualified(self, task: Task) -> bool:
        """
        判断 Agent 是否是合格投标者
        :param task: 任务
        :return: 是否是合格投标者
        """
        return self.online and task.type in self.skills

    def calculate_bid_score(self, task: Task, current_time: float, max_latency: float = 1000.0, 
                            deadline_buffer: float = 5.0, weights: Tuple[float, float, float, float, float] = (0.3, 0.2, 0.2, 0.15, 0.15)) -> Optional[float]:
        """
        计算投标得分
        :param task: 任务
        :param current_time: 当前时间（时间戳，单位：秒）
        :param max_latency: 最大允许延迟（单位：毫秒）
        :param deadline_buffer: 截止时间缓冲（单位：秒）
        :param weights: 权重（w1, w2, w3, w4, w5）
        :return: 投标得分，如果不符合要求则返回 None
        """
        w1, w2, w3, w4, w5 = weights
        # 检查权重是否合法
        if not (0.0 <= w1 <= 1.0 and 0.0 <= w2 <= 1.0 and 0.0 <= w3 <= 1.0 and 0.0 <= w4 <= 1.0 and 0.0 <= w5 <= 1.0 and abs(w1 + w2 + w3 + w4 + w5 - 1.0) < 1e-6):
            raise ValueError("权重必须在 0-1 之间，且总和为 1")

        # 计算成功率得分
        if task.complexity >= 0.7:
            s_success = self.success_rate_complex
        else:
            s_success = self.success_rate * (1 - task.complexity) + task.complexity * self.success_rate_complex

        # 计算延迟得分
        expected_latency = self.avg_latency * (1 + task.complexity)
        if expected_latency >= max_latency:
            return None
        s_latency = 1 - expected_latency / max_latency

        # 计算优先级得分
        s_priority = task.priority / 5.0

        # 计算截止时间得分
        expected_finish_time = current_time + expected_latency / 1000.0
        if expected_finish_time > task.deadline:
            return None
        time_left = task.deadline - expected_finish_time
        s_deadline = 1 - max(0.0, (deadline_buffer - time_left) / deadline_buffer)

        # 计算负载得分
        s_load = 1 - self.load

        # 计算总得分
        total_score = w1 * s_success + w2 * s_latency + w3 * s_priority + w4 * s_deadline + w5 * s_load
        return total_score

    def execute_task(self, task: Task) -> Tuple[bool, float]:
        """
        执行任务（模拟）
        :param task: 任务
        :return: (是否成功, 实际延迟（单位：毫秒）)
        """
        # 模拟延迟：在平均延迟的基础上增加 ±20% 的随机波动
        actual_latency = self.avg_latency * (0.8 + 0.4 * random.random())
        # 模拟成功率：复杂度越高，成功率越低
        if task.complexity >= 0.7:
            success_prob = self.success_rate_complex
        else:
            success_prob = self.success_rate * (1 - task.complexity) + task.complexity * self.success_rate_complex
        is_success = random.random() < success_prob

        # 更新统计信息
        self.total_tasks += 1
        if is_success:
            self.successful_tasks += 1
        self.total_l