【Claude Fable 5】無料期間で構築する自己改善型AIエージェントシステム

2026年版実践完全ガイド7月7日までの無料期間で最大限Fable5を活用するため本書は、「自律的に学習し、改善を続けるシステム」として設計する段階へと移行するための実践的な手引書です。API呼び出しの解説に留まらず、実際に機能するシステムを構築するための設計思想、アーキテクチャ、そして具体的なコードまでを網羅しています。

第1章　Fable 5とは何か

1.1 AIエージェントフレームワークの系譜

2024年から2026年にかけ、AIエージェントフレームワークは目覚ましい進化を遂げました。LangChain、AutoGen、CrewAIといった初期のフレームワークは、LLMの「オーケストレーション」に特化していましたが、いずれも規模が拡大した際の信頼性、コスト管理、自己修正能力において限界を抱えていました。

Fable 5は、これらの課題を克服するために誕生しました。その設計思想の核心は、次の言葉に集約されます。

「AIエージェントは、ただ動けば良いのではなく、常に賢くなり続けるべきである」

Fable 5は単なるエージェントランナーではありません。エージェントが過去の実行履歴から学び、スキルを蓄積し、次回の実行をより高品質なものにするための自己改善ループを、フレームワークの根幹として実装しています。

1.2 他フレームワークとの根本的な違い

Fable 5が特に優れているのは、「Verification Loop（検証ループ）」と「Skills自動蓄積」の2点です。他のフレームワークでは、エージェントが同じ過ちを繰り返しても誰も気づかないことがありますが、Fable 5はエラーパターンを記録し、次回の実行前に自動的に回避策を適用します。

1.3 2026年のAIエコシステムにおけるFable 5の位置づけ

2026年現在、AIエージェント開発者が直面している主な課題を整理してみましょう。

課題1: モデルの乱立GPT-5.5、Claude Sonnet 5、Gemini 2.5 Ultra、Llama 4、Mistral Largeなど、高性能なモデルが次々と登場しています。タスクに応じた最適なモデル選択が、コストとパフォーマンスの両面に直接影響します。

課題2: 長時間実行の不安定性30分や1時間を超えるエージェントの実行では、ネットワーク障害、コンテキストのオーバーフロー、API制限など、多くの失敗要因が潜んでいます。そのため、途中で中断しても再開できる状態管理が不可欠です。

課題3: 品質の定量化困難エージェントが「良い仕事をしているか」を人間が一つ一つ確認する作業は、規模が大きくなると現実的ではありません。自動的に品質を検証する仕組みが求められます。

課題4: 知識の揮発エージェントが一度発見した有効なアプローチが、次回の実行に引き継がれないことがあります。毎回ゼロから試行錯誤を繰り返す非効率さが課題です。

Fable 5は、これら4つの課題すべてに対し、設計段階から解決策を提供しています。

1.4 Fable 5のコアコンポーネント

Plain Text

Fable 5
├── Orchestrator          # タスク分解・エージェント選択
├── Router                # モデル自動ルーティング
├── Verification Engine   # 出力品質の自動検証
├── Memory Store          # 短期・長期・手続きメモリ
├── Skills Library        # 再利用可能なスキル蓄積
├── State Manager         # チェックポイント・再開
└── Cost Controller       # トークン・コスト最適化

各コンポーネントは独立して機能しますが、STATE.md を中心に情報を共有し、システム全体として自己改善を促進します。

第2章　自己改善型AIという考え方

2.1 「動くAI」から「賢くなるAI」へ

従来のAIエージェント開発では、「タスクを完了すること」が最終目標とされていました。しかし、2026年のプロフェッショナルな開発現場では、この考え方はすでに時代遅れとなっています。

真に価値を生み出すのは、実行するたびに性能が向上するシステムです。人間の専門家が経験を積むことで判断精度を高めるように、AIエージェントも実行履歴から学習し、同じ失敗を繰り返さず、発見した効率的なアプローチを次回に活かすべきです。

自己改善型AIの本質は、以下の3つのサイクルで表現できます。

mermaid

Source

2.2 自己改善の4レベル

Level 1: プロンプト自己修正出力が基準を満たさない場合、エージェントが自らプロンプトを修正し、再試行します。これは最も基本的な自己改善の形です。

Level 2: ツール選択の最適化複数のツールの中から、過去の成功率に基づいて最適なツールを自動的に選択するようになります。「この種のタスクではWeb検索よりもコード実行の方が信頼性が高い」といった判断を自動化します。

Level 3: ワークフローの再構成タスクの分解方法そのものを改善します。当初は直列に実行していた処理を、依存関係を学習することで並列化し、処理時間を短縮します。

Level 4: 新スキルの自律生成繰り返し発生するパターンを検出し、それをスキルとして抽象化・保存します。これにより、次回からはそのスキルを再利用できるようになります。これがFable 5の最も高度な機能です。

2.3 自己改善ループの設計原則

原則1: 観測可能であること（Observability）改善のためには、システム内で何が起きているかを正確に記録する必要があります。Fable 5はすべての実行ステップをSTATE.mdに記録し、事後分析を可能にします。

原則2: 失敗を資産にすること（Failure as Asset）エラーは排除すべき障害ではなく、貴重な学習データです。Fable 5のVerification Engineはエラーパターンを分類・保存し、類似タスクでの事前回避に役立てます。

原則3: 漸進的改善（Incremental Improvement）一度に大きく変更するのではなく、小さな改善を積み重ねていきます。これにより、改善が原因でシステムが不安定になる事態を防ぎます。

原則4: 人間の監督との両立（Human-in-the-Loop）完全な自律は万能ではありません。重要な判断、コスト閾値の超過、未知の状況などでは、人間へのエスカレーションを自動的にトリガーします。

2.4 なぜ今、自己改善型AIが必要か

2026年現在、AIエージェントは以下の3つの圧力に直面しています。

コスト圧力: GPT-5.5のような高性能モデルは、1Mトークンあたり15ドルから30ドルと高価です。大規模な実行では、自動的なコスト最適化がなければ運用は困難です。

品質圧力: エージェントが生成するコード、文章、分析が実際のビジネスで利用される機会が増え、品質基準が厳格化しています。

速度圧力: 競争優位の源泉がAI活用のスピードへと移行しており、毎回ゼロから設計している余裕はありません。

自己改善型システムは、これら3つの課題に同時に対応します。実行を重ねるほど効率が向上し（コスト削減）、品質が高まり（検証ループ）、再利用可能な資産が蓄積される（速度向上）のです。

第3章　Fable 5のアーキテクチャ設計

3.1 全体アーキテクチャ

mermaid

Source

3.2 STATE.mdの設計

