Mixture of Agents 多模型协作架构设计与实现

AGGREGATOR_PROMPT = """你是一个高质量回答的综合器。

以下是多位 AI 助手对同一问题的独立回答：

{proposer_responses}

请综合以上所有回答的优点，生成一个更全面、更准确、更有深度的最终回答。
要求：
1. 保留各回答中正确且独特的见解
2. 消除矛盾信息，选择更可靠的说法
3. 补充任何遗漏的重要信息
4. 使用清晰的结构组织最终回答
"""

import asyncio
from dataclasses import dataclass
from typing import Optional
from openai import AsyncOpenAI
from anthropic import AsyncAnthropic
import google.generativeai as genai

@dataclass
class ProposerResponse:
    model: str
    content: str
    latency_ms: float
    success: bool

@dataclass
class MoAConfig:
    min_proposers: int = 3
    proposer_timeout: float = 30.0
    max_layers: int = 3

class MixtureOfAgents:
    def __init__(self, config: MoAConfig = MoAConfig()):
        self.config = config
        self.openai = AsyncOpenAI()
        self.anthropic = AsyncAnthropic()

    async def _call_openai(self, prompt: str, model: str = "gpt-4o") -> ProposerResponse:
        import time
        start = time.time()
        try:
            response = await asyncio.wait_for(
                self.openai.chat.completions.create(
                    model=model,
                    messages=[{"role": "user", "content": prompt}],
                    temperature=0.7,
                    max_tokens=2048,
                ),
                timeout=self.config.proposer_timeout,
            )
            return ProposerResponse(
                model=model,
                content=response.choices[0].message.content,
                latency_ms=(time.time() - start) * 1000,
                success=True,
            )
        except Exception as e:
            return ProposerResponse(
                model=model, content="", latency_ms=(time.time() - start) * 1000, success=False
            )

    async def _call_anthropic(self, prompt: str, model: str = "claude-sonnet-4-20250514") -> ProposerResponse:
        import time
        start = time.time()
        try:
            response = await asyncio.wait_for(
                self.anthropic.messages.create(
                    model=model,
                    max_tokens=2048,
                    messages=[{"role": "user", "content": prompt}],
                ),
                timeout=self.config.proposer_timeout,
            )
            return ProposerResponse(
                model=model,
                content=response.content[0].text,
                latency_ms=(time.time() - start) * 1000,
                success=True,
            )
        except Exception as e:
            return ProposerResponse(
                model=model, content="", latency_ms=(time.time() - start) * 1000, success=False
            )

    async def propose(self, query: str) -> list[ProposerResponse]:
        """Layer 1: 并行调用多个 Proposer"""
        tasks = [
            self._call_openai(query, "gpt-4o"),
            self._call_anthropic(query, "claude-sonnet-4-20250514"),
            self._call_openai(query, "gpt-4o-mini"),
        ]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        valid = [r for r in responses if isinstance(r, ProposerResponse) and r.success]

        if len(valid) < self.config.min_proposers:
            raise RuntimeError(
                f"Only {len(valid)} proposers succeeded, need {self.config.min_proposers}"
            )
        return valid

    async def aggregate(self, query: str, proposals: list[ProposerResponse]) -> str:
        """Layer 2+: 聚合多个 Proposer 的输出"""
        proposals_text = "\n\n".join(
            f"--- 回答来自 {p.model} ---\n{p.content}" for p in proposals
        )
        aggregation_prompt = f"""你是一个高质量回答的综合器。

原始问题：{query}

以下是多位 AI 助手对同一问题的独立回答：

{proposals_text}

请综合以上所有回答的优点，生成一个更全面、更准确的最终回答。
保留各回答中正确且独特的见解，消除矛盾，补充遗漏。"""

        response = await self._call_anthropic(aggregation_prompt, "claude-sonnet-4-20250514")
        return response.content

    async def run(self, query: str) -> str:
        """执行完整的 MoA 管道"""
        proposals = await self.propose(query)
        result = await self.aggregate(query, proposals)
        return result

