0Pricing
AI Agents · レッスン

リフレクションと自己批評のループ

エージェントが自身の出力を評価し、改善案を生成する仕組みを構築します。

「リフレクションと自己批評のループ」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。

エージェントの自己リフレクションとは

自己リフレクションとは、エージェントに、生成したばかりの出力をユーザーに返す前、または返した直後に評価させる実践です。エージェント自身が批評役を務めます。

これは、熟練した人間が自分の作業を見直す方法、つまり下書き → 批評 → 改訂という流れに似ています。このループをエージェントに追加すると、基盤モデルを変更しなくても出力品質が向上することがよくあります。

リフレクションプロンプトのパターン

エージェントが応答を生成した後、その応答を構造化されたリフレクションプロンプトとともにモデルへ再入力します。モデルは弱点を特定し、改善案を提案します。

import anthropic

client = anthropic.Anthropic(api_key='YOUR_API_KEY')

def reflect_on_response(task: str, response: str) -> str:
    reflection_prompt = (
        'You just completed the following task:\n\n'
        f'TASK: {task}\n\n'
        f'YOUR RESPONSE:\n{response}\n\n'
        'Please reflect on your performance by answering:\n'
        '1. What did you do well?\n'
        '2. What could be improved?\n'
        '3. What would you do differently if you had to redo this?\n'
        'Be specific and honest.'
    )
    result = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=512,
        messages=[{'role': 'user', 'content': reflection_prompt}]
    )
    return result.content[0].text

構造化されたリフレクション出力

構造化されていないリフレクションの文章は、プログラムで処理するのが困難です。スコアやアクションアイテムを確実に抽出できるよう、構造化されたJSON形式のリフレクションをモデルに生成させてください。

STRUCTURED_REFLECTION_PROMPT = '''
Reflect on the task and response above. Return ONLY valid JSON:
{
  "strengths": ["..."],
  "weaknesses": ["..."],
  "alternative_approach": "...",
  "quality_score": 0.0,
  "retry_recommended": false
}
quality_score: 0.0 (terrible) to 1.0 (excellent).
retry_recommended: true if quality_score < 0.6.
'''

import json

def structured_reflect(task: str, response: str, client) -> dict:
    prompt = f'TASK: {task}\n\nRESPONSE: {response}\n\n{STRUCTURED_REFLECTION_PROMPT}'
    result = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=512,
        messages=[{'role': 'user', 'content': prompt}]
    )
    text = result.content[0].text.strip()
    # strip markdown code fences if present
    if text.startswith('```'):
        text = text.split('```')[1].lstrip('json').strip()
    return json.loads(text)

if __name__ == '__main__':
    class FakeContent:
        def __init__(self, text):
            self.text = text

    class FakeResponse:
        def __init__(self, text):
            self.content = [FakeContent(text)]

    class FakeMessages:
        def create(self, **kwargs):
            return FakeResponse(
                '{"strengths": ["clear"], "weaknesses": ["too long"], '
                '"alternative_approach": "be more concise", '
                '"quality_score": 0.7, "retry_recommended": false}'
            )

    class FakeClient:
        def __init__(self):
            self.messages = FakeMessages()

    result = structured_reflect('Summarize the article', 'A very long response...', FakeClient())
    print('quality_score:', result['quality_score'])
    print('weaknesses:', result['weaknesses'])

自己批評ループ：低スコア時の再試行

リフレクションのスコアがしきい値を下回った場合は、リフレクションで示された弱点と別のアプローチを追加のコンテキストとして使用し、タスクを自動的に再試行します。これにより、1回のエージェント実行の中に、フィードバックによる改善ループが作られます。

def agent_with_self_critique(task: str, client, max_retries: int = 2) -> str:
    response = run_agent(task, client)

    for attempt in range(max_retries):
        reflection = structured_reflect(task, response, client)
        print(f'Attempt {attempt+1} quality: {reflection["quality_score"]:.2f}')

        if not reflection['retry_recommended']:
            break

        # Enrich the task with reflection insights
        improved_task = (
            f'{task}\n\n'
            'Previous attempt weaknesses:\n'
            + '\n'.join(f'- {w}' for w in reflection['weaknesses'])
            + f'\n\nSuggested approach: {reflection["alternative_approach"]}'
        )
        response = run_agent(improved_task, client)

    return response

def run_agent(task: str, client) -> str:
    result = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=1024,
        messages=[{'role': 'user', 'content': task}]
    )
    return result.content[0].text

リフレクションのためのエピソード記憶

1回のリフレクションは一度しか役立ちませんが、保存されたリフレクションは、セッションをまたいでエージェントの学習を助けるエピソード記憶になります。各リフレクションは、タスクのコンテキスト、起きたこと、エージェントが学んだことからなる1つのエピソードです。

from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class ReflectionEpisode:
    episode_id: str
    task_type: str          # e.g. 'summarize', 'code_review', 'translate'
    task_summary: str       # short description (not full text)
    quality_score: float
    strengths: list
    weaknesses: list
    alternative_approach: str
    timestamp: str = ''

    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = datetime.utcnow().isoformat()

    def to_dict(self) -> dict:
        return asdict(self)

