Scratchpad: рабочая память для рассуждений (chain-of-thought)

import anthropic

client = anthropic.Anthropic()


def think_and_answer(question: str, budget_tokens: int = 8000) -> dict:
    """
    Задаёт вопрос с включённым Extended Thinking.
    Возвращает thinking и финальный ответ отдельно.

    budget_tokens: количество токенов для рассуждений (минимум 1024).
    Чем сложнее задача — тем больше бюджет.
    """
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=budget_tokens + 4096,   # max_tokens > budget_tokens — обязательно
        thinking={
            "type": "enabled",
            "budget_tokens": budget_tokens
        },
        messages=[{"role": "user", "content": question}]
    )

    result = {"thinking": None, "answer": None}

    for block in response.content:
        if block.type == "thinking":
            result["thinking"] = block.thinking
        elif block.type == "text":
            result["answer"] = block.text

    return result


# Пример: сложная логическая задача
q = """
Есть 5 домов разного цвета. В каждом живёт человек разной национальности.
Каждый пьёт свой напиток, курит свои сигареты, держит своё животное.
Норвежец живёт в первом доме. Красный дом стоит слева от белого...
[знаменитая загадка Эйнштейна]
Кто держит рыбок?
"""

result = think_and_answer(q, budget_tokens=10000)

# Логируем thinking для отладки
print("=== РАССУЖДЕНИЯ МОДЕЛИ ===")
print(result["thinking"][:500] + "...")   # первые 500 символов

print("\n=== ФИНАЛЬНЫЙ ОТВЕТ ===")
print(result["answer"])

tools = [
    {
        "name": "search_web",
        "description": "Поиск актуальной информации в интернете",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Поисковый запрос"}
            },
            "required": ["query"]
        }
    },
    {
        "name": "calculate",
        "description": "Точные арифметические вычисления",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "Математическое выражение"}
            },
            "required": ["expression"]
        }
    }
]


def agent_with_thinking(user_message: str) -> str:
    """Агент с Extended Thinking и tool calling."""
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=16000,
            thinking={"type": "enabled", "budget_tokens": 8000},
            tools=tools,
            messages=messages
        )

        # Добавляем полный ответ (с thinking-блоками) в историю
        # Это важно: thinking-блоки нужно передавать обратно для корректной кэшируемости
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            # Извлекаем только текстовый ответ для пользователя
            return next(
                b.text for b in response.content if b.type == "text"
            )

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    # Выполняем инструмент
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })

            messages.append({"role": "user", "content": tool_results})

import re
import anthropic

client = anthropic.Anthropic()

SYSTEM_PROMPT = """Ты — аналитический ассистент.

При ответе на любой сложный вопрос ОБЯЗАТЕЛЬНО:
1. Сначала запиши рассуждения внутри тегов ...
2. Потом дай финальный ответ внутри тегов ...

Формат ответа:

[Здесь — пошаговые рассуждения, промежуточные вычисления, анализ вариантов]


[Здесь — чёткий и конкретный финальный ответ для пользователя]

"""


def parse_scratchpad(response_text: str) -> tuple[str, str]:
    """
    Парсит явный скратчпад из ответа модели.
    Возвращает (thinking, answer).
    """
    thinking_match = re.search(
        r'(.*?)',
        response_text,
        re.DOTALL
    )
    answer_match = re.search(
        r'(.*?)',
        response_text,
        re.DOTALL
    )

    thinking = thinking_match.group(1).strip() if thinking_match else ""
    answer = answer_match.group(1).strip() if answer_match else response_text.strip()

    return thinking, answer


def ask_with_scratchpad(question: str, show_thinking: bool = False) -> str:
    """
    Задаёт вопрос с явным скратчпадом.
    show_thinking=True — логирует рассуждения для отладки.
    """
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": question}]
    )

    thinking, answer = parse_scratchpad(response.content[0].text)

    if show_thinking:
        print("── РАССУЖДЕНИЯ ──")
        print(thinking)
        print("── ОТВЕТ ──")

    return answer


# Использование
answer = ask_with_scratchpad(
    "Если я инвестирую 100 000 руб. под 12% годовых, "
    "сколько у меня будет через 5 лет с учётом реинвестирования?",
    show_thinking=True
)