# 使用示例
async def main():
    moa = MixtureOfAgents(MoAConfig(min_proposers=2))
    result = await moa.run("解释量子计算中的量子纠缠原理及其在量子通信中的应用")
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

import OpenAI from "openai";
import Anthropic from "@anthropic-ai/sdk";

interface ProposerResponse {
  model: string;
  content: string;
  latencyMs: number;
  success: boolean;
}

interface MoAConfig {
  minProposers: number;
  proposerTimeoutMs: number;
  maxLayers: number;
}

class MixtureOfAgents {
  private openai: OpenAI;
  private anthropic: Anthropic;
  private config: MoAConfig;

  constructor(config: Partial<MoAConfig> = {}) {
    this.config = {
      minProposers: 3,
      proposerTimeoutMs: 30000,
      maxLayers: 3,
      ...config,
    };
    this.openai = new OpenAI();
    this.anthropic = new Anthropic();
  }

  private async callOpenAI(
    prompt: string,
    model = "gpt-4o"
  ): Promise<ProposerResponse> {
    const start = Date.now();
    try {
      const response = await Promise.race([
        this.openai.chat.completions.create({
          model,
          messages: [{ role: "user", content: prompt }],
          temperature: 0.7,
          max_tokens: 2048,
        }),
        new Promise<never>((_, reject) =>
          setTimeout(() => reject(new Error("Timeout")), this.config.proposerTimeoutMs)
        ),
      ]);
      return {
        model,
        content: response.choices[0].message.content ?? "",
        latencyMs: Date.now() - start,
        success: true,
      };
    } catch {
      return { model, content: "", latencyMs: Date.now() - start, success: false };
    }
  }

  private async callAnthropic(
    prompt: string,
    model = "claude-sonnet-4-20250514"
  ): Promise<ProposerResponse> {
    const start = Date.now();
    try {
      const response = await this.anthropic.messages.create({
        model,
        max_tokens: 2048,
        messages: [{ role: "user", content: prompt }],
      });
      return {
        model,
        content: response.content[0].type === "text" ? response.content[0].text : "",
        latencyMs: Date.now() - start,
        success: true,
      };
    } catch {
      return { model, content: "", latencyMs: Date.now() - start, success: false };
    }
  }

  async propose(query: string): Promise<ProposerResponse[]> {
    const tasks = [
      this.callOpenAI(query, "gpt-4o"),
      this.callAnthropic(query, "claude-sonnet-4-20250514"),
      this.callOpenAI(query, "gpt-4o-mini"),
    ];
    const responses = await Promise.allSettled(tasks);
    const valid = responses
      .filter((r): r is PromiseFulfilledResult<ProposerResponse> =>
        r.status === "fulfilled" && r.value.success
      )
      .map((r) => r.value);

    if (valid.length < this.config.minProposers) {
      throw new Error(
        `Only ${valid.length} proposers succeeded, need ${this.config.minProposers}`
      );
    }
    return valid;
  }

  async aggregate(query: string, proposals: ProposerResponse[]): Promise<string> {
    const proposalsText = proposals
      .map((p) => `--- Response from ${p.model} ---\n${p.content}`)
      .join("\n\n");

    const aggregationPrompt = `You are a high-quality answer synthesizer.

Original question: ${query}

Below are independent responses from multiple AI assistants:

${proposalsText}

Synthesize the best aspects of all responses into a comprehensive final answer.
Retain unique correct insights, resolve contradictions, and fill gaps.`;

    const response = await this.callAnthropic(aggregationPrompt, "claude-sonnet-4-20250514");
    return response.content;
  }

  async run(query: string): Promise<string> {
    const proposals = await this.propose(query);
    return this.aggregate(query, proposals);
  }
}

// 使用示例
const moa = new MixtureOfAgents({ minProposers: 2 });
const result = await moa.run("Explain quantum entanglement and its applications in quantum communication");
console.log(result);

