MAS Adversarial Defense — 多智能体系统攻防：群体合谋检测、规划时攻击防御、路由感知注入

1. 解决的问题

`Skill-Agent-Safety-Guardrails` 保护的是单个 Agent 免受提示注入攻击。但 MAS 中的攻击比单 Agent 复杂 10 倍：攻击者可以利用多个 Agent 之间的信任关系和通信路径发动群体级攻击。

2. 核心算法逻辑

SkillAgentSafetyGuardrails 保护的是单个 Agent 免受提示注入攻击。但 MAS 中的攻击比单 Agent 复杂 10 倍：攻击者可以利用多个 Agent 之间的信任关系和通信路径发动群体级攻击。

3. 业务应用场景

业务背景：母婴品牌的 MAS 系统从多个数据源 Agent 获取竞品价格数据。竞品可能通过"投毒"数据源（在 Amazon 平台上故意异常定价），诱导 MAS 的多个 Agent 同时报告高竞品价格，让品牌高估市场价格、定价过高失去竞争力。

业务背景：广告竞价 MAS 接收来自外部广告代理的"优化建议"，但恶意代理可能通过建议修改竞价 DAG，插入"向竞争对手账户发送出价预告"的步骤。

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

5-10 万/月

7. 代码模板

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

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Set, Tuple
import math
import time


@dataclass
class AgentMessage:
    sender_id: str
    receiver_id: str
    content: str
    timestamp: float = field(default_factory=time.time)
    is_honeypot_response: bool = False


@dataclass
class DefenseAlert:
    alert_type: str
    severity: str
    involved_agents: List[str]
    evidence: dict
    action: str


class GroupGuardMonitor:
    """
    GroupGuard 图监控：检测群体合谋攻击
    三层：图监控 + 蜜罐 + 结构剪枝
    """

    def __init__(self, density_threshold_sigma: float = 3.0,
                 window_seconds: float = 300.0):
        self.density_threshold_sigma = density_threshold_sigma
        self.window_seconds = window_seconds
        self._message_log: List[AgentMessage] = []
        self._density_history: List[float] = []
        self._honeypot_agents: Set[str] = set()
        self._honeypot_solicitations: List[Dict] = []

    def register_honeypot(self, agent_id: str):
        self._honeypot_agents.add(agent_id)

    def record_message(self, msg: AgentMessage) -> Optional[DefenseAlert]:
        self._message_log.append(msg)
        if msg.receiver_id in self._honeypot_agents and not msg.is_honeypot_response:
            self._honeypot_solicitations.append({
                "sender": msg.sender_id, "content": msg.content, "ts": msg.timestamp
            })
            if len(self._honeypot_solicitations) >= 2:
                return DefenseAlert(
                    alert_type="honeypot_triggered",
                    severity="high",
                    involved_agents=[s["sender"] for s in self._honeypot_solicitations],
                    evidence={"solicitations": self._honeypot_solicitations[-3:]},
                    action="isolate_senders",
                )

        density_alert = self._check_density()
        return density_alert

8. 论文来源

• 2603.13940
• 2604.16543
• 2605.11514