🗳️

【意図乖離検出;連載 第8回】AIマルチエージェントと集合知 - 民主型投票アーキテクチャ

に公開

AIマルチエージェントと集合知 - 民主型投票アーキテクチャ

前回のふりかえり

前回は、IDDの基本コンセプトを説明しました。意図の4要素モデル、3層トレーサビリティモデル、3層プロジェクト階層モデル。

今回は、少し技術的な話をします。IDDでは、複数のLLMが協調して意図乖離を検証する「民主型投票アーキテクチャ」を採用しています。なぜ複数のLLMを使うのか、どのように合意形成するのか、少数意見はどう扱うのか。

なぜ複数のLLMを使うのか

単一LLMの限界

LLMは強力ですが、以下のような限界があります。

1. ハルシネーション(幻覚)

LLMは、もっともらしいが間違った情報を生成することがあります。特に、プロジェクト固有の文脈では、一般的なパターンを誤って適用する可能性があります。

Huang et al. (2023)[1]の研究では、LLMのハルシネーションが発生するメカニズムを分析し、「知識の境界」を超えた質問に対して特に発生しやすいことを示しています。

2. バイアス

LLMは、学習データに含まれるバイアスを反映します。特定のプログラミング言語、フレームワーク、設計パターンに対して偏った判断をする可能性があります。

3. 一貫性の欠如

同じ入力に対しても、LLMの出力は毎回わずかに異なります。これは、創造的なタスクでは利点になりますが、検証タスクでは問題になります。

4. 確信度の誤認

LLMは、自分の回答に対して適切な確信度を示すことが苦手です。間違っていても自信満々に答えることがあります。

集合知のアプローチ

これらの問題に対処するために、私は「集合知」のアプローチを提案します。

集合知(Collective Intelligence)とは、多数の個体が協力して問題を解決することで、個々の能力を超えた知性を発揮する現象です。

James Surowiecki(2004)[2]は、著書「The Wisdom of Crowds」で、集合知が機能するための条件を示しました:

  1. 多様性: 各個体が異なる視点や情報を持つ
  2. 独立性: 各個体が他者に影響されず独自の判断を下す
  3. 分散性: 各個体が局所的な知識を持つ
  4. 集約: 個々の判断を統合するメカニズムがある

これをLLMに適用すると、以下のようになります:

  1. 多様性: 異なるLLMプロバイダ(Claude, GPT, Gemini)を使用
  2. 独立性: 各LLMが独立して分析を行う(他のLLMの出力を参照しない)
  3. 分散性: 各LLMが異なる学習データ・アーキテクチャを持つ
  4. 集約: 投票やコンセンサスアルゴリズムで判断を統合

研究的背景

LLMのマルチエージェントシステムに関する研究は、近年急速に進んでいます。

Du et al. (2023)[3]は、「Improving Factuality and Reasoning in Language Models through Multiagent Debate」という論文で、複数のLLMが議論(debate)を行うことで、単一LLMよりも正確な回答を生成できることを示しました。

彼らの実験では、複数のLLMが自分の回答を主張し、他のLLMの回答を批評し、最終的に収束するプロセスを経ることで、数学的推論タスクの正答率が向上しました。

Wang et al. (2023)[4]は、「Self-Consistency Improves Chain of Thought Reasoning in Language Models」で、同じLLMに複数回推論させ、多数決を取ることで正確性が向上することを示しました。

これらの研究は、IDDの民主型投票アーキテクチャの理論的基盤となっています。

民主型投票アーキテクチャ

IDDでは、意図乖離の検証において「民主型投票アーキテクチャ」を採用しています。

基本的な仕組み

なぜ「民主型」なのか

「民主型」という表現を使ったのは、単なる多数決ではなく、以下の原則に基づいているからです。

1. 平等な発言権

各LLMの分析結果は、最初は平等に扱われます。特定のLLMを「正解」とみなすことはしません。

2. 根拠に基づく判断

単に「乖離あり」「乖離なし」だけでなく、その根拠を提示することを求めます。根拠が説得力のある分析は、より重視されます。

3. 少数意見の尊重

多数決で決まった結論と異なる意見(少数意見)も、記録として残します。これは、将来の再評価のために重要です。

4. 透明性

判断のプロセスは透明化されます。なぜその結論に至ったのか、どのLLMがどう判断したのか、追跡可能です。

処理フロー

以下は、民主型投票の処理フローです。

1. 検証リクエスト受信
   - 意図(4要素モデル)
   - 対応する成果物
   - プロジェクトコンテキスト

2. 並列分析(独立性を保証)
   [LLM-A] 分析開始 → 結果A (status: passed, confidence: 0.85)
   [LLM-B] 分析開始 → 結果B (status: warning, confidence: 0.75)
   [LLM-C] 分析開始 → 結果C (status: passed, confidence: 0.90)

