วงจรการไตร่ตรองและวิจารณ์ตนเอง

การไตร่ตรองตนเองของเอเจนต์คืออะไร

รูปแบบคำสั่งสำหรับการไตร่ตรอง

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

ผลลัพธ์การไตร่ตรองแบบมีโครงสร้าง

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

วงจรวิจารณ์ตนเอง: ลองใหม่เมื่อคะแนนต่ำ

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

ความจำแบบเป็นตอน ๆ สำหรับการไตร่ตรอง

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

การจัดเก็บการไตร่ตรองอย่างถาวร

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

แบบทดสอบความรู้

ทบทวน: วงจรการไตร่ตรองและการวิจารณ์ตนเอง