MAS Dynamic Trust — 多智能体动态信任图：抵御 Sleeper Agent 与历史感知可信聚合

1. 解决的问题

MAS 中 Agent 之间传递消息，但消息的可信度并不相同——某个 Agent 可能已被攻击者控制（Sleeper Agent），在积累足够信任后才触发恶意行为。动态信任管理解决的问题是：在没有可信第三方的情况下，每个 Agent 如何评估其他 Agent 发来消息的可信度，并据此决定接受或拒绝。

2. 核心算法逻辑

MAS 中 Agent 之间传递消息，但消息的可信度并不相同——某个 Agent 可能已被攻击者控制（Sleeper Agent），在积累足够信任后才触发恶意行为。动态信任管理解决的问题是：在没有可信第三方的情况下，每个 Agent 如何评估其他 Agent 发来消息的可信度，并据此决定接受或拒绝。

3. 业务应用场景

业务背景：AgenticPay 采购谈判系统由多个 Agent 协同工作——价格谈判 Agent、合规检查 Agent、市场行情 Agent。若某个市场行情 Agent 被错误数据污染（如竞品恶意干扰），所有下游决策将偏差。

数据要求： - Agent 间消息历史（JSON 格式）：`{agent_id, message_type, content, outcome}` - 决策结果反馈：`{decision_id, outcome, correct: bool}`

业务背景：AIM-RM 库存管理系统接收来自多个数据源的库存预测（本地仓、海外仓、第三方 3PL）。当大促期间某个数据源出现系统故障，产生异常预测值，如何自动隔离该数据源？

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

5-30 万元

7. 代码模板

代码块数量：4 · 路径：未检测到

"""
MAS Dynamic Trust Management
整合 DynaTrust (Bayesian信任图) + A-Trust (Attention量化) + ECL (历史感知聚合)
"""

import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from scipy.stats import beta as beta_dist


@dataclass
class TrustEdge:
    """Alpha-beta Bayesian 信任边"""
    agent_id: str
    alpha: float = 1.0   # 成功交互次数（+1 先验）
    beta: float = 1.0    # 失败交互次数（+1 先验）
    history: List[bool] = field(default_factory=list)  # True=可靠, False=不可靠

    @property
    def mean(self) -> float:
        return self.alpha / (self.alpha + self.beta)

    @property
    def confidence(self) -> float:
        """交互次数越多，置信度越高"""
        n = self.alpha + self.beta - 2  # 减去先验
        return min(1.0, n / 20)         # 20次交互达到满置信

    def update(self, outcome: bool):
        if outcome:
            self.alpha += 1
        else:
            self.beta += 1
        self.history.append(outcome)

    def sample(self) -> float:
        return np.random.beta(self.alpha, self.beta)


class DynamicTrustGraph:
    """
    DynaTrust 动态信任图
    - 每个 Agent 维护对其他所有 Agent 的信任边
    - 接收端评分（非发送端自报）
    - 陪审团共识用于高风险决策
    """

    def __init__(
        self,
        agent_id: str,
        trust_threshold: float = 0.6,
        jury_threshold: float = 0.5,
        jury_size: int = 2,
    ):
        self.agent_id = agent_id
        self.trust_threshold = trust_threshold
        self.jury_threshold = jury_threshold
        self.jury_size = jury_size
        self.edges: Dict[str, TrustEdge] = {}

8. 论文来源

• 2506.02546
• 2601.21742
• 2603.15661