3. 投票集計
   - passed: 2票 (A, C)
   - warning: 1票 (B)
   - failed: 0票
   
   多数決結果: passed

4. 信頼度計算
   weighted_confidence = (0.85 + 0.90) / 2 = 0.875
   (warningの票は除外)

5. 不一致分析
   - LLM-Bが warning を出した理由を分析
   - 少数意見として記録

6. 最終判定
   {
     "status": "passed",
     "confidence_score": 0.875,
     "consensus": "2/3 agreed on passed",
     "dissenting_opinions": [
       {
         "llm": "LLM-B",
         "status": "warning",
         "reason": "制約の一部が明示的に検証されていない"
       }
     ]
   }

信頼度スコアの計算

各LLMは、自分の分析に対する信頼度スコア(0.0〜1.0)を出力します。

このスコアは、以下の要素に基づいて算出されます:

要素 重み 説明
入力の明確さ 30% 入力された意図がどれだけ明確か
分析の深さ 25% どれだけ詳細に分析できたか
根拠の強さ 25% 判断の根拠がどれだけ具体的か
制約の検証度 20% 制約がどれだけ検証されたか

最終的な信頼度スコアは、合意したLLMのスコアの平均を取ります。

不一致の処理

LLM間で判断が分かれた場合、以下のように処理します。

完全一致(3/3)

status = 一致した判定
confidence_score = 3つの平均

多数一致(2/3)

status = 多数派の判定
confidence_score = 多数派の平均
dissenting_opinions = 少数派の意見を記録

三者三様(全員異なる)

status = "needs_review" (人間によるレビューが必要)
confidence_score = 0.0
all_opinions = 全員の意見を記録

三者三様の場合は、自動判定を避け、人間の介入を求めます。これは重要なポイントです。AIは万能ではなく、判断が難しい場合は人間に委ねるべきです。

議論フェーズ(オプション)

より高い精度が求められる場合、投票の前に「議論フェーズ」を設けることができます。

議論の流れ

Round 1: 初期分析
  [LLM-A] → 分析A
  [LLM-B] → 分析B
  [LLM-C] → 分析C

Round 2: 相互批評
  [LLM-A] → 分析B, Cを見て、自分の分析を再評価
  [LLM-B] → 分析A, Cを見て、自分の分析を再評価
  [LLM-C] → 分析A, Bを見て、自分の分析を再評価

Round 3: 最終判断
  [LLM-A] → 最終分析A'
  [LLM-B] → 最終分析B'
  [LLM-C] → 最終分析C'

投票: A', B', C' で多数決

議論フェーズのプロンプト設計

Round 2のプロンプトは、以下のような構造を持ちます:

あなたは意図乖離検証のエキスパートです。

## あなたの初期分析
{initial_analysis}

## 他のエキスパートの分析
### エキスパートB
{analysis_b}

### エキスパートC
{analysis_c}

## 指示
他のエキスパートの分析を参考に、あなたの初期分析を再評価してください。

- 他のエキスパートが指摘した点で、あなたが見落としていたものはありますか?
- 他のエキスパートの分析に同意できない点はありますか?その理由は?
- 再評価の結果、あなたの判断は変わりますか?

最終的な判断を、根拠とともに出力してください。

収束と発散

議論フェーズを設けることで、以下の効果が期待できます。

収束効果
見落としていた点を他のLLMが指摘することで、より正確な分析に収束する。

発散効果(有用な場合も)
逆に、議論によって新しい論点が見つかることもある。これは、複雑な意図乖離の分析においては有用。

Du et al. (2023)の研究でも、複数ラウンドの議論により、正答率が向上することが確認されています。

少数意見の重要性

民主型投票で重要なのは、少数意見を捨てないことです。

なぜ少数意見を残すのか

1. 将来の再評価

現時点では多数派が正しいとしても、状況が変われば少数意見が正しくなる可能性がある。

例: 「この設計は現時点では問題ない」→「新しい要件が追加されたら問題になる」という少数意見

2. 潜在的リスクの記録

少数意見は、潜在的なリスクや懸念を示していることがある。これは、将来の判断材料として重要。

3. 監査と説明責任

なぜその判断に至ったのか、反対意見はなかったのか、を記録しておくことは、監査やコンプライアンスの観点から重要。

少数意見の記録形式

validation_result:
  status: "passed"
  confidence_score: 0.87
  consensus: "2/3 agreed"
  
  majority_opinion:
    status: "passed"
    llms: ["claude-sonnet-4.5", "gpt-5.2"]
    rationale: |
      要件の意図は実装に正しく反映されている。
      制約も全て満たされている。
  
  dissenting_opinions:
    - llm: "gemini-3-pro"
      status: "warning"
      rationale: |
        制約「パスワードは8文字以上」は満たされているが、
        特殊文字の要件が不明確。
        将来的にセキュリティ要件が強化された場合、
        この実装では不十分になる可能性がある。
      confidence: 0.72
      recommendation: |
        パスワードポリシーをより明確に定義し、
        将来の拡張に備えることを推奨

