ハイパーエージェント: 編集可能なメタ手続を通じた再帰的自己改善

ほとんどの自己改善型AIシステムは根本的な制限に直面しています。メタレベルの改善メカニズムが固定されているのです。強化学習エージェントはポリシーを改善できますが、報酬関数は静的です。プログラム合成エージェントはコードを生成できますが、合成手続そのものは変わることがありません。

ハイパーエージェントはシンプルながら深い洞察を通じてこの問題を解決します。改善メカニズム自体を含め、すべてのものを修正の対象にするのです。これにより、システムはタスクパフォーマンスだけでなく、改善を探索する方法も向上させる再帰的構造が生まれます。その結果、開放型能力成長が実現されます。各反復がシステムのさらなる改善能力を向上させるのです。

コアコンセプト

ハイパーエージェントは完全な編集可能性を備えたダーウィン・ゲーデル機械（DGM）を実装します。

タスクエージェント: ターゲット問題を解決し、修正可能です。

メタエージェント: 自身とタスクエージェントを修正でき、同時に修正も可能です。

主要な革新: メタ修正手続自体が編集可能です。ヒトが自己改善アルゴリズムを設計するのではなく、システムが独自の改善戦略を進化させます。

システムは自身のバリアント（異なるメタレベル手続を含む）を生成し、どのバリアントが最も高いパフォーマンスを発揮するかを評価し、改善を保持します。これはシステムが文字通り自身のソースコードを改善する再帰ループを作成します。

アーキテクチャの概要

• 統一されたプログラム表現: タスクとメタエージェントが編集可能なプログラムとして保存される
• 自己バリアント生成: プログラムコードを体系的に変異・再結合する
• 評価フレームワーク: 各バリアントをテストしてパフォーマンス向上を測定する
• 永続メモリ: 成功した修正を実行全体で再利用するために保存する
• メタ認知ループ: メタ改善は反復全体で複合する
• ドメイン非依存: 計算可能なタスク評価を備えたあらゆるドメインで機能する

実装ステップ

ステップ1: エージェントを編集可能なプログラムとして表現する

タスクとメタロジックの両方を修正可能なコードとして保存します。

from dataclasses import dataclass
from typing import Callable, List, Dict, Optional, Any
import hashlib
import copy

@dataclass
class AgentProgram:
    """
    Editable representation of an agent.
    Can be task logic or meta-improvement logic.
    """
    code: str  # Python code defining the agent
    metadata: Dict[str, Any]  # Task params, performance history
    version: int  # Iteration when created
    performance: float = 0.0  # Cached performance score
    creation_time: Optional[str] = None

    def compute_hash(self) -> str:
        """Unique identifier for this program."""
        return hashlib.sha256(self.code.encode()).hexdigest()[:8]

    def execute(self, task_input: Any, env_context: Dict = None) -> Any:
        """Run this agent program on a task."""
        # Create execution namespace
        namespace = {'task_input': task_input}
        if env_context:
            namespace.update(env_context)

        try:
            exec(self.code, namespace)
            return namespace.get('output', None)
        except Exception as e:
            return {'error': str(e), 'output': None}

    def clone(self, mutation: Optional[str] = None) -> 'AgentProgram':
        """Create a variant (optionally with code mutation)."""
        new_code = mutation if mutation else self.code
        new_program = AgentProgram(
            code=new_code,
            metadata=copy.deepcopy(self.metadata),
            version=self.version + 1,
            performance=0.0
        )
        return new_program

class HyperagentSystem:
    """
    System that recursively improves via self-modification.
    """

    def __init__(self, initial_task_agent: AgentProgram,
                 initial_meta_agent: AgentProgram):
        self.task_agent = initial_task_agent
        self.meta_agent = initial_meta_agent
        self.improvement_history = []  # Track successful modifications
        self.persistent_modifications = {}  # Reusable code patterns

    def create_initial_agents(self, task_description: str) - None:
        """Initialize task and meta agents for a domain."""

        # Task agent: solves the target task
        task_code = f"""
# Task Agent for: {task_description}
def task_agent(task_input):
    # Initial strategy: simple heuristic
    output = solve_task(task_input)
    return output

def solve_task(x):
    # Basic implementation (will be improved)
    return x
"""

        # Meta agent: improves the task agent
        meta_code = """
# Meta Agent: Improves task agent
def meta_agent(task_agent_code, evaluation_results):
    # Strategy 1: Add logging for debugging
    # Strategy 2: Try parameter variations
    # Strategy 3: Apply remembered successful modifications

    if 'high_variance' in evaluation_results:
        modified_code = add_parameter_sweep(task_agent_code)
    elif 'low_performance' in evaluation_results:
        modified_code = apply_heuristic_refinement(task_agent_code)
    else:
        modified_code = apply_random_local_search(task_agent_code)

    return modified_code

def add_parameter_sweep(code):
    # Insert parameter tuning
    return code.replace(
        'return x',
        'return x * 1.1'  # Simple modification
    )

def apply_heuristic_refinement(code):
    # Add domain-specific logic
    return code.replace(
        'return x',
        'return x if x > 0 else -x'
    )

def apply_random_local_search(code):
    # Small syntactic variations
    return code
"""

        self.task_agent = AgentProgram(
            code=task_code,
            metadata={'task': task_description},
            version=0
        )

        self.meta_agent = AgentProgram(
            code=meta_code,
            metadata={'role': 'meta_agent'},
            version=0
        )

