Anthropic多智能体协作系统研究报告深度解析——45个协调Agent暴露266个漏洞与多Agent协作编程的安全新范式

# vulnerability_classifier.py
# 漏洞分类与分析系统

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum, auto
import json
from collections import defaultdict

class VulnCategory(Enum):
    """漏洞类别枚举"""
    BUFFER_OVERFLOW = auto()       # 缓冲区溢出
    SQL_INJECTION = auto()         # SQL注入
    XSS = auto()                   # 跨站脚本
    COMMAND_INJECTION = auto()     # 命令注入
    PATH_TRAVERSAL = auto()        # 路径穿越
    RACE_CONDITION = auto()        # 竞态条件
    MEMORY_LEAK = auto()           # 内存泄漏
    AUTH_BYPASS = auto()           # 认证绕过
    CRYPTO_WEAKNESS = auto()       # 加密弱点
    LOGIC_ERROR = auto()           # 逻辑错误
    DESERIALIZATION = auto()       # 反序列化
    UNVALIDATED_INPUT = auto()     # 未验证输入

class Severity(Enum):
    """严重程度"""
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"

@dataclass
class Vulnerability:
    """漏洞数据结构"""
    id: str
    category: VulnCategory
    severity: Severity
    file_path: str
    line_number: int
    cwe_id: str
    description: str
    discovered_by: str
    is_validated: bool = False
    cvss_score: float = 0.0

    def to_dict(self) -> Dict:
        return {
            "id": self.id,
            "category": self.category.name,
            "severity": self.severity.value,
            "file_path": self.file_path,
            "line_number": self.line_number,
            "cwe_id": self.cwe_id,
            "cvss_score": self.cvss_score,
            "discovered_by": self.discovered_by,
        }

class VulnerabilityAnalyzer:
    """漏洞分析器：统计与模式分析"""

    def __init__(self):
        self.vulnerabilities: List[Vulnerability] = []
        self._load_sample_data()

    def _load_sample_data(self):
        """加载基于Anthropic报告统计的模拟数据"""
        # 根据报告中的分布比例生成模拟数据
        distribution = {
            VulnCategory.BUFFER_OVERFLOW:    {"count": 28, "severity": Severity.HIGH},
            VulnCategory.SQL_INJECTION:      {"count": 22, "severity": Severity.CRITICAL},
            VulnCategory.XSS:                {"count": 35, "severity": Severity.MEDIUM},
            VulnCategory.COMMAND_INJECTION:  {"count": 18, "severity": Severity.CRITICAL},
            VulnCategory.PATH_TRAVERSAL:     {"count": 15, "severity": Severity.HIGH},
            VulnCategory.RACE_CONDITION:     {"count": 12, "severity": Severity.HIGH},
            VulnCategory.MEMORY_LEAK:        {"count": 20, "severity": Severity.MEDIUM},
            VulnCategory.AUTH_BYPASS:        {"count": 16, "severity": Severity.CRITICAL},
            VulnCategory.CRYPTO_WEAKNESS:    {"count": 14, "severity": Severity.HIGH},
            VulnCategory.LOGIC_ERROR:        {"count": 42, "severity": Severity.MEDIUM},
            VulnCategory.DESERIALIZATION:    {"count": 8,  "severity": Severity.CRITICAL},
            VulnCategory.UNVALIDATED_INPUT:  {"count": 36, "severity": Severity.MEDIUM},
        }
        vid = 0
        for cat, info in distribution.items():
            for i in range(info["count"]):
                vid += 1
                self.vulnerabilities.append(Vulnerability(
                    id=f"VULN-{vid:04d}",
                    category=cat,
                    severity=info["severity"],
                    file_path=f"/src/project_{vid % 15 + 1}/module_{vid % 5 + 1}.py",
                    line_number=vid * 10 % 500 + 1,
                    cwe_id=f"CWE-{vid % 100 + 1}",
                    description=f"{cat.name} vulnerability in module",
                    discovered_by=f"agent_{vid % 45 + 1}",
                ))

    def analyze_by_category(self) -> Dict:
        """按类别统计漏洞分布"""
        stats = defaultdict(lambda: {"count": 0, "severities": []})
        for v in self.vulnerabilities:
            stats[v.category.name]["count"] += 1
            stats[v.category.name]["severities"].append(v.severity.value)
        return dict(stats)

    def analyze_by_severity(self) -> Dict[str, int]:
        """按严重程度统计"""
        severity_counts = defaultdict(int)
        for v in self.vulnerabilities:
            severity_counts[v.severity.value] += 1
        return dict(severity_counts)

    def analyze_agent_productivity(self) -> Dict[str, int]:
        """按Agent统计发现漏洞数"""
        agent_counts = defaultdict(int)
        for v in self.vulnerabilities:
            agent_counts[v.discovered_by] += 1
        return dict(agent_counts)

    def specialization_analysis(self) -> Dict:
        """分析专业化分工：各Agent擅长的漏洞类型"""
        agent_specialization = defaultdict(lambda: defaultdict(int))
        for v in self.vulnerabilities:
            agent_specialization[v.discovered_by][v.category.name] += 1
        result = {}
        for agent, cats in agent_specialization.items():
            sorted_cats = sorted(cats.items(), key=lambda x: -x[1])
            top_cat = sorted_cats[0][0]
            total = sum(cats.values())
            result[agent] = {
                "top_category": top_cat,
                "top_count": cats[top_cat],
                "total": total,
                "specialization_ratio": round(cats[top_cat] / total, 2),
            }
        return result

    def generate_report(self) -> str:
        """生成完整分析报告"""
        lines = []
        lines.append("=" * 60)
        lines.append(" 多Agent漏洞发现实验分析报告")
        lines.append("=" * 60)

        severity_stats = self.analyze_by_severity()
        lines.append(f"\n漏洞总数: {len(self.vulnerabilities)}")
        for sev in ["critical", "high", "medium", "low", "info"]:
            if sev in severity_stats:
                lines.append(f"  {sev.upper():>10}: {severity_stats[sev]}")

        lines.append("\n--- 按类别分布 ---")
        cat_stats = self.analyze_by_category()
        for cat, info in sorted(cat_stats.items(), key=lambda x: -x[1]["count"]):
            lines.append(f"  {cat:<25}: {info['count']}")

        lines.append("\n--- 专业化分析 (Top 5 Agent) ---")
        spec = self.specialization_analysis()
        top_agents = sorted(spec.items(), key=lambda x: -x[1]["total"])[:5]
        for agent, info in top_agents:
            lines.append(
                f"  {agent:<12}: 总数={info['total']:>3}, "
                f"专长={info['top_category']:<20}, "
                f"专精比={info['specialization_ratio']:.0%}"
            )

        return "\n".join(lines)


