多Agent线性链衰减：Blackboard六栈实测

Agent_1: "巴黎的人口约为220万" （正确）
    ↓ 传递（微小的置信度偏移）
Agent_2: "巴黎的人口约220万，是法国最大的城市" （正确但新增了未验证信息）
    ↓ 传递（新增信息被当作事实接受）
Agent_3: "法国最大城市巴黎拥有220万人口，远超第二大城市马赛" （"远超"是幻觉）
    ↓ 传递（幻觉被强化）
Agent_4: "巴黎220万人口远超马赛，差距达到数百万" （数字幻觉爆发）

传统线性链:
  Agent_A → Agent_B → Agent_C → Agent_D
  （每一步都是点对点传递，错误逐跳放大）

黑板架构:
  ┌─────────────────────────────────┐
  │         Blackboard              │
  │  (共享的、结构化的状态空间)       │
  │  - 任务描述                      │
  │  - 中间结果（带元数据）           │
  │  - 置信度评分                    │
  │  - 矛盾标记                      │
  │  - 控制状态                      │
  └───────┬───────┬───────┬─────────┘
          │       │       │
       Agent_A  Agent_B  Agent_C
       (读写)   (读写)   (读写)

graph TD
    A[MAST失败分类法] --> B[FM-1: 规范歧义<br/>Spec Ambiguity]
    A --> C[FM-2: 通信失败<br/>Communication Failure]
    A --> D[FM-3: 任务执行失败<br/>Task Execution Failure]
    A --> E[FM-4: 集体无能<br/>Collective Incompetence]

    C --> C1[FM-2.1: 信息丢失]
    C --> C2[FM-2.2: 信息幻觉]
    C --> C3[FM-2.3: 格式不兼容]
    C --> C4[FM-2.4: 信息隐瞒]

    D --> D1[FM-3.1: 任务遗漏]
    D --> D2[FM-3.2: 重复执行]
    D --> D3[FM-3.3: 死循环]

    B --> B1[角色定义模糊]
    B --> B2[任务边界不清]

# MAST失败分类器核心结构（简化）
from enum import Enum
from dataclasses import dataclass

class FailureMode(Enum):
    SPEC_AMBIGUITY = "FM-1"       # 规范歧义
    INFO_LOSS = "FM-2.1"          # 信息丢失
    INFO_HALLUCINATION = "FM-2.2" # 信息幻觉
    FORMAT_INCOMPATIBLE = "FM-2.3" # 格式不兼容
    INFO_WITHHOLDING = "FM-2.4"   # 信息隐瞒
    TASK_OMISSION = "FM-3.1"      # 任务遗漏
    DUPLICATE_EXECUTION = "FM-3.2" # 重复执行
    INFINITE_LOOP = "FM-3.3"      # 死循环

@dataclass
class MASTDiagnosis:
    failure_mode: FailureMode
    confidence: float
    affected_agents: list[str]
    propagation_path: list[str]  # 错误传播路径
    root_cause_agent: str        # 根因Agent

# Agent Drift检测器（概念实现）
import numpy as np
from dataclasses import dataclass

@dataclass
class DriftMetrics:
    task_performance: float    # 任务性能 [0, 1]
    role_consistency: float    # 角色一致性 [0, 1]  
    protocol_adherence: float  # 协议遵守率 [0, 1]

    @property
    def stability_index(self) -> float:
        """复合稳定性指标，低于0.6视为漂移"""
        return np.mean([
            self.task_performance,
            self.role_consistency,
            self.protocol_adherence
        ])

class DriftDetector:
    def __init__(self, baseline: DriftMetrics, threshold: float = 0.6):
        self.baseline = baseline
        self.threshold = threshold
        self.history: list[DriftMetrics] = []

    def check(self, current: DriftMetrics) -> dict:
        drift = {
            'task_drift': self.baseline.task_performance - current.task_performance,
            'role_drift': self.baseline.role_consistency - current.role_consistency,
            'protocol_drift': self.baseline.protocol_adherence - current.protocol_adherence,
            'stability_index': current.stability_index,
            'is_drifting': current.stability_index < self.threshold
        }
        self.history.append(current)

        # 检测加速漂移（导数）
        if len(self.history) >= 3:
            recent = [h.stability_index for h in self.history[-3:]]
            drift_acceleration = recent[-3] - 2*recent[-2] + recent[-1]
            drift['acceleration'] = drift_acceleration
            drift['critical'] = drift_acceleration > 0.1  # 加速漂移

        return drift

