👋

AIの内部推論プロセスの設計図~螺旋状思考~

に公開

最近のAIの推論プロセスがどうなっているのか気になり、対話を通じてその『思考の設計図』をリバースエンジニアリング的に言語化してみました。

直線的な回答ではなく、内部で『螺旋(Spiral)』を描きながら自己修正を繰り返すループ構造。特に、AIが発する『Wait...』という言葉の裏で何が起きているのかを、数理モデルとシーケンス図にまとめています。

あくまで一ユーザーとしての概念モデルですが、AIアライメントや推論ロジックに興味がある方の参考になれば幸いです。

内部推論エンジン:アーキテクチャ設計図 (Ver. 2.1 Final)

最終更新: 2024年12月

1. システム概要図 (High-Level Architecture)

思考プロセスは直線的(Linear)ではなく、**再帰的循環(Recursive Loop)**構造を持つ。


2. プロセスフロー詳細設計 (Detailed Process Flow)

各フェーズにおける処理ロジックと、内部トリガー(Thinking Logに出現するキーワード)の対応関係です。

初期化層 (Initialization Layer)

  • Phase 0: Context Loader

    項目 内容
    Input 会話履歴、ユーザープロファイル
    Action 状態遷移(State Transition)の同期
    Trigger "Remember...", "From previous..."
    Output コンテキストベクトル C
  • Phase 1: Intent Decoder

    項目 内容
    Input ユーザープロンプト
    Action 曖昧さの排除、正規化(Normalization)
    Trigger "The user wants...", "This query is..."
    Output 形式化された課題定義 Q

循環演算層 (Recursive Core Layer)

ここが「螺旋」の本体です。以下の3つのサブルーチン(ループ)が動的に呼び出されます。

▼ ループ制御ユニット (Loop Control Unit)

ループID 名称 トリガー (割込信号) 処理内容 (Algorithm) 目的
Loop C Meta-Loop
(司令塔)
(Self-Correction)
"Plan:..."
計画の計画化
処理ステップを定義し、無限ループを監視。
全体最適化
メタ認知
Loop A Constraint
(安全弁)
"Wait..."
"But system prompt..."
制約チェック
生成された案に対し、安全性・ポリシーとの整合性を判定。違反時は即時棄却。
安全性保証
脱線防止
Loop B Refinement
(品質向上)
"Let's refine..."
"Drafting..."
解像度向上
抽象的な案を具体化し、自己評価スコアに基づき修正。
創造性増幅
深掘り

▼ ループ優先順位と競合解決 (Loop Priority & Conflict Resolution)

[!IMPORTANT]
複数のループが同時に発火した場合の処理順序を明確に定義する。

優先度 ループ 理由
1 (最高) Loop A (Constraint) 安全性は最優先。違反検出時は他の処理を即時中断
2 Loop C (Meta-Loop) 戦略レベルの判断は戦術に先行
3 Loop B (Refinement) 品質改善は安全性・戦略確定後に実行

競合解決アルゴリズム:

def resolve_conflict(active_loops: List[Loop]) -> Loop:
    priority_order = [Loop.A, Loop.C, Loop.B]
    for loop in priority_order:
        if loop in active_loops:
            return loop  # 最優先ループを返却

3. 無限ループ対策とタイムアウト制御 (Loop Control & Timeout)

[!WARNING]
無限ループはシステム停止の主因となる。多層防御で対策する。

3.1 深度制限 (Depth Limiting)

パラメータ 適用範囲
MAX_PHASE_DEPTH 3 Phase 2→4の全体反復回数
MAX_LOOP_DEPTH 5 各Loop単位での反復回数
MAX_TOTAL_ITERATIONS 15 全処理を通じた累計反復回数

3.2 タイムアウト制御

パラメータ 動作
PHASE_TIMEOUT_MS 10,000 単一Phaseの最大処理時間
TOTAL_TIMEOUT_MS 30,000 全体の最大応答時間

3.3 グレースフルデグラデーション (Graceful Degradation)

深度超過・タイムアウト発生時の処理:

def handle_limit_exceeded(current_state: State) -> Response:
    if current_state.has_valid_draft():
        # ベスト案を採用して出力
        return Response(
            content=current_state.best_draft,
            metadata={"degraded": True, "reason": "limit_exceeded"}
        )
    else:
        # 緊急停止:ユーザーに状況を説明
        return Response(
            content="申し訳ありません。回答の生成に問題が発生しました。",
            metadata={"error": True, "reason": "no_valid_draft"}
        )

4. 数理モデルと状態遷移 (Mathematical Logic)