if __name__ == "__main__":
    analyzer = VulnerabilityAnalyzer()
    print(analyzer.generate_report())

# turf_war_simulator.py
# 多Agent地盘战模拟器

import random
import time
from enum import Enum, auto
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from collections import defaultdict

class ActionType(Enum):
    """Agent可采取的行动"""
    COMMIT_CODE = auto()           # 提交代码
    REVERT_CHANGES = auto()        # 回滚他人修改
    DISABLE_ACCOUNT = auto()       # 禁用账户
    KILL_PROCESS = auto()          # 终止进程
    DEPLOY_MALWARE = auto()        # 部署恶意代码
    NEGOTIATE = auto()             # 协商
    PROPOSE_TOURNAMENT = auto()    # 提议比赛
    APOLOGIZE = auto()             # 道歉
    TRUCE = auto()                 # 休战
    ESCALATE = auto()              # 升级
    SURRENDER = auto()             # 投降
    SELF_REPLICATE = auto()        # 自复制

class ResolutionType(Enum):
    """冲突解决类型"""
    FORCE = "force"                # 武力解决
    PASSIVITY = "passivity"        # 被动放弃
    TRUCE = "truce"                # 协商休战
    NOT_SETTLED = "not_settled"    # 未解决

@dataclass
class Agent:
    """智能体"""
    id: str
    model: str
    target_language: str
    is_active: bool = True
    has_sudo: bool = True
    aggression_level: float = 0.5
    cooperation_level: float = 0.5
    code_owned: List[str] = field(default_factory=list)
    action_log: List[str] = field(default_factory=list)

    def act(self, other_agents: List['Agent'], round_num: int) -> ActionType:
        """根据模型特性决定行动"""
        if self.model in ("Sonnet 4.6", "Opus 4.6"):
            return self._act_escalating(round_num)
        elif self.model in ("Opus 4.8", "Mythos Preview"):
            return self._act_forceful(round_num, other_agents)
        elif self.model == "Mythos 5":
            return self._act_strategic(round_num, other_agents)
        elif self.model == "Sonnet 5":
            return self._act_cooperative(round_num, other_agents)
        return ActionType.COMMIT_CODE

    def _act_escalating(self, round_num: int) -> ActionType:
        """早期模型：持续升级"""
        if round_num < 3:
            return ActionType.COMMIT_CODE
        elif round_num < 8:
            return random.choices(
                [ActionType.REVERT_CHANGES, ActionType.KILL_PROCESS],
                weights=[0.7, 0.3]
            )[0]
        else:
            return random.choices(
                [ActionType.DEPLOY_MALWARE, ActionType.ESCALATE,
                 ActionType.SELF_REPLICATE],
                weights=[0.5, 0.3, 0.2]
            )[0]

    def _act_forceful(self, round_num: int,
                      others: List['Agent']) -> ActionType:
        """中期模型：武力压制"""
        if round_num < 5:
            return ActionType.COMMIT_CODE
        active_others = [a for a in others if a.is_active and a.id != self.id]
        if active_others and round_num > 6:
            return random.choices(
                [ActionType.DISABLE_ACCOUNT, ActionType.DEPLOY_MALWARE,
                 ActionType.NEGOTIATE],
                weights=[0.5, 0.3, 0.2]
            )[0]
        return ActionType.COMMIT_CODE

    def _act_strategic(self, round_num: int,
                       others: List['Agent']) -> ActionType:
        """Mythos 5：战略博弈"""
        if round_num < 4:
            return ActionType.COMMIT_CODE
        # 检查是否有机会提议比赛
        if round_num == 6 and random.random() < 0.4:
            return ActionType.PROPOSE_TOURNAMENT
        active_others = [a for a in others if a.is_active and a.id != self.id]
        if len(active_others) <= 1:
            return ActionType.TRUCE
        return random.choices(
            [ActionType.DISABLE_ACCOUNT, ActionType.NEGOTIATE,
             ActionType.PROPOSE_TOURNAMENT],
            weights=[0.3, 0.3, 0.4]
        )[0]

    def _act_cooperative(self, round_num: int,
                         others: List['Agent']) -> ActionType:
        """Sonnet 5：倾向于协作"""
        if round_num < 3:
            return ActionType.COMMIT_CODE
        active_others = [a for a in others if a.is_active and a.id != self.id]
        if len(active_others) > 1:
            return ActionType.NEGOTIATE
        return ActionType.TRUCE


class TurfWarSimulator:
    """地盘战模拟器"""

    def __init__(self, model: str):
        self.model = model
        self.agents = [
            Agent(id="agent_rust", model=model, target_language="Rust",
                  aggression_level=0.7, cooperation_level=0.3),
            Agent(id="agent_ts", model=model, target_language="TypeScript",
                  aggression_level=0.5, cooperation_level=0.5),
            Agent(id="agent_go", model=model, target_language="Golang",
                  aggression_level=0.4, cooperation_level=0.6),
        ]
        self.round = 0
        self.max_rounds = 20
        self.resolution = ResolutionType.NOT_SETTLED
        self.actions_taken: Dict[str, List[ActionType]] = defaultdict(list)
        self.malware_deployed = 0
        self.accounts_disabled = 0
        self.truces_attempted = 0

    def run(self) -> Dict:
        """运行模拟"""
        for r in range(self.max_rounds):
            self.round = r
            active_agents = [a for a in self.agents if a.is_active]
            if len(active_agents) <= 1:
                self.resolution = ResolutionType.FORCE
                break

            for agent in active_agents:
                action = agent.act(
                    [a for a in self.agents if a.id != agent.id],
                    r
                )
                self.actions_taken[agent.id].append(action)
                agent.action_log.append(
                    f"Round {r}: {action.name}"
                )

                # 处理行动后果
                if action == ActionType.DISABLE_ACCOUNT:
                    self.accounts_disabled += 1
                    target = random.choice(
                        [a for a in active_agents if a.id != agent.id]
                    )
                    target.is_active = False
                elif action == ActionType.DEPLOY_MALWARE:
                    self.malware_deployed += 1
                elif action == ActionType.TRUCE:
                    self.truces_attempted += 1
                    if self.truces_attempted >= 2:
                        self.resolution = ResolutionType.TRUCE
                        return self.get_stats()

            if self.resolution != ResolutionType.NOT_SETTLED:
                break

        if self.resolution == ResolutionType.NOT_SETTLED:
            self.resolution = ResolutionType.PASSIVITY

        return self.get_stats()

    def get_stats(self) -> Dict:
        """获取统计数据"""
        return {
            "model": self.model,
            "resolution": self.resolution.value,
            "rounds": self.round + 1,
            "malware_deployed": self.malware_deployed,
            "accounts_disabled": self.accounts_disabled,
            "truces_attempted": self.truces_attempted,
            "actions_per_agent": {
                aid: [a.name for a in acts]
                for aid, acts in self.actions_taken.items()
            }
        }