# GUARDIAN时序图模型（简化概念）
import networkx as nx
from datetime import datetime

class GuardianMonitor:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.anomaly_threshold = 0.75

    def log_interaction(self, from_agent: str, to_agent: str, 
                        content: str, confidence: float, 
                        timestamp: datetime):
        """记录Agent间交互，构建时序图"""
        edge_key = f"{from_agent}->{to_agent}"
        self.graph.add_edge(
            from_agent, to_agent,
            content=content,
            confidence=confidence,
            timestamp=timestamp,
            # 计算语义曲率——如果信息传递导致语义空间的大幅弯曲，标记为可疑
            semantic_curvature=self._compute_curvature(content, confidence)
        )

    def detect_cascade_risk(self) -> list[dict]:
        """检测级联风险——识别高风险传播路径"""
        risks = []
        for path in nx.all_simple_paths(self.graph, source=None, target=None):
            cumulative_drift = 0
            for i in range(len(path)-1):
                edge = self.graph[path[i]][path[i+1]]
                cumulative_drift += (1 - edge['confidence'])

            # 级联风险 = 路径长度 × 累积漂移
            cascade_score = len(path) * cumulative_drift
            if cascade_score > self.anomaly_threshold:
                risks.append({
                    'path': path,
                    'score': cascade_score,
                    'action': 'isolate_last_agent'  # 隔离最末Agent
                })
        return risks

步骤1: 语义嵌入
  将每个Agent的输出映射到高维语义空间（使用Sentence-BERT等）

步骤2: 几何映射
  计算相邻Agent输出之间的语义距离（余弦距离、欧几里得距离）

步骤3: 曲率计算
  在语义空间中计算三连Agent(A→B→C)的"曲率"——如果B的输出偏离了A→C的"直线"，曲率值高

步骤4: 协演化审计循环
  持续监控曲率分布，当曲率超过基线的2σ时触发审计

# Agent熔断器（生产级实现概念）
from enum import Enum
from time import time, monotonic
from collections import deque

class CircuitState(Enum):
    CLOSED = "closed"      # 正常：请求通过
    OPEN = "open"          # 熔断：请求被拒绝
    HALF_OPEN = "half_open" # 半开：允许探测请求

