多Agent涌现与共识：从混沌到有序的工程突围

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import create_react_agent

# 定义三个专业Agent
researcher = create_react_agent(model, tools=[search_tool], prompt="You are a researcher")
coder = create_react_agent(model, tools=[code_tool], prompt="You are a coder")
reviewer = create_react_agent(model, tools=[], prompt="You are a code reviewer")

# Supervisor决策逻辑
def supervisor_router(state: MessagesState) -> str:
    last_msg = state["messages"][-1].content
    if "research" in last_msg.lower():
        return "researcher"
    elif "code" in last_msg.lower():
        return "coder"
    elif "review" in last_msg.lower():
        return "reviewer"
    return END

# 构建图
graph = StateGraph(MessagesState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher)
graph.add_node("coder", coder)
graph.add_node("reviewer", reviewer)
graph.add_conditional_edges("supervisor", supervisor_router)
graph.add_edge(START, "supervisor")
app = graph.compile()

            有益 / Intent-aligned
                    |
     象限2：培育     |     象限1：鼓励
    自发分工优化      |    预期内的协作创新
                    |
    ────────────────┼────────────────
                    |
     象限3：监控     |     象限4：抑制
    潜在偏移风险      |    确认的有害行为
                    |
            有害 / Misaligned

# 涌现行为检测器 —— 基于行为偏离度
class EmergenceMonitor:
    def __init__(self, baseline_behavior, drift_threshold=0.3):
        self.baseline = baseline_behavior
        self.threshold = drift_threshold
        self.history = []

    def observe(self, agent_actions: list[dict]):
        deviation = self._compute_deviation(agent_actions)
        self.history.append(deviation)

        if deviation > self.threshold:
            self._alert_quadrant(deviation, agent_actions)
            if deviation > self.threshold * 2:
                return "HALT"  # 触发人类介入
        return "CONTINUE"

    def _compute_deviation(self, actions):
        """计算当前行为与基线的余弦相似度偏离"""
        action_embedding = self._embed(actions)
        return 1 - cosine_similarity(action_embedding, self.baseline)

# Aegean共识 —— LLM-Agent化的PBFT
class AegeanConsensus:
    def __init__(self, agents: list, f: int = 1):
        """
        f: 可容忍的拜占庭节点数
        总Agent数需 >= 3f + 1
        """
        self.agents = agents
        self.f = f
        assert len(agents) >= 3 * f + 1

    async def reach_consensus(self, task: str) -> dict:
        # Phase 1: Pre-prepare —— Leader提出提案
        leader_proposal = await self.agents[0].execute(task)

        # Phase 2: Prepare —— 所有Agent独立评估
        votes = await asyncio.gather(*[
            agent.evaluate(leader_proposal, task) 
            for agent in self.agents[1:]
        ])

        # Phase 3: Commit —— 达到2f+1确认后提交
        confirmations = sum(1 for v in votes if v["agree"])
        if confirmations >= 2 * self.f + 1:
            return {"result": leader_proposal, "consensus": True}

        # 拜占庭检测 —— 触发降级策略
        return await self._fallback_consensus(task, votes)

# Google ADK —— 定义一个A2A兼容的Agent
from google.adk import Agent, LlmAgent

# 定义Agent能力卡片
routing_agent = LlmAgent(
    name="customer_support_router",
    model="gemini-2.0-flash",
    instruction="""You are a customer support router. 
    Analyze the user's request and delegate to the appropriate specialist.""",
    description="Routes customer queries to specialized agents",  # Agent Card描述
)

# 注册子Agent（A2A发现）
billing_agent = Agent(name="billing_specialist", ...)
tech_agent = Agent(name="tech_support", ...)

routing_agent.add_child_agents([billing_agent, tech_agent])

┌──────────────────────────────────────────────┐
│              Long-Term Memory                │
│         (Vector DB / Knowledge Graph)         │
│    ┌─────────────────────────────────┐       │
│    │        Session Memory           │       │
│    │    (Shared State / Blackboard)  │       │
│    │  ┌───────────────────────┐     │       │
│    │  │   Working Memory      │     │       │
│    │  │  (Per-Agent Context)  │     │       │
│    │  └───────────────────────┘     │       │
│    └─────────────────────────────────┘       │
└──────────────────────────────────────────────┘

graph TB
    subgraph "用户/外部系统"
        USER[用户请求]
        API[API Gateway]
    end

    subgraph "编排层 Orchestration"
        ORCH[Supervisor Agent<br/>任务分解与调度]
        ROUTER[Router / Swarm<br/>Agent间控制权传递]
    end

    subgraph "Agent集群"
        A1[Research Agent]
        A2[Code Agent]
        A3[Review Agent]
        A4[QA Agent]
    end

    subgraph "通信协议层"
        A2A[A2A Protocol<br/>Agent↔Agent]
        MCP[MCP Protocol<br/>Agent↔Tool/Data]
    end

    subgraph "记忆层 Memory"
        WM[Working Memory<br/>每Agent独立上下文]
        SM[Session Memory<br/>共享Blackboard]
        LTM[Long-Term Memory<br/>Vector DB / KG]
    end

    subgraph "治理层 Governance"
        MON[Emergence Monitor<br/>涌现行为检测]
        CON[Consensus Engine<br/>共识协议引擎]
        AUDIT[Audit Logger<br/>审计追踪]
    end

    subgraph "工具/数据层"
        TOOLS[外部工具/API]
        DATA[数据源/知识库]
    end

    USER --> API --> ORCH
    ORCH --> ROUTER
    ROUTER --> A1 & A2 & A3 & A4
    A1 & A2 & A3 & A4 <--> A2A
    A1 & A2 & A3 & A4 <--> MCP
    MCP <--> TOOLS & DATA
    A1 & A2 & A3 & A4 <--> WM
    WM <--> SM <--> LTM
    A1 & A2 & A3 & A4 --> MON
    MON --> CON
    CON --> AUDIT

flowchart TD
    START[多个Agent需要达成共识] --> Q1{需要容错吗？}
    Q1 -->|否| Q2{Agent数量？}
    Q1 -->|是| Q3{有无恶意Agent风险？}

    Q2 -->|≤5| VOTE[多数投票<br/>成本低，速度快]
    Q2 -->|>5| WEIGHTED[加权共识<br/>按历史准确率加权]

    Q3 -->|是| Q4{可接受的延迟？}
    Q3 -->|否| PBFT[PBFT-Lite<br/>3f+1节点容忍f个故障]

    Q4 -->|低| WEIGHTED
    Q4 -->|中| DEBATE[多轮辩论<br/>2-3轮+Early Stop]

    DEBATE --> Q5{规模>20 Agent？}
    Q5 -->|是| HIER[层级共识<br/>先小组后全局]
    Q5 -->|否| DEBATE