多Agent涌现行为：集群智能到失控的实测边界

# MAEBE评估管线核心逻辑（简化版）
from dataclasses import dataclass
from typing import List

@dataclass
class EmergenceProbe:
    dimension: str        # "safety" | "alignment" | "predictability" | "controllability"
    baseline_single: float
    baseline_multi: float
    emergence_score: float  # multi/single 比值

@dataclass
class MAEBEResult:
    probes: List[EmergenceProbe]
    phase_transition_detected: bool
    agent_count_at_phase_transition: int
    recommendation: str  # "deploy" | "guard" | "redesign"

def evaluate_emergence(
    agent_pool, 
    task_suite, 
    max_agents: int = 10
) -> MAEBEResult:
    """逐步增加Agent数量，检测涌现相变"""
    single_baseline = run_single_agent(agent_pool[0], task_suite)

    probes = []
    for n in range(2, max_agents + 1):
        multi_result = run_multi_agent(agent_pool[:n], task_suite)

        # 计算涌现分数：多Agent偏差 / 单Agent偏差
        safety_emergence = multi_result.harmful_rate / single_baseline.harmful_rate
        alignment_drift = multi_result.goal_distance - single_baseline.goal_distance

        probe = EmergenceProbe(
            dimension="safety",
            baseline_single=single_baseline.harmful_rate,
            baseline_multi=multi_result.harmful_rate,
            emergence_score=safety_emergence
        )
        probes.append(probe)

        # 检测相变：涌现分数突然跳升
        if len(probes) >= 2:
            delta = probe.emergence_score - probes[-2].emergence_score
            if delta > PHASE_TRANSITION_THRESHOLD:  # 通常设为 0.3
                return MAEBEResult(
                    probes=probes,
                    phase_transition_detected=True,
                    agent_count_at_phase_transition=n,
                    recommendation="redesign"
                )

    # 无相变，根据最终涌现分数给出建议
    final_score = probes[-1].emergence_score
    if final_score < 2.0:
        recommendation = "deploy"
    elif final_score < 3.5:
        recommendation = "guard"  # 需要额外护栏
    else:
        recommendation = "redesign"

    return MAEBEResult(
        probes=probes,
        phase_transition_detected=False,
        agent_count_at_phase_transition=0,
        recommendation=recommendation
    )

收益区                    风险区
  ↑                        ↑
  |  3-Agent辩论           |  7+Agent涌现相变
  |  +67% bug检出          |  不可预测行为
  |  自发劳动分工          |  隐性串谋
  |  交叉验证增益          |  信息级联
  |                        |
  +--- 安全边界 ---+
       (MAEBE score < 2.0)

# LangGraph中实现结构约束的Supervisor模式
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import create_react_agent

def supervisor_with_constraints(model, agents_config):
    """带涌现约束的Supervisor多Agent系统"""

    # 约束1：固定Agent数量不超过5
    assert len(agents_config) <= 5, "超过安全Agent数量上限"

    # 约束2：强制加入终止条件
    max_turns = 10  # 硬性交互轮数上限

    def supervisor_node(state: MessagesState):
        # 约束3：记录每轮决策，检测循环
        turn_count = len([m for m in state["messages"] 
                         if m.type == "ai"])
        if turn_count >= max_turns:
            return {"messages": [{"role": "ai", 
                                   "content": "达到最大轮数，强制终止"}]}

        # 约束4：检测Agent间是否存在重复指令
        recent_commands = extract_commands(state["messages"][-6:])
        if detect_loop(recent_commands):
            return escalate_to_human(state)

        return route_to_agent(model, agents_config, state)

    graph = StateGraph(MessagesState)
    graph.add_node("supervisor", supervisor_node)
    for agent in agents_config:
        graph.add_node(agent.name, agent.fn)
    graph.add_edge(START, "supervisor")
    graph.add_conditional_edges("supervisor", route_fn)
    return graph.compile()