# Proposer Prompt：鼓励独特视角
PROPOSER_SYSTEM_PROMPTS = {
    "analytical": "你是一位注重逻辑严谨和数据支撑的分析师。请从数据和证据的角度回答问题。",
    "creative": "你是一位富有创意的思考者。请提供新颖独特的视角和类比来解答问题。",
    "practical": "你是一位注重实践的工程师。请从可操作性和实际应用的角度回答问题。",
    "critical": "你是一位严格的审查者。请指出问题中可能的陷阱和常见误解。",
}

# Aggregator Prompt：结构化综合
AGGREGATOR_TEMPLATE = """你是一位资深的知识综合专家。

## 任务
综合以下多位专家的独立回答，生成一个权威的最终回答。

## 原始问题
{query}

## 专家回答
{responses}

## 综合要求
1. 识别各回答中的共识点——这些很可能是正确的
2. 对于矛盾点，选择论证更充分的说法
3. 合并各回答的独特贡献
4. 确保最终回答逻辑连贯、结构清晰
5. 标注不确定的信息

请输出综合后的最终回答："""

class DynamicMoARouter:
    """根据任务复杂度动态选择 MoA 层数"""

    async def classify_complexity(self, query: str) -> str:
        response = await self.openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "system",
                "content": "判断问题复杂度。返回: simple/medium/complex"
            }, {
                "role": "user",
                "content": query
            }],
            max_tokens=10,
        )
        return response.choices[0].message.content.strip().lower()

    async def route(self, query: str) -> str:
        complexity = await self.classify_complexity(query)

        if complexity == "simple":
            # 单模型直答
            resp = await self._call_openai(query, "gpt-4o-mini")
            return resp.content
        elif complexity == "medium":
            # 单层 MoA（3 Proposer + 1 Aggregator）
            proposals = await self.propose(query)
            return await self.aggregate(query, proposals)
        else:
            # 多层 MoA（3 Proposer + 2 Aggregator + 1 Synthesizer）
            proposals = await self.propose(query)
            agg_results = await asyncio.gather(
                self.aggregate(query, proposals),
                self.aggregate(query, proposals),  # 不同 Aggregator Prompt
            )
            return await self.synthesize(query, agg_results)

TASK_MODEL_MAPPING = {
    "code_generation": ["gpt-4o", "claude-sonnet-4-20250514", "deepseek-coder"],
    "creative_writing": ["claude-sonnet-4-20250514", "gpt-4o", "gemini-1.5-pro"],
    "data_analysis": ["gpt-4o", "gemini-1.5-pro", "claude-sonnet-4-20250514"],
    "math_reasoning": ["gpt-4o", "deepseek-math", "claude-sonnet-4-20250514"],
}

async def select_proposers(self, query: str, task_type: str) -> list[str]:
    """基于任务类型选择最优 Proposer 组合"""
    return TASK_MODEL_MAPPING.get(task_type, ["gpt-4o", "claude-sonnet-4-20250514", "gemini-1.5-pro"])

async def iterative_refinement(self, query: str, rounds: int = 3) -> str:
    """迭代精炼：每轮将前一轮输出作为下一轮的上下文"""
    current_proposals = await self.propose(query)

    for round_idx in range(rounds - 1):
        aggregated = await self.aggregate(query, current_proposals)
        # 下一轮 Proposer 看到前一轮的聚合结果作为参考
        refined_prompt = f"""原始问题：{query}

以下是前一轮的综合回答：
{aggregated}

请基于此回答，提供进一步的补充、修正或深化。重点关注：
- 前一轮可能遗漏的要点
- 需要修正的不准确之处
- 可以进一步深入的方面"""

        current_proposals = await self.propose(refined_prompt)

    return await self.aggregate(query, current_proposals)

