Самоулучшающиеся системы запросов

Цикл самоулучшения

Обзор архитектуры цикла

import anthropic

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

class SelfImprovingPromptSystem:
    def __init__(self, initial_prompt, test_cases, eval_fn, max_iterations=5):
        self.current_prompt = initial_prompt
        self.test_cases = test_cases
        self.eval_fn = eval_fn
        self.max_iterations = max_iterations
        self.history = []  # [(iteration, prompt, score, critique)]

    def run(self):
        for i in range(self.max_iterations):
            print(f'=== Iteration {i+1}/{self.max_iterations} ===')
            score = self._evaluate_prompt()
            print(f'Score: {score:.2f}')
            self.history.append((i, self.current_prompt, score))
            if score >= 0.95:
                print('Target score reached. Stopping.')
                break
            critique = self._critique_prompt(score)
            self.current_prompt = self._rewrite_prompt(critique)
        return self.best_prompt()

    def best_prompt(self):
        return max(self.history, key=lambda x: x[2])[1]

Исполнитель: запуск запроса

def _execute_prompt(self, prompt):
    outputs = []
    for case in self.test_cases:
        response = client.messages.create(
            model='claude-haiku-4-5',  # use cheaper model for execution
            max_tokens=500,
            messages=[
                {'role': 'user', 'content': prompt + '\n\nInput: ' + case['input']}
            ]
        )
        outputs.append({
            'case_id': case['id'],
            'input': case['input'],
            'expected': case['expected'],
            'actual': response.content[0].text
        })
    return outputs

# Bind method to class (demonstration)
SelfImprovingPromptSystem._execute = _execute_prompt

# Sample test cases
test_cases = [
    {'id': 1, 'input': 'The meeting was cancelled.', 'expected': 'negative'},
    {'id': 2, 'input': 'Great product, love it!', 'expected': 'positive'},
    {'id': 3, 'input': 'It arrived on time.', 'expected': 'neutral'},
]
print(f'Test suite: {len(test_cases)} cases')

Оценщик: оценка результатов

def _evaluate_prompt(self):
    outputs = self._execute(self.current_prompt)
    correct = 0
    failed_cases = []
    for out in outputs:
        # Exact match for classification tasks
        if out['expected'].lower() in out['actual'].lower():
            correct += 1
        else:
            failed_cases.append(out)
    score = correct / len(outputs)
    self._last_failed_cases = failed_cases
    return score

# For open-ended tasks: LLM-as-judge evaluator
JUDGE_PROMPT = '''Rate the quality of this AI response (1-5).
Task: {task_description}
Input: {input}
Expected approach: {expected}
Actual response: {actual}

Return only the integer score (1-5). No explanation.'''

def llm_judge_score(task_desc, input_text, expected, actual):
    response = client.messages.create(
        model='claude-haiku-4-5', max_tokens=5,
        messages=[{'role': 'user', 'content':
            JUDGE_PROMPT.format(
                task_description=task_desc, input=input_text,
                expected=expected, actual=actual
            )}]
    )
    try:
        return int(response.content[0].text.strip()) / 5.0
    except ValueError:
        return 0.5

Критик: выявление слабых мест запроса

CRITIC_PROMPT = '''You are a prompt engineering expert analyzing why a prompt fails.

Current prompt:
{current_prompt}

Failed test cases (where the prompt gave wrong outputs):
{failed_cases}

For each failure, explain:
1. What went wrong in the output
2. Which part of the prompt caused or failed to prevent this
3. A specific, actionable fix

End with a prioritized list of the top 3 prompt improvements to make.
Be specific — quote the relevant prompt section and suggest the exact replacement.'''

def _critique_prompt(self, score):
    failed_json = '\n'.join(
        f'Input: {c["input"]}\nExpected: {c["expected"]}\nActual: {c["actual"]}'
        for c in self._last_failed_cases[:5]
    )
    response = client.messages.create(
        model='claude-opus-4-5', max_tokens=1000,
        messages=[{'role': 'user', 'content':
            CRITIC_PROMPT.format(
                current_prompt=self.current_prompt,
                failed_cases=failed_json
            )}]
    )
    return response.content[0].text

Переписыватель: улучшение запроса

REWRITER_PROMPT = '''You are a prompt engineer. Improve the prompt based on the critique below.

Current prompt:
{current_prompt}

Critique and suggested improvements:
{critique}

Rules for rewriting:
1. Make TARGETED changes based on the critique — do not rewrite everything
2. Keep all parts of the prompt that were working well
3. Apply all prioritized fixes from the critique
4. Do not add unnecessary verbosity — conciseness is a quality
5. Output ONLY the improved prompt, no explanation

Improved prompt:'''

def _rewrite_prompt(self, critique):
    response = client.messages.create(
        model='claude-opus-4-5', max_tokens=1000,
        messages=[{'role': 'user', 'content':
            REWRITER_PROMPT.format(
                current_prompt=self.current_prompt,
                critique=critique
            )}]
    )
    new_prompt = response.content[0].text.strip()
    print(f'Prompt updated. Length: {len(new_prompt)} chars '
          f'(was {len(self.current_prompt)} chars)')
    return new_prompt

Полный цикл: объединение компонентов

