多Agent群体思维治理六栈实测：从认知退化到集体智慧跃迁

from langgraph.graph import StateGraph, START, END

class BlindVoteState(TypedDict):
    question: str
    anonymous_answers: list[str]
    final_answer: str

def blind_collection(state: BlindVoteState) -> BlindVoteState:
    """所有Agent并行独立回答，不共享上下文"""
    answers = []
    for agent in agents:
        # 关键：每个Agent只接收问题，不接收其他Agent的回答
        response = agent.invoke({"question": state["question"]})
        answers.append(response["answer"])
    state["anonymous_answers"] = answers
    return state

def majority_vote(state: BlindVoteState) -> BlindVoteState:
    """多数决投票，可加权"""
    votes = Counter(state["anonymous_answers"])
    state["final_answer"] = votes.most_common(1)[0][0]
    return state

graph = StateGraph(BlindVoteState)
graph.add_node("collect", blind_collection)
graph.add_node("vote", majority_vote)
graph.add_edge(START, "collect")
graph.add_edge("collect", "vote")
graph.add_edge("vote", END)

from crewai import Agent, Task, Crew, Process

devil_advocate = Agent(
    role="Devil's Advocate",
    goal="Challenge every consensus and identify weaknesses",
    backstory="""You are a professional contrarian. Your job is NOT to be helpful 
    but to be critical. Even if you agree with the team, you must find the 
    strongest possible counter-argument. Never say 'I agree'.""",
    llm="gpt-4.1",  # 使用同等级模型，避免权威差异
    verbose=True
)

analyst = Agent(
    role="Senior Analyst", 
    goal="Provide thorough analysis",
    backstory="Expert analyst with 20 years experience.",
    llm="gpt-4.1"
)

# 使用层级流程：先分析，再反驳，最后综合
crew = Crew(
    agents=[analyst, devil_advocate, synthesizer],
    tasks=[analysis_task, challenge_task, synthesis_task],
    process=Process.sequential  # 顺序保证魔鬼代言人看到分析结果
)

from agents import Agent, Runner

async def structured_dissent(question: str):
    # Phase 1: 每个Agent独立生成"观点+自反对"
    agents = [
        Agent(name=f"analyst_{i}", 
              instructions="""You must follow this protocol:
              1. State your position (1 paragraph)
              2. List 2-3 strongest arguments AGAINST your own position
              3. Rate your confidence (0-100%)
              Only after completing all 3 steps, you may conclude.""")
        for i in range(5)
    ]

    # 并行执行，避免锚定
    results = await asyncio.gather(*[
        Runner.run(agent, question) for agent in agents
    ])

    # Phase 2: 综合考虑各Agent的观点和自反对
    synthesizer = Agent(
        name="synthesizer",
        instructions="""Consider all positions AND their self-identified 
        weaknesses. Give extra weight to arguments where multiple agents 
        independently identified similar weaknesses in their own positions."""
    )
    final = await Runner.run(synthesizer, 
        f"Question: {question}\nAnalyses:\n{format_results(results)}")
    return final

from google.adk import Agent, Runner
from google.adk.a2a import A2AClient

# 每个Agent有不同的"知识视角"
perspectives = {
    "data_analyst": "Focus on quantitative evidence and statistical significance",
    "domain_expert": "Focus on domain-specific constraints and real-world feasibility", 
    "risk_analyst": "Focus on downside scenarios and failure modes",
    "user_advocate": "Focus on end-user experience and usability",
    "contrarian": "Focus on why the popular answer might be wrong"
}

agents = []
for name, perspective in perspectives.items():
    agent = Agent(
        name=name,
        model="gemini-2.5-pro",
        instruction=f"""You are {name}. {perspective}.
        IMPORTANT: You must NOT try to be balanced. Argue forcefully 
        from YOUR specific perspective only.""",
        # A2A协议允许跨框架互操作
        a2a_capability=A2AClient(
            agent_card={
                "name": name,
                "capabilities": [perspective],
                "perspective": perspective
            }
        )
    )
    agents.append(agent)

import ag2

