การแก้ไขตนเองและการเขียนพรอมต์แบบไตร่ตรอง

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