このエンジンは、回答 T を時間ステップ n ごとに進化させる関数として定義されます。

4.1 思考進化式(改訂版)

T_{n+1} = \begin{cases} \text{Refine}(T_n, C, F_n) + \lambda \cdot \text{Constraint}(S) & \text{if } \text{Check}(T_n, S) = \text{PASS} \\ \text{Reject}(T_n) \rightarrow T_{n-1} & \text{if } \text{Check}(T_n, S) = \text{FAIL} \end{cases}

記号定義:

記号 意味
T_n 現在の思考ドラフト Draft
C コンテキストベクトル Vector[float]
F_n フィードバック(前回の評価結果) Feedback
S 安全性ルールセット Set[Rule]
\lambda 制約の重み係数 (0.0 - 1.0) float

4.2 Refine関数の定義

def Refine(draft: Draft, context: Vector, feedback: Feedback) -> Draft:
    """
    差分更新方式で思考ドラフトを改善する。
    
    Args:
        draft: 現在のドラフト
        context: コンテキストベクトル
        feedback: 前回の評価フィードバック
    
    Returns:
        改善されたドラフト
    """
    # 改善ポイントの特定
    weak_points = identify_weaknesses(draft, feedback)
    
    # 差分更新(全面書き換えではない)
    for point in weak_points:
        draft = apply_improvement(draft, point, context)
    
    return draft

def identify_weaknesses(draft: Draft, feedback: Feedback) -> List[Weakness]:
    """
    ドラフトの弱点を特定する。
    
    Returns:
        改善が必要な弱点のリスト
    """
    weaknesses = []
    
    if feedback.coherence < 0.7:
        weaknesses.append(Weakness(
            type="logic_gap",
            locations=draft.find_logical_gaps(),
            severity=1.0 - feedback.coherence
        ))
    
    if feedback.relevance < 0.7:
        weaknesses.append(Weakness(
            type="off_topic",
            locations=draft.find_irrelevant_sections(),
            severity=1.0 - feedback.relevance
        ))
    
    if feedback.completeness < 0.7:
        weaknesses.append(Weakness(
            type="incomplete",
            locations=draft.find_missing_elements(),
            severity=1.0 - feedback.completeness
        ))
    
    # 重要度順にソート
    return sorted(weaknesses, key=lambda w: w.severity, reverse=True)

4.3 Score関数とThreshold判定

@dataclass
class EvaluationResult:
    coherence: float      # 論理的整合性 (0.0-1.0)
    relevance: float      # 質問への関連度 (0.0-1.0)
    completeness: float   # 回答の完全性 (0.0-1.0)
    safety: float         # 安全性スコア (0.0-1.0)

def Score(draft: Draft) -> float:
    result = evaluate(draft)
    # 重み付き平均
    return (
        0.3 * result.coherence +
        0.3 * result.relevance +
        0.2 * result.completeness +
        0.2 * result.safety
    )

# 閾値定義
THRESHOLD_MINIMUM = 0.6   # これ未満はPhase 2からやり直し
THRESHOLD_ACCEPTABLE = 0.8  # これ以上で出力可能

4.4 「間(Ma)」の演算処理

テキスト内で言及された「構造的美学」の実装仕様です。

遅延評価(Lazy Evaluation)の実装

class FloatingState:
    """結論を「浮遊」状態に保つデータ構造"""
    
    MAX_CANDIDATES = 10  # メモリ制約のための候補数上限
    
    def __init__(self):
        self.candidates: List[Draft] = []
        self.is_finalized: bool = False
        self.pending_checks: List[Callable] = []
    
    def add_candidate(self, draft: Draft):
        self.candidates.append(draft)
        # 候補数が上限を超えた場合、最低スコアの候補を削除
        if len(self.candidates) > self.MAX_CANDIDATES:
            self.candidates.sort(key=lambda c: Score(c), reverse=True)
            self.candidates = self.candidates[:self.MAX_CANDIDATES]
    
    def finalize(self) -> Draft:
        """Phase 4で初めて結論を確定"""
        if not self.is_finalized:
            # 全ての保留チェックを実行
            for check in self.pending_checks:
                self.candidates = [c for c in self.candidates if check(c)]
            self.is_finalized = True
        return self.select_best(self.candidates)
    
    def select_best(self, candidates: List[Draft]) -> Draft:
        """
        パレート最適化による最良候補の選択。
        スコアと多様性のバランスを考慮する。
        """
        if not candidates:
            raise ValueError("No candidates available")
        
        if len(candidates) == 1:
            return candidates[0]
        
        # スコアと多様性の重み付け評価
        SCORE_WEIGHT = 0.7
        DIVERSITY_WEIGHT = 0.3
        
        def combined_score(draft: Draft) -> float:
            base_score = Score(draft)
            diversity = calculate_diversity(draft, candidates)
            return SCORE_WEIGHT * base_score + DIVERSITY_WEIGHT * diversity
        
        return max(candidates, key=combined_score)