ステップ2: 自己バリアントを生成する

タスクとメタエージェントの変異を作成します。

import random
import re

class VariantGenerator:
    """Generate program variants through mutation."""

    def __init__(self):
        self.mutation_operators = [
            self.add_parameter,
            self.refactor_logic,
            self.add_early_exit,
            self.add_memoization,
            self.modify_constants
        ]

    def generate_variants(self, program: AgentProgram,
                         num_variants: int = 5) -> List[AgentProgram]:
        """Create multiple variants of a program."""
        variants = []

        for _ in range(num_variants):
            mutation_fn = random.choice(self.mutation_operators)
            mutated_code = mutation_fn(program.code)
            variant = program.clone(mutation=mutated_code)
            variants.append(variant)

        return variants

    def add_parameter(self, code: str) -> str:
        """Introduce tunable parameters."""
        # Find return statements and add a parameter multiplier
        modified = re.sub(
            r'return ([^;\n]+)',
            r'return param * (\1)',
            code
        )
        # Initialize parameter
        modified = 'param = 1.0\n' + modified
        return modified

    def refactor_logic(self, code: str) -> str:
        """Simplify or restructure logic."""
        # Try to identify and refactor loops/conditionals
        if 'while' in code or 'for' in code:
            # Suggest vectorization
            modified = code.replace('for ', '# (refactored) for ')
            return modified
        return code

    def add_early_exit(self, code: str) -> str:
        """Add early termination conditions."""
        modified = code.replace(
            'def solve_task(',
            'def solve_task(\n    # Early exit on impossible inputs\n'
        )
        modified = re.sub(
            r'return ([^;\n]+)',
            r'if check_feasibility(): return \1\nelse: return None',
            modified
        )
        return modified

    def add_memoization(self, code: str) -> str:
        """Cache computation results."""
        modified = 'cache = {}\n' + code
        modified = modified.replace(
            'def solve_task(',
            'def solve_task(\n    if task_input in cache: return cache[task_input]\n'
        )
        modified = modified.replace(
            'return ',
            'cache[task_input] = result; return result\n    result = '
        )
        return modified

    def modify_constants(self, code: str) -> str:
        """Adjust numerical constants."""
        def adjust_number(match):
            num = float(match.group())
            # Random ±10% variation
            adjusted = num * (0.9 + 0.2 * random.random())
            return str(adjusted)

        modified = re.sub(r'\d+\.?\d*', adjust_number, code)
        return modified

ステップ3: バリアントを評価する

バリアントをテストして改善を特定します。

class PerformanceEvaluator:
    """Measure agent performance on tasks."""

    def __init__(self, test_tasks: List[Dict], success_metric: Callable):
        self.test_tasks = test_tasks
        self.success_metric = success_metric

    def evaluate_agent(self, agent: AgentProgram,
                      num_evals: int = 10) -> Dict[str, float]:
        """
        Run agent on test tasks and measure performance.
        Returns: {accuracy, speed, stability, etc.}
        """

        results = {
            'success_rate': 0.0,
            'avg_latency': 0.0,
            'error_count': 0,
            'consistency': 0.0
        }

        successes = 0
        latencies = []
        outputs = []

        for task in self.test_tasks[:num_evals]:
            import time
            start = time.time()

            try:
                output = agent.execute(task)
                latency = time.time() - start

                if self.success_metric(output, task):
                    successes += 1

                latencies.append(latency)
                outputs.append(output)

            except Exception as e:
                results['error_count'] += 1

        # Aggregate metrics
        results['success_rate'] = successes / max(num_evals, 1)
        results['avg_latency'] = sum(latencies) / max(len(latencies), 1)

        # Consistency: do we get same output for same input?
        consistency_score = 1.0 if len(set(str(o) for o in outputs)) == 1 else 0.5
        results['consistency'] = consistency_score

        # Combined score
        results['combined_score'] = (
            0.6 * results['success_rate'] +
            0.2 * (1.0 / max(results['avg_latency'], 0.01)) +
            0.2 * results['consistency']
        )

        return results

    def select_best_variant(self, variants: List[AgentProgram]) -> AgentProgram:
        """Evaluate all variants and return the best."""
        best_variant = None
        best_score = -float('inf')

        for variant in variants:
            results = self.evaluate_agent(variant)
            score = results['combined_score']

            if score > best_score:
                best_score = score
                best_variant = variant

        if best_variant:
            best_variant.performance = best_score

        return best_variant