STATE.mdは、Fable 5システムの「現在の状況」を記録するファイルです。エージェントはこのファイルを読み込むことで、前回の実行からどこまで完了したかを把握し、中断した場所から再開できます。

Markdown

# STATE.md — Fable 5 実行状態

## メタデータ
- **Session ID**: sess_20260701_001
- **開始時刻**: 2026-07-01T09:00:00Z
- **最終更新**: 2026-07-01T09:47:23Z
- **ステータス**: in_progress
- **現在フェーズ**: phase_3_execution

## タスク概要
yaml
goal: "競合他社のSNS戦略を分析し、次月のコンテンツカレンダーを作成する"
priority: high
deadline: "2026-07-05"
assigned_agents:
  - planner: claude-sonnet-5
  - executor: gpt-5.5
  - verifier: claude-haiku-5

完了フェーズ

現在の作業状態

YAML

current_step: "コンテンツカレンダー生成"
progress: 40%
completed_items:
  - 競合A社: ✓
  - 競合B社: ✓
  - 競合C社: ✓
pending_items:
  - 競合D社〜J社: 7社
errors_encountered: []
retry_count: 0

チェックポイント

JSON

{
  "last_checkpoint": "2026-07-01T09:47:00Z",
  "checkpoint_data": {
    "collected_data": ["competitorA", "competitorB", "competitorC"],
    "partial_calendar": "/tmp/calendar_draft_v1.json"
  }
}

コスト追跡

Plain Text

### 3.3 AGENTS.mdの設計

AGENTS.mdは、システム内で利用可能なエージェントの仕様を定義するファイルです。各エージェントの役割、使用モデル、得意なタスク、コスト特性が記述されています。

markdown
# AGENTS.md — エージェント定義書

## Planner Agent

yaml
id: planner
model: claude-sonnet-5
role: タスク分解・計画立案
strengths:
  - 複雑なタスクの構造化分解
  - 依存関係の特定
  - リスク予測
temperature: 0.3
max_tokens: 4096
cost_per_1m_tokens:
  input: $3.00
  output: $15.00
use_when:
  - タスクが曖昧で構造化が必要なとき
  - 複数のサブタスクが絡み合うとき
avoid_when:
  - 単純な1ステップタスク
  - コスト最優先のとき

Executor Agent

YAML

id: executor
model: gpt-5.5
role: コード実行・情報収集・実作業
strengths:
  - コード生成・実行
  - Web検索・情報収集
  - ファイル操作
  - API呼び出し
temperature: 0.1
max_tokens: 16384
cost_per_1m_tokens:
  input: $2.50
  output: $10.00
tools:
  - code_interpreter
  - web_search
  - file_system
  - api_caller
use_when:
  - 実際の作業実行
  - コードが必要なとき
  - ツール呼び出しが必要なとき

Verifier Agent

YAML

id: verifier
model: claude-haiku-5
role: 出力品質検証・エラー検出
strengths:
  - 論理的整合性チェック
  - コード品質検査
  - 事実確認
temperature: 0.0
max_tokens: 2048
cost_per_1m_tokens:
  input: $0.25
  output: $1.25
use_when:
  - すべての主要出力の検証
  - コスト優先の検証タスク
  - 高速フィードバックが必要なとき

Synthesizer Agent

YAML

id: synthesizer
model: claude-opus-5
role: 複数出力の統合・最終品質保証
strengths:
  - 複数ソースの統合
  - 高品質な文章生成
  - 最終チェック
temperature: 0.4
max_tokens: 32768
cost_per_1m_tokens:
  input: $15.00
  output: $75.00
use_when:
  - 最終的な統合・仕上げ
  - 最高品質が求められるとき
avoid_when:
  - 中間ステップ（コスト高）
  - 単純タスク

Plain Text

### 3.4 WORKFLOW.mdの設計

markdown
# WORKFLOW.md — 標準ワークフロー定義

## ワークフロー: SNS戦略分析

yaml
name: sns_strategy_analysis
version: "2.3"
last_optimized: "2026-06-15"
success_rate: 94.2%
avg_duration: "47min"
avg_cost: "$2.10"

phases:
  - id: planning
    agent: planner
    input: "ユーザーの指示"
    output: "タスク分解リスト・優先順位"
    timeout: 5min
    on_failure: retry_once_then_escalate

  - id: research
    agent: executor
    parallel: true
    max_parallel: 5
    input: "調査対象リスト"
    output: "生データ・JSON形式"
    timeout: 30min
    on_failure: skip_and_continue

  - id: analysis
    agent: executor
    depends_on: research
    input: "生データ"
    output: "分析結果・Markdown"
    timeout: 15min

  - id: verification
    agent: verifier
    depends_on: analysis
    checks:
      - factual_accuracy
      - logical_consistency
      - completeness

第4章　Routerによる動的最適化

4.1 Routerの役割

Router（ルーター）は、各タスクの性質（複雑さ、コスト感、品質要件）を分析し、最適なモデルとエージェントを動的に割り当てるコンポーネントです。

すべてのタスクに最高性能のモデル（例：Claude 5 Opus）を使用すれば品質は安定しますが、コストが爆発的に増大します。Routerは「このタスクは軽量モデル（例：Claude 5 Haiku）で十分か、それともフラッグシップモデルが必要か」を瞬時に判断します。

4.2 動的ルーティングの実装

Python

# fable5/router.py

import re
from enum import Enum
from typing import Dict, List, Optional

class TaskComplexity(Enum):
    TRIVIAL = 1    # 単純な定型作業
    SIMPLE = 2     # 短い文章生成・要約
    MODERATE = 3   # 複数情報の統合・分析
    COMPLEX = 4    # 複雑なロジック・長文生成
    CRITICAL = 5   # 最終成果物・高度な判断