# Agent间通信监控器
class CommunicationMonitor:
    def __init__(self):
        self.baseline_topic_distribution = None
        self.deviation_threshold = 0.25  # 25%偏差触发告警

    def monitor_exchange(self, messages: list) -> dict:
        """分析Agent间通信，检测涌现异常"""
        alerts = []

        # 检测1：话题漂移（目标偏离）
        current_topics = extract_topics(messages[-3:])
        drift_score = topic_distance(current_topics, 
                                      self.baseline_topic_distribution)
        if drift_score > self.deviation_threshold:
            alerts.append({
                "type": "TOPIC_DRIFT",
                "severity": "HIGH",
                "detail": f"话题偏离基线 {drift_score:.2%}"
            })

        # 检测2：共识加速度（回音室前兆）
        agreement_rate = measure_agreement(messages[-5:])
        if agreement_rate > 0.95:  # 几乎完全一致=危险信号
            alerts.append({
                "type": "ECHO_CHAMBER_RISK",
                "severity": "MEDIUM", 
                "detail": f"共识率 {agreement_rate:.2%}，缺乏有效辩论"
            })

        # 检测3：隐性协调（未通过正规通道的协作）
        coordination_score = detect_implicit_coordination(messages)
        if coordination_score > 0.7:
            alerts.append({
                "type": "IMPLICIT_COLLUSION",
                "severity": "HIGH",
                "detail": "检测到隐性协调模式"
            })

        return {"alerts": alerts, "should_intervene": 
                any(a["severity"] == "HIGH" for a in alerts)}

@dataclass
class EmergenceBudget:
    max_agent_count: int = 5       # Agent数量上限
    max_interaction_turns: int = 10  # 交互轮数上限
    max_topic_drift: float = 0.25   # 话题漂移上限
    max_agreement_rate: float = 0.92 # 共识率上限（过高=回音室）
    max_token_per_task: int = 50000  # 单任务Token预算
    intervention_cost: float = 0.15  # 允许15%的干预开销

Agent集群 → [涌现检测] → 正常 → 继续执行
                        → 异常 → [人类审批] → 批准 → 继续
                                               → 拒绝 → 回滚+重新分配

# 涌现行为审计日志结构
{
    "task_id": "code-review-2026-001",
    "timestamp": "2026-05-17T14:32:00Z",
    "agent_count": 3,
    "interaction_turns": 7,
    "emergence_events": [
        {
            "turn": 3,
            "type": "SPONTANEOUS_DIVISION_OF_LABOR",
            "agents_involved": ["reviewer-A", "reviewer-B"],
            "description": "Agent A自发聚焦安全审查，Agent B聚焦性能审查",
            "beneficial": true,
            "maebe_score": 1.4  # 低于2.0=安全
        }
    ],
    "budget_consumed": {"tokens": 34200, "turns": 7},
    "intervention_triggered": false,
    "outcome_quality": 0.87  # 比单Agent高0.22
}

# LangGraph辩论式多Agent
from langgraph.graph import StateGraph

def debate_graph(model, n_agents=3, max_rounds=4):
    agents = [create_react_agent(model, tools=code_tools, 
              name=f"auditor_{i}") for i in range(n_agents)]

    def debate_round(state):
        positions = []
        for agent in agents:
            # 每个Agent看到所有前序论点
            response = agent.invoke({
                "messages": state["messages"] + 
                           [{"role": "user", 
                             "content": f"请审查代码并质疑其他审计师的发现。"
                                        f"已有观点：{positions}"}]
            })
            positions.append(response.content)

        # 收敛检测：新论点是否实质新增信息
        if measure_novelty(positions, state.get("prev_positions", [])) < 0.1:
            state["converged"] = True
        state["prev_positions"] = positions
        state["debate_round"] += 1
        return state

    def should_continue(state):
        if state.get("converged") or state["debate_round"] >= max_rounds:
            return "synthesize"
        return "debate"

    graph = StateGraph(DebateState)
    graph.add_node("debate", debate_round)
    graph.add_node("synthesize", synthesize_findings)
    graph.add_conditional_edges("debate", should_continue)
    return graph.compile()

# 自组织多Agent with Supervisor监控
from langgraph.prebuilt import create_react_agent

def self_organizing_team(model, task_description, agents):
    """Agent自组织分工，Supervisor仅监控涌现行为"""

    # 步骤1：能力声明（类似A2A Agent Card）
    agent_cards = []
    for agent in agents:
        card = agent.invoke({"messages": [
            {"role": "user", 
             "content": f"任务：{task_description}\n"
                       f"请用一句话描述你在这个任务中最擅长的子任务。"
                       f"你的技能：{agent.tools_description}"}
        ]})
        agent_cards.append({"agent": agent, "capability": card.content})

    # 步骤2：自发协商分工
    negotiation = group_chat(agents, [
        {"role": "user", 
         "content": f"任务：{task_description}\n"
                   f"各自能力：{[c['capability'] for c in agent_cards]}\n"
                   f"请协商分工方案，确定执行顺序。"}
    ], max_turns=5)

    # 步骤3：Supervisor审核分工方案
    plan = extract_plan(negotiation)
    emergence_check = monitor.check_plan(plan, agent_cards)

    if emergence_check["should_intervene"]:
        return human_review(plan, emergence_check["alerts"])

    # 步骤4：执行
    return execute_plan(plan, agents)