ステップ4: 再帰的メタ改善

メタエージェントが自身を改善できるようにします。

class RecursiveImprover:
    """
    Enables the meta-agent to improve itself.
    This is the key to open-ended growth.
    """

    def __init__(self, evaluator: PerformanceEvaluator):
        self.evaluator = evaluator
        self.meta_improvement_history = []
        self.variant_generator = VariantGenerator()

    def improve_task_agent(self, task_agent: AgentProgram,
                          meta_agent: AgentProgram) -> AgentProgram:
        """Use meta-agent to improve task-agent."""

        # Get meta-agent's suggested improvement
        improvement_suggestion = meta_agent.execute(
            task_input=task_agent.code,
            env_context={'evaluation_results': {}}
        )

        if improvement_suggestion and 'error' not in improvement_suggestion:
            improved_code = improvement_suggestion
        else:
            # Fallback: use variant generator
            improved_code = self.variant_generator.add_parameter(task_agent.code)

        improved_agent = task_agent.clone(mutation=improved_code)
        return improved_agent

    def improve_meta_agent(self, meta_agent: AgentProgram,
                          improvement_history: List[Dict]) -> AgentProgram:
        """
        CRITICALLY: Improve the meta-agent itself.
        This enables open-ended improvement.
        """

        # Generate variants of the meta-agent
        meta_variants = self.variant_generator.generate_variants(
            meta_agent, num_variants=3
        )

        # Evaluate each meta-variant by using it to improve a task agent
        # This is slower but demonstrates true meta-improvement

        best_meta_variant = None
        best_meta_score = -float('inf')

        for meta_variant in meta_variants:
            # Quick evaluation: does it produce sensible modifications?
            # (Full evaluation would be expensive)

            suggested_modification = meta_variant.execute(
                task_input={'sample': 'task_code'},
                env_context={}
            )

            # Score: preference for non-error modifications
            is_valid = suggested_modification and 'error' not in str(suggested_modification)
            meta_score = 1.0 if is_valid else 0.0

            if meta_score > best_meta_score:
                best_meta_score = meta_score
                best_meta_variant = meta_variant

        if best_meta_variant and best_meta_score > 0:
            return best_meta_variant

        return meta_agent

    def run_improvement_loop(self, task_agent: AgentProgram,
                            meta_agent: AgentProgram,
                            num_iterations: int = 10) -> Dict[str, AgentProgram]:
        """
        Run recursive improvement loop.
        Each iteration: improve task agent, then improve meta-agent.
        """

        current_task = task_agent
        current_meta = meta_agent

        for iteration in range(num_iterations):
            print(f"\n=== Iteration {iteration} ===")

            # Phase 1: Improve task agent
            improved_task = self.improve_task_agent(current_task, current_meta)
            task_perf = self.evaluator.evaluate_agent(improved_task)
            print(f"Task agent score: {task_perf['combined_score']:.3f}")

            # Phase 2: Improve meta-agent (the key step!)
            improved_meta = self.improve_meta_agent(current_meta, self.meta_improvement_history)
            print(f"Meta-agent improved (v{improved_meta.version})")

            # Record improvement
            self.meta_improvement_history.append({
                'iteration': iteration,
                'task_score': task_perf['combined_score'],
                'task_agent_hash': improved_task.compute_hash(),
                'meta_agent_hash': improved_meta.compute_hash()
            })

            current_task = improved_task
            current_meta = improved_meta

        return {
            'final_task_agent': current_task,
            'final_meta_agent': current_meta,
            'improvement_history': self.meta_improvement_history
        }

実践的なガイダンス

ハイパーパラメータ:

• 反復ごとのバリアント数: 3～7個（探索と計算のバランスを取る）
• 変異タイプ: 3～5種類の異なるオペレータを使用（多様性が探索を改善）
• バリアントごとの評価予算: 10～50テストケース（高速な評価でより多くの反復が可能）
• メタ改善の頻度: タスク反復ごとに2～5回

使用すべき場合:

• 継続的な改善が価値のある長時間実行システム
• 改善戦略自体が変動するドメイン
• 開放型学習を探索する研究環境
• 再帰的評価のための計算予算がある場合

使用してはいけない場合:

• リアルタイムシステム（再帰的改善はレイテンシを追加）
• 安全関連ドメイン（制御されない自己修正はリスク）
• シングルショットタスク（改善オーバーヘッドが正当化されない）
• 形式検証が必要なシステム（自己修正は分析が困難）

落とし穴:

• 暴走変異: チェックなしではコードが無意味に発散する可能性があります。構造を検証してください
• 評価ノイズ: わずかなパフォーマンス差が無作為な方向につながる。複数実行を使用してください
• 正のフィードバックループ: 改善戦略が機能すると支配的になる。多様化してください
• 無限のコード成長: プログラムが肥大化する可能性があります。コードサイズを追跡し、必要に応じてペナルティを課してください

参考資料

論文: arxiv.org/abs/2603.19461