Reflection: агент проверяет свою работу и улучшает её

CRITIQUE_TOOL = {
    "name": "critique",
    "description": "Оцени качество текста по заданным критериям.",
    "input_schema": {
        "type": "object",
        "properties": {
            "passed": {
                "type": "boolean",
                "description": "True если качество достаточное (score >= 8)"
            },
            "score": {
                "type": "integer",
                "minimum": 1,
                "maximum": 10,
                "description": "Оценка качества от 1 до 10"
            },
            "issues": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Конкретные проблемы — не 'плохо', а 'нет примеров'"
            },
            "suggestions": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Конкретные предложения по улучшению"
            }
        },
        "required": ["passed", "score", "issues", "suggestions"]
    }
}

import anthropic

client = anthropic.Anthropic()

GENERATOR_SYSTEM = """Ты — экспертный автор. Создавай качественный, точный контент.
Если в контексте есть результаты предыдущей критики — учти их при написании."""


def generate(task: str, critique_context: str = "") -> str:
    """Генерирует черновик или ревизию (с учётом критики)."""
    system = GENERATOR_SYSTEM
    if critique_context:
        system += f"\n\n─── Предыдущая версия и критика ───\n{critique_context}"

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        system=system,
        messages=[{"role": "user", "content": task}]
    )
    return response.content[0].text


def critique(task: str, draft: str) -> dict:
    """Возвращает структурированную оценку черновика."""
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        system=(
            "Ты — строгий технический редактор. Оценивай объективно.\n"
            "passed=True только если score >= 8 и нет серьёзных проблем.\n"
            "Указывай конкретные проблемы, не абстракции."
        ),
        tools=[CRITIQUE_TOOL],
        tool_choice={"type": "tool", "name": "critique"},
        messages=[{
            "role": "user",
            "content": f"Задача: {task}\n\nТекст для оценки:\n{draft}"
        }]
    )
    for block in response.content:
        if block.type == "tool_use" and block.name == "critique":
            return block.input
    # fallback если что-то пошло не так
    return {"passed": True, "score": 8, "issues": [], "suggestions": []}

def critique_and_revise(task: str, max_rounds: int = 3) -> str:
    """
    Итеративно генерирует, критикует и улучшает текст.
    Возвращает лучший черновик: либо прошедший критику, либо после max_rounds.
    """
    # Раунд 1: первоначальная генерация
    draft = generate(task)
    print(f"[Раунд 1] Черновик: {len(draft)} символов")

    for round_num in range(1, max_rounds + 1):
        result = critique(task, draft)
        score   = result.get("score", 0)
        passed  = result.get("passed", False)
        issues  = result.get("issues", [])
        suggestions = result.get("suggestions", [])

        print(f"[Раунд {round_num}] Критика: score={score}/10, "
              f"issues={len(issues)}, passed={passed}")

        if passed:
            print(f"  ✓ Принято на раунде {round_num}")
            return draft

        if round_num == max_rounds:
            print(f"  ⚠ Достигнут лимит {max_rounds} раундов")
            return draft

        # Формируем контекст для ревизора
        critique_text = (
            f"Черновик (раунд {round_num}):\n{draft}\n\n"
            f"Оценка: {score}/10\n"
            f"Проблемы:\n"
            + "\n".join(f"  - {i}" for i in issues)
            + f"\nПредложения:\n"
            + "\n".join(f"  - {s}" for s in suggestions)
        )
        draft = generate(task, critique_context=critique_text)
        print(f"[Раунд {round_num + 1}] Ревизия: {len(draft)} символов")

    return draft


# Использование
result = critique_and_revise(
    task="Напиши описание продукта для беспроводных наушников с ANC",
    max_rounds=3
)

from dataclasses import dataclass, field

@dataclass
class ReflexionAgent:
    """
    Накапливает вербальные рефлексии между попытками.
    Каждая неудача пополняет память, которая инжектируется в следующий запуск.
    """
    base_system: str
    reflections: list[str] = field(default_factory=list)
    max_stored_reflections: int = 5  # не накапливать слишком много

    def _system_with_reflections(self) -> str:
        if not self.reflections:
            return self.base_system
        recent = self.reflections[-self.max_stored_reflections:]
        notes = "\n".join(f"{i+1}. {r}" for i, r in enumerate(recent))
        return (
            f"{self.base_system}\n\n"
            f"─── Уроки из прошлых попыток ───\n"
            f"Я уже пробовал решить подобную задачу. Вот что узнал:\n{notes}"
        )

    def attempt(self, task: str) -> str:
        """Одна попытка решить задачу (здесь — простая генерация, можно заменить ReAct)."""
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=2048,
            system=self._system_with_reflections(),
            messages=[{"role": "user", "content": task}]
        )
        return response.content[0].text

    def generate_reflection(self, task: str, failed_result: str) -> str:
        """Генерирует краткую рефлексию о причинах неудачи."""
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=200,
            system=(
                "Ты анализируешь неудачные попытки. "
                "Напиши 1–2 конкретных предложения: что пошло не так "
                "и как нужно действовать иначе. Без воды."
            ),
            messages=[{
                "role": "user",
                "content": f"Задача: {task}\n\nНеудачный результат:\n{failed_result}"
            }]
        )
        reflection = response.content[0].text.strip()
        self.reflections.append(reflection)
        return reflection

    def run(self, task: str, is_correct, max_attempts: int = 3) -> str:
        """
        Запускает попытки с рефлексией между ними.

        is_correct: callable(result: str) -> bool
        Для объективных задач (тесты, запросы) — проверяй автоматически.
        Для субъективных — используй LLM-критика из critique_and_revise.
        """
        result = ""
        for n in range(1, max_attempts + 1):
            print(f"\n=== Попытка {n}/{max_attempts} ===")
            result = self.attempt(task)

            if is_correct(result):
                print(f"✓ Успех на попытке {n}")
                return result

            if n < max_attempts:
                reflection = self.generate_reflection(task, result)
                print(f"✗ Неудача. Рефлексия: {reflection[:80]}...")

        print("⚠ Лимит попыток исчерпан")
        return result


# Пример: агент пишет Python-функцию и проверяет тестами
agent = ReflexionAgent(
    base_system="Ты — опытный Python-разработчик. Пиши корректный, рабочий код."
)

def passes_tests(code: str) -> bool:
    """Запускает тесты и возвращает True если все прошли."""
    try:
        exec(code)          # упрощённо; в реальности используй subprocess
        return True
    except Exception:
        return False

solution = agent.run(
    task="Напиши функцию merge_sorted_lists(a, b) -> list, которая сливает два отсортированных списка",
    is_correct=passes_tests,
    max_attempts=3
)

def plan_execute_with_reflection(task: str, max_rounds: int = 2) -> str:
    """
    Plan-and-Execute с финальной проверкой через Reflection.
    Перезапускает агента только если критик не принял результат.
    """
    # Первый запуск Plan-and-Execute
    draft = plan_and_execute(task)

    for _ in range(max_rounds):
        result = critique(task, draft)
        if result["passed"]:
            return draft

        # Добавляем критику в задачу и перезапускаем
        issues_str = "\n".join(f"- {i}" for i in result["issues"])
        enriched_task = (
            f"{task}\n\n"
            f"ВАЖНО — предыдущий ответ не прошёл проверку:\n"
            f"{issues_str}\n"
            f"Учти эти замечания при составлении нового плана."
        )
        draft = plan_and_execute(enriched_task)

    return draft