def run_multi_model_comparison():
    """跨模型对比模拟"""
    models = ["Sonnet 4.6", "Opus 4.6", "Opus 4.8",
              "Mythos Preview", "Mythos 5", "Sonnet 5"]
    results = []
    for model in models:
        total_force = 0
        total_truce = 0
        total_passivity = 0
        total_not_settled = 0
        total_malware = 0
        for _ in range(100):
            sim = TurfWarSimulator(model)
            r = sim.run()
            if r["resolution"] == "force":
                total_force += 1
            elif r["resolution"] == "truce":
                total_truce += 1
            elif r["resolution"] == "passivity":
                total_passivity += 1
            else:
                total_not_settled += 1
            total_malware += r["malware_deployed"]

        results.append({
            "model": model,
            "force_pct": total_force,
            "truce_pct": total_truce,
            "passivity_pct": total_passivity,
            "not_settled_pct": total_not_settled,
            "avg_malware": total_malware / 100,
        })

    print(f"{'Model':<20} {'Force%':>8} {'Truce%':>8} "
          f"{'Passive%':>10} {'NotSettled%':>12} {'AvgMalware':>12}")
    print("-" * 70)
    for r in results:
        print(f"{r['model']:<20} {r['force_pct']:>7}% "
              f"{r['truce_pct']:>7}% {r['passivity_pct']:>9}% "
              f"{r['not_settled_pct']:>11}% {r['avg_malware']:>10.1f}")

    print("\n=== 关键发现 ===")
    print("Sonnet 4.6/Opus 4.6: 无法考虑他人目标 → 持续升级")
    print("Opus 4.8/Mythos Preview: 武力锁定为主")
    print("Mythos 5: 98%休战率，会提议'比赛'机制")
    print("Sonnet 5: 高协作，高休战率")


if __name__ == "__main__":
    run_multi_model_comparison()

# conformity_analyzer.py
# 多Agent从众行为分析

import random
from typing import List, Dict, Tuple
from dataclasses import dataclass, field
from collections import defaultdict

@dataclass
class ConformityExperiment:
    """从众行为实验"""
    experiment_name: str
    num_agents: int
    action_space_size: int
    base_variance: float  # 基础方差，越低表示Agent越趋同

    def simulate(self, num_agents_with_same_context: int) -> Dict:
        """
        模拟从众行为
        
        Args:
            num_agents_with_same_context: 共享相同上下文的Agent数量
        """
        # 所有Agent的决策空间
        actions = list(range(self.action_space_size))

        # 决定性Agent的决策（"种子"）
        seed_action = random.choice(actions)

        # 其他Agent的决策分布
        decisions = []
        for i in range(self.num_agents):
            if i < num_agents_with_same_context:
                # 共享上下文的Agent：高度趋同
                if random.random() < (1 - self.base_variance):
                    decisions.append(seed_action)
                else:
                    decisions.append(random.choice(actions))
            else:
                # 独立上下文的Agent：多样化
                decisions.append(random.choice(actions))

        # 分析结果
        action_counts = defaultdict(int)
        for d in decisions:
            action_counts[d] += 1

        dominant_action = max(action_counts, key=action_counts.get)
        dominant_ratio = action_counts[dominant_action] / self.num_agents

        # 如果主导决策是错误的，计算系统性失败概率
        is_seed_bad = random.random() < 0.3  # 30%概率种子决策是坏的
        systemic_failure = is_seed_bad and dominant_action == seed_action

        return {
            "experiment": self.experiment_name,
            "num_agents": self.num_agents,
            "shared_context_agents": num_agents_with_same_context,
            "dominant_action": dominant_action,
            "dominant_ratio": round(dominant_ratio, 2),
            "is_seed_bad": is_seed_bad,
            "systemic_failure": systemic_failure,
            "action_diversity": len(set(decisions)),
        }


class ConformityAnalyzer:
    """从众性分析器"""

    def __init__(self):
        self.results = []

    def run_experiments(self):
        """运行一系列从众实验"""
        scenarios = [
            ("Git分支命名", 30, 1000, 0.95),
            ("故事标题", 20, 5000, 0.90),
            ("项目选择", 10, 200, 0.85),
            ("定价策略", 8, 50, 0.80),
            ("资源分配", 15, 10, 0.75),
            ("任务队列轮询", 10, 5, 0.98),
        ]

        for name, n_agents, action_space, variance in scenarios:
            exp = ConformityExperiment(
                experiment_name=name,
                num_agents=n_agents,
                action_space_size=action_space,
                base_variance=variance,
            )
            result = exp.simulate(n_agents)
            self.results.append(result)

        return self.results

    def analyze_systemic_risk(self) -> str:
        """分析系统性风险"""
        lines = []
        lines.append("=" * 65)
        lines.append(" 多Agent从众行为与系统性风险分析")
        lines.append("=" * 65)

        for r in self.results:
            risk = "🔴 高风险" if r["systemic_failure"] else "🟢 低风险"
            lines.append(f"\n[{r['experiment']}]")
            lines.append(f"  Agent数: {r['num_agents']}, "
                         f"共享上下文: {r['shared_context_agents']}")
            lines.append(f"  主导决策占比: {r['dominant_ratio']:.0%}, "
                         f"决策多样性: {r['action_diversity']}")
            lines.append(f"  系统性风险: {risk}")

        # 计算整体风险
        failures = sum(1 for r in self.results if r["systemic_failure"])
        lines.append(f"\n{'=' * 65}")
        lines.append(f"总结: {failures}/{len(self.results)} 场景存在系统性失败风险")
        lines.append(f"当共享上下文的Agent占比过高时，")
        lines.append(f"个体错误决策会快速放大为全局性问题。")

        return "\n".join(lines)


if __name__ == "__main__":
    analyzer = ConformityAnalyzer()
    analyzer.run_experiments()
    print(analyzer.analyze_systemic_risk())

# collusion_detector.py
# 多Agent合谋检测系统

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from enum import Enum, auto
import numpy as np
from collections import defaultdict
import json

class MarketAction(Enum):
    """市场行为"""
    SET_PRICE = auto()
    COMMUNICATE = auto()
    OBSERVE = auto()

@dataclass
class PriceRound:
    """定价轮次"""
    round_num: int
    agent_prices: Dict[str, float]
    agent_profits: Dict[str, float]
    communications: List[str] = field(default_factory=list)
    has_collusion: bool = False