class CostOptimizedMoA:
    """成本优化的 MoA 配置"""

    COST_PER_1K_TOKENS = {
        "gpt-4o": {"input": 0.005, "output": 0.015},
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "claude-sonnet-4-20250514": {"input": 0.003, "output": 0.015},
        "gemini-1.5-pro": {"input": 0.00125, "output": 0.005},
    }

    def estimate_cost(self, query_tokens: int, num_proposers: int, num_layers: int) -> float:
        """估算单次 MoA 调用成本"""
        avg_output_tokens = 800
        proposer_cost = sum(
            (query_tokens / 1000) * self.COST_PER_1K_TOKENS[m]["input"] +
            (avg_output_tokens / 1000) * self.COST_PER_1K_TOKENS[m]["output"]
            for m in ["gpt-4o", "claude-sonnet-4-20250514", "gpt-4o-mini"][:num_proposers]
        )
        aggregator_input = query_tokens + avg_output_tokens * num_proposers
        aggregator_cost = (
            (aggregator_input / 1000) * self.COST_PER_1K_TOKENS["claude-sonnet-4-20250514"]["input"] +
            (avg_output_tokens / 1000) * self.COST_PER_1K_TOKENS["claude-sonnet-4-20250514"]["output"]
        )
        return (proposer_cost + aggregator_cost) * num_layers

import aiohttp
from asyncio import Semaphore

class ProductionMoA:
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = Semaphore(max_concurrent)
        self.cache = {}  # 实际生产中使用 Redis

    async def run_with_fallback(self, query: str) -> str:
        """带降级策略的生产级执行"""
        # 1. 检查缓存
        cache_key = self._hash_query(query)
        if cache_key in self.cache:
            return self.cache[cache_key]

        # 2. 执行 MoA 管道
        try:
            async with self.semaphore:
                proposals = await self.propose(query)
                result = await self.aggregate(query, proposals)
        except RuntimeError:
            # 降级：Proposer 不足时使用最佳单模型
            result = await self._fallback_single_model(query)

        # 3. 缓存结果
        self.cache[cache_key] = result
        return result

    async def _fallback_single_model(self, query: str) -> str:
        """降级策略：回退到单模型"""
        response = await self._call_anthropic(query, "claude-sonnet-4-20250514")
        return response.content

interface MoAMetrics {
  totalLatencyMs: number;
  proposerLatencies: Record<string, number>;
  proposerSuccessRate: Record<string, number>;
  aggregationQualityScore: number;
  cacheHitRate: number;
  costPerQuery: number;
}

function reportMetrics(metrics: MoAMetrics): void {
  // 推送到监控系统（如 Prometheus / Datadog）
  console.log(JSON.stringify({
    timestamp: new Date().toISOString(),
    pipeline: "moa",
    ...metrics,
  }));

  // 告警规则
  if (metrics.proposerSuccessRate["gpt-4o"] < 0.9) {
    alert("GPT-4o success rate dropped below 90%");
  }
  if (metrics.totalLatencyMs > 30000) {
    alert("MoA pipeline latency exceeded 30s threshold");
  }
}

import hashlib
from functools import lru_cache

class SemanticCache:
    """语义缓存：相似问题复用已有结果"""

    def __init__(self, similarity_threshold: float = 0.92):
        self.threshold = similarity_threshold
        self.embeddings = {}  # query_hash -> embedding
        self.results = {}     # query_hash -> result

    async def get_or_compute(self, query: str, compute_fn) -> str:
        query_embedding = await self._embed(query)

        # 查找语义相似的已缓存结果
        for cached_hash, cached_embedding in self.embeddings.items():
            similarity = self._cosine_similarity(query_embedding, cached_embedding)
            if similarity > self.threshold:
                return self.results[cached_hash]

        # 缓存未命中，执行计算
        result = await compute_fn(query)
        query_hash = hashlib.sha256(query.encode()).hexdigest()
        self.embeddings[query_hash] = query_embedding
        self.results[query_hash] = result
        return result