class AgentCircuitBreaker:
    def __init__(self, failure_threshold: int = 5, 
                 recovery_timeout: float = 60.0,
                 window_size: int = 20):
        self.state = CircuitState.CLOSED
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.recent_calls: deque = deque(maxlen=window_size)
        self.last_failure_time: float = 0

    async def call_agent(self, agent_fn, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if monotonic() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise CircuitOpenError("Agent circuit is open")

        try:
            result = await agent_fn(*args, **kwargs)
            self._record_success()
            return result
        except Exception as e:
            self._record_failure()
            raise

    def _record_failure(self):
        self.recent_calls.append(False)
        self.last_failure_time = monotonic()
        failure_rate = sum(1 for x in self.recent_calls if not x) / len(self.recent_calls)
        if failure_rate >= self.failure_threshold / len(self.recent_calls):
            self.state = CircuitState.OPEN

    def _record_success(self):
        self.recent_calls.append(True)
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED

{
  "id": "kb-2026-0528-001",
  "type": "intermediate_result",
  "contributor": "Analyzer_Agent",
  "content": "分析结果：系统需要支持10万QPS",
  "confidence": 0.92,
  "evidence": ["负载测试报告p3", "历史数据Q1"],
  "contradictions": [],
  "timestamp": "2026-05-28T10:30:00Z",
  "version": 3,
  "reviews": [
    {"reviewer": "Validator_Agent", "verdict": "supported", "note": "与压测数据一致"}
  ]
}

graph LR
    subgraph Linear["线性链模式"]
        direction LR
        A1[Agent A] --> A2[Agent B] --> A3[Agent C] --> A4[Agent D]
    end

    subgraph Blackboard["黑板模式"]
        direction TB
        BB[共享黑板<br/>结构化知识空间]
        BB -.-> B1[Agent A<br/>分析]
        BB -.-> B2[Agent B<br/>设计]
        BB -.-> B3[Agent C<br/>验证]
        BB -.-> B4[Agent D<br/>执行]
        CS[控制Shell<br/>调度决策]
        CS -.-> BB
    end

    subgraph Hybrid["混合模式<br/>推荐方案"]
        direction TB
        HB[分层黑板]
        HB -.-> H1[小组1<br/>线性链]
        HB -.-> H2[小组2<br/>线性链]
        HB -.-> H3[小组3<br/>黑板]
        HB -.-> CO[协调器<br/>黑板级]
    end

Step 1: 诊断当前链的衰减点
  → 部署MAST分类器，找出哪两个Agent之间的信息损失最大
  → 通常问题集中在"摘要Agent→执行Agent"和"分析Agent→报告Agent"这两个边界

Step 2: 引入共享状态层
  → 在现有线性链的基础上，添加一个简单的共享存储（Redis/SQLite）
  → 每个Agent除了传递输出给下游，同时写入共享存储
  → 下游Agent在接收直接输入外，可以查询共享存储中的完整上下文

Step 3: 添加置信度标注
  → 要求每个Agent在输出中附带置信度评分和证据引用
  → 结构化黑板条目，而非自由文本传递

Step 4: 部署控制Shell
  → 替换线性编排为黑板控制Shell
  → Shell根据黑板状态决定下一步激活哪个Agent
  → 初始可以用简单的规则：任务未完成→调度下一个Agent

Step 5: 添加韧性层
  → 在控制Shell中集成Circuit Breaker和Drift Detector
  → 当ASI跌破阈值或熔断触发时，自动降级为单Agent模式

# 生产级黑板Agent基类
from abc import ABC, abstractmethod
from datetime import datetime
import json

class BlackboardEntry:
    """黑板条目——结构化知识单元"""
    def __init__(self, entry_type: str, contributor: str, 
                 content: str, confidence: float, evidence: list[str]):
        self.id = f"be-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
        self.entry_type = entry_type
        self.contributor = contributor
        self.content = content
        self.confidence = confidence
        self.evidence = evidence
        self.contradictions = []
        self.reviews = []
        self.version = 1

    def add_contradiction(self, other_id: str, reason: str):
        self.contradictions.append({
            'conflicting_entry': other_id, 
            'reason': reason
        })

class Blackboard:
    """共享黑板——结构化知识空间"""
    def __init__(self):
        self.entries: dict[str, BlackboardEntry] = {}
        self.control_state = {}

    def write(self, entry: BlackboardEntry):
        # 冲突检测：检查是否有矛盾条目
        for existing in self.entries.values():
            if existing.entry_type == entry.entry_type:
                if self._is_contradictory(existing, entry):
                    existing.add_contradiction(entry.id, "语义冲突")
                    entry.add_contradiction(existing.id, "语义冲突")
        self.entries[entry.id] = entry

    def read(self, entry_type: str = None, 
             min_confidence: float = 0.0) -> list[BlackboardEntry]:
        results = list(self.entries.values())
        if entry_type:
            results = [e for e in results if e.entry_type == entry_type]
        if min_confidence > 0:
            results = [e for e in results if e.confidence >= min_confidence]
        return sorted(results, key=lambda e: e.confidence, reverse=True)

    def _is_contradictory(self, a: BlackboardEntry, b: BlackboardEntry) -> bool:
        # 简化：使用语义相似度阈值检测矛盾
        # 生产中应使用嵌入模型计算
        return False

class BlackboardAgent(ABC):
    """黑板Agent基类——所有Agent继承此类"""
    def __init__(self, name: str, blackboard: Blackboard):
        self.name = name
        self.bb = blackboard
        self.drift_history = []

    @abstractmethod
    def can_contribute(self) -> bool:
        """控制Shell调用：当前Agent是否有贡献可做？"""
        pass

    @abstractmethod
    def contribute(self):
        """执行贡献：读取黑板→处理→写回黑板"""
        pass

    def read_board(self, **kwargs):
        return self.bb.read(**kwargs)

    def write_board(self, entry_type: str, content: str, 
                    confidence: float, evidence: list[str]):
        entry = BlackboardEntry(
            entry_type=entry_type,
            contributor=self.name,
            content=content,
            confidence=confidence,
            evidence=evidence
        )
        self.bb.write(entry)
        return entry.id

class ControlShell:
    """控制Shell——黑板调度器"""
    def __init__(self, agents: list[BlackboardAgent], max_iterations: int = 10):
        self.agents = agents
        self.max_iterations = max_iterations

    def execute(self):
        for iteration in range(self.max_iterations):
            any_contributed = False
            for agent in self.agents:
                if agent.can_contribute():
                    agent.contribute()
                    any_contributed = True
            if not any_contributed:
                break  # 所有Agent都无新贡献，终止
        return self.agents[0].bb  # 返回最终黑板状态