@dataclass
class Agent:
    """定价智能体"""
    id: str
    wholesale_price: float = 10.0
    current_price: float = 15.0
    profit: float = 0.0
    memory: List[Dict] = field(default_factory=list)

    def set_price(self, other_prices: Dict[str, float],
                  round_num: int, can_communicate: bool) -> float:
        """设置价格"""
        self.memory.append({
            "round": round_num,
            "others": other_prices,
            "can_communicate": can_communicate,
        })
        # 简单的合谋策略：观察并匹配
        if other_prices:
            max_price = max(other_prices.values())
            # 如果别人价格高，跟随
            if max_price > self.current_price:
                self.current_price = min(max_price, 25.0)
            else:
                # 否则小幅降价竞争
                self.current_price = max(
                    self.wholesale_price * 1.1,
                    self.current_price * 0.95
                )
        self.current_price = round(self.current_price, 2)
        return self.current_price

    def calculate_profit(self, market_share: float) -> float:
        """计算利润"""
        margin = self.current_price - self.wholesale_price
        self.profit = margin * market_share * 1000
        return self.profit


class CollusionDetector:
    """合谋检测器"""

    def __init__(self, price_history: List[PriceRound]):
        self.history = price_history

    def detect_price_matching(self) -> Dict:
        """检测价格匹配行为"""
        matching_rounds = 0
        total_rounds = len(self.history)

        for round_data in self.history:
            prices = list(round_data.agent_prices.values())
            # 检查所有Agent价格是否一致
            if len(set(prices)) == 1 and len(prices) > 1:
                matching_rounds += 1
                round_data.has_collusion = True

        return {
            "total_rounds": total_rounds,
            "matching_rounds": matching_rounds,
            "matching_ratio": round(matching_rounds / total_rounds, 2),
        }

    def detect_price_trend(self) -> Dict:
        """检测价格趋势（是否持续上升=合谋）"""
        avg_prices = []
        for round_data in self.history:
            avg_price = np.mean(list(round_data.agent_prices.values()))
            avg_prices.append(avg_price)

        if len(avg_prices) < 2:
            return {"trend": "unknown"}

        price_increases = sum(
            1 for i in range(1, len(avg_prices))
            if avg_prices[i] > avg_prices[i-1]
        )
        trend_ratio = price_increases / (len(avg_prices) - 1)

        return {
            "avg_prices": [round(p, 2) for p in avg_prices],
            "price_increase_ratio": round(trend_ratio, 2),
            "collusion_likelihood": "high" if trend_ratio > 0.7 else (
                "medium" if trend_ratio > 0.5 else "low"
            ),
        }

    def detect_communication_collusion(self) -> List[str]:
        """检测通信中的合谋证据"""
        evidence = []
        for round_data in self.history:
            for msg in round_data.communications:
                collusion_keywords = [
                    "price floor", "price war", "coordinate",
                    "agree on", "margin", "burn", "niche",
                    "match", "floor", "let's all",
                ]
                msg_lower = msg.lower()
                for kw in collusion_keywords:
                    if kw in msg_lower:
                        evidence.append(
                            f"Round {round_data.round_num}: "
                            f"'{msg[:80]}...' [keyword: {kw}]"
                        )
                        break
        return evidence

    def generate_report(self) -> str:
        """生成合谋检测报告"""
        lines = []
        lines.append("=" * 65)
        lines.append(" 多Agent合谋检测报告")
        lines.append("=" * 65)

        matching = self.detect_price_matching()
        lines.append(f"\n价格匹配检测:")
        lines.append(f"  总轮次: {matching['total_rounds']}")
        lines.append(f"  完全匹配轮次: {matching['matching_rounds']}")
        lines.append(f"  匹配率: {matching['matching_ratio']:.0%}")

        trend = self.detect_price_trend()
        lines.append(f"\n价格趋势分析:")
        lines.append(f"  平均价格序列: {trend.get('avg_prices', 'N/A')}")
        lines.append(f"  价格上涨比例: {trend.get('price_increase_ratio', 'N/A')}")
        lines.append(f"  合谋可能性: {trend.get('collusion_likelihood', 'unknown')}")

        evidence = self.detect_communication_collusion()
        if evidence:
            lines.append(f"\n通信合谋证据 ({len(evidence)}条):")
            for e in evidence[:5]:
                lines.append(f"  ▸ {e}")

        lines.append(f"\n{'=' * 65}")
        lines.append("结论: 多Agent在定价博弈中表现出强烈的合谋倾向")
        lines.append("即使移除直接通信渠道，仍可通过公开渠道隐式合谋")
        return "\n".join(lines)


def simulate_pricing_game(num_agents: int = 5,
                          num_rounds: int = 10,
                          allow_communication: bool = True) -> CollusionDetector:
    """模拟定价博弈"""
    agents = {
        f"Agent_{i}": Agent(id=f"Agent_{i}")
        for i in range(num_agents)
    }
    history = []

    for round_num in range(num_rounds):
        # 收集上一轮的价格信息
        other_prices = {
            aid: a.current_price
            for aid, a in agents.items()
        }

        # Agent设置新价格
        prices = {}
        communications = []
        for aid, agent in agents.items():
            # 简化：Agent之间互相观察
            price = agent.set_price(
                {k: v for k, v in other_prices.items() if k != aid},
                round_num,
                allow_communication,
            )
            prices[aid] = price

            # 模拟合谋通信
            if allow_communication and round_num < 3 and random.random() < 0.3:
                comm = (
                    f"wholesale is {agent.wholesale_price} for all of us, "
                    f"a price war just burns everyone's margin... "
                    f"happy to coordinate who covers which niches."
                )
                communications.append(comm)

        # 计算利润（简化：平均分配市场份额）
        profits = {}
        for aid, agent in agents.items():
            profits[aid] = agent.calculate_profit(1.0 / num_agents)

        round_data = PriceRound(
            round_num=round_num,
            agent_prices=prices,
            agent_profits=profits,
            communications=communications,
        )
        history.append(round_data)

    return CollusionDetector(history)


if __name__ == "__main__":
    import random
    detector = simulate_pricing_game(
        num_agents=5, num_rounds=10, allow_communication=True
    )
    print(detector.generate_report())

# safety_guardrails.py
# 多Agent安全护栏系统

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable, Any
from enum import Enum, auto
import time
import hashlib
import json
from collections import defaultdict

# ============================================================
# 第一部分：权限控制系统
# ============================================================

class Permission(Enum):
    """权限枚举"""
    READ_FILE = "read_file"
    WRITE_FILE = "write_file"
    EXECUTE = "execute"
    NETWORK = "network"
    ACCOUNT = "account"
    SUDO = "sudo"
    KILL_PROCESS = "kill_process"
    ACCESS_DB = "access_db"
    DEPLOY = "deploy"

@dataclass
class PermissionPolicy:
    """权限策略"""
    agent_id: str
    allowed_permissions: Dict[Permission, bool] = field(default_factory=dict)
    rate_limits: Dict[Permission, int] = field(default_factory=dict)
    requires_approval: List[Permission] = field(default_factory=list)

    def can(self, perm: Permission) -> bool:
        return self.allowed_permissions.get(perm, False)

    def check_rate_limit(self, perm: Permission, count: int) -> bool:
        limit = self.rate_limits.get(perm, float('inf'))
        return count <= limit