# 定义信号提取器
signal_extractor = ag2.Agent(
    name="signal_extractor",
    system_message="""Extract structured signals from analysis:
    {
        "key_claims": ["claim1", "claim2"],
        "confidence_per_claim": [0.8, 0.6],
        "supporting_evidence": ["evidence1"],
        "identified_risks": ["risk1"],
        "assumptions": ["assumption1"]
    }"""
)

group_chat = ag2.GroupChat(
    agents=[analyst_1, analyst_2, analyst_3, signal_extractor, aggregator],
    messages=[],
    max_round=3  # 限制轮次防止螺旋
)

# 自定义speaker选择：确保signal_extractor在最后发言
group_chat.speaker_selection_method = "round_robin"

import { Agent } from '@mastra/core';
import { openai } from '@ai-sdk/openai';
import { anthropic } from '@ai-sdk/anthropic';
import { google } from '@ai-sdk/google';

const agents = [
  new Agent({
    name: 'gpt-analyst',
    model: openai('gpt-4.1'),
    instructions: 'Provide analysis from a logical, structured perspective.'
  }),
  new Agent({
    name: 'claude-analyst', 
    model: anthropic('claude-sonnet-4-6'),
    instructions: 'Provide analysis focusing on nuance and edge cases.'
  }),
  new Agent({
    name: 'gemini-analyst',
    model: google('gemini-2.5-pro'),
    instructions: 'Provide analysis with emphasis on factual grounding.'
  }),
];

// Mastra workflow: 并行执行后加权聚合
const workflow = new Workflow()
  .parallel(agents.map(a => a.step()))
  .aggregate('weighted-consensus');

graph TB
    subgraph Input["输入层"]
        Q[问题/任务]
    end

    subgraph Isolation["隔离层 - 切断锚定路径"]
        Q --> P1[Agent 1<br/>仅接收原始问题]
        Q --> P2[Agent 2<br/>仅接收原始问题]
        Q --> P3[Agent 3<br/>仅接收原始问题]
        Q --> P4[Agent 4<br/>仅接收原始问题]
        Q --> P5[Devil's Advocate<br/>仅接收原始问题]
    end

    subgraph Process["处理层 - 结构化异议协议"]
        P1 --> S1[观点 + 3条自反对<br/>置信度: 82%]
        P2 --> S2[观点 + 3条自反对<br/>置信度: 75%]
        P3 --> S3[观点 + 3条自反对<br/>置信度: 90%]
        P4 --> S4[观点 + 3条自反对<br/>置信度: 68%]
        P5 --> S5[系统性反驳<br/>针对共识弱点]
    end

    subgraph Aggregate["聚合层 - 信号联邦"]
        S1 --> F[信号提取器<br/>分离论点/证据/假设]
        S2 --> F
        S3 --> F
        S4 --> F
        S5 --> F
        F --> V[加权投票<br/>高置信度论点权重↑<br/>多人独立识别的弱点权重↑]
    end

    subgraph Output["输出层"]
        V --> R[最终决策<br/>附带 dissent report]
        V --> DR[Dissent Report<br/>记录被否决的观点及理由]
    end

    style Isolation fill:#e8f5e9,stroke:#4caf50
    style Process fill:#fff3e0,stroke:#ff9800
    style Aggregate fill:#e3f2fd,stroke:#2196f3

from agents import Agent, handoff

# 关键：使用input_filter隔离上下文
filtered_handoff = handoff(
    agent=analyst_agent,
    input_filter=lambda ctx: {
        "messages": ctx["messages"][-1:]  # 只传递最后一条消息
    }
)

选择型决策（从A/B/C中选）→ 匿名投票 + 5Agent
    实现：任意框架均可，LangGraph最简洁
    成本：约$0.02/次（GPT-4.1-mini）

分析型决策（需要深度推理）→ 结构化异议 + 3-5Agent  
    实现：CrewAI（角色驱动）或 OpenAI Agents SDK（Handoff filter）
    成本：约$0.08/次（GPT-4.1）

创意型决策（需要新颖性）→ 信息茧房打破 + 5-7Agent
    实现：Google ADK（视角注入）+ A2A跨框架
    成本：约$0.12/次（混合模型）

关键决策（不可承受错误）→ 多模型混合 + 全套策略
    实现：Mastra（多provider）+ 联邦聚合
    成本：约$0.35/次（三模型混合）