# Example
episode = ReflectionEpisode(
    episode_id='ep_001',
    task_type='summarize',
    task_summary='Summarize a 5-page financial report',
    quality_score=0.55,
    strengths=['Identified key figures'],
    weaknesses=['Too verbose', 'Missed conclusion'],
    alternative_approach='Lead with the executive summary first'
)
print(episode.to_dict())

リフレクションの永続保存

リフレクションのエピソードをJSONファイルまたはデータベースに永続化します。起動時に、同じタスク種別の最近のエピソードを読み込み、コンテキストとして注入します。エージェントは過去の自身のパフォーマンスから学習できるようになります。

import json
import os

MEMORY_FILE = 'agent_episodic_memory.json'

def save_episode(episode: ReflectionEpisode):
    episodes = load_all_episodes()
    episodes.append(episode.to_dict())
    with open(MEMORY_FILE, 'w') as f:
        json.dump(episodes, f, indent=2)

def load_all_episodes() -> list:
    if not os.path.exists(MEMORY_FILE):
        return []
    with open(MEMORY_FILE) as f:
        return json.load(f)

def load_recent_episodes(task_type: str, n: int = 3) -> list:
    all_ep = load_all_episodes()
    matching = [e for e in all_ep if e['task_type'] == task_type]
    # Sort by timestamp descending, take most recent n
    matching.sort(key=lambda e: e['timestamp'], reverse=True)
    return matching[:n]

過去のリフレクションのコンテキストへの注入

タスクを開始する前に、そのタスク種別に関する最新のエピソードリフレクションを取得し、システムプロンプトに含めます。これでエージェントは前回のミスを把握し、事前に回避できるようになります。

def build_system_prompt_with_memory(task_type: str) -> str:
    base = 'You are a helpful AI assistant. Complete the task carefully.'
    episodes = load_recent_episodes(task_type, n=3)

    if not episodes:
        return base

    memory_block = '\n\nYour recent performance on similar tasks:\n'
    for ep in episodes:
        memory_block += (
            f'- Score {ep["quality_score"]:.2f}: '
            f'Weaknesses: {ep["weaknesses"]}. '
            f'Better approach: {ep["alternative_approach"]}\n'
        )
    memory_block += '\nApply these lessons to your current response.'
    return base + memory_block

# Before each task:
system_prompt = build_system_prompt_with_memory('summarize')
print(system_prompt[:300])

ツール使用に関するリフレクション

リフレクションは、ツールを使用するエージェントにとってさらに価値があります。エージェントはツール呼び出しの戦略について、適切なツールを適切な順序で、適切なパラメーターとともに使用したかを振り返れるためです。

TOOL_REFLECTION_PROMPT = '''
You completed a multi-step task using tools. Reflect on your tool usage:

Tool call log:
{tool_log}

Final result: {result}

Answer:
1. Were all tool calls necessary?
2. Were there redundant or incorrect calls?
3. What is the optimal tool sequence for this task type?

Return JSON:
{{
  "redundant_calls": [],
  "incorrect_calls": [],
  "optimal_sequence": [],
  "efficiency_score": 0.0
}}
'''

def reflect_on_tool_use(tool_log: list, result: str, client) -> dict:
    import json
    log_str = json.dumps(tool_log, indent=2)
    prompt = TOOL_REFLECTION_PROMPT.format(
        tool_log=log_str, result=result
    )
    response = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=512,
        messages=[{'role': 'user', 'content': prompt}]
    )
    return json.loads(response.content[0].text)

if __name__ == '__main__':
    class FakeContent:
        def __init__(self, text):
            self.text = text

    class FakeResponse:
        def __init__(self, text):
            self.content = [FakeContent(text)]

    class FakeMessages:
        def create(self, **kwargs):
            return FakeResponse(
                '{"redundant_calls": ["search(x)"], "incorrect_calls": [], '
                '"optimal_sequence": ["search", "summarize"], "efficiency_score": 0.8}'
            )

    class FakeClient:
        def __init__(self):
            self.messages = FakeMessages()

    tool_log = [{'tool': 'search', 'args': {'q': 'x'}}, {'tool': 'search', 'args': {'q': 'x'}}]
    reflection = reflect_on_tool_use(tool_log, 'Found the answer', FakeClient())
    print('Efficiency score:', reflection['efficiency_score'])
    print('Redundant calls:', reflection['redundant_calls'])

エピソード記憶の減衰とプルーニング

世界の状況が変化したりモデルが更新されたりすると、古いリフレクションは陳腐化します。減衰を実装し、最近のエピソードに大きな重みを付けてください。また、しきい値より古いエピソードや、品質スコアが非常に低いエピソード（外れ値である可能性があります）をプルーニングします。

from datetime import datetime, timedelta

def prune_old_episodes(
    episodes: list,
    max_age_days: int = 30,
    min_quality: float = 0.0
) -> list:
    cutoff = datetime.utcnow() - timedelta(days=max_age_days)
    kept = []
    for ep in episodes:
        ep_time = datetime.fromisoformat(ep['timestamp'])
        if ep_time >= cutoff and ep['quality_score'] >= min_quality:
            kept.append(ep)
    return kept