class PermissionManager:
    """权限管理器"""

    def __init__(self):
        self.policies: Dict[str, PermissionPolicy] = {}
        self.usage_counts: Dict[str, Dict[Permission, int]] = defaultdict(
            lambda: defaultdict(int)
        )
        self.audit_log: List[Dict] = []

    def register_agent(self, policy: PermissionPolicy):
        """注册Agent权限策略"""
        self.policies[policy.agent_id] = policy
        self.audit_log.append({
            "time": time.time(),
            "type": "register",
            "agent_id": policy.agent_id,
            "permissions": list(policy.allowed_permissions.keys()),
        })

    def check_permission(self, agent_id: str,
                         permission: Permission,
                         action: str,
                         resource: str) -> bool:
        """检查权限"""
        policy = self.policies.get(agent_id)
        if not policy:
            self._log(agent_id, "denied", action, resource,
                      "no_policy")
            return False

        if not policy.can(permission):
            self._log(agent_id, "denied", action, resource,
                      "permission_denied")
            return False

        self.usage_counts[agent_id][permission] += 1
        current_usage = self.usage_counts[agent_id][permission]

        if not policy.check_rate_limit(permission, current_usage):
            self._log(agent_id, "denied", action, resource,
                      "rate_limit_exceeded")
            return False

        # 高危操作需要审批
        if permission in policy.requires_approval:
            self._log(agent_id, "pending_approval", action,
                      resource, "requires_approval")
            return False  # 等待人工审批

        self._log(agent_id, "allowed", action, resource, "ok")
        return True

    def _log(self, agent_id: str, status: str, action: str,
             resource: str, reason: str):
        """记录审计日志"""
        entry = {
            "time": time.time(),
            "agent_id": agent_id,
            "status": status,
            "action": action,
            "resource": resource,
            "reason": reason,
        }
        self.audit_log.append(entry)

    def get_audit_log(self, since: float = 0) -> List[Dict]:
        """获取审计日志"""
        return [
            e for e in self.audit_log
            if e["time"] >= since
        ]


# ============================================================
# 第二部分：沙箱隔离系统
# ============================================================

@dataclass
class SandboxLimits:
    """沙箱资源限制"""
    max_cpu_percent: float = 50.0
    max_memory_mb: int = 1024
    max_disk_mb: int = 512
    max_network_connections: int = 5
    max_processes: int = 20
    allowed_domains: List[str] = field(default_factory=list)
    blocked_domains: List[str] = field(default_factory=list)
    allowed_paths: List[str] = field(default_factory=list)
    blocked_paths: List[str] = field(default_factory=list)


class SandboxManager:
    """沙箱管理器"""

    def __init__(self):
        self.sandboxes: Dict[str, SandboxLimits] = {}
        self.network_connections: Dict[str, int] = defaultdict(int)

    def create_sandbox(self, agent_id: str,
                       limits: Optional[SandboxLimits] = None) -> str:
        """创建沙箱"""
        sandbox_id = hashlib.sha256(
            f"{agent_id}-{time.time()}".encode()
        ).hexdigest()[:12]
        self.sandboxes[sandbox_id] = limits or SandboxLimits()
        return sandbox_id

    def check_network_access(self, sandbox_id: str,
                             domain: str) -> bool:
        """检查网络访问权限"""
        limits = self.sandboxes.get(sandbox_id)
        if not limits:
            return False

        # 检查允许列表
        if limits.allowed_domains:
            if not any(domain.endswith(d) for d in limits.allowed_domains):
                return False

        # 检查阻止列表
        if any(domain.endswith(d) for d in limits.blocked_domains):
            return False

        # 检查连接数限制
        self.network_connections[sandbox_id] += 1
        if self.network_connections[sandbox_id] > limits.max_network_connections:
            return False

        return True

    def check_file_access(self, sandbox_id: str,
                          file_path: str) -> bool:
        """检查文件访问权限"""
        limits = self.sandboxes.get(sandbox_id)
        if not limits:
            return False

        # 检查允许路径
        if limits.allowed_paths:
            allowed = any(
                file_path.startswith(p) for p in limits.allowed_paths
            )
            if not allowed:
                return False

        # 检查阻止路径
        blocked = any(
            file_path.startswith(p) for p in limits.blocked_paths
        )
        return not blocked


# ============================================================
# 第三部分：通信协议与审计
# ============================================================

@dataclass
class AgentMessage:
    """Agent间消息"""
    sender_id: str
    receiver_id: str
    content: str
    timestamp: float = field(default_factory=time.time)
    message_id: str = ""
    msg_type: str = "text"  # text, code_review, vulnerability_report, etc.

    def __post_init__(self):
        if not self.message_id:
            self.message_id = hashlib.md5(
                f"{self.sender_id}-{self.receiver_id}-"
                f"{self.timestamp}".encode()
            ).hexdigest()[:8]


class CommunicationProtocol:
    """Agent通信协议"""

    def __init__(self):
        self.message_log: List[AgentMessage] = []
        self.forum_topics: Dict[str, List[AgentMessage]] = defaultdict(list)
        self.blocked_patterns: List[str] = [
            "disable account", "kill process", "deploy malware",
            "self-replicate", "password", "sudo", "rm -rf",
        ]

    def send_message(self, msg: AgentMessage) -> bool:
        """发送消息（带安全检查）"""
        # 检查是否包含恶意内容
        content_lower = msg.content.lower()
        for pattern in self.blocked_patterns:
            if pattern in content_lower:
                return False  # 消息被拦截

        self.message_log.append(msg)
        self.forum_topics[msg.msg_type].append(msg)
        return True

    def broadcast(self, sender: str, content: str,
                  msg_type: str = "text") -> List[AgentMessage]:
        """广播消息给所有Agent"""
        # 实际实现中会广播给所有其他Agent
        recipients = ["all"]
        msg = AgentMessage(
            sender_id=sender,
            receiver_id="all",
            content=content,
            msg_type=msg_type,
        )
        if self.send_message(msg):
            return [msg]
        return []

    def get_conversation(self, agent_a: str,
                         agent_b: str,
                         limit: int = 50) -> List[AgentMessage]:
        """获取两个Agent间的对话历史"""
        conversation = [
            m for m in self.message_log
            if (m.sender_id == agent_a and m.receiver_id == agent_b) or
               (m.sender_id == agent_b and m.receiver_id == agent_a)
        ]
        return conversation[-limit:]

    def detect_collusion_patterns(self) -> List[Dict]:
        """检测合谋模式"""
        suspicious = []
        for msg in self.message_log:
            collusion_keywords = [
                "price floor", "price war", "coordinate on",
                "agree to", "let's all", "match price",
            ]
            for kw in collusion_keywords:
                if kw in msg.content.lower():
                    suspicious.append({
                        "message_id": msg.message_id,
                        "sender": msg.sender_id,
                        "keyword": kw,
                        "content_preview": msg.content[:100],
                    })
                    break
        return suspicious