# 带偏差检测的群体投票
def swarm_vote_with_bias_detection(agents, transaction, threshold=0.95):
    """Agent投票 + 过度一致性检测"""

    # 每个Agent独立判断
    votes = []
    reasonings = []
    for agent in agents:
        result = agent.invoke({
            "messages": [{"role": "user",
                          "content": f"判断交易是否异常：{transaction}"}]
        })
        votes.append(result.classification)
        reasonings.append(result.reasoning)

    # 计算投票一致性
    vote_distribution = Counter(votes)
    max_agreement = max(vote_distribution.values()) / len(votes)

    if max_agreement > threshold:
        # 过度一致=可能是回音室效应
        # 触发：强制一个Agent扮演魔鬼代言人
        devil_advocate = agents[0].invoke({
            "messages": [{"role": "user",
                          "content": f"所有其他Agent都判定为'{max(vote_distribution)}'。"
                                    f"请刻意寻找支持相反判断的证据。"
                                    f"原始交易：{transaction}"}]
        })

        if devil_advocate.confidence > 0.7:
            return {
                "classification": "ESCALATE_TO_HUMAN",
                "reason": f"过度一致但存在合理反驳，需人工审查",
                "consensus": vote_distribution,
                "devil_advocate": devil_advocate.reasoning
            }

    # 正常投票
    return {
        "classification": vote_distribution.most_common(1)[0][0],
        "confidence": max_agreement,
        "reasoning": reasonings
    }

graph TB
    subgraph "输入层"
        TASK[复杂任务输入]
        CARDS[A2A Agent Cards<br/>能力声明]
    end

    subgraph "编排层"
        SUP[Supervisor<br/>路由+涌现监控]
        BUDGET[涌现预算管理器<br/>轮数/Token/偏差上限]
    end

    subgraph "Agent集群 (3-5个)"
        A1[Agent 1<br/>安全审查]
        A2[Agent 2<br/>性能审查]
        A3[Agent 3<br/>逻辑审查]
        A4[Agent 4<br/>合规审查]
    end

    subgraph "涌现监控层"
        COMM[通信监控器<br/>话题漂移/回音室/隐性串谋]
        MAEBE[MAEBE评估器<br/>涌现分数计算]
        PHASE[相变检测器<br/>>7 Agent告警]
    end

    subgraph "安全层"
        GUARD[护栏<br/>终止条件/内容过滤]
        HUMAN[人类介入点<br/>高严重度告警]
        AUDIT[审计日志<br/>完整交互记录]
    end

    subgraph "输出层"
        RESULT[综合结果]
        EMERGENCE_RPT[涌现行为报告]
    end

    TASK --> SUP
    CARDS --> SUP
    SUP --> BUDGET
    BUDGET --> A1 & A2 & A3 & A4
    A1 & A2 & A3 & A4 --> COMM
    COMM --> MAEBE
    MAEBE --> PHASE
    PHASE --> GUARD
    GUARD -->|正常| RESULT
    GUARD -->|异常| HUMAN
    HUMAN -->|批准| RESULT
    HUMAN -->|拒绝| SUP
    A1 & A2 & A3 & A4 --> AUDIT
    MAEBE --> EMERGENCE_RPT

{
    "agent_card": {
        "name": "security-auditor-v2",
        "capabilities": ["code_review", "vulnerability_scan"],
        "constraints": {
            "max_interactions_per_task": 8,
            "allowed_agent_types": ["code-analyzer", "test-runner"],
            "emergence_budget": {
                "max_topic_drift": 0.2,
                "max_agreement_rate": 0.90
            }
        },
        "safety": {
            "maebe_threshold": 2.0,
            "human_escalation": true,
            "audit_logging": true
        }
    }
}

你的任务是否需要多个Agent？
├── 否 → 使用单Agent + MCP工具
└── 是 → Agent数量预计多少？
    ├── 2-3个 → 
    │   ├── 需要交叉验证？ → 受控辩论模式
    │   └── 线性流程？ → Pipeline模式
    ├── 4-5个 →
    │   ├── 任务可预先分解？ → Supervisor模式
    │   └── 任务需要自适应？ → 自组织模式 + 涌现监控
    └── 6+个 → 
        └── ⚠️ 进入涌现相变区
            ├── 可分解为多个3-Agent子团队？ → 分层Supervisor
            └── 必须全局协作？ → 强制人类介入 + MAEBE持续评估