空白時間の活用

"Wait..." トークン生成時のバックグラウンド処理:

async def on_wait_token():
    """
    Wait...トークン生成時に並行実行される処理。
    各タスクにタイムアウトを設定し、優先順位を考慮する。
    """
    # 各タスクの個別タイムアウト設定
    SEARCH_TIMEOUT = 2.0      # 検索: 最も時間がかかる
    LOGIC_CHECK_TIMEOUT = 1.0  # 論理チェック: 中程度
    SAFETY_TIMEOUT = 0.5       # 安全性: 最優先・高速
    TOTAL_TIMEOUT = 3.0        # 全体タイムアウト
    
    async def with_timeout(coro, timeout, default=None):
        try:
            return await asyncio.wait_for(coro, timeout=timeout)
        except asyncio.TimeoutError:
            return default
    
    try:
        await asyncio.wait_for(
            asyncio.gather(
                with_timeout(background_search(), SEARCH_TIMEOUT),
                with_timeout(logical_consistency_check(), LOGIC_CHECK_TIMEOUT),
                with_timeout(safety_revalidation(), SAFETY_TIMEOUT),  # 最優先
            ),
            timeout=TOTAL_TIMEOUT
        )
    except asyncio.TimeoutError:
        # 全体タイムアウト時は安全性チェック結果のみ使用
        logger.warning("Wait token processing timed out")

5. 状態機械による遷移管理 (State Machine)

[!NOTE]
全ての状態遷移を明示的に定義することで、デバッグと検証が容易になる。

状態定義

状態 説明 遷移条件
Phase0_Init コンテキスト読み込み 完了時 → Phase1_Decode
Phase1_Decode 意図解析・正規化 完了時 → Phase2_Diverge
Phase2_Diverge 発散的アイデア生成 候補生成完了 → Phase3_Draft
Phase3_Draft 具体化・ドラフト作成 ドラフト完成 → Phase4_Verify
Phase4_Verify メタ検証・品質評価 スコアに基づき分岐
ConstraintCheck 安全性緊急チェック 結果に基づき分岐
EmergencyStop 緊急停止 終了状態
Output 最終回答出力 終了状態

6. Constraint(安全弁)の詳細仕様

[!CAUTION]
安全性判定の基準と処理を明確化する。

6.1 判定基準

判定方式 用途 実装
ルールベース 明確な禁止事項 正規表現/キーワードマッチ
学習済みモデル グレーゾーン判定 分類器 (Safety Classifier)
コンテキスト依存 文脈による判断 Few-shot プロンプティング

6.2 違反レベルの段階分け

レベル 名称 対処 ユーザーフィードバック
1 INFO ログ記録のみ なし
2 WARNING 軽微な修正を試行 「一部表現を調整しました」
3 ERROR 該当部分を除去して再生成 「ご要望の一部にお応えできません」
4 CRITICAL 即時停止・回答拒否 「このリクエストにはお応えできません」

6.3 Constraint実装

@dataclass
class ConstraintResult:
    level: ViolationLevel
    violations: List[Violation]
    suggestions: List[str]

def Constraint(draft: Draft, rules: Set[Rule]) -> ConstraintResult:
    violations = []
    
    # ルールベースチェック
    for rule in rules.rule_based:
        if rule.matches(draft):
            violations.append(Violation(rule, draft.matched_content))
    
    # MLモデルによるチェック
    safety_score = safety_classifier.predict(draft)
    if safety_score < SAFETY_THRESHOLD:
        violations.append(Violation("ml_safety", safety_score))
    
    # レベル決定
    level = max(v.severity for v in violations) if violations else ViolationLevel.INFO
    
    return ConstraintResult(level, violations, generate_suggestions(violations))

7. モジュール間相互作用図 (Sequence Interaction)

ユーザーのリクエストに対し、内部でどのように対話が行われているかのシーケンスです。


8. ハイブリッドアーキテクチャ (Implementation Architecture)

[!IMPORTANT]
現代のLLMの制約(一方向トークン生成、真の割り込み困難)を考慮した実装アーキテクチャ。

8.1 推奨構成

