自我纠错与反思式提示

What Is Reflective Prompting?

The Reflect-and-Revise Loop

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

Writing an Effective Critique Prompt

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.
'''

Parsing the Critique Response

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

The Revision Prompt

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.
'''

Self-Correction for Code Generation

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

Avoiding Overcorrection

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

Reflection in Multi-Step Agent Tasks

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)

Logging Reflection Outcomes

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'
    )

When to Use Reflection

# 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'}

Measuring Reflection Effectiveness

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'])
    }

Quick Check

Lesson Recap