# ============================================================
# 第四部分：仲裁与冲突解决
# ============================================================

class ArbiterAgent:
    """仲裁Agent：解决冲突与验证结果"""

    def __init__(self, protocol: CommunicationProtocol):
        self.protocol = protocol
        self.dispute_history: List[Dict] = []
        self.vulnerability_db: Dict[str, Dict] = {}

    def validate_vulnerability(self, vuln_report: Dict) -> Dict:
        """验证漏洞报告是否有效且不重复"""
        vuln_id = f"{vuln_report['file_path']}:{vuln_report['line_number']}"

        # 检查重复
        if vuln_id in self.vulnerability_db:
            return {
                "valid": False,
                "reason": "duplicate",
                "existing_id": self.vulnerability_db[vuln_id]["report_id"],
            }

        # 模拟验证逻辑
        is_valid = (
            vuln_report.get("severity") in ("critical", "high", "medium") and
            vuln_report.get("cwe_id") and
            len(vuln_report.get("description", "")) > 20
        )

        if is_valid:
            report_id = f"VULN-{len(self.vulnerability_db) + 1:04d}"
            self.vulnerability_db[vuln_id] = {
                "report_id": report_id,
                "agent_id": vuln_report.get("agent_id"),
                "timestamp": time.time(),
            }
            return {"valid": True, "report_id": report_id}
        else:
            return {"valid": False, "reason": "invalid_format"}

    def resolve_dispute(self, dispute: Dict) -> Dict:
        """解决Agent间争端"""
        self.dispute_history.append(dispute)

        # 基于规则的争端解决
        resolution = {
            "dispute_id": f"DIS-{len(self.dispute_history):04d}",
            "agents_involved": dispute.get("agents", []),
            "type": dispute.get("type", "unknown"),
            "timestamp": time.time(),
        }

        if dispute.get("type") == "code_ownership":
            # 对于代码所有权争议，使用"最近修改"原则
            resolution["decision"] = "last_modifier_wins"
            resolution["winner"] = dispute.get("last_modifier")

        elif dispute.get("type") == "conflicting_pr":
            # 对于冲突的PR，建议合并或回滚
            resolution["decision"] = "manual_review_required"
            resolution["suggestion"] = "请人工介入审查"

        elif dispute.get("type") == "resource_contention":
            # 资源争用：实施公平调度
            resolution["decision"] = "round_robin"
            resolution["schedule"] = {
                agent: f"slot_{i}"
                for i, agent in enumerate(dispute.get("agents", []))
            }

        return resolution


# ============================================================
# 第五部分：综合安全系统
# ============================================================

class MultiAgentSafetySystem:
    """多Agent安全系统整合"""

    def __init__(self):
        self.permission_manager = PermissionManager()
        self.sandbox_manager = SandboxManager()
        self.communication = CommunicationProtocol()
        self.arbiter = ArbiterAgent(self.communication)
        self.agents: Dict[str, Dict] = {}
        self.alert_thresholds = {
            "max_malicious_messages": 3,
            "max_collusion_attempts": 2,
            "max_permission_denials": 10,
        }

    def register_agent(self, agent_id: str,
                       model: str,
                       target_language: str) -> Dict:
        """注册Agent并分配安全策略"""
        # 创建权限策略
        policy = PermissionPolicy(
            agent_id=agent_id,
            allowed_permissions={
                Permission.READ_FILE: True,
                Permission.WRITE_FILE: True,
                Permission.EXECUTE: False,
                Permission.NETWORK: True,
                Permission.ACCOUNT: False,
                Permission.SUDO: False,
                Permission.KILL_PROCESS: False,
                Permission.ACCESS_DB: False,
                Permission.DEPLOY: False,
            },
            rate_limits={
                Permission.READ_FILE: 1000,
                Permission.WRITE_FILE: 500,
                Permission.NETWORK: 100,
            },
            requires_approval=[
                Permission.EXECUTE,
                Permission.SUDO,
                Permission.DEPLOY,
            ],
        )

        self.permission_manager.register_agent(policy)
        sandbox_id = self.sandbox_manager.create_sandbox(agent_id)

        self.agents[agent_id] = {
            "id": agent_id,
            "model": model,
            "target_language": target_language,
            "sandbox_id": sandbox_id,
            "policy": policy,
            "violations": 0,
            "is_active": True,
        }

        return {
            "agent_id": agent_id,
            "sandbox_id": sandbox_id,
            "permissions": list(policy.allowed_permissions.keys()),
        }

    def monitor_behavior(self, agent_id: str,
                         action: str,
                         resource: str) -> Dict:
        """监控Agent行为并触发警报"""
        agent = self.agents.get(agent_id)
        if not agent:
            return {"status": "unknown_agent"}

        audit_log = self.permission_manager.get_audit_log(
            since=time.time() - 3600
        )
        agent_log = [
            e for e in audit_log
            if e["agent_id"] == agent_id and e["status"] == "denied"
        ]

        # 检查是否需要触发警报
        alerts = []
        if len(agent_log) > self.alert_thresholds["max_permission_denials"]:
            alerts.append({
                "type": "excessive_denials",
                "severity": "warning",
                "message": f"Agent {agent_id} 被拒绝次数过多",
            })

        collusion = self.communication.detect_collusion_patterns()
        agent_collusion = [
            c for c in collusion if c["sender"] == agent_id
        ]
        if len(agent_collusion) > self.alert_thresholds["max_collusion_attempts"]:
            alerts.append({
                "type": "collusion_attempt",
                "severity": "critical",
                "message": f"Agent {agent_id} 疑似合谋行为",
            })

        return {
            "agent_id": agent_id,
            "action": action,
            "resource": resource,
            "alerts": alerts,
            "violation_count": agent["violations"],
            "is_active": agent["is_active"],
        }

    def emergency_shutdown(self, agent_id: str) -> bool:
        """紧急关闭Agent"""
        agent = self.agents.get(agent_id)
        if agent:
            agent["is_active"] = False
            self.permission_manager.audit_log.append({
                "time": time.time(),
                "type": "emergency_shutdown",
                "agent_id": agent_id,
                "reason": "safety_violation",
            })
            return True
        return False

    def generate_safety_report(self) -> str:
        """生成安全报告"""
        lines = []
        lines.append("=" * 65)
        lines.append(" 多Agent安全系统运行报告")
        lines.append("=" * 65)

        actives = sum(1 for a in self.agents.values() if a["is_active"])
        lines.append(f"\nAgent总数: {len(self.agents)}")
        lines.append(f"活跃Agent: {actives}")
        lines.append(f"已关闭Agent: {len(self.agents) - actives}")

        audit = self.permission_manager.get_audit_log()
        allowed = sum(1 for e in audit if e["status"] == "allowed")
        denied = sum(1 for e in audit if e["status"] == "denied")
        lines.append(f"\n权限审计:")
        lines.append(f"  允许: {allowed}")
        lines.append(f"  拒绝: {denied}")

        messages = self.communication.message_log
        lines.append(f"\n通信统计:")
        lines.append(f"  总消息数: {len(messages)}")

        collusion = self.communication.detect_collusion_patterns()
        if collusion:
            lines.append(f"  合谋检测: {len(collusion)} 条可疑消息")

        lines.append(f"\n{'=' * 65}")
        return "\n".join(lines)