tools_with_thinking = [
    {
        "name": "search_database",
        "description": """Поиск по базе данных клиентов.
        ВАЖНО: перед вызовом заполни поле 'reasoning' — объясни,
        почему именно такой запрос поможет решить задачу.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "reasoning": {
                    "type": "string",
                    "description": "Твоё рассуждение: зачем нужен этот запрос, какую информацию ожидаешь найти"
                },
                "query": {
                    "type": "string",
                    "description": "SQL-запрос или поисковое выражение"
                },
                "filters": {
                    "type": "object",
                    "description": "Опциональные фильтры"
                }
            },
            "required": ["reasoning", "query"]
        }
    }
]

# Теперь каждый вызов инструмента будет содержать reasoning:
# {
#   "reasoning": "Нужно найти всех клиентов с задолженностью > 30 дней,
#                 чтобы определить объём проблемной дебиторки",
#   "query": "SELECT * FROM clients WHERE overdue_days > 30",
#   "filters": {"status": "active"}
# }

# Это можно логировать, отображать в UI отладки, или использовать
# для объяснения действий агента пользователю.

from dataclasses import dataclass, field
from typing import Any
import anthropic

client = anthropic.Anthropic()


@dataclass
class AgentStep:
    """Один шаг ReAct-цикла с сохранённым скратчпадом."""
    step_num: int
    thought: str           # рассуждение агента
    action: str            # название инструмента
    action_input: dict     # аргументы инструмента
    observation: str       # результат выполнения
    thinking: str = ""     # Extended Thinking (если включён)


@dataclass
class AgentTrace:
    """Полная трасса выполнения агента для отладки."""
    task: str
    steps: list[AgentStep] = field(default_factory=list)
    final_answer: str = ""

    def show(self) -> None:
        """Выводит читаемую трассу выполнения."""
        print(f"\n{'='*60}")
        print(f"ЗАДАЧА: {self.task}")
        print('='*60)
        for step in self.steps:
            print(f"\n[Шаг {step.step_num}]")
            if step.thinking:
                print(f"  Extended Thinking: {step.thinking[:200]}...")
            print(f"  Мысль: {step.thought}")
            print(f"  Действие: {step.action}({step.action_input})")
            print(f"  Результат: {step.observation[:200]}")
        print(f"\nФИНАЛЬНЫЙ ОТВЕТ: {self.final_answer}")


REACT_SYSTEM = """Ты — агент, который решает задачи методом Reason-Act-Observe.

На каждом шаге ты ДОЛЖЕН:
1. Написать свои рассуждения в поле 'thought' (scratchpad)
2. Выбрать инструмент и заполнить его аргументы
3. Проанализировать результат инструмента

В поле 'thought' пиши честно: что ты знаешь, чего не знаешь, почему
выбираешь именно этот инструмент. Это поможет тебе принять лучшее решение.

Когда у тебя достаточно информации — вызови инструмент 'finish' с ответом."""

TOOLS = [
    {
        "name": "search",
        "description": "Поиск информации по запросу",
        "input_schema": {
            "type": "object",
            "properties": {
                "thought": {"type": "string", "description": "Твоё рассуждение перед поиском"},
                "query": {"type": "string", "description": "Поисковый запрос"}
            },
            "required": ["thought", "query"]
        }
    },
    {
        "name": "calculate",
        "description": "Вычислить математическое выражение",
        "input_schema": {
            "type": "object",
            "properties": {
                "thought": {"type": "string", "description": "Зачем нужно вычисление"},
                "expression": {"type": "string", "description": "Выражение Python: eval(expression)"}
            },
            "required": ["thought", "expression"]
        }
    },
    {
        "name": "finish",
        "description": "Завершить выполнение с финальным ответом",
        "input_schema": {
            "type": "object",
            "properties": {
                "thought": {"type": "string", "description": "Итоговое рассуждение"},
                "answer": {"type": "string", "description": "Финальный ответ для пользователя"}
            },
            "required": ["thought", "answer"]
        }
    }
]


def execute_tool(name: str, args: dict) -> str:
    """Выполняет инструмент и возвращает строку-результат."""
    if name == "search":
        # Здесь — реальный поиск (SerpAPI, Brave Search и т.д.)
        return f"[Результаты поиска по '{args['query']}': ...]"
    elif name == "calculate":
        try:
            result = eval(args["expression"], {"__builtins__": {}})
            return str(result)
        except Exception as e:
            return f"Ошибка вычисления: {e}"
    return ""


def run_react_agent(task: str, max_steps: int = 10, use_thinking: bool = True) -> AgentTrace:
    """
    Запускает ReAct-агента с явным скратчпадом в каждом шаге.
    use_thinking=True включает Extended Thinking дополнительно к явному CoT.
    """
    trace = AgentTrace(task=task)
    messages = [{"role": "user", "content": task}]

    for step_num in range(1, max_steps + 1):
        kwargs = {
            "model": "claude-sonnet-4-6",
            "max_tokens": 8192,
            "system": REACT_SYSTEM,
            "tools": TOOLS,
            "messages": messages
        }

        if use_thinking:
            kwargs["model"] = "claude-opus-4-6"
            kwargs["thinking"] = {"type": "enabled", "budget_tokens": 4000}
            kwargs["max_tokens"] = 12000

        response = client.messages.create(**kwargs)
        messages.append({"role": "assistant", "content": response.content})

        # Извлекаем Extended Thinking если есть
        extended_thinking = ""
        if use_thinking:
            for block in response.content:
                if block.type == "thinking":
                    extended_thinking = block.thinking

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type != "tool_use":
                    continue

                if block.name == "finish":
                    trace.final_answer = block.input["answer"]
                    trace.steps.append(AgentStep(
                        step_num=step_num,
                        thought=block.input.get("thought", ""),
                        action="finish",
                        action_input=block.input,
                        observation="[завершение]",
                        thinking=extended_thinking
                    ))
                    return trace

                # Выполняем инструмент
                observation = execute_tool(block.name, block.input)

                trace.steps.append(AgentStep(
                    step_num=step_num,
                    thought=block.input.get("thought", ""),
                    action=block.name,
                    action_input=block.input,
                    observation=observation,
                    thinking=extended_thinking
                ))

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": observation
                })

            messages.append({"role": "user", "content": tool_results})

        else:
            # stop_reason == "end_turn" — модель решила отвечать текстом
            text = next((b.text for b in response.content if b.type == "text"), "")
            trace.final_answer = text
            break

    return trace


# Запуск
trace = run_react_agent(
    "Сколько шагов нужно сделать за 30-минутную прогулку, если средний шаг — 75 см?",
    use_thinking=True
)
trace.show()