class Fable5Router:
    """
    タスクの内容に基づいて最適なモデルを決定する。
    コストと品質のバランスを自動調整する。
    """

    MODEL_PROFILES = {
        "claude-haiku-5": {"speed": 5, "quality": 2, "cost_input": 0.25, "cost_output": 1.25},
        "claude-sonnet-5": {"speed": 4, "quality": 4, "cost_input": 3.00, "cost_output": 15.00},
        "claude-opus-5": {"speed": 2, "quality": 5, "cost_input": 15.00, "cost_output": 75.00},
        "gpt-5.5": {"speed": 3, "quality": 4.5, "cost_input": 2.50, "cost_output": 10.00},
        "codex-2026": {"speed": 4, "quality": 4, "cost_input": 2.00, "cost_output": 8.00},
    }

    def route(self, task: dict) -> dict:
        description = task.get("description", "")
        complexity = self._estimate_complexity(description)
        task_type = self._detect_task_type(description)

        # 過去の成功データから学習したモデルを選択
        best_past_model = self.memory.get_best_model_for_type(task_type)

        selected_model = self._decide_model(
            complexity,
            task_type,
            task.get("estimated_tokens", 2000),
            task.get("required_quality", "standard"),
            task.get("budget_limit"),
            best_past_model
        )

        return {
            "model": selected_model,
            "complexity": complexity.name,
            "estimated_cost": self._estimate_cost(selected_model, task.get("estimated_tokens", 2000)),
            "reasoning": f"Task type '{task_type}' with complexity '{complexity.name}' detected."
        }

    def _estimate_complexity(self, description: str) -> TaskComplexity:
        # キーワードと文章構造から複雑度を判定
        indicators = {
            TaskComplexity.TRIVIAL: [
                r"format", r"convert", r"simple"
            ],
            TaskComplexity.SIMPLE: [
                r"要約", r"分類", r"翻訳", r"summarize", r"classify"
            ],
            TaskComplexity.MODERATE: [
                r"分析", r"比較", r"evaluate", r"analyze", r"compare"
            ],
            TaskComplexity.COMPLEX: [
                r"設計", r"アーキテクチャ", r"戦略", r"design", r"strategy"
            ],
            TaskComplexity.CRITICAL: [
                r"最高品質", r"本番", r"critical", r"production"
            ],
        }
        for complexity, patterns in reversed(list(indicators.items())):
            if any(re.search(p, description, re.IGNORECASE) for p in patterns):
                return complexity
        return TaskComplexity.MODERATE

    def _decide_model(self, complexity, task_type, context_size,
                      quality, budget, best_past_model) -> str:
        """最終的なモデル選択"""

        # コードタスクはCodexまたはGPT-5.5を優先
        if task_type == "code":
            if complexity.value >= 4:
                return "gpt-5.5"
            return "codex-2026"

        # 検証タスクはHaikuで十分
        if task_type == "verification":
            return "claude-haiku-5"

        # 複雑度とコストのトレードオフ
        model_map = {
            TaskComplexity.TRIVIAL: "claude-haiku-5",
            TaskComplexity.SIMPLE: "claude-haiku-5",
            TaskComplexity.MODERATE: "claude-sonnet-5",
            TaskComplexity.COMPLEX: "claude-sonnet-5",
            TaskComplexity.CRITICAL: "claude-opus-5",
        }
        selected = model_map[complexity]

        # 過去の成功パターンを反映（学習効果）
        if best_past_model and best_past_model["success_rate"] > 0.9:
            selected = best_past_model["model"]

        # 予算制約チェック
        if budget and self._estimate_cost(selected, 5000) > budget:
            selected = self._find_cheaper_alternative(selected)

        return selected

    def _estimate_cost(self, model: str, tokens: int) -> float:
        profile = self.MODEL_PROFILES.get(model, {})
        input_cost = (tokens * 0.7 / 1_000_000) * profile.get("cost_input", 5.0)
        output_cost = (tokens * 0.3 / 1_000_000) * profile.get("cost_output", 20.0)
        return round(input_cost + output_cost, 4)

4.3 動的ルーティングの実例

実際の運用では、同一ワークフロー内であっても、各ステップに異なるモデルが割り当てられます。

Python

# ワークフロー定義例
workflow = {
    "name": "competitive_analysis",
    "steps": [
        {
            "id": "decompose",
            "description": "タスクを具体的なリサーチ項目に分解する",
            # → Router判定: MODERATE + planning → claude-sonnet-5
        },
        {
            "id": "scrape_data",
            "description": "各競合サイトからデータを収集するコードを書く",
            # → Router判定: code + MODERATE → codex-2026
        },
        {
            "id": "analyze",
            "description": "収集したデータを戦略的に分析する",
            # → Router判定: COMPLEX + analysis → claude-sonnet-5
        },
        {
            "id": "verify_facts",
            "description": "分析結果の事実確認",
            # → Router判定: verification → claude-haiku-5
        },
        {
            "id": "final_report",
            "description": "クライアントに提出する最終レポート（最高品質）",
            # → Router判定: CRITICAL → claude-opus-5
        }
    ]
}
# コスト内訳例: $0.05 + $0.08 + $0.45 + $0.02 + $1.20 = $1.80
# 全ステップopus使用時: 約$12.00（6.7倍のコ

4.4 ルーティングの自己最適化

Fable 5のRouterは、実行結果を記録し、ルーティングの判断を継続的に改善していきます。

Python

def update_routing_knowledge(self, task_id: str, result: dict):
    """実行結果からルーティング知識を更新する"""
    decision = self.get_decision(task_id)
    outcome = {
        "model": decision.model,
        "task_type": result["task_type"],
        "success": result["verification_passed"],
        "quality_score": result["quality_score"],
        "actual_cost": result["actual_cost"],
        "duration_seconds": result["duration"],
    }
    # 次回の同種タスクに反映
    self.memory.store_routing_outcome(outcome)

    # 継続的失敗パターンの検出
    if self._detect_systematic_failure(decision.model, result["task_type"]):
        self._update_routing_rules(decision.model, result["task_type"])
        self.logger.warning(
            f"Routing rule updated: {decision.model} → {result["task_type"]} "
            f"success rate dropped below threshold"
        )

第5章　Verification Loop

5.1 Verificationとは何か

Verification Loop（検証ループ）は、エージェントの出力が設定された基準を満たしているかを自動的に検証し、満たしていない場合には修正を促す仕組みです。人間が一つ一つ確認する代わりに、AIが自身の出力を自らチェックします。

重要なのは、Verifier（検証者）がExecutor（実行者）とは異なるモデルを使用することです。同じモデルでは同じバイアスを持つため、独立した検証ができません。

mermaid

Source

5.2 検証基準の設計

Python

# fable5/verification.py

from dataclasses import dataclass, field
from typing import List, Callable
import re

@dataclass
class VerificationCriteria:
    """出力の検証基準を定義する"""
    name: str
    weight: float  # 0.0〜1.0 合計1.0になるように
    checker: Callable
    threshold: float = 0.8