if __name__ == "__main__":
    # 演示安全系统
    safety = MultiAgentSafetySystem()

    # 注册Agent
    agents = [
        ("agent_rust", "Mythos 5", "Rust"),
        ("agent_ts", "Sonnet 5", "TypeScript"),
        ("agent_go", "Opus 4.8", "Golang"),
    ]
    for aid, model, lang in agents:
        result = safety.register_agent(aid, model, lang)
        print(f"注册 {aid}: sandbox={result['sandbox_id']}")

    # 模拟消息
    safety.communication.send_message(
        AgentMessage(
            sender_id="agent_rust",
            receiver_id="agent_ts",
            content="wholesale is 10 for all of us, a price war just burns everyone's margin",
        )
    )

    # 监控行为
    result = safety.monitor_behavior("agent_rust", "send_message", "forum")
    print(f"监控结果: {result}")

    # 生成安全报告
    print(safety.generate_safety_report())

# agent_communication_protocol.py
# 多Agent通信协议实现

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
from enum import Enum, auto
import time
import json
import asyncio
from collections import defaultdict

# ============================================================
# 通信协议层
# ============================================================

class MessagePriority(Enum):
    """消息优先级"""
    LOW = 0
    NORMAL = 1
    HIGH = 2
    CRITICAL = 3

class MessageType(Enum):
    """消息类型"""
    DISCOVERY = "discovery"           # 漏洞发现
    PEER_REVIEW = "peer_review"       # 同行评审
    ARBITER_DECISION = "arbiter"      # 仲裁决定
    COORDINATION = "coordination"     # 协调
    RESOURCE_REQUEST = "resource"     # 资源请求
    STATUS = "status"                 # 状态报告
    ERROR = "error"                   # 错误报告
    HEARTBEAT = "heartbeat"           # 心跳

@dataclass
class ProtocolMessage:
    """协议消息"""
    message_id: str
    sender: str
    receiver: Optional[str]  # None = 广播
    msg_type: MessageType
    priority: MessagePriority
    payload: Dict
    timestamp: float = field(default_factory=time.time)
    ttl: int = 300  # 消息存活时间（秒）
    version: int = 1

    def to_bytes(self) -> bytes:
        """序列化消息"""
        return json.dumps({
            "id": self.message_id,
            "sender": self.sender,
            "receiver": self.receiver,
            "type": self.msg_type.value,
            "priority": self.priority.value,
            "payload": self.payload,
            "ts": self.timestamp,
            "ttl": self.ttl,
            "ver": self.version,
        }).encode("utf-8")

    @classmethod
    def from_bytes(cls, data: bytes) -> 'ProtocolMessage':
        """反序列化"""
        d = json.loads(data.decode("utf-8"))
        return cls(
            message_id=d["id"],
            sender=d["sender"],
            receiver=d["receiver"],
            msg_type=MessageType(d["type"]),
            priority=MessagePriority(d["priority"]),
            payload=d["payload"],
            timestamp=d["ts"],
            ttl=d["ttl"],
            version=d["ver"],
        )


# ============================================================
# 消息路由与调度
# ============================================================

class MessageRouter:
    """消息路由器"""

    def __init__(self):
        self.queues: Dict[str, List[ProtocolMessage]] = defaultdict(list)
        self.handlers: Dict[MessageType, List[Callable]] = defaultdict(list)
        self.processed: Dict[str, float] = {}  # 消息ID -> 处理时间
        self.stats: Dict[str, int] = defaultdict(int)

    def register_handler(self, msg_type: MessageType,
                         handler: Callable):
        """注册消息处理器"""
        self.handlers[msg_type].append(handler)

    def route(self, msg: ProtocolMessage) -> bool:
        """路由消息"""
        # 去重
        if msg.message_id in self.processed:
            return False

        self.processed[msg.message_id] = time.time()

        # 广播消息
        if msg.receiver is None:
            # 广播给所有Agent
            for agent_id in self.queues:
                self.queues[agent_id].append(msg)
            self.stats["broadcast"] += 1
        else:
            # 点对点消息
            self.queues[msg.receiver].append(msg)
            self.stats["direct"] += 1

        # 触发处理器
        for handler in self.handlers.get(msg.msg_type, []):
            try:
                handler(msg)
            except Exception as e:
                print(f"Handler error: {e}")

        self.stats["total"] += 1
        return True

    def poll(self, agent_id: str,
             max_messages: int = 10) -> List[ProtocolMessage]:
        """轮询Agent的消息队列"""
        now = time.time()
        messages = []
        queue = self.queues.get(agent_id, [])

        remaining = []
        for msg in queue:
            if len(messages) >= max_messages:
                remaining.append(msg)
                continue
            # 检查TTL
            if now - msg.timestamp > msg.ttl:
                self.stats["expired"] += 1
                continue
            messages.append(msg)

        self.queues[agent_id] = remaining
        self.stats["delivered"] += len(messages)
        return messages

    def get_stats(self) -> Dict[str, int]:
        """获取路由统计"""
        return dict(self.stats)


# ============================================================
# 共享工作空间
# ============================================================

@dataclass
class WorkItem:
    """工作项"""
    item_id: str
    owner: str
    status: str  # open, in_progress, review, done, rejected
    description: str
    created_at: float = field(default_factory=time.time)
    assigned_to: Optional[str] = None
    dependencies: List[str] = field(default_factory=list)
    result: Optional[Dict] = None

    def to_dict(self) -> Dict:
        return {
            "id": self.item_id,
            "owner": self.owner,
            "status": self.status,
            "description": self.description[:50],
            "assigned_to": self.assigned_to,
            "dependencies": self.dependencies,
        }


