LLM Council実践ガイド：複数AIモデルの合議システムで実現する高精度判断の設計パターン

# 基本ライブラリ
pip install openai>=1.0.0
pip install anthropic>=0.18.0
pip install google-generativeai>=0.3.0

# 非同期処理用
pip install asyncio
pip install aiohttp

# 結果集約用
pip install numpy
pip install scipy

# .envファイルの例
OPENAI_API_KEY=sk-xxx
ANTHROPIC_API_KEY=sk-ant-xxx
GOOGLE_API_KEY=AIzaxxx

import asyncio
from typing import List, Dict, Any
from openai import AsyncOpenAI
from anthropic import AsyncAnthropic
import google.generativeai as genai

class LLMCouncil:
    def __init__(self):
        self.openai_client = AsyncOpenAI()
        self.anthropic_client = AsyncAnthropic()
        genai.configure()
        self.gemini_model = genai.GenerativeModel('gemini-pro')

    async def query_gpt4(self, prompt: str) -> str:
        response = await self.openai_client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content

    async def query_claude(self, prompt: str) -> str:
        response = await self.anthropic_client.messages.create(
            model="claude-3-sonnet-20240229",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text

    async def query_gemini(self, prompt: str) -> str:
        response = await asyncio.to_thread(
            self.gemini_model.generate_content, prompt
        )
        return response.text

    async def vote(self, prompt: str) -> Dict[str, Any]:
        # 並列で全モデルに問い合わせ
        results = await asyncio.gather(
            self.query_gpt4(prompt),
            self.query_claude(prompt),
            self.query_gemini(prompt)
        )

        # 回答を集計
        votes = {}
        for i, result in enumerate(results):
            model_name = ["GPT-4", "Claude", "Gemini"][i]
            votes[model_name] = result

        return {
            "individual_responses": votes,
            "consensus": self._find_consensus(results)
        }

    def _find_consensus(self, responses: List[str]) -> str:
        # 最も類似度の高い回答を選択（簡易実装）
        # 実際にはセマンティック類似度を使用
        return max(set(responses), key=responses.count)

from sklearn.metrics.pairwise import cosine_similarity
from sentence_transformers import SentenceTransformer

class SemanticVoter:
    def __init__(self):
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')

    def find_consensus(self, responses: List[str], threshold: float = 0.8) -> str:
        # 各回答をベクトル化
        embeddings = self.encoder.encode(responses)

        # 類似度行列を計算
        similarity_matrix = cosine_similarity(embeddings)

        # 最も他の回答と類似度が高い回答を選択
        avg_similarities = similarity_matrix.mean(axis=1)
        best_idx = avg_similarities.argmax()

        return responses[best_idx]

class WeightedCouncil(LLMCouncil):
    def __init__(self, weights: Dict[str, float] = None):
        super().__init__()
        # デフォルトの重み設定
        self.weights = weights or {
            "GPT-4": 0.4,      # 汎用タスクで高評価
            "Claude": 0.35,    # 論理的推論で高評価
            "Gemini": 0.25     # 速度重視のタスクで使用
        }

    def weighted_consensus(self, responses: Dict[str, str]) -> str:
        # 各回答にスコアを付与
        scores = {}
        for model, response in responses.items():
            weight = self.weights.get(model, 0.33)
            if response not in scores:
                scores[response] = 0
            scores[response] += weight

        # 最高スコアの回答を返す
        return max(scores.items(), key=lambda x: x[1])[0]

class DebateCouncil(LLMCouncil):
    async def debate(self, question: str, rounds: int = 2) -> Dict[str, Any]:
        # 第1ラウンド：各モデルが独立して回答
        initial_responses = await self.vote(question)

        debate_history = [initial_responses]

        for round_num in range(rounds):
            # 他のモデルの回答を含めた再質問
            critique_prompt = self._build_critique_prompt(
                question,
                initial_responses["individual_responses"]
            )

            # 各モデルが批評を踏まえて再回答
            refined_responses = await self.vote(critique_prompt)
            debate_history.append(refined_responses)
            initial_responses = refined_responses

        return {
            "final_consensus": initial_responses["consensus"],
            "debate_history": debate_history
        }

    def _build_critique_prompt(self, question: str, responses: Dict[str, str]) -> str:
        prompt = f"質問: {question}\n\n"
        prompt += "他のAIモデルの回答:\n"
        for model, response in responses.items():
            prompt += f"- {model}: {response}\n"
        prompt += "\n上記の回答を踏まえ、最も適切な回答を再検討してください。"
        return prompt