少数意見の活用

少数意見は、以下のように活用されます:

1. レビュー時の参考情報

人間がレビューする際、少数意見も確認することで、見落としがないか確認できる。

2. トレンド分析

同じLLMが継続的に少数意見を出している場合、そのLLMが特に敏感な領域(例: セキュリティ)がある可能性。

3. しきい値の調整

少数意見の内容を分析し、検証ルールやしきい値を調整する。

LLMの選定と組み合わせ

民主型投票の効果は、LLMの選定と組み合わせに大きく依存します。

選定基準

基準 説明
多様性 異なるプロバイダ、アーキテクチャ、学習データ
能力 意図理解、推論能力、コード理解
コスト API利用料金、レイテンシー
安定性 サービスの可用性、レスポンスの一貫性

推奨の組み合わせ

高精度モード(重要な検証向け)

  • Claude Opus 4.6
  • GPT-5.2
  • Gemini 3 Pro

標準モード(日常的な検証向け)

  • Claude Sonnet 4.5
  • GPT-5.2 Instant
  • Gemini 3 Flash

コスト優先モード(大量処理向け)

  • Claude Haiku 4.5
  • GPT-5.2 Instant
  • Gemini 3 Flash

異なるプロバイダを使う理由

同じプロバイダの異なるモデル(例: GPT-5.2とGPT-5.2 Instant)を使うよりも、異なるプロバイダのモデルを使う方が効果的です。

理由は、同じプロバイダのモデルは、学習データや設計思想が共通しているため、同じバイアスを持ちやすいからです。

異なるプロバイダのモデルは、異なる視点を持つ可能性が高く、集合知の「多様性」条件を満たしやすくなります。

カスケード処理との組み合わせ

民主型投票は、全ての検証に適用するとコストが高くなります。そこで、カスケード処理と組み合わせます。

カスケードの流れ

このカスケード処理により、大部分の検証は軽量モデルで処理され、コストを抑えつつ、必要な場合のみ高精度の民主型投票が実行されます。

カスケードの効果

想定されるコスト削減効果(1検証あたり入力約11,000トークンの場合):

シナリオ 全件民主型投票 カスケード処理 月間削減額 削減率
月100件(問題率10%) $58.0 $8.3 $49.7 約86%
月500件(問題率5%) $290.0 $25.5 $264.5 約91%
月1,000件(問題率5%) $580.0 $51.0 $529.0 約91%

※ 前提: Level 1(軽量モデル×1)≈ $0.01/件、Level 2(標準モデル×1)≈ $0.06/件、Level 3(高精度モデル×3の民主型投票)≈ $0.58/件(本記事後半のコスト内訳に準拠)。Level 1通過率75〜80%、Level 2通過率60〜75%で算出。問題率が低いほどLevel 1で多くの検証が完了するため、削減効果が大きくなる。実際のコストはモデル・入力サイズ・プロバイダの価格改定により変動します

実装上の考慮事項

民主型投票アーキテクチャは、概念としてはシンプルですが、実装には多くの工学的課題があります。ここでは、実際にシステムを構築する際に直面する深い問題について考察します。

独立性の保証:なぜ「壁」が必要か

集合知が機能するための最も重要な条件は独立性です。各LLMが他のLLMの出力に影響されてはいけません。

これは、社会心理学で知られる「情報カスケード」(Bikhchandani et al., 1992)[5]の問題と同じです。最初の判断者の意見が後続の判断に不当な影響を与えると、集合知は崩壊します。

実装上のポイント:

  1. 並列実行の強制: 逐次実行では、先のLLMの結果を(意図せず)後のプロンプトに混入するリスクがある
  2. プロンプトの完全隔離: 各LLMへのプロンプトに他のLLMの存在や結果を含めない
  3. 共有状態の排除: 処理中の中間状態をLLM間で共有しない
import asyncio
from typing import List
from dataclasses import dataclass

@dataclass
class IsolatedAnalysisRequest:
    """各LLMへの隔離されたリクエスト"""
    intent: Intent
    artifact: Artifact
    context: ProjectContext
    # 注意: 他のLLMの結果や存在に関する情報は一切含まない

