自己修正とリフレクティブプロンプティング

内省的プロンプトとは

内省と修正のループ

async def reflect_and_revise(task: str, max_rounds: int = 2) -> str:
    response = await generate_initial(task)
    for round_num in range(max_rounds):
        critique = await critique_response(task, response)
        if critique.is_satisfactory:
            break
        response = await revise_response(task, response, critique.feedback)
    return response

効果的な批評プロンプトの作成

CRITIQUE_PROMPT = '''
You are reviewing an AI-generated response to this task: {task}

Response to evaluate:
{response}

Check each criterion and provide specific feedback:
1. COMPLETENESS: Does it address all parts of the task?
2. ACCURACY: Are all factual claims correct?
3. CONCISENESS: Is there unnecessary padding or repetition?
4. FORMAT: Does it match the requested output format?
5. ACTIONABILITY: Can the user act on this response?

For each issue found, state exactly what to fix.
If the response is satisfactory on all criteria, say APPROVE.
'''

批評応答の解析

from pydantic import BaseModel
from typing import List, Literal

class Issue(BaseModel):
    criterion: str
    description: str
    severity: Literal['critical', 'moderate', 'minor']

class Critique(BaseModel):
    is_satisfactory: bool
    issues: List[Issue]
    overall_verdict: str

# is_satisfactory=True means no revision needed
# is_satisfactory=False means issues must be addressed

修正プロンプト

def build_revision_prompt(task: str, response: str, critique: Critique) -> str:
    issues_text = '\n'.join(
        f'- [{i.severity.upper()}] {i.criterion}: {i.description}'
        for i in critique.issues
    )
    return f'''
Original task: {task}

Your previous response:
{response}

Issues to fix:
{issues_text}

Write an improved response that fixes ONLY the issues listed above.
Do not change parts that were not flagged as problems.
'''

コード生成における自己修正

import subprocess
import sys

async def self_correct_code(task: str, max_rounds: int = 3) -> str:
    code = await generate_code(task)
    for _ in range(max_rounds):
        # Write code to temp file and run mypy
        with open('/tmp/agent_code.py', 'w') as f:
            f.write(code)
        result = subprocess.run(
            [sys.executable, '-m', 'mypy', '/tmp/agent_code.py', '--ignore-missing-imports'],
            capture_output=True, text=True
        )
        if result.returncode == 0:
            break  # No type errors
        code = await fix_code(code, result.stdout + result.stderr)
    return code

過剰修正の回避

from difflib import SequenceMatcher

def safe_revision(original: str, revised: str, max_change_ratio: float = 0.7) -> str:
    similarity = SequenceMatcher(None, original, revised).ratio()
    if similarity < (1 - max_change_ratio):
        print(f'Revision changed too much (similarity: {similarity:.2f}). Keeping original.')
        return original
    return revised

複数ステップのエージェントタスクにおけるリフレクション

async def research_with_reflection(question: str) -> str:
    # Phase 1: gather evidence
    evidence = await gather_evidence(question)

    # Reflection checkpoint
    assessment = await assess_evidence_completeness(question, evidence)
    if not assessment.is_complete:
        for gap in assessment.gaps:
            more_evidence = await targeted_search(gap.search_query)
            evidence.extend(more_evidence)

    # Phase 2: synthesize
    return await synthesize_answer(question, evidence)

リフレクション結果のログ記録

import structlog

log = structlog.get_logger()

def log_reflection_round(task_id: str, round_num: int, critique: Critique, action: str):
    log.info(
        'reflection_round',
        task_id=task_id,
        round=round_num,
        is_satisfactory=critique.is_satisfactory,
        issue_count=len(critique.issues),
        critical_issues=sum(1 for i in critique.issues if i.severity == 'critical'),
        action=action  # 'approved', 'revised', 'max_rounds_reached'
    )

リフレクションを使用するタイミング

# Reflection decision matrix:
# Task type:              Use reflection?
# SQL query generation    YES (run+verify)
# Final report writing    YES (review before delivery)
# Tool argument prep      NO  (tool result verifies it)
# Short factual answer    MAYBE (if accuracy is critical)
# Internal agent thought  NO   (intermediate, not final)
# Code generation         YES  (run linter/tests)

USE_REFLECTION = {'report', 'code', 'email', 'analysis'}

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

async def reflection_ab_test(tasks: list) -> dict:
    import random
    results = {'with_reflection': [], 'without_reflection': []}
    for task in tasks:
        if random.random() < 0.5:
            response = await reflect_and_revise(task, max_rounds=2)
            group = 'with_reflection'
        else:
            response = await generate_initial(task)
            group = 'without_reflection'
        score = await judge(task, response)
        results[group].append(score.overall)
    return {
        'mean_with': sum(results['with_reflection']) / len(results['with_reflection']),
        'mean_without': sum(results['without_reflection']) / len(results['without_reflection'])
    }

理解度チェック

レッスンのまとめ