def run_improvement_loop(initial_prompt, test_cases, max_iterations=5,
                          target_score=0.90):
    history = []
    current_prompt = initial_prompt
    last_failed_cases = []

    for i in range(max_iterations):
        print(f'\n--- Iteration {i+1} ---')

        # Execute
        outputs = execute_prompt(current_prompt, test_cases)

        # Evaluate
        score, failed_cases = evaluate_outputs(outputs)
        last_failed_cases = failed_cases
        print(f'Score: {score:.2f} ({len(failed_cases)} failures)')
        history.append({'iteration': i, 'prompt': current_prompt, 'score': score})

        if score >= target_score:
            print(f'Target score {target_score} reached!')
            break

        if not failed_cases:
            print('No failures to learn from. Stopping.')
            break

        # Critique and rewrite
        critique = critique_prompt(current_prompt, failed_cases)
        current_prompt = rewrite_prompt(current_prompt, critique)

    best = max(history, key=lambda x: x['score'])
    print(f'\nBest prompt at iteration {best["iteration"]+1} with score {best["score"]:.2f}')
    return best['prompt'], history

Управление затратами в циклах самоулучшения

# Cost optimization strategies

# 1. Use cheap model for execution, expensive model for critique/rewrite
def execute_prompt(prompt, test_cases):
    # Use cheapest capable model
    model = 'claude-haiku-4-5'
    # ... execute ...
    pass

def critique_prompt(prompt, failed_cases):
    # Use best model for reasoning about why the prompt fails
    model = 'claude-opus-4-5'
    # ... critique ...
    pass

# 2. Limit test suite size (sample from larger set)
import random

def get_evaluation_sample(full_test_suite, sample_size=20):
    if len(full_test_suite) <= sample_size:
        return full_test_suite
    return random.sample(full_test_suite, sample_size)

# 3. Early stopping: stop if score doesn't improve
def has_converged(history, patience=2, min_delta=0.02):
    if len(history) < patience + 1:
        return False
    recent_scores = [h['score'] for h in history[-patience:]]
    best_recent = max(recent_scores)
    baseline = history[-(patience+1)]['score']
    return (best_recent - baseline) < min_delta

print('Cost optimization: cheap model for execution, expensive for critique')

Отслеживание происхождения запросов

class PromptLineage:
    def __init__(self):
        self.lineage = []

    def record(self, iteration, prompt, score, critique=None,
                failed_case_ids=None):
        self.lineage.append({
            'iteration': iteration,
            'prompt': prompt,
            'score': score,
            'critique_summary': critique[:100] if critique else None,
            'failed_case_ids': failed_case_ids or [],
            'prompt_length': len(prompt.split())
        })

    def print_history(self):
        print('Prompt improvement history:')
        for entry in self.lineage:
            print(
                f'  Iter {entry["iteration"]}: '
                f'score={entry["score"]:.2f} '
                f'words={entry["prompt_length"]} '
                f'failures={len(entry["failed_case_ids"])}'
            )

    def get_best(self):
        return max(self.lineage, key=lambda x: x['score'])

lineage = PromptLineage()
lineage.record(0, 'Initial simple prompt', 0.60)
lineage.record(1, 'Improved with examples', 0.75, 'Missing edge cases')
lineage.record(2, 'Added edge case handling', 0.92, 'Minor format issue')
lineage.print_history()

Визуализация истории версий запроса

def visualize_improvement_history(history):
    '''
    history: list of (iteration, prompt, score)
    Prints an ASCII chart of score progression.
    '''
    if not history:
        print('No history to visualize.')
        return

    max_score = 1.0
    bar_width = 40
    print('\nScore Progression:')
    print('-' * (bar_width + 20))

    for iteration, prompt, score in history:
        filled = int(score * bar_width)
        bar = '#' * filled + '-' * (bar_width - filled)
        marker = ' <-- BEST' if score == max(h[2] for h in history) else ''
        print(f'Iter {iteration:2d}: [{bar}] {score:.3f}{marker}')

    final_score = history[-1][2]
    best_score = max(h[2] for h in history)
    gain = best_score - history[0][2]
    print('-' * (bar_width + 20))
    print(f'Initial: {history[0][2]:.3f} -> Best: {best_score:.3f} (gain: +{gain:.3f})')

# Example
sample_history = [
    (0, 'v0', 0.60),
    (1, 'v1', 0.72),
    (2, 'v2', 0.78),
    (3, 'v3', 0.77),
    (4, 'v4', 0.85)
]
visualize_improvement_history(sample_history)

Выявление регрессий

def detect_regressions(old_outputs, new_outputs):
    old_results = {o['case_id']: o['correct'] for o in old_outputs}
    new_results = {o['case_id']: o['correct'] for o in new_outputs}

    regressions = []
    improvements = []

    for case_id in old_results:
        was_correct = old_results[case_id]
        is_correct = new_results.get(case_id, False)
        if was_correct and not is_correct:
            regressions.append(case_id)
        elif not was_correct and is_correct:
            improvements.append(case_id)

    print(f'Fixed: {len(improvements)} cases. Regressed: {len(regressions)} cases.')
    if regressions:
        print(f'WARNING: Regression on cases: {regressions}')
        print('Consider reverting to previous prompt version.')
    return regressions, improvements

# Example
old = [{'case_id': 1, 'correct': True}, {'case_id': 2, 'correct': False}]
new = [{'case_id': 1, 'correct': False}, {'case_id': 2, 'correct': True}]
detect_regressions(old, new)  # Fixed: 1, Regressed: 1

Быстрая проверка

Итоги: системы самоулучшения запросов