構造化された経験の蓄積によるエージェント進化

デプロイされた言語モデルエージェントは通常静的です。一度トレーニングされると、実世界のインタラクションから改善されません。FLEXはグラディエントフリーの継続的学習を通じてこの問題を解決します。エージェントは成功、失敗およびそのコンテキストを記録する構造化された経験ライブラリを維持します。その後のインタラクションの際に、エージェントは関連する過去の経験を検索して反省し、リトレーニングなしにこれらの教訓をプロンプティングに組み込みます。

このアプローチは大幅な改善を実証しています。数学推論では23%の改善(AIME25)、化学合成では10%、タンパク質工学では14%——すべてデプロイ中の自己改善から得られたもので、追加のトレーニングではありません。

コアコンセプト

FLEXはデプロイされたエージェントの改善を、パラメータ最適化ではなく構造化された経験管理の問題として扱います。このシステムは3つのコンポーネントを維持します。

1. 経験ライブラリ: 過去のインタラクション(状態、アクション、結果、反省)の構造化レコード
2. 検索メカニズム: 現在の問題に対する関連する先例の探索
3. 自己反省: エージェントが成功/失敗を分析し、プロンプティングコンテキストとして教訓を抽出する

このアプローチが特に強力な理由は、学習中にグラディエント計算、モデルのリトレーニング、外部LLMへのAPIコールが不要——推論中の構造化された反省のみが必要だからです。

アーキテクチャ概要

• 経験キャプチャモジュール: インタラクション(問題、ソリューション試行、結果、文脈因子)を記録
• 構造化ライブラリ: 経験を問題ドメイン、難度、技術タイプによって整理
• セマンティック検索: エンベッディング類似度またはキーワードマッチングを使用して関連する過去の経験を検索
• 反省エンジン: ソリューションが成功/失敗した理由の自然言語要約を生成
• プロンプト拡張: 検索された経験をコンテキスト内の例として組み込む
• パフォーマンス追跡: 時間経過による改善を測定し、学習プラトーを識別

実装ステップ

ステップ1: 経験データ構造

エージェントのインタラクションを記録および検索するための構造化フォーマットを定義します。

from dataclasses import dataclass
from typing import List, Dict, Any
import json
from datetime import datetime

@dataclass
class Experience:
    """Single interaction record in the experience library."""
    problem: str                    # Problem description
    solution_attempt: str           # Agent's attempted solution
    ground_truth: str              # Correct answer (if available)
    is_correct: bool               # Did the solution succeed?
    domain: str                    # Problem domain (math, coding, etc.)
    difficulty: str                # Estimated difficulty
    timestamp: str                 # When this occurred
    techniques_used: List[str]     # Techniques employed (e.g., 'divide-and-conquer')
    failure_reason: str            # Why it failed (if applicable)
    reflection: str                # Agent's own analysis of the attempt
    metadata: Dict[str, Any]       # Additional context (tokens used, latency, etc.)

    def to_dict(self):
        return {
            'problem': self.problem,
            'solution': self.solution_attempt,
            'correct': self.is_correct,
            'domain': self.domain,
            'difficulty': self.difficulty,
            'timestamp': self.timestamp,
            'techniques': self.techniques_used,
            'failure_reason': self.failure_reason,
            'reflection': self.reflection,
            'metadata': self.metadata
        }

class ExperienceLibrary:
    """Maintains structured experience collection."""

    def __init__(self, storage_path='./experience_library.jsonl'):
        self.storage_path = storage_path
        self.experiences: List[Experience] = []
        self.load_from_disk()

    def add_experience(self, exp: Experience):
        """Record a new experience."""
        self.experiences.append(exp)
        # Append to disk for persistence
        with open(self.storage_path, 'a') as f:
            f.write(json.dumps(exp.to_dict()) + '\n')

    def retrieve_relevant(self, problem: str, domain: str, k=3) -> List[Experience]:
        """
        Find most relevant past experiences for a given problem.

        Args:
            problem: Current problem description
            domain: Problem domain
            k: Number of experiences to retrieve

        Returns:
            relevant_experiences: Top-k similar past experiences
        """
        from sklearn.feature_extraction.text import TfidfVectorizer
        from sklearn.metrics.pairwise import cosine_similarity

        # Filter by domain first
        domain_exps = [e for e in self.experiences if e.domain == domain]

        if len(domain_exps) < k:
            return domain_exps

        # Compute similarity between current problem and past problems
        all_problems = [e.problem for e in domain_exps] + [problem]
        vectorizer = TfidfVectorizer(max_features=100)
        tfidf = vectorizer.fit_transform(all_problems)

        # Similarity of current problem to all past problems
        similarities = cosine_similarity(tfidf[-1:], tfidf[:-1])[0]

        # Sort by similarity and return top-k
        top_indices = similarities.argsort()[-k:][::-1]
        return [domain_exps[i] for i in top_indices]

    def load_from_disk(self):
        """Load experiences from persistent storage."""
        try:
            with open(self.storage_path, 'r') as f:
                for line in f:
                    exp_dict = json.loads(line)
                    self.experiences.append(Experience(**exp_dict))
        except FileNotFoundError:
            pass  # First run: empty library