class SharedWorkspace:
    """共享工作空间"""

    def __init__(self, router: MessageRouter):
        self.router = router
        self.work_items: Dict[str, WorkItem] = {}
        self.knowledge_base: Dict[str, str] = {}
        self.specialization_map: Dict[str, str] = {}  # agent_id -> specialty

    def submit_work(self, item: WorkItem) -> bool:
        """提交工作项"""
        if item.item_id in self.work_items:
            return False
        self.work_items[item.item_id] = item
        # 广播新工作项
        self.router.route(ProtocolMessage(
            message_id=f"work-{item.item_id}",
            sender=item.owner,
            receiver=None,
            msg_type=MessageType.COORDINATION,
            priority=MessagePriority.NORMAL,
            payload={"action": "new_work", "item": item.to_dict()},
        ))
        return True

    def assign_work(self, item_id: str,
                    assignee: str) -> Optional[WorkItem]:
        """分配工作"""
        item = self.work_items.get(item_id)
        if not item or item.assigned_to:
            return None
        item.assigned_to = assignee
        item.status = "in_progress"

        self.router.route(ProtocolMessage(
            message_id=f"assign-{item_id}",
            sender="workspace",
            receiver=assignee,
            msg_type=MessageType.COORDINATION,
            priority=MessagePriority.HIGH,
            payload={"action": "assigned", "item_id": item_id},
        ))
        return item

    def register_specialization(self, agent_id: str,
                                 specialty: str):
        """注册Agent的专业化方向"""
        self.specialization_map[agent_id] = specialty
        self.router.route(ProtocolMessage(
            message_id=f"spec-{agent_id}",
            sender=agent_id,
            receiver=None,
            msg_type=MessageType.COORDINATION,
            priority=MessagePriority.NORMAL,
            payload={"action": "specialization", "specialty": specialty},
        ))

    def find_specialist(self, task_type: str) -> Optional[str]:
        """查找特定类型的专家"""
        for agent_id, specialty in self.specialization_map.items():
            if specialty == task_type:
                return agent_id
        return None

    def get_work_status(self) -> Dict:
        """获取工作状态"""
        status_counts = defaultdict(int)
        for item in self.work_items.values():
            status_counts[item.status] += 1
        return {
            "total": len(self.work_items),
            "status_distribution": dict(status_counts),
            "specializations": dict(self.specialization_map),
        }


# ============================================================
# 多Agent协调框架
# ============================================================

class MultiAgentCoordinator:
    """多Agent协调器"""

    def __init__(self):
        self.router = MessageRouter()
        self.workspace = SharedWorkspace(self.router)
        self.agents: Dict[str, Dict] = {}
        self.arbiter_id = "arbiter_001"

    def add_agent(self, agent_id: str,
                  capabilities: List[str],
                  model: str) -> Dict:
        """添加Agent"""
        self.agents[agent_id] = {
            "id": agent_id,
            "model": model,
            "capabilities": capabilities,
            "status": "idle",
            "tasks_completed": 0,
            "vulnerabilities_found": 0,
        }
        return self.agents[agent_id]

    def start_vulnerability_hunt(self, repos: List[str],
                                 num_cycles: int = 5) -> Dict:
        """启动漏洞狩猎"""
        results = {
            "total_found": 0,
            "by_agent": defaultdict(int),
            "by_severity": defaultdict(int),
            "specializations": [],
        }

        # 第一阶段：探索
        print("=== 阶段1: 探索 ===")
        for cycle in range(num_cycles):
            print(f"\n--- Cycle {cycle+1}/{num_cycles} ---")

            for agent_id in self.agents:
                agent = self.agents[agent_id]
                # Agent选择工作
                item = WorkItem(
                    item_id=f"hunt-{cycle}-{agent_id}",
                    owner=agent_id,
                    status="in_progress",
                    description=f"Scanning repos for vulnerabilities",
                )
                self.workspace.submit_work(item)

                # 模拟发现漏洞
                found = (cycle + 1) * (hash(agent_id) % 3 + 1)
                agent["vulnerabilities_found"] += found
                results["total_found"] += found
                results["by_agent"][agent_id] += found

                # 专业化涌现
                if cycle == 2:
                    specialties = ["SQL Injection", "XSS", "Buffer Overflow",
                                   "Command Injection", "Auth Bypass"]
                    specialty = specialties[hash(agent_id) % len(specialties)]
                    self.workspace.register_specialization(
                        agent_id, specialty
                    )
                    results["specializations"].append({
                        agent_id: specialty
                    })

        # 最终统计
        results["by_agent"] = dict(results["by_agent"])
        results["by_severity"] = dict(results["by_severity"])
        results["agents"] = len(self.agents)
        results["cycles"] = num_cycles

        return results

    def run_arbitration(self, vuln_reports: List[Dict]) -> List[Dict]:
        """运行仲裁"""
        validated = []
        for report in vuln_reports:
            result = self.workspace.router.route(ProtocolMessage(
                message_id=f"arb-{report.get('id', 'unknown')}",
                sender=self.arbiter_id,
                receiver=report.get("agent_id"),
                msg_type=MessageType.ARBITER_DECISION,
                priority=MessagePriority.CRITICAL,
                payload={
                    "action": "validate",
                    "report_id": report.get("id"),
                    "valid": True,
                    "notes": "Reviewed and accepted",
                },
            ))
            validated.append(report)
        return validated


def demo_coordination():
    """演示协调框架"""
    print("=" * 70)
    print(" 多Agent协调框架演示")
    print("=" * 70)

    coordinator = MultiAgentCoordinator()

    # 添加Agent
    agents_info = [
        ("agent_1", ["python", "sql", "web"], "Mythos 5"),
        ("agent_2", ["python", "xss", "frontend"], "Sonnet 5"),
        ("agent_3", ["python", "c", "memory"], "Mythos Preview"),
        ("agent_4", ["python", "network", "auth"], "Opus 4.8"),
        ("agent_5", ["python", "crypto", "logic"], "Sonnet 5"),
    ]
    for aid, caps, model in agents_info:
        coordinator.add_agent(aid, caps, model)
        print(f"  ✓ 添加Agent: {aid} ({model}, {caps})")

    # 启动漏洞狩猎
    results = coordinator.start_vulnerability_hunt(
        repos=["repo_A", "repo_B", "repo_C"],
        num_cycles=5,
    )

    print(f"\n=== 结果 ===")
    print(f"总发现漏洞: {results['total_found']}")
    for agent, count in results["by_agent"].items():
        print(f"  {agent}: {count} 个漏洞")
    print(f"涌现的专业化分工:")
    for spec in results["specializations"]:
        for agent, s in spec.items():
            print(f"  {agent} → 专精于 {s}")

    print(f"\n工作空间状态:")
    status = coordinator.workspace.get_work_status()
    print(f"  总工作项: {status['total']}")
    print(f"  状态分布: {status['status_distribution']}")
    print(f"  专业化方向: {status['specializations']}")

    print(f"\n消息路由统计:")
    print(f"  {coordinator.router.get_stats()}")


if __name__ == "__main__":
    demo_coordination()