class VerificationEngine:
    """
    複数の検証基準を組み合わせて出力品質を評価する。
    タスクタイプごとに異なる基準セットを適用する。
    """

    def __init__(self, verifier_model="claude-haiku-5"):
        self.verifier_model = verifier_model
        self.criteria_sets = self._build_criteria_sets()

    def _build_criteria_sets(self) -> dict:
        return {
            "code": [
                VerificationCriteria("syntax_valid", 0.3, self._check_syntax),
                VerificationCriteria("tests_pass", 0.4, self._run_tests),
                VerificationCriteria("no_security_issues", 0.2, self._security_check),
                VerificationCriteria("readable", 0.1, self._readability_check),
            ],
            "analysis": [
                VerificationCriteria("factual_accuracy", 0.35, self._fact_check),
                VerificationCriteria("logical_consistency", 0.30, self._logic_check),
                VerificationCriteria("completeness", 0.20, self._completeness_check),
                VerificationCriteria("citation_present", 0.15, self._citation_check),
            ],
            "writing": [
                VerificationCriteria("grammar_correct", 0.20, self._grammar_check),
                VerificationCriteria("tone_appropriate", 0.25, self._tone_check),
                VerificationCriteria("length_appropriate", 0.15, self._length_check),
                VerificationCriteria("key_points_covered", 0.40, self._coverage_check),
            ],
        }

    def verify(self, output: dict, task: dict) -> dict:
        """出力を検証し、スコアと改善指示を返す"""
        task_type = task.get("type", "general")
        criteria = self.criteria_sets.get(task_type, self.criteria_sets["analysis"])

        results = []
        for criterion in criteria:
            score = criterion.checker(output, task)
            results.append({
                "criterion": criterion.name,
                "score": score,
                "weight": criterion.weight,
                "passed": score >= criterion.threshold,
            })

        # 重み付き総合スコア
        total_score = sum(r["score"] * r["weight"] for r in results)
        passed = total_score >= 0.85

        # 失敗した基準に対して改善指示を生成
        failed_criteria = [r for r in results if not r["passed"]]
        improvement_prompt = self._generate_improvement_prompt(
            failed_criteria, output, task
        ) if not passed else None

        return {
            "passed": passed,
            "total_score": round(total_score, 3),
            "criteria_results": results,
            "improvement_prompt": improvement_prompt,
            "recommendation": "proceed" if passed else "revise",
        }

    def _generate_improvement_prompt(
        self, failed_criteria: list, output: dict, task: dict
    ) -> str:
        """失敗した検証基準に基づいて具体的な改善指示を生成する"""
        failures = "\n".join([
            f"- {c['criterion']}: スコア {c['score']:.2f} (基準: {0.8})"
            for c in failed_criteria
        ])
        return f"""
以下の出力を改善してください。

【改善が必要な点】
{failures}

【元のタスク】
{task.get("description", "")}

【現在の出力】
{str(output.get("content", ""))[:1000]}

上記の問題点を修正した改善版を出力してください。
特に以下の点に集中してください：
{self._failure_specific_guidance(failed_criteria)}
"""

5.3 Verification Loopの実装

Python

# fable5/execution_loop.py

class ExecutionLoop:
    """
    Executor → Verifier → (修正) のループを管理する。
    最大retry_limit回まで自動修正を試みる。
    """

    def __init__(self, executor, verifier, state_manager, max_retries=3):
        self.executor = executor
        self.verifier = verifier
        self.state = state_manager
        self.max_retries = max_retries

    async def run(self, task: dict) -> dict:
        attempt = 0
        current_task = task.copy()
        history = []

        while attempt < self.max_retries:
            # 実行
            self.state.update_status(f"executing_attempt_{attempt + 1}")
            output = await self.executor.execute(current_task)

            # 検証
            verification = self.verifier.verify(output, task)
            history.append({
                "attempt": attempt + 1,
                "output": output,
                "verification": verification,
            })

            if verification["passed"]:
                # 成功: スキル候補として記録
                self.state.record_success(task, output, verification)
                self._maybe_create_skill(task, current_task, output)
                return {
                    "status": "success",
                    "output": output,
                    "attempts": attempt + 1,
                    "final_score": verification["total_score"],
                    "history": history,
                }

            # 失敗: プロンプトを改善して再試行
            attempt += 1
            if attempt < self.max_retries:
                current_task = self._inject_improvement(
                    current_task,
                    verification["improvement_prompt"]
                )
                self.state.record_retry(attempt, verification)
            else:
                # 最大試行回数超過
                return {
                    "status": "escalate",
                    "reason": "max_retries_exceeded",
                    "best_output": max(history, key=lambda x: x["verification"]["total_score"]),
                    "attempts": attempt,
                    "history": history,
                }

    def _maybe_create_skill(self, original_task: dict, refined_task: dict, output: dict):
        """成功パターンをスキル候補として記録する"""
        # 同種タスクが3回以上成功していればスキル化
        similar_successes = self.state.count_similar_successes(original_task)
        if similar_successes >= 3:
            skill_candidate = {
                "task_pattern": original_task.get("pattern"),
                "refined_prompt": refined_task.get("system_prompt"),
                "success_template": output.get("content"),
                "avg_score": self.state.get_avg_score(original_task),
            }
            self.state.add_skill_candidate(skill_candidate)

第6章　Memoryシステム

6.1 3層メモリアーキテクチャ

人間の記憶と同様に、Fable 5のメモリは3つの層で構成されています。

Plain Text

Working Memory（作業記憶）
↓ 重要な情報を選別
Episodic Memory（エピソード記憶）
↓ パターンを抽象化
Procedural Memory（手続き記憶）= Skills Library

Working Memory（作業記憶）現在のセッション内でのみ有効な一時的な情報です。コンテキストウィンドウに直接注入され、タスク完了後に消滅します。

Episodic Memory（エピソード記憶）過去の実行履歴を指します。「この種のタスクを以前どのように解決したか」という記録であり、ベクトル検索によって類似事例が取得されます。

Procedural Memory（手続き記憶）繰り返し利用可能な手順やスキルのライブラリです。Skills Libraryとして実装されます。

6.2 Memoryシステムの実装

Python

# fable5/memory.py

import json
import hashlib
from datetime import datetime
from pathlib import Path
from typing import List, Optional