def weighted_episodes(episodes: list) -> list:
    now = datetime.utcnow()
    for ep in episodes:
        age_days = (now - datetime.fromisoformat(ep['timestamp'])).days
        # Recency weight: 1.0 today, halves every 7 days
        ep['weight'] = 0.5 ** (age_days / 7)
    return sorted(episodes, key=lambda e: e['weight'], reverse=True)

if __name__ == '__main__':
    now = datetime.utcnow()
    episodes = [
        {'timestamp': (now - timedelta(days=2)).isoformat(), 'quality_score': 0.9, 'content': 'recent good episode'},
        {'timestamp': (now - timedelta(days=45)).isoformat(), 'quality_score': 0.8, 'content': 'old episode'},
        {'timestamp': (now - timedelta(days=10)).isoformat(), 'quality_score': 0.3, 'content': 'low quality episode'},
    ]
    kept = prune_old_episodes(episodes, max_age_days=30, min_quality=0.5)
    print(f'Kept {len(kept)} of {len(episodes)} episodes after pruning')
    for ep in weighted_episodes(kept):
        print(f"  weight={ep['weight']:.3f} content={ep['content']}")

リフレクションの有効性の測定

自己批評によって実際に結果が改善したかどうかを、最初の試行と最終的な（リフレクション後の）試行の品質スコアを比較して追跡します。改善が小さい、または悪化している場合は、リフレクションプロンプトの調整が必要な可能性があります。

def measure_reflection_gain(run_log: list) -> dict:
    """
    run_log: list of dicts with keys 'attempt', 'quality_score'
    e.g. [{'attempt': 1, 'quality_score': 0.55}, {'attempt': 2, 'quality_score': 0.78}]
    """
    if not run_log:
        return {}

    first_score = run_log[0]['quality_score']
    best_score = max(r['quality_score'] for r in run_log)
    final_score = run_log[-1]['quality_score']

    return {
        'first_attempt_score': first_score,
        'final_score': final_score,
        'best_score': best_score,
        'absolute_gain': final_score - first_score,
        'relative_gain_pct': ((final_score - first_score) / max(first_score, 0.001)) * 100,
        'retries': len(run_log) - 1
    }

log = [
    {'attempt': 1, 'quality_score': 0.55},
    {'attempt': 2, 'quality_score': 0.78}
]
print(measure_reflection_gain(log))

リフレクションループの安全策

制限がないと、リフレクションループが無限に実行される可能性があります。必ず、最大再試行回数、早期終了のための最低スコアしきい値、時間予算を設定してください。ループの動作を監査できるよう、すべてのリフレクションをログに記録します。

import time

def safe_reflection_loop(
    task: str,
    client,
    max_retries: int = 3,
    quality_target: float = 0.75,
    time_budget_seconds: float = 30.0
) -> dict:
    start = time.time()
    response = run_agent(task, client)
    run_log = []

    for attempt in range(max_retries + 1):
        if time.time() - start > time_budget_seconds:
            print('Time budget exceeded, returning best result')
            break
        reflection = structured_reflect(task, response, client)
        run_log.append({'attempt': attempt + 1,
                        'quality_score': reflection['quality_score']})

        if reflection['quality_score'] >= quality_target:
            print(f'Quality target reached at attempt {attempt + 1}')
            break
        if attempt < max_retries:
            response = run_agent(task + '\n' + reflection['alternative_approach'], client)

    return {'response': response, 'run_log': run_log,
            'gain': measure_reflection_gain(run_log)}

理解度チェック

リフレクションのエピソードをエピソード記憶として保存する主なメリットは何ですか？

総括：リフレクションと自己批評ループ

すばらしいです。このレッスンでは次のことを学びました。

  • リフレクションプロンプト：強み、弱点、品質スコア、再試行フラグを含む構造化JSON
  • 自己批評ループ：低スコア時に再試行し、リフレクションの知見でタスクを強化
  • エピソード記憶：タスク種別ごとにタイムスタンプ付きのエピソードとしてリフレクションを保存
  • メモリ注入：各実行前に最近のエピソードをシステムプロンプトへ読み込み
  • 安全策：最大再試行回数、時間予算、品質目標達成時の早期終了

次は、より深い自己改善のために、成功した軌跡と失敗した軌跡を活用する方法です。

よくある質問

「リフレクションと自己批評のループ」レッスンは無料ですか？

はい。「リフレクションと自己批評のループ」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。

「リフレクションと自己批評のループ」で何を学びますか？

エージェントが自身の出力を評価し、改善案を生成する仕組みを構築します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Agentsを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。

「リフレクションと自己批評のループ」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Agentsレッスンでコードを書いて実行できますか？

はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. フィードバックの収集と保存
  2. リフレクションと自己批評のループ
  3. 軌跡ベースの自己改善
  4. 自己改善がうまくいかないとき
← AI Agentsに戻る