ステップ2: 自己反省エンジン

試行が成功したまたは失敗した理由に関する構造化された反省を生成します。

def generate_reflection(problem: str, solution: str, is_correct: bool,
                       ground_truth: str = None, llm_api=None) -> str:
    """
    Generate agent's reflection on an attempt.

    Args:
        problem: Original problem
        solution: Agent's attempted solution
        is_correct: Whether solution was correct
        ground_truth: Correct solution (if available)
        llm_api: LLM API for generating reflection (e.g., GPT-4, Claude)

    Returns:
        reflection: Natural language analysis
    """
    if is_correct:
        prompt = f"""Analyze why this solution was correct:

Problem: {problem}

Solution: {solution}

Provide a brief reflection on what techniques made this solution work:"""
    else:
        prompt = f"""Analyze why this solution failed:

Problem: {problem}

Your solution: {solution}

Correct solution: {ground_truth}

Identify the key mistake or misconception:"""

    # Call LLM to generate reflection
    if llm_api:
        reflection = llm_api.generate(prompt, max_tokens=200)
    else:
        # Fallback: simple pattern matching
        if "ValueError" in solution or "TypeError" in solution:
            reflection = "Code had syntax or type error"
        elif is_correct:
            reflection = "Solution approach was sound"
        else:
            reflection = "Solution logic was flawed"

    return reflection

ステップ3: 経験拡張プロンプティング

推論中に検索された経験をプロンプトに組み込みます。

def augment_prompt_with_experiences(
        original_prompt: str,
        relevant_experiences: List[Experience],
        include_failures: bool = True) -> str:
    """
    Create augmented prompt including relevant past experiences.

    Args:
        original_prompt: User's problem description
        relevant_experiences: Retrieved past experiences
        include_failures: Whether to include negative examples

    Returns:
        augmented_prompt: Enhanced prompt with examples
    """
    augmented = "You have access to relevant past experiences. Use insights from successes:\n\n"

    successful_exps = [e for e in relevant_experiences if e.is_correct]
    for i, exp in enumerate(successful_exps):
        augmented += f"Example {i+1} - Success:\n"
        augmented += f"Problem: {exp.problem}\n"
        augmented += f"Solution: {exp.solution_attempt}\n"
        augmented += f"Key insight: {exp.reflection}\n\n"

    if include_failures:
        failed_exps = [e for e in relevant_experiences if not e.is_correct]
        if failed_exps:
            augmented += "Learn from past mistakes:\n\n"
            for i, exp in enumerate(failed_exps):
                augmented += f"Past Mistake {i+1}:\n"
                augmented += f"Problem: {exp.problem}\n"
                augmented += f"Failed attempt: {exp.solution_attempt}\n"
                augmented += f"Why it failed: {exp.failure_reason}\n\n"

    augmented += f"Now solve this new problem:\n{original_prompt}"
    return augmented

ステップ4: エージェントデプロイメントループ

デプロイ中の経験キャプチャと検索を統合するメインループです。