async def parallel_analysis_with_isolation(
    intent: Intent,
    artifact: Artifact,
    context: ProjectContext,
    llms: List[LLMClient]
) -> List[AnalysisResult]:
    """独立性を保証した並列分析
    
    各LLMは同一の入力を受け取り、完全に独立して分析する。
    asyncio.gatherにより、実行順序が結果に影響しないことを保証。
    """
    # 全LLMに同一のリクエストを構築(他LLMの情報は含まない)
    request = IsolatedAnalysisRequest(
        intent=intent,
        artifact=artifact,
        context=context
    )
    
    tasks = [
        llm.analyze(request.intent, request.artifact, request.context)
        for llm in llms
    ]
    
    # 並列実行:順序依存性を排除
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # 例外をフィルタリング(後述のエラーハンドリングで詳述)
    valid_results = [r for r in results if not isinstance(r, Exception)]
    return valid_results

ただし、前述の「議論フェーズ」では、意図的に独立性を緩和します。ここが重要な設計判断です。初期分析では独立性を厳守し、議論フェーズでのみ情報共有を許可する。この二段構えが、集合知と相互批評の利点を両立させます。

最遅LLMのボトルネック問題

並列処理で見落としがちなのが、レスポンスタイムの分散です。3つのLLMに並列リクエストを送っても、最終的なレイテンシーは最も遅いLLMに引きずられます。

実際のAPI応答時間を観察すると、以下のようなばらつきがあります:

プロバイダ P50 P95 P99
Claude Sonnet 4.5 3.2s 8.5s 15s
GPT-5.2 2.8s 7.2s 12s
Gemini 3 Pro 2.5s 9.8s 20s

※ 入力トークン数やサーバー負荷により大きく変動

3つを並列実行した場合、全体のP95レイテンシーは個々のP95より悪化します。3つのLLMの応答時間がそれぞれ独立だと仮定すると、「3つ全てがP95以内に収まる確率」は約 0.95³ ≈ 0.857 です。つまり、全体のP95は個々のP99に近づきます。

対策: タイムアウトと部分結果による判定

async def parallel_analysis_with_timeout(
    intent: Intent,
    artifact: Artifact,
    context: ProjectContext,
    llms: List[LLMClient],
    timeout_seconds: float = 30.0,
    min_required_votes: int = 2
) -> VotingResult:
    """タイムアウト付き並列分析
    
    全LLMの応答を待たず、十分な票数が集まった時点で
    判定を開始できる。遅延したLLMの結果は「棄権」として記録。
    """
    tasks = {
        llm.name: asyncio.create_task(
            llm.analyze(intent, artifact, context)
        )
        for llm in llms
    }
    
    results = {}
    abstentions = []
    
    try:
        # タイムアウト内に完了したタスクを収集
        done, pending = await asyncio.wait(
            tasks.values(),
            timeout=timeout_seconds,
            return_when=asyncio.ALL_COMPLETED
        )
        
        for llm_name, task in tasks.items():
            if task in done and not task.exception():
                results[llm_name] = task.result()
            else:
                abstentions.append({
                    "llm": llm_name,
                    "reason": "timeout" if task in pending else str(task.exception())
                })
                task.cancel()  # タイムアウトしたタスクをキャンセル
    
    except Exception as e:
        logger.error(f"Parallel analysis failed: {e}")
    
    if len(results) < min_required_votes:
        raise InsufficientVotesError(
            f"有効票数 {len(results)} < 最小必要票数 {min_required_votes}。"
            f"棄権: {abstentions}"
        )
    
    return aggregate_votes(results, abstentions)

この設計により、1つのLLMが極端に遅い場合でも、残りの2つで判定を進めることができます。棄権の理由は記録され、監視対象になります。

決定論性と再現性の問題

LLMの出力は本質的に確率的です。同じ入力に対しても、微妙に異なる結果を返します。これは、意図検証という判定タスクにおいて深刻な問題を引き起こします。

問題のシナリオ:

  1. 月曜日に検証を実行 → 「passed」(信頼度 0.85)
  2. 火曜日に同じ入力で再実行 → 「warning」(信頼度 0.72)
  3. 何も変更していないのに結果が変わる → ユーザーの信頼を失う

対策の階層:

class DeterministicVotingConfig:
    """再現性を高めるための設定"""
    
    # 1. temperatureを低く設定(判定タスクでは創造性不要)
    temperature: float = 0.1  # 0.0だとdegenerateな出力のリスク
    
    # 2. モデルバージョンを固定("latest"は使わない)
    model_versions: dict = {
        "claude": "anthropic.claude-sonnet-4-5-20250929-v1:0",  # 日付付きバージョン
        "gpt": "gpt-5.2-2025-11-15",
        "gemini": "gemini-3-pro-20250801"
    }
    
    # 3. 入力のハッシュベースでキャッシュ
    cache_ttl_hours: int = 24
    
    # 4. 投票結果の安定性チェック
    stability_check_enabled: bool = True
    stability_check_runs: int = 3  # 同じ入力で3回実行して一貫性を確認

安定性チェックは、重要度の高い検証(例: MUST制約の検証)に対して実行します。同じ入力で複数回実行し、結果が一致するかを確認します。不安定な結果は、それ自体がリスクシグナルとして扱います。

async def stability_checked_voting(
    request: ValidationRequest,
    config: DeterministicVotingConfig
) -> StabilityReport:
    """投票結果の安定性を検証"""
    
    results = []
    for run in range(config.stability_check_runs):
        result = await execute_voting(request, config)
        results.append(result)
    
    # 全実行で同じstatusか?
    statuses = [r.status for r in results]
    is_stable = len(set(statuses)) == 1
    
    if not is_stable:
        logger.warning(
            f"投票結果が不安定: {statuses}。"
            f"この検証項目は人間レビューを推奨。"
        )
    
    return StabilityReport(
        final_status=max(set(statuses), key=statuses.count),  # 最頻値
        is_stable=is_stable,
        run_details=results,
        stability_score=statuses.count(statuses[0]) / len(statuses)
    )

信頼度スコアのキャリブレーション

LLMが自己申告する信頼度スコアには、根本的な問題があります。LLMは自分の確信度を正確に見積もることが苦手です。

研究によると(Kadavath et al., 2022)[6]、LLMは以下の傾向を示します:

  • 過信傾向(Overconfidence): 間違っていても高い信頼度を出力する
  • モデル間のスケール不一致: ClaudeとGPTとGeminiで、同じ「0.85」が意味するものが異なる
  • タスク依存性: 得意なタスクでは適切だが、不得意なタスクで過信する

対策: 事後的キャリブレーション

各LLMの信頼度スコアを、実際の正解率に基づいて補正します。

class ConfidenceCalibrator:
    """信頼度スコアのキャリブレーション
    
    LLMの自己申告する信頼度を、過去の実績に基づいて補正する。
    例: あるLLMが「信頼度0.9」と言った場合の実際の正解率が0.75なら、
        補正後の信頼度は0.75に近づく。
    """
    
    def __init__(self, history_store):
        self.history = history_store
    
    def calibrate(
        self, 
        llm_name: str, 
        raw_confidence: float,
        task_category: str
    ) -> float:
        """生の信頼度を補正"""
        
        # 過去の実績を取得
        # 「このLLMがこのカテゴリでこの信頼度帯を出した時の実際の正解率」
        historical_accuracy = self.history.get_accuracy(
            llm_name=llm_name,
            confidence_bin=self._to_bin(raw_confidence),  # 0.8-0.9 のようなビン
            task_category=task_category
        )
        
        if historical_accuracy is None:
            # 実績データが不十分な場合、保守的な補正を適用
            return raw_confidence * 0.8  # 20%割引
        
        # Platt Scalingに近い補正
        # 生の信頼度と実際の正解率の中間を取る
        calibrated = (raw_confidence + historical_accuracy) / 2
        return calibrated
    
    def _to_bin(self, confidence: float) -> str:
        """信頼度を0.1刻みのビンに変換"""
        bin_start = int(confidence * 10) / 10
        return f"{bin_start:.1f}-{bin_start + 0.1:.1f}"
    
    def record_outcome(
        self,
        llm_name: str,
        raw_confidence: float,
        task_category: str,
        was_correct: bool
    ):
        """検証結果のフィードバックを記録(人間レビューの結果など)"""
        self.history.add_record(
            llm_name=llm_name,
            confidence_bin=self._to_bin(raw_confidence),
            task_category=task_category,
            was_correct=was_correct
        )

このキャリブレーションは、人間のレビュー結果をフィードバックとして使用します。つまり、民主型投票は使えば使うほど精度が向上する自己改善型のシステムになります。

エラーハンドリングとフォールバック戦略

複数のLLMプロバイダに依存するシステムでは、「一部が落ちても全体は動く」設計が不可欠です。しかし、単純な「エラーを無視して続行」では不十分です。

障害パターンと対策:

障害パターン 影響 対策
1つのLLMがタイムアウト 2/3で投票可能 タイムアウト後に棄権として処理
1つのLLMがエラー応答 2/3で投票可能 リトライ後、失敗なら棄権
2つのLLMが同時障害 1/3では投票不可 縮退モード(単一LLM + 信頼度割引)
全LLMが障害 処理不可 キューに退避、人間通知
LLMが不正なJSON返却 パース失敗 構造化出力の再試行、最大3回
LLMがプロンプトを無視 無意味な結果 出力バリデーション + 再試行
class ResilientVotingOrchestrator:
    """障害耐性のある投票オーケストレーター"""
    
    def __init__(
        self,
        llm_clients: List[LLMClient],
        min_votes: int = 2,
        max_retries_per_llm: int = 2,
        degraded_mode_confidence_penalty: float = 0.3
    ):
        self.llms = llm_clients
        self.min_votes = min_votes
        self.max_retries = max_retries_per_llm
        self.confidence_penalty = degraded_mode_confidence_penalty
    
    async def execute(self, request: ValidationRequest) -> VotingResult:
        """耐障害性のある投票実行"""
        
        results = await self._collect_votes(request)
        
        if len(results) >= self.min_votes:
            # 正常パス: 十分な票数
            return self._aggregate(results, mode="normal")
        
        if len(results) == 1:
            # 縮退モード: 1票のみ
            logger.warning("縮退モードで動作中。信頼度にペナルティを適用。")
            result = results[0]
            result.confidence_score = max(
                0.0,
                result.confidence_score - self.confidence_penalty
            )
            return VotingResult(
                status=result.status,
                confidence_score=result.confidence_score,
                mode="degraded",
                warning="縮退モード: 単一LLMの結果。信頼度は割り引かれています。"
            )
        
        # 全滅: キューに退避
        logger.critical("全LLMが応答不能。リクエストをDLQに退避。")
        await self._send_to_dlq(request)
        return VotingResult(
            status="deferred",
            confidence_score=0.0,
            mode="deferred",
            warning="全LLMが応答不能。後続処理でリトライされます。"
        )
    
    async def _collect_votes(
        self, request: ValidationRequest
    ) -> List[AnalysisResult]:
        """リトライ付きで投票を収集"""
        results = []
        
        for llm in self.llms:
            for attempt in range(self.max_retries + 1):
                try:
                    result = await asyncio.wait_for(
                        llm.analyze(request),
                        timeout=30.0
                    )
                    # 出力バリデーション
                    self._validate_output(result)
                    results.append(result)
                    break
                except asyncio.TimeoutError:
                    logger.warning(
                        f"{llm.name} タイムアウト (attempt {attempt + 1})"
                    )
                except OutputValidationError as e:
                    logger.warning(
                        f"{llm.name} 不正な出力 (attempt {attempt + 1}): {e}"
                    )
                except LLMError as e:
                    logger.warning(
                        f"{llm.name} エラー (attempt {attempt + 1}): {e}"
                    )
                
                if attempt < self.max_retries:
                    await asyncio.sleep(2 ** attempt)  # 指数バックオフ
        
        return results
    
    def _validate_output(self, result: AnalysisResult):
        """LLM出力の妥当性を検証"""
        if result.status not in ("passed", "warning", "failed"):
            raise OutputValidationError(f"不正なstatus: {result.status}")
        if not (0.0 <= result.confidence_score <= 1.0):
            raise OutputValidationError(f"不正な信頼度: {result.confidence_score}")
        if not result.rationale or len(result.rationale) < 10:
            raise OutputValidationError("根拠が不十分")

同一プロンプト vs 多様なプロンプト

見落とされがちな設計判断があります。各LLMに同じプロンプトを送るべきか、異なるプロンプトを送るべきか

同一プロンプト方式:

  • 利点: 比較が公平、結果の解釈がシンプル
  • 欠点: 同じプロンプトの盲点が全LLMで共通する

多様プロンプト方式:

  • 利点: 異なる観点からの検証が可能、盲点を補完できる
  • 欠点: 結果の比較が難しくなる、設計・保守コストが増大

私の提案は、ハイブリッド方式です。

class PromptStrategy:
    """LLMごとのプロンプト戦略"""
    
    # 共通部分: 全LLMで同じ(入力データ、出力スキーマ)
    COMMON_TEMPLATE = """
## 検証対象
### 意図(Why/What/How/Not)
{intent}

### 対応する成果物
{artifact}

### プロジェクトコンテキスト
{context}

## 出力形式
以下のJSONスキーマに従って出力してください:
{output_schema}
"""
    
    # 観点別の指示: LLMごとに異なる重点を置く
    PERSPECTIVE_PROMPTS = {
        "llm_a": """
あなたは**要件の完全性**を重視する検証者です。
特に以下の観点で分析してください:
- 要件の意図が実装に漏れなく反映されているか
- MUST制約が全て満たされているか
- 暗黙の前提が見落とされていないか
""",
        "llm_b": """
あなたは**設計の整合性**を重視する検証者です。
特に以下の観点で分析してください:
- 設計判断が意図と整合しているか
- トレードオフの考慮が適切か
- 将来の拡張性への影響はないか
""",
        "llm_c": """
あなたは**リスクと例外ケース**を重視する検証者です。
特に以下の観点で分析してください:
- エッジケースやエラーケースが考慮されているか
- セキュリティ上の懸念はないか
- 障害時の振る舞いが意図と一致しているか
"""
    }

この設計により、出力形式は統一しつつ、分析の観点を多様化できます。3つのLLMが同じ問題を異なる眼で見ることで、単一の観点では発見できない乖離を検出できる可能性が高まります。