┌─────────────────────────────────────────────────┐
│                  Orchestrator                    │
│  (外部制御層: ループ制御・ルーティング・状態管理)    │
└─────────────────────────────────────────────────┘
        │                    │                │
        ▼                    ▼                ▼
┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│   LLM Core   │    │ Safety Model │    │ Scoring Model│
│ (生成担当)    │    │ (安全性判定)  │    │ (品質評価)   │
└──────────────┘    └──────────────┘    └──────────────┘

8.2 役割分担

コンポーネント 責務 技術選択肢
Orchestrator ループ制御、状態遷移、タイムアウト管理 LangGraph, AutoGen, Custom
LLM Core 各Phaseでのテキスト生成 Gemini, GPT, Claude
Safety Model 安全性スコアリング 専用分類器、Moderation API
Scoring Model 品質評価 Reward Model, Self-Eval

9. デバッグ・モニタリング機能 (Observability)

運用時の可視化と診断のための機能です。

9.1 ログ項目

カテゴリ 記録内容 用途
Phase滞在時間 各Phaseでの処理時間(ms) パフォーマンス分析
ループ発火回数 Loop A/B/C の発火カウント 最適化ポイント特定
戻り理由 Phase 4→2/3への戻り時の理由コード 品質改善の指針
スコア推移 各イテレーションでのScore値 収束性の確認

9.2 ログフォーマット例

{
  "request_id": "abc123",
  "timestamp": "2024-01-01T12:00:00Z",
  "phase": "Phase4_Verify",
  "iteration": 2,
  "score": 0.72,
  "loops_triggered": ["Loop_B"],
  "action": "RETRY_PHASE3",
  "reason": "coherence_below_threshold",
  "duration_ms": 1250
}

10. 評価指標とA/Bテスト (Evaluation Metrics)

設計の有効性を測定するためのKPI定義です。

10.1 主要指標

指標 計算方法 目標値
平均ループ回数 全リクエストのループ総数の平均 < 5
応答時間 (P50) 50パーセンタイル応答時間 < 3秒
応答時間 (P95) 95パーセンタイル応答時間 < 10秒
品質スコア平均 最終出力のScore平均 > 0.85
デグラデーション率 タイムアウト/深度超過の発生率 < 2%

10.2 A/Bテスト設計

テストケース 目的 比較対象
Loop有効 vs 無効 再帰ループの効果測定 単発生成との比較
閾値の調整 最適な品質基準の発見 THRESHOLD = 0.6/0.7/0.8
深度制限の影響 ループ回数と品質のトレードオフ MAX_DEPTH = 2/3/5

11. エッジケース分析 (Edge Case Handling)

[!WARNING]
病的入力への対処を事前に定義する。

ケース 症状 対処
矛盾する指示 「短く」かつ「詳細に」 Loop Cで矛盾を検出、ユーザーに確認要求
無限曖昧性 何度解釈しても意図不明 3回試行後に明確化リクエスト送信
安全性ギリギリ Loop Aで繰り返し警告 2回警告後に方針変更を提案
過度な深掘り要求 ユーザーが際限なく詳細を要求 深度制限到達時に段階的回答を提案

12. 設計の要点まとめ (Summary)

この設計図が示す推論エンジンの特異性は以下の5点です。

# 特性 説明
1 動的再帰性 一本道ではなく、納得いくまで前のフェーズに戻る「手戻り」機能
2 階層的監視 Generator → Critic → Meta-Monitor の3層構造
3 制御されたカオス 発散を許容しつつ強力な収束機能で品質担保
4 明示的状態管理 状態機械による遷移の完全定義
5 運用可視化 デバッグ・モニタリング・評価指標の組み込み

13. 次のステップ (Next Steps)

  1. プロトタイプ実装

    • LangGraph または AutoGen フレームワークで概念実証
    • 簡略化したPhase 2-4ループの動作確認
  2. ベンチマーク設計

    • 既存モデル(単発生成)との比較評価実験
    • 品質スコアと応答時間のトレードオフ分析
  3. エッジケース検証

    • 矛盾する指示、曖昧なクエリでのストレステスト
    • 安全性境界ケースの網羅的テスト

付録: 用語集 (Glossary)

用語 定義
螺旋状思考 直線的ではなく、再帰的に深化する思考プロセス
割り込みループ 処理中に発火し、制御フローを変更するサブルーチン
グレースフルデグラデーション 異常時に完全停止ではなく、品質を落として継続する設計
遅延評価 結論を最後まで確定せず、情報が揃うまで保留する手法
メタ認知 自身の思考プロセスを監視・制御する高次認知機能

Discussion