class DeployedAgent:
    """LLM agent that learns from deployment experiences."""

    def __init__(self, base_model, experience_library, domain='general'):
        self.model = base_model
        self.library = experience_library
        self.domain = domain
        self.success_count = 0
        self.total_attempts = 0

    def solve_problem(self, problem: str, ground_truth: str = None) -> Dict[str, Any]:
        """
        Solve a problem, recording experience for future learning.

        Args:
            problem: Problem description
            ground_truth: Correct answer (if available for offline validation)

        Returns:
            result: {solution, is_correct, experience_recorded}
        """
        # Step 1: Retrieve relevant past experiences
        relevant_exps = self.library.retrieve_relevant(problem, self.domain, k=3)

        # Step 2: Augment prompt with relevant experiences
        augmented_prompt = augment_prompt_with_experiences(
            problem, relevant_exps, include_failures=True
        )

        # Step 3: Generate solution
        solution = self.model.generate(augmented_prompt, max_tokens=2048)

        # Step 4: Validate (if ground truth available)
        is_correct = False
        if ground_truth:
            is_correct = self._validate_solution(solution, ground_truth)

        self.total_attempts += 1
        if is_correct:
            self.success_count += 1

        # Step 5: Generate reflection
        reflection = generate_reflection(problem, solution, is_correct, ground_truth)

        # Step 6: Record experience
        failure_reason = None
        if not is_correct:
            failure_reason = self._analyze_failure(solution, ground_truth)

        experience = Experience(
            problem=problem,
            solution_attempt=solution,
            ground_truth=ground_truth or '',
            is_correct=is_correct,
            domain=self.domain,
            difficulty=self._estimate_difficulty(problem),
            timestamp=datetime.now().isoformat(),
            techniques_used=self._extract_techniques(solution),
            failure_reason=failure_reason,
            reflection=reflection,
            metadata={
                'model': self.model.name,
                'num_examples': len(relevant_exps),
                'accuracy_rate': self.success_count / self.total_attempts
            }
        )

        self.library.add_experience(experience)

        return {
            'solution': solution,
            'is_correct': is_correct,
            'experience_recorded': True
        }

    def _validate_solution(self, solution: str, ground_truth: str) -> bool:
        """Check if solution matches ground truth."""
        # Simple string matching; extend for domain-specific validation
        return solution.strip() == ground_truth.strip()

    def _analyze_failure(self, solution: str, ground_truth: str) -> str:
        """Identify type of failure."""
        if "ValueError" in solution or "TypeError" in solution:
            return "Syntax/Type Error"
        elif len(solution) == 0:
            return "No Output Generated"
        else:
            return "Incorrect Logic"

    def _estimate_difficulty(self, problem: str) -> str:
        """Estimate problem difficulty."""
        word_count = len(problem.split())
        if word_count < 50:
            return "easy"
        elif word_count < 200:
            return "medium"
        else:
            return "hard"

    def _extract_techniques(self, solution: str) -> List[str]:
        """Identify reasoning techniques used."""
        techniques = []
        if "divide" in solution.lower() or "split" in solution.lower():
            techniques.append("divide-and-conquer")
        if "recursion" in solution.lower():
            techniques.append("recursion")
        if "dynamic" in solution.lower():
            techniques.append("dynamic-programming")
        if "greedy" in solution.lower():
            techniques.append("greedy")
        return techniques

ステップ5: 継続的な監視と適応

時間経過によるエージェントのパフォーマンスの学習トレンドを追跡します。

def monitor_agent_learning(agent: DeployedAgent, window_size: int = 100):
    """
    Monitor improvement trends in agent performance.

    Args:
        agent: Deployed agent instance
        window_size: Number of recent attempts to analyze

    Yields:
        metrics: Performance statistics
    """
    while True:
        recent_exps = agent.library.experiences[-window_size:]

        if len(recent_exps) > 0:
            success_rate = sum(1 for e in recent_exps if e.is_correct) / len(recent_exps)
            avg_reflection_length = sum(
                len(e.reflection.split()) for e in recent_exps
            ) / len(recent_exps)

            metrics = {
                'success_rate': success_rate,
                'sample_count': agent.total_attempts,
                'improvement': success_rate,  # Compare to baseline if available
                'avg_reflection_length': avg_reflection_length,
                'unique_domains': len(set(e.domain for e in recent_exps))
            }

            yield metrics

            # If plateau detected, could trigger additional strategies
            if success_rate > 0.9:
                print("Agent has reached high performance; consider expanding domain")

        import time
        time.sleep(60)  # Monitor every minute

実践的なガイダンス

FLEXを使用すべき場合:

• 多様で変化するタスクを扱うデプロイされたエージェント(継続的学習が必須)
• リトレーニングが高コストまたは時間がかかるシナリオ
• 解釈可能性が重要なシステム(経験反省は人間が読める)

FLEXを使用しないべき場合:

• 静的バッチ学習(デプロイメントまたは評価フィードバックがない)
• 明確な成功/失敗シグナルがないタスク(反省は結果検証が必要)
• プライバシー機密アプリケーション(経験ライブラリが過去のインタラクションを保存)

ハイパーパラメータと設定:

• 検索k: 問題あたり3～5の例(コンテキストサイズと多様性のバランス)
• 反省モデル: 同じLLMまたはコスト削減のための小規模モデルを使用
• ライブラリ保持: 初期段階はすべての経験を保持し、スケールアップ後に冗長なものを削除
• ドメイン粒度: 異なる問題タイプに対して個別のライブラリ(数学、コード、推論)

避けるべき落とし穴:

1. 古い経験 - デプロイ初期の古い経験は陳腐化する可能性があります。最近の成功を優先してください
2. 確認バイアス - 成功した経験に過度の重みをつけます。間違ったパターンの学習を避けるために失敗を含めてください
3. ライブラリの肥大化 - 経験が蓄積します。スケールアップ後に重複排除またはアーカイブ戦略を実装してください
4. 検証がない - グラウンドトゥルースがない場合、不正確に分類された成功がライブラリを汚します。可能な限り検証を追加してください

---

参考文献: https://arxiv.org/abs/2511.06449