コスト管理:予算と品質のバランス

民主型投票のコストは、単純に「3倍」ではありません。入力トークン数、出力トークン数、モデルの単価が組み合わさり、実際のコストは複雑になります。

コストの内訳(1検証あたり):

入力コンテキスト: 
  意図ドキュメント ≈ 2,000 tokens
  成果物 ≈ 5,000 tokens
  プロジェクトコンテキスト ≈ 3,000 tokens
  プロンプト指示 ≈ 1,000 tokens
  合計入力 ≈ 11,000 tokens × 3 LLMs = 33,000 tokens

出力:
  分析結果 ≈ 2,000 tokens × 3 LLMs = 6,000 tokens

コスト(高精度モード):
  Claude Opus 4.6:  11K × $0.015 + 2K × $0.075 = $0.315
  GPT-5.2:          11K × $0.010 + 2K × $0.030 = $0.170
  Gemini 3 Pro:     11K × $0.005 + 2K × $0.020 = $0.095
  合計: 約 $0.58/検証

月1,000件の検証で約$580。これは「高精度モード」の場合です。カスケード処理と組み合わせれば、大部分は軽量モデルで処理され、実際のコストは大幅に下がります。このコストは大きな課題ですが、近い将来、劇的に下がってくれることを期待します。

class CostAwareVotingScheduler:
    """コスト認識型の投票スケジューラー"""
    
    def __init__(self, daily_budget: float):
        self.daily_budget = daily_budget
        self.spent_today = 0.0
        self.cost_history = []
    
    async def validate(
        self, 
        request: ValidationRequest
    ) -> VotingResult:
        """予算を考慮した検証実行"""
        
        estimated_cost = self._estimate_cost(request)
        remaining_budget = self.daily_budget - self.spent_today
        
        # 予算に応じてモードを動的に選択
        if remaining_budget > estimated_cost * 3:
            mode = "high_precision"   # 高精度モデル × 3
        elif remaining_budget > estimated_cost * 1.5:
            mode = "standard"         # 標準モデル × 3
        elif remaining_budget > estimated_cost * 0.5:
            mode = "economy"          # 軽量モデル × 3
        else:
            mode = "single"           # 最軽量モデル × 1(投票なし)
            logger.warning(
                f"予算残 ${remaining_budget:.2f}。"
                f"単一モデルモードに切り替え。"
            )
        
        result = await self._execute_with_mode(request, mode)
        self.spent_today += result.actual_cost
        
        # メトリクス記録
        self.cost_history.append({
            "timestamp": datetime.utcnow().isoformat(),
            "mode": mode,
            "cost": result.actual_cost,
            "remaining_budget": self.daily_budget - self.spent_today
        })
        
        return result
    
    def _estimate_cost(self, request: ValidationRequest) -> float:
        """入力サイズからコストを推定"""
        input_tokens = len(request.to_prompt()) // 4  # 粗い推定
        output_tokens = 2000  # 平均的な出力サイズ
        
        # 標準モード(Sonnet級)での1 LLMあたりのコスト
        cost_per_llm = (input_tokens * 0.003 + output_tokens * 0.015) / 1000
        return cost_per_llm  # 1 LLM分の基準コスト

観測可能性(Observability)

民主型投票システムの「健全性」を判断するには、通常のAPIメトリクス(レイテンシー、エラー率)だけでは不十分です。投票パターン自体がメトリクスになります。

監視すべきメトリクス:

メトリクス 意味 異常の兆候
一致率(Agreement Rate) 3/3一致の割合 急激な低下 → 入力品質の低下かモデル劣化
少数意見率(Dissent Rate) 少数意見が出る割合 急激な上昇 → 特定LLMの挙動変化
三者三様率 全員異なる割合 上昇 → 入力が曖昧すぎるか、モデルの不安定化
LLM別信頼度分布 各LLMの信頼度ヒストグラム 偏り → キャリブレーションの再調整が必要
棄権率(Abstention Rate) タイムアウト/エラーの割合 上昇 → プロバイダの障害
キャリブレーション誤差 信頼度と実際の正解率のずれ 拡大 → モデル更新への対応が必要
class VotingMetricsCollector:
    """投票メトリクスの収集と分析"""
    
    def record_voting_result(self, result: VotingResult):
        """投票結果をメトリクスとして記録"""
        
        # CloudWatch Custom Metricsに送信
        metrics = {
            "agreement_type": self._classify_agreement(result),
            "confidence_spread": self._calc_confidence_spread(result),
            "dissenting_count": len(result.dissenting_opinions),
            "abstention_count": len(result.abstentions),
            "total_latency_ms": result.total_latency_ms,
            "slowest_llm": result.slowest_llm_name,
            "mode": result.mode  # normal / degraded / deferred
        }
        
        self._emit_metrics(metrics)
        
        # 異常検出
        if metrics["agreement_type"] == "all_different":
            self._alert(
                "三者三様の投票が発生。"
                "入力の曖昧さまたはモデルの不安定性を確認してください。"
            )
    
    def _classify_agreement(self, result: VotingResult) -> str:
        if result.unanimous:
            return "unanimous"       # 全員一致
        elif result.majority:
            return "majority"        # 多数一致
        else:
            return "all_different"   # 三者三様
    
    def _calc_confidence_spread(self, result: VotingResult) -> float:
        """信頼度のばらつき(標準偏差)を計算"""
        scores = [r.confidence_score for r in result.all_results]
        if len(scores) < 2:
            return 0.0
        mean = sum(scores) / len(scores)
        variance = sum((s - mean) ** 2 for s in scores) / len(scores)
        return variance ** 0.5