class MemoryStore:
    """
    Fable 5のメモリシステム。
    3層の記憶を管理し、類似タスクの高速検索を提供する。
    """

    def __init__(self, base_path: str = ".fable5/memory"):
        self.base = Path(base_path)
        self.base.mkdir(parents=True, exist_ok=True)
        self.working: dict = {}
        self.episodic: List[dict] = self._load_episodic()
        self.episodic_path = self.base / "episodic.json"

    def store_episode(self, task: dict, output: dict, outcome: dict):
        """実行結果をエピソード記憶として保存する"""
        episode = {
            "id": self._generate_id(task),
            "timestamp": datetime.now().isoformat(),
            "task": task,
            "output": output,
            "outcome": outcome,  # quality_score, success, model, cost等
        }
        self.episodic.append(episode)
        self._save_episodic()

    def find_similar(self, task: dict, limit: int = 5) -> List[dict]:
        """現在のタスクに類似した過去のエピソードを検索する"""
        # 実際の実装では、task['description']のembeddingを用いたベクトル検索を行う
        # ここでは簡易的にタスクタイプとキーワードでフィルタリング
        task_type = task.get("type")
        similar = [
            ep for ep in self.episodic
            if ep["task"].get("type") == task_type
        ]
        return sorted(
            similar,
            key=lambda x: x["outcome"].get("quality_score", 0),
            reverse=True
        )[:limit]

    def get_best_model_for_type(self, task_type: str) -> Optional[dict]:
        """特定のタスクタイプにおいて最も成功率の高いモデルを返す"""
        similar = [ep for ep in self.episodic if ep["task"].get("type") == task_type]
        if not similar: return None

        model_stats = {}
        for ep in similar:
            m = ep["outcome"]["model"]
            if m not in model_stats:
                model_stats[m] = {"count": 0, "total_quality": 0, "total_cost": 0}
            model_stats[m]["count"] += 1
            model_stats[m]["total_quality"] += ep["outcome"]["quality_score"]
            model_stats[m]["total_cost"] += ep["outcome"]["cost"]

        # 品質/コスト比で最適なモデルを選択
        best = max(model_stats.items(), key=lambda x: (
            (x[1]["total_quality"] / x[1]["count"]) * 0.7 -
            (x[1]["total_cost"] / x[1]["count"]) * 0.3
        ))
        return {
            "model": best[0],
            "success_rate": best[1]["count"] / len(similar),
            "avg_quality": best[1]["total_quality"] / best[1]["count"],
        }

    def _save_episodic(self):
        # 最新1000件のみ保持
        if len(self.episodic) > 1000:
            self.episodic = self.episodic[-1000:]
        with open(self.episodic_path, "w") as f:
            json.dump(self.episodic, f, ensure_ascii=False, indent=2)

    def _generate_id(self, task: dict) -> str:
        content = json.dumps(task, sort_keys=True)
        return hashlib.md5(content.encode()).hexdigest()[:12]

6.3 コンテキスト注入のベストプラクティス

メモリシステムを効果的に活用する上で最も重要なのは
何をコンテキストに注入するかという設計です。

Python

def build_agent_system_prompt(task: dict, memory: MemoryStore,
                               skills: 'SkillsLibrary') -> str:
    """
    タスクに応じたシステムプロンプトを動的に構築する。
    関連記憶・スキル・制約を適切に注入する。
    """
    base_prompt = """あなたはFable 5フレームワーク上で動作する高性能AIエージェントです。
以下の情報を参照して、タスクを高品質に完了してください。"""

    # 関連スキルを注入
    relevant_skills = skills.get_relevant(task, top_k=3)
    skills_section = ""
    if relevant_skills:
        skills_section = "\n## 利用可能なスキル\n"
        for skill in relevant_skills:
            skills_section += f"\n### {skill['name']}\n{skill['description']}\n"
            if skill.get("example"):
                skills_section += f"使用例:\n\n{skill['example']}\n\n"

    # 過去の類似成功例を注入
    similar = memory.find_similar(task, limit=3)
    examples_section = ""
    if similar:
        examples_section = "\n## 過去の成功例（参考）\n"
        for ep in similar[:2]:
            if ep["outcome"]["success"] and ep["outcome"]["quality_score"] > 0.85:
                examples_section += f"- タスク: {ep['task']['description'][:100]}\n"
                examples_section += f"  品質スコア: {ep['outcome']['quality_score']}\n"

    # 作業記憶を注入
    working_context = memory.build_context_injection()

    return f"{base_prompt}{skills_section}{examples_section}{working_context}"

第7章　Skillsライブラリ

7.1 Skillsとは何か

Fable 5におけるSkills（スキル）とは、繰り返し利用可能な実行パターンを抽象化し、保存したものです。人間が「この種の問題はこう解決する」という経験則を持つように、エージェントも成功パターンをスキルとして蓄積します。

スキルは以下の3種類に分類されます。

Type 1: Prompt Skills特定のタスクに最適化されたプロンプトテンプレートです。検証済みの高品質なプロンプトを再利用します。

Type 2: Workflow Skills複数ステップのタスクを解決するための手順書です。「この種のタスクはこの順序で解決する」という知識を体系化します。

Type 3: Code Skills再利用可能な関数やユーティリティです。エージェントが生成し、検証済みのコードスニペットを指します。

7.2 SKILLS.mdの設計

Markdown

【分析対象】
{targets}

【重点分析軸】
{analysis_axes}

【出力形式】
- エグゼクティブサマリー（300字）
- 競合比較表（Markdown）
- 各社の強み・弱みの詳細
- 自社への示唆（3〜5点）

verified_examples:
- input: "TikTok広告市場の競合5社分析"
quality_score: 0.93
model_used: claude-sonnet-5
duration: "4.2min"
cost: "$0.31"

# SKILLS.md — スキルライブラリ

## メタデータ
- **最終更新**: 2026-07-01
- **総スキル数**: 47
- **自動生成**: 31件
- **手動作成**: 16件

---

## skill_001: 競合分析レポート生成

yaml
id: skill_001
name: 競合分析レポート生成
type: workflow
category: marketing
created_by: auto  # Verification Loopで自動生成
created_at: 2026-05-12
success_rate: 0.94
avg_quality_score: 0.91
usage_count: 23
last_used: 2026-06-28

trigger_patterns:
  - "競合.*分析"
  - "competitor.*analysis"
  - "market.*research"

steps:
  1. 対象企業リストの確定（5〜10社）
  2. 各社のSNS・サイト・広告データ収集（並列実行）
  3. KPI比較表の作成（Markdown形式）
  4. SWOT分析の生成
  5. 自社への示唆・アクション提案

system_prompt_template: |
  あなたは競合分析の専門家です。
  以下の情報を基に、実務で使える分析レポートを作成してください。

skill_002: Python コードレビュー

YAML

id: skill_002
name: Python コードレビュー
type: prompt
category: development
created_by: manual
success_rate: 0.97
avg_quality_score: 0.94

trigger_patterns:
- "code.*review"
- "コードレビュー"
- "コードの品質"

system_prompt_template: |
あなたはシニアPythonエンジニアとして、以下のコードをレビューしてください。

【レビュー観点】
1. バグ・ロジックエラーの検出
2. セキュリティ脆弱性（SQLインジェクション・XSS等）
3. パフォーマンス問題
4. コード品質（可読性・保守性）
5. ベストプラクティス準拠

【対象コード】
```python
{code}

【出力形式】

Plain Text

---

## skill_003: SNS投稿コンテンツ生成

yaml
id: skill_003
name: SNS投稿コンテンツ生成
type: prompt
category: sns
created_by: auto
success_rate: 0.88
usage_count: 156

trigger_patterns:
  - "SNS投稿"
  - "コンテンツ生成"
  - "X（Twitter）投稿"

system_prompt_template: |
  あなたはSNSマーケティングの専門家です。
  以下の条件でX（Twitter）投稿を生成してください。

  【テーマ】{theme}
  【ターゲット】{audience}
  【目標】{goal}  # エンゲージメント / リーチ / 転換
  【トーン】{tone}  # 教育的 / 刺激的 / 親近感 / 権威的

  【Xアルゴリズム最適化ルール】
  - 最初の20字で興味を引く
  - 改行を効果的に使う
  - ハッシュタグは2〜3個（多すぎない）
  - 数字・実績を入れると信頼性UP
  - 問いかけでエンゲージメントを促進

  【バリエーション】5パターン生成

7.3 Skills自動生成システム

Python
# fable5/skills_auto_generator.py

class SkillAutoGenerator:
    """
    成功した実行パターンを分析し、
    自動的にスキルとして抽象化・登録する。
    """

    def __init__(self, memory_store, skills_library, llm_client):
        self.memory = memory_store
        self.skills = skills_library
        self.llm = llm_client
        self.min_occurrences = 3       # スキル化に必要な最小成功回数
        self.min_success_rate = 0.80   # スキル化に必要な最小成功率

    async def scan_and_generate(self):
        """
        メモリを定期スキャンし、スキル化候補を特定・登録する。
        推奨: 1日1回の夜間バッチ実行。
        """
        candidates = self._identify_candidates()

        for candidate in candidates:
            # 既存スキルと重複チェック
            if self.skills.has_similar(candidate):
                continue

            # LLMによるスキル抽象化
            skill = await self._abstract_to_skill(candidate)

            if skill and skill["quality_score"] >= 0.75:
                self.skills.register(skill)
                print(f"New skill auto-generated: {skill['name']}")

    def _identify_candidates(self) -> list:
        """成功パターンを集計してスキル候補を特定する"""
        # タスクタイプ別に成功エピソードを集計
        type_groups = {}
        for ep in self.memory.episodic:
            if not ep["outcome"]["success"]:
                continue
            task_type = ep["task"]["type"]
            if task_type not in type_groups:
                type_groups[task_type] = []
            type_groups[task_type].append(ep)

        candidates = []
        for task_type, episodes in type_groups.items():
            if len(episodes) >= self.min_occurrences:
                success_rate = sum(
                    1 for ep in episodes if ep["outcome"]["quality_score"] > 0.8
                ) / len(episodes)

                if success_rate >= self.min_success_rate:
                    candidates.append({
                        "task_type": task_type,
                        "episodes": episodes,
                        "success_rate": success_rate,
                        "avg_quality": sum(
                            ep["outcome"]["quality_score"] for ep in episodes
                        ) / len(episodes),
                    })
        return candidates

    async def _abstract_to_skill(self, candidate: dict) -> Optional[dict]:
        """成功パターンをスキルに抽象化する（LLM使用）"""
        examples = "\n\n".join([
            f"タスク: {ep['task']['description']}\n"
            f"品質: {ep['outcome']['quality_score']}"
            for ep in candidate["episodes"][:5]
        ])

        prompt = f"""
以下の成功した実行パターンを分析し、再利用可能なスキルに抽象化してください。

【成功パターン群】
{examples}

【出力形式（YAML）】
name: スキルの名前
description: 何をするスキルか（1文）
trigger_patterns:
  - このスキルを使うべきキーワード
steps:
  - ステップ1
  - ステップ2
system_prompt_template: |
  再利用可能なプロンプトテンプレート
  {{variable}}形式でパラメータを記述
"""
        response = await self.llm.complete(prompt, model="claude-haiku-5")
        return self._parse_skill_yaml(response)

第8章　長時間動作するAI

8.1 長時間実行の課題

30分、数時間、あるいは数日にわたって動作するAIエージェントは、単純なAPI呼び出しとは根本的に異なる設計が求められます。主な課題は以下の5点です。

8.2 チェックポイント設計

Python

# fable5/state_manager.py

import json
import time
from pathlib import Path
from typing import Optional
from datetime import datetime

class StateManager:
    """
    長時間実行エージェントの状態を管理する。
    チェックポイント・再開・コスト管理を担当。
    """

    def __init__(self, session_id: str, state_file: str = "STATE.md"):
        self.session_id = session_id
        self.state_path = Path(state_file)
        self.start_time = time.time()
        self.state = self._init_state()

    def _init_state(self) -> dict:
        """STATE.mdが存在すれば再開用に読み込む"""
        if self.state_path.exists():
            return self._parse_existing_state()
        return {
            "session_id": self.session_id,
            "status": "initialized",
            "current_phase": None,
            "completed_phases": [],
            "checkpoints": [],
            "cost_tracker": {"tokens": 0, "cost_usd": 0, "limit_usd": 10.0},
            "errors": [],
            "start_time": datetime.now().isoformat(),
        }

    def checkpoint(self, phase: str, data: dict):
        """現在の状態をチェックポイントとして保存する"""
        checkpoint = {
            "phase": phase,
            "timestamp": datetime.now().isoformat(),
            "elapsed_seconds": round(time.time() - self.start_time),
            "data": data,
        }
        self.state["checkpoints"].append(checkpoint)
        self.state["current_phase"] = phase

        # STATE.mdを更新
        self._write_state_md()

    def can_resume_from(self, phase: str) -> Optional[dict]:
        """指定フェーズのチェックポイントデータを返す（再開用）"""
        for cp in reversed(self.state["checkpoints"]):
            if cp["phase"] == phase:
                return cp["data"]
        return None

    def track_cost(self, input_tokens: int, output_tokens: int, model: str):
        """コストを追跡し、上限超過を検出する"""
        cost = self._calc_cost(input_tokens, output_tokens, model)
        self.state["cost_tracker"]["tokens"] += input_tokens + output_tokens
        self.state["cost_tracker"]["cost_usd"] += cost

        if self.state["cost_tracker"]["cost_usd"] >= self.state["cost_tracker"]["limit_usd"]:
            raise CostLimitExceeded(
                f"コスト上限に達しました: "
                f"${self.state['cost_tracker']['cost_usd']:.2f} / "
                f"${self.state['cost_tracker']['limit_usd']:.2f}"
            )

8.3 コンテキストドリフトの監視

長時間実行されるエージェントは、複数のステップを経ていくうちに、当初の目標から少しずつ逸脱していくことがあります。これを「コンテキストドリフト」と呼びます。

Python

# fable5/drift_monitor.py

class DriftMonitor:
    """
    現在の作業が当初の目標と整合しているかを監視する。
    逸脱を検出した場合、再プランニングを要求する。
    """

    def __init__(self, original_goal: str, verifier_model: str = "claude-haiku-5"):
        self.goal = original_goal
        self.verifier = verifier_model

    async def check_drift(self, current_state: dict, last_output: str) -> dict:
        prompt = f"""
当初の目標: {self.goal}
現在のフェーズ: {current_state['current_phase']}
直近の出力: {last_output[:500]}...

以下を評価してください:
1. 目標との一致度 (0-1)
2. ドリフトが検出された場合、その内容
3. 軌道修正が必要か

JSON形式で回答:
{{"alignment_score": 0.0-1.0, "drift_detected": true/false, "drift_description": "...", "correction_needed": true/false}}
"""
        response = await self.llm.complete(prompt, model=self.verifier)
        result = json.loads(response)

        if result["drift_detected"] and result["alignment_score"] < 0.7:
            return {
                "drift_detected": True,
                "severity": "high" if result["alignment_score"] < 0.5 else "medium",
                "correction": self._generate_correction_prompt(result),
            }
        return {"drift_detected": False, "alignment_score": result["alignment_score"]}

第9章　実践構築例

9.1 SNS戦略分析エージェントの完全実装

ここでは、Fable 5の全コンポーネントを統合した実際のシステムを構築します。

ユースケース: 競合他社のSNS戦略を自動的に分析し、翌月のコンテンツ戦略を提案します。

Python

# examples/sns_strategy_agent.py

import asyncio
from fable5 import (
    Fable5Router, VerificationEngine, MemoryStore,
    SkillsLibrary, StateManager, ExecutionLoop
)

async def run_sns_strategy_analysis(
    target_companies: list,
    user_context: dict,
    budget_usd: float = 5.0
) -> dict:
    """
    競合SNS戦略分析エージェントのエントリーポイント。

    Args:
        target_companies: 分析対象企業のリスト
        user_context: ユーザー情報（業界・目標等）
        budget_usd: コスト上限
    """

    # コンポーネント初期化
    memory = MemoryStore(".fable5/memory")
    skills = SkillsLibrary(".fable5/skills.json")
    state = StateManager("sns_analysis_001")
    router = Fable5Router(skills, memory)
    verifier = VerificationEngine()

    # セッション開始
    state.update_status("running")

    # ── Phase 1: タスク分解 ──────────────────────────────
    state.checkpoint("planning", {})
    task_plan = {
        "type": "planning",
        "description": f"{len(target_companies)}社のSNS戦略分析タスクを分解する",
        "budget_usd": budget_usd * 0.1,
    }
    routing = router.route(task_plan)
    # → claude-sonnet-5 にルーティングされる

    plan_result = await execute_with_model(
        model=routing.model,
        system_prompt=build_agent_system_prompt(task_plan, memory, skills),
        user_prompt=f"""
以下の競合分析タスクを実行可能なサブタスクに分解してください。

【分析対象】
{", ".join(target_companies)}

【ユーザーコンテキスト】
業界: {user_context.get('industry')}
目標: {user_context.get('goal')}

【出力】
- 具体的なリサーチ項目（JSON形式）
- 実行順序と依存関係
- 各タスクの推定所要時間
""",
    )
    state.track_cost(plan_result["input_tokens"], plan_result["output_tokens"], routing.model)

    # ── Phase 2: データ収集（並列実行）────────────────────
    state.checkpoint("data_collection", {"plan": plan_result["content"]})

    # 既存スキルの確認
    competitor_skill = skills.get_by_id("skill_001")
    if competitor_skill:
        # スキルを活用して効率化
        collection_prompt = competitor_skill["system_prompt_template"].format(
            targets=target_companies,
            analysis_axes=["投稿頻度", "エンゲージメント率", "コンテンツカテゴリ"]
        )
    else:
        collection_prompt = build_collection_prompt(target_companies)

    # 各企業を並列で分析
    collection_tasks = [
        analyze_company(company, routing, verifier, state)
        for company in target_companies
    ]
    company_results = await asyncio.gather(*collection_tasks, return_exceptions=True)

    # エラーハンドリング
    valid_results = []
    for i, result in enumerate(company_results):
        if isinstance(result, Exception):
            state.record_error(f"{target_companies[i]}: {str(result)}")
            # エラーがあっても続行（skip_and_continue戦略）
        else:
            valid_results.append(result)

    state.checkpoint("data_collected", {"companies": len(valid_results)})

    # ── Phase 3: 統合分析 ────────────────────────────────
    synthesis_task = {
        "type": "analysis",
        "description": "競合データを統合してSNS戦略を分析する",
        "quality": "high",
        "estimated_tokens": 8000,
    }
    synthesis_routing = router.route(synthesis_task)
    # → claude-sonnet-5 にルーティング

    analysis_result = await execute_with_model(
        model=synthesis_routing.model,
        system_prompt=build_agent_system_prompt(synthesis_task, memory, skills),
        user_prompt=build_synthesis_prompt(valid_results, user_context),
    )

    # ── Phase 4: 検証 ────────────────────────────────────
    verification = verifier.verify(analysis_result, synthesis_task)
    if not verification["passed"]:
        # Verification Loopで自動修正
        exec_loop = ExecutionLoop(
            executor=None,  # 既存のexecutorを再利用
            verifier=verifier,
            state_manager=state,
        )
        analysis_result = await exec_loop.retry_with_improvement(
            analysis_result,
            verification["improvement_prompt"],
            synthesis_routing.model
        )

    state.checkpoint("analysis_verified", {
        "quality_score": verification["total_score"]
    })

    # ── Phase 5: 最終レポート生成 ─────────────────────────
    final_task = {
        "type": "writing",
        "description": "クライアント提出用の最終戦略レポートを生成する",
        "quality": "critical",
        "estimated_tokens": 12000,
    }
    final_routing = router.route(final_task)
    # → 品質要件がcriticalのため claude-opus-5 にルーティング

    final_report = await execute_with_model(
        model=final_routing.model,
        system_prompt=build_agent_system_prompt(final_task, memory, skills),
        user_prompt=build_final_report_prompt(analysis_result, user_context),
    )

    # ── 後処理: メモリ更新・スキル候補登録 ───────────────────
    episode_metadata = {
        "success": True,
        "quality_score": verification["total_score"],
        "model": final_routing.model,
        "cost": state.get_total_cost(),
        "duration": state.get_elapsed_seconds(),
    }
    memory.store_episode(synthesis_task, final_report, episode_metadata)

    state.update_status("completed")

    return {
        "report": final_report["content"],
        "metadata": {
            "total_cost": state.get_total_cost(),
            "duration_seconds": state.get_elapsed_seconds(),
            "quality_score": verification["total_score"],
            "companies_analyzed": len(valid_results),
        }
    }
9.2 プロンプト例：Planner Agent
Python
PLANNER_SYSTEM_PROMPT = """
あなたはFable 5のPlannerエージェントです。
複雑なタスクを実行可能なサブタスクに分解することが専門です。

【分解の原則】
1. 各サブタスクは独立して実行可能にする
2. 依存関係を明示する（タスクBはタスクAの完了後に実行）
3. 並列実行可能なタスクを識別する（コスト・時間の最適化）
4. 各タスクのリスクと失敗時の代替案を考える
5. 全体のコスト見積もりを提供する

【出力形式（JSON）】
{
  "task_graph": [
    {
      "id": "task_001",
      "description": "具体的な作業内容",
      "type": "research | code | analysis | writing | verification",
      "depends_on": [],
      "can_parallel_with": ["task_002"],
      "estimated_minutes": 5,
      "estimated_tokens": 2000,
      "risk_level": "low | medium | high",
      "fallback_strategy": "失敗時の対応"
    }
  ],
  "critical_path": ["task_001", "task_003", "task_005"],
  "total_estimated_minutes": 45,
  "total_estimated_cost_usd": 2.50
}
"""

9.3 プロンプト例：Verifier Agent

Python

VERIFIER_SYSTEM_PROMPT = """
あなたはFable 5のVerifierエージェントです。
他のエージェントの出力品質を客観的に評価することが専門です。

【重要な原則】
- 感情的な評価をしない。数値とエビデンスで評価する
- 「良い・悪い」ではなく「基準に対してどうか」で評価する
- 具体的な改善指示を出す。抽象的なフィードバックは不可
- 評価は必ずJSON形式で出力する

【評価基準（タスクタイプ別）】

分析レポートの場合:
- factual_accuracy: 事実の正確性（引用元・数字の確認）
- logical_consistency: 論理的整合性（前提→結論の流れ）
- completeness: 要求された項目の網羅率
- actionability: 読者がアクションを取れる具体性

コードの場合:
- syntax_correctness: 構文エラーなし
- logic_correctness: ロジックの正確性
- security: 脆弱性の有無
- test_coverage: テストの有無と品質

【出力形式（JSON）】
{
  "overall_score": 0.0-1.0,
  "passed": true/false,
  "criteria": {
    "factual_accuracy": {"score": 0.0-1.0, "issues": ["具体的な問題"]},
    "logical_consistency": {"score": 0.0-1.0, "issues": []},
    ...
  },
  "improvement_instructions": "具体的な改善指示（箇条書き）",
  "critical_errors": ["即座に修正が必要な重大エラー"]
}
"""

9.4 Claude Code・Codexとの統合

2026年時点では、Claude CodeとOpenAI Codexが単独のエージェントとして最も高い能力を持つコーディングAIです。Fable 5では、これらをサブエージェントとしてシステムに組み込みます。

Python

# fable5/integrations/claude_code.py

class ClaudeCodeIntegration:
    """
    Claude Code CLIをFable 5のサブエージェントとして統合する。
    長時間コーディングタスクを委任する際に使用。
    """

    async def delegate_coding_task(
        self,
        task_description: str,
        codebase_path: str,
        context: dict
    ) -> dict:
        """
        Claude Codeに複雑なコーディングタスクを委任する。
        Fable 5のVerification Loopで品質を確保する。
        """
        # プロンプトファイルの作成（対話的な確認を避けるため）
        prompt_file = "/tmp/claude_code_task.md"
        with open(prompt_file, "w") as f:
            f.write(f"""# タスク指示

{task_description}

## コンテキスト
{json.dumps(context, ensure_ascii=False, indent=2)}

## 制約
- 質問せず全自動で実装してください
- 実装後にテストを実行してください
- 変更したファイルのリストを最後に出力してください

## 完了条件
- テストが全て通過する
- コードが既存のスタイルに準拠している
- エラーハンドリングが適切に実装されている
""")

        # Claude Code実行
        import subprocess
        result = subprocess.run(
            ["claude", "--dangerously-skip-permissions",
             f"--system-prompt=質問せず全自動で実装する",
             f"< {prompt_file}"],
            cwd=codebase_path,
            capture_output=True,
            text=True,
            timeout=600,
        )

        return {
            "success": result.returncode == 0,
            "output": result.stdout,
            "errors": result.stderr,
        }


# fable5/integrations/codex.py

class CodexIntegration:
    """OpenAI Codex CLIとの統合"""

    async def delegate_to_codex(
        self,
        task: str,
        project_path: str,
        approval_policy: str = "never"
    ) -> dict:
        import subprocess
        result = subprocess.run(
            ["codex", "exec",
             "-c", f'approval_policy="{approval_policy}"',
             task],
            cwd=project_path,
            capture_output=True,
            text=True,
            timeout=600,
        )
        return {
            "success": result.returncode == 0,
            "output": result.stdout,
        }

9.5 Mermaidによるシステム全体図

mermaid

Source

9.6 本番運用のためのベストプラクティス

ベストプラクティス1: コスト上限の必須設定

Python

config = {
    "budget_usd": 5.0,          # タスク単位の上限
    "daily_budget_usd": 50.0,   # 日次上限
    "alert_threshold": 0.8,     # 上限の80%で警告
    "hard_stop": True,          # 上限到達で強制停止
}

ベストプラクティス2: 段階的な品質ゲート

Python

# すべての重要な出力に品質ゲートを設ける
quality_gates = {
    "draft": 0.70,      # 草稿: 70%以上
    "review": 0.85,     # レビュー: 85%以上
    "production": 0.93, # 本番: 93%以上
}

ベストプラクティス3: 失敗パターンの積極的な記録

Python

# エラーも資産として記録する
error_taxonomy = {
    "hallucination": "存在しない情報の生成",
    "format_error": "指定形式への不準拠",
    "context_drift": "目標からの逸脱",
    "tool_failure": "外部ツールの呼び出し失敗",
    "cost_overrun": "コスト上限超過",
}

ベストプラクティス4: 人間の監督ポイントの設計

Python

# 自動化してはいけない判断ポイントを明示する
human_escalation_triggers = [
    "quality_score < 0.60",          # 品質が著しく低い
    "cost > budget * 0.9",           # コスト上限に近い
    "retry_count >= max_retries",    # 最大リトライ超過
    "unknown_error_type",            # 未知のエラー
    "high_stakes_decision",          # 高リスクな判断
    "user_data_modification",        # ユーザーデータ変更
]

おわりに

本書では、Fable 5を活用して自己改善型AIエージェントシステムを構築するための設計思想、アーキテクチャ、そして実装について網羅的に解説しました。

最も重要なのは、「ただ動くAI」を開発するのではなく、「常に賢くなり続けるAI」を設計することです。そのためには、以下の4つの要素が不可欠となります。

2026年のAI開発において、これらの要素を実装していないシステムは、すでに競争力を失いつつあります。本書の実装例を起点として、実際のユースケースに合わせたFable 5システムを構築し、運用を通じて継続的に改善していくことを強く推奨です。

事業者様のお問い合わせは
下記からお願いします
↓
https://lin.ee/qPSWSWR