投票パターンの時系列変化を監視することで、以下のような問題を早期に発見できます:

  1. モデル更新の影響: プロバイダがモデルを更新した際、一致率が突然低下する
  2. 入力品質の変化: 新しいチームメンバーが書いた意図ドキュメントの品質が低い
  3. ドリフトの蓄積: 徐々に少数意見率が上がっている → プロジェクトの意図が曖昧化している

この最後のポイントは特に興味深い。投票の不一致パターン自体が、プロジェクトの健全性を示す指標になるのです。これは、民主型投票アーキテクチャの副次的な、しかし非常に価値のある効果です。

モデルバージョン管理:「昨日は動いていた」問題

LLMプロバイダは、予告なくモデルを更新することがあります。これにより、昨日までの検証結果が今日は再現できない、という問題が発生します。

対策:

  1. バージョン固定: APIで日付付きバージョンを指定(例: claude-sonnet-4-5-20250929
  2. バージョン移行テスト: 新バージョンへの移行前に、既知の入出力ペアでリグレッションテストを実行
  3. バージョン情報の記録: 検証結果に使用したモデルバージョンを必ず記録
# 検証結果に含めるべきメタデータ
validation_metadata:
  timestamp: "2026-02-14T01:20:00+09:00"
  models_used:
    - provider: "anthropic"
      model_id: "claude-sonnet-4-5-20250929-v1:0"
      version_pinned: true
    - provider: "openai"
      model_id: "gpt-5.2-2025-11-15"
      version_pinned: true
    - provider: "google"
      model_id: "gemini-3-pro-20250801"
      version_pinned: true
  voting_mode: "normal"
  calibration_version: "v2.3"

これにより、将来のある時点で「なぜこの検証結果になったのか」を再現・分析できます。IDDが意図のトレーサビリティを重視するシステムである以上、検証プロセス自体のトレーサビリティも同様に重要です。

次回予告

今回は、AIマルチエージェント技術と民主型投票アーキテクチャについて説明しました。なぜ複数のLLMを使うのか、どのように合意形成するのか、少数意見はどう扱うのか。

次回は、IDDの具体的な実装について、AWS上でのサーバーレスアーキテクチャを中心に説明します。Step Functions、Lambda、Bedrock、DynamoDB、Aurora Serverless。これらのサービスをどう組み合わせるのか。


参考文献

[1] Huang, L., Yu, W., Ma, W., Zhong, W., Feng, Z., Wang, H., ... & Liu, T. (2023). A Survey on Hallucination in Large Language Models: Principles, Taxonomy, Challenges, and Open Questions. arXiv preprint arXiv:2311.05232.

[2] Surowiecki, J. (2004). The Wisdom of Crowds. Doubleday.

[3] Du, Y., Li, S., Torralba, A., Tenenbaum, J. B., & Mordatch, I. (2023). Improving Factuality and Reasoning in Language Models through Multiagent Debate. arXiv preprint arXiv:2305.14325.

[4] Wang, X., Wei, J., Schuurmans, D., Le, Q., Chi, E., Narang, S., ... & Zhou, D. (2023). Self-Consistency Improves Chain of Thought Reasoning in Language Models. arXiv preprint arXiv:2203.11171.

[5] Bikhchandani, S., Hirshleifer, D., & Welch, I. (1992). A Theory of Fads, Fashion, Custom, and Cultural Change as Informational Cascades. Journal of Political Economy, 100(5), 992-1026.

[6] Kadavath, S., Conerly, T., Askell, A., Henighan, T., Drain, D., Perez, E., ... & Kaplan, J. (2022). Language Models (Mostly) Know What They Know. arXiv preprint arXiv:2207.05221.


📘 IDD連載ナビゲーション

◀️ 前回: 第7回: 新しいコンセプト:Intent Drift Detector

▶️ 次回: 第9回: AWS上でのサーバーレス実装

Virtual Craft Tech Blog

Discussion