Chain-of-Thought Prompting

import anthropic

client = anthropic.AsyncAnthropic()

# Самая известная фраза — из оригинальной статьи Kojima et al. 2022
ZERO_SHOT_COT_TRIGGERS = [
    "Думай пошагово.",                          # самый простой
    "Let's think step by step.",                # оригинал (EN)
    "Реши задачу шаг за шагом, затем дай ответ.",
    "Сначала разбери задачу на шаги, потом ответь.",
    "Объясни своё рассуждение, прежде чем дать финальный ответ.",
]

async def solve_with_cot(problem: str, extract_answer: bool = True) -> dict:
    """
    Решаем задачу с zero-shot CoT.
    Если extract_answer=True — делаем второй запрос для извлечения ответа.
    """
    # Шаг 1: рассуждение
    reasoning_response = await client.messages.create(
        model="claude-opus-4-6",
        system="Ты — точный и аккуратный решатель задач.",
        messages=[{
            "role": "user",
            "content": f"{problem}\n\nДумай пошагово.",
        }],
        max_tokens=1024,
        temperature=0,
    )
    reasoning = reasoning_response.content[0].text

    if not extract_answer:
        return {"reasoning": reasoning, "answer": None}

    # Шаг 2: извлечение финального ответа
    extract_response = await client.messages.create(
        model="claude-opus-4-6",
        messages=[
            {"role": "user",      "content": f"{problem}\n\nДумай пошагово."},
            {"role": "assistant", "content": reasoning},
            {"role": "user",      "content": "Исходя из рассуждения выше, дай только финальный ответ — одно число или краткую фразу."},
        ],
        max_tokens=32,
        temperature=0,
    )
    answer = extract_response.content[0].text.strip()

    return {"reasoning": reasoning, "answer": answer}

# Пример
result = await solve_with_cot(
    "В магазине 23 яблока. Купили ещё столько же минус 10. "
    "Потом съели треть всех яблок. Сколько осталось?"
)
print("Рассуждение:\n", result["reasoning"])
print("\nОтвет:", result["answer"])

import anthropic

client = anthropic.AsyncAnthropic()

# Примеры с цепочками рассуждений
MATH_COT_EXAMPLES = [
    (
        "У Пети 5 конфет. Маша дала ему ещё 3. Потом он съел 2. Сколько осталось?",
        "Шаг 1: Начало — 5 конфет.\n"
        "Шаг 2: Маша дала 3 → 5 + 3 = 8.\n"
        "Шаг 3: Съел 2 → 8 − 2 = 6.\n"
        "Ответ: 6",
    ),
    (
        "В корзине 12 яблок и 8 апельсинов. Убрали половину яблок. Сколько фруктов стало?",
        "Шаг 1: Яблок 12, апельсинов 8, итого 20.\n"
        "Шаг 2: Половина яблок: 12 / 2 = 6.\n"
        "Шаг 3: Яблок после: 12 − 6 = 6.\n"
        "Шаг 4: Всего фруктов: 6 + 8 = 14.\n"
        "Ответ: 14",
    ),
]

def build_cot_messages(
    examples: list[tuple[str, str]],
    question: str,
) -> list[dict]:
    messages = []
    for q, reasoning in examples:
        messages.append({"role": "user",      "content": q})
        messages.append({"role": "assistant", "content": reasoning})
    messages.append({"role": "user", "content": question})
    return messages

async def few_shot_cot(question: str) -> str:
    messages = build_cot_messages(MATH_COT_EXAMPLES, question)
    response = await client.messages.create(
        model="claude-opus-4-6",
        system="Решай задачи строго пошагово, следуя формату из примеров.",
        messages=messages,
        max_tokens=512,
        temperature=0,
    )
    return response.content[0].text

result = await few_shot_cot(
    "В магазине 23 яблока. Купили ещё столько же минус 10. "
    "Потом съели треть всех яблок. Сколько осталось?"
)
print(result)
# Шаг 1: Купить ещё: 23 − 10 = 13.
# Шаг 2: Итого: 23 + 13 = 36.
# Шаг 3: Съесть треть: 36 / 3 = 12.
# Шаг 4: Остаток: 36 − 12 = 24.
# Ответ: 24

import anthropic
import re

client = anthropic.AsyncAnthropic()

AGENT_COT_SYSTEM = """
Ты — агент для анализа данных. При получении задачи следуй строгому формату:


  Что я вижу в задаче? Какие данные есть?
  Что означают эти данные? Какие закономерности?
  Какие шаги нужно выполнить?
  Выполняю план шаг за шагом...



Финальный ответ здесь.


Всегда используй этот формат. Не пропускай блоки.
"""

def parse_structured_cot(response_text: str) -> dict:
    """Разбираем структурированный CoT-ответ."""
    result = {}

    # Извлекаем блоки thinking и answer
    thinking_match = re.search(r'(.*?)', response_text, re.DOTALL)
    answer_match   = re.search(r'(.*?)',   response_text, re.DOTALL)

    if thinking_match:
        thinking = thinking_match.group(1)
        for tag in ['observe', 'analyze', 'plan', 'execute']:
            m = re.search(rf'<{tag}>(.*?)', thinking, re.DOTALL)
            result[tag] = m.group(1).strip() if m else ""

    result['answer'] = answer_match.group(1).strip() if answer_match else response_text

    return result

async def agent_cot(task: str) -> dict:
    response = await client.messages.create(
        model="claude-opus-4-6",
        system=AGENT_COT_SYSTEM,
        messages=[{"role": "user", "content": task}],
        max_tokens=1024,
        temperature=0,
    )
    return parse_structured_cot(response.content[0].text)

# Пример задачи
result = await agent_cot(
    "Продажи: январь 100к, февраль 85к, март 120к, апрель 95к. "
    "Есть ли тренд? Какой прогноз на май?"
)
print("Наблюдения:", result.get('observe', '')[:100])
print("Анализ:",     result.get('analyze', '')[:100])
print("Ответ:",      result.get('answer',  '')[:200])

import anthropic
import re

client = anthropic.AsyncAnthropic()

# Паттерн: скрытое рассуждение + публичный ответ
SCRATCHPAD_SYSTEM = """
Используй тег  для рассуждения перед ответом.
Содержимое scratchpad видно только тебе — думай там свободно, черновики, расчёты.
После scratchpad дай чистый финальный ответ пользователю.
"""

async def think_then_answer(question: str) -> tuple[str, str]:
    """
    Возвращает (reasoning, answer).
    reasoning — скрытый scratchpad, answer — финальный ответ.
    """
    response = await client.messages.create(
        model="claude-opus-4-6",
        system=SCRATCHPAD_SYSTEM,
        messages=[{"role": "user", "content": question}],
        max_tokens=1024,
        temperature=0,
    )
    full = response.content[0].text

    # Разделяем scratchpad и ответ
    sp_match = re.search(r'(.*?)', full, re.DOTALL)
    reasoning = sp_match.group(1).strip() if sp_match else ""

    # Ответ — всё после закрывающего тега
    if sp_match:
        answer = full[sp_match.end():].strip()
    else:
        answer = full.strip()

    return reasoning, answer

reasoning, answer = await think_then_answer(
    "Если у меня 3 кошки и каждая ловит 2 мыши в день, "
    "сколько мышей поймают за рабочую неделю (5 дней)?"
)
print(f"[Рассуждение]: {reasoning}")
print(f"[Ответ]: {answer}")

import anthropic

client = anthropic.AsyncAnthropic()

async def extended_thinking(question: str, budget_tokens: int = 5000) -> dict:
    """
    Используем Extended Thinking Claude.

    budget_tokens — максимум токенов для внутреннего рассуждения.
    Чем сложнее задача, тем больше нужно (обычно 1000–16000).
    """
    response = await client.messages.create(
        model="claude-sonnet-4-6",          # Extended Thinking доступен с 3.7+
        max_tokens=16000,                    # должен быть > budget_tokens
        thinking={
            "type": "enabled",
            "budget_tokens": budget_tokens,  # сколько токенов отдать на рассуждение
        },
        messages=[{"role": "user", "content": question}],
    )

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

    for block in response.content:
        if block.type == "thinking":
            result["thinking"] = block.thinking   # внутреннее рассуждение
        elif block.type == "text":
            result["answer"] = block.text          # финальный ответ

    return result

# Задача, требующая глубокого рассуждения
result = await extended_thinking(
    "Логическая задача: Пять друзей (Аня, Боря, Вася, Галя, Дима) живут "
    "в домах разных цветов, содержат разных животных и пьют разные напитки. "
    "Аня живёт в красном доме. Боря держит собаку. Вася пьёт кофе. "
    "Галин дом стоит рядом с синим. Владелец зелёного дома пьёт чай. "
    "Кто держит рыбку?",
    budget_tokens=8000,
)
print("Внутреннее рассуждение (первые 300 символов):")
print(result["thinking"][:300], "...")
print("\nОтвет:", result["answer"])


# ── Streaming с Extended Thinking ──
async def extended_thinking_stream(question: str, budget_tokens: int = 4000):
    """Стримим Extended Thinking с разделением thinking/text блоков."""
    async with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=8000,
        thinking={"type": "enabled", "budget_tokens": budget_tokens},
        messages=[{"role": "user", "content": question}],
    ) as stream:

        current_block_type = None

        async for event in stream:
            if event.type == "content_block_start":
                current_block_type = event.content_block.type
                if current_block_type == "thinking":
                    print("\n[Thinking...]", end="", flush=True)
                elif current_block_type == "text":
                    print("\n[Answer]: ", end="", flush=True)

            elif event.type == "content_block_delta":
                delta = event.delta
                if hasattr(delta, "thinking"):
                    print(".", end="", flush=True)  # точки пока думает
                elif hasattr(delta, "text"):
                    print(delta.text, end="", flush=True)

    print()

import asyncio
import re
from collections import Counter
import anthropic

client = anthropic.AsyncAnthropic()

async def single_cot_sample(question: str, temperature: float = 0.7) -> str:
    """Один CoT-прогон с заданной температурой."""
    response = await client.messages.create(
        model="claude-opus-4-6",
        system="Решай задачи пошагово. В конце выведи: 'Ответ: <число>'",
        messages=[{"role": "user", "content": question}],
        max_tokens=512,
        temperature=temperature,
    )
    return response.content[0].text

def extract_final_answer(cot_text: str) -> str | None:
    """Извлекаем финальный ответ из CoT-вывода."""
    # Ищем паттерн "Ответ: <значение>"
    patterns = [
        r'Ответ:\s*([^\n.]+)',
        r'Answer:\s*([^\n.]+)',
        r'итого[:\s]+(\d+)',
        r'=\s*(\d+)\s*$',
    ]
    for pattern in patterns:
        m = re.search(pattern, cot_text, re.IGNORECASE | re.MULTILINE)
        if m:
            return m.group(1).strip()
    return None

async def self_consistency(
    question: str,
    n_samples: int = 5,
    temperature: float = 0.7,
) -> dict:
    """
    Self-Consistency: n независимых CoT-рассуждений → majority vote.
    """
    # Генерируем все образцы параллельно
    samples = await asyncio.gather(*[
        single_cot_sample(question, temperature)
        for _ in range(n_samples)
    ])

    # Извлекаем ответы
    answers = []
    for s in samples:
        ans = extract_final_answer(s)
        if ans:
            answers.append(ans.lower().strip())

    if not answers:
        return {"answer": None, "confidence": 0, "samples": samples}

    # Majority vote
    counter = Counter(answers)
    best_answer, count = counter.most_common(1)[0]
    confidence = count / len(answers)

    return {
        "answer": best_answer,
        "confidence": confidence,
        "vote_counts": dict(counter),
        "n_samples": n_samples,
        "samples": samples,
    }

# Пример
result = await self_consistency(
    "В магазине 23 яблока. Купили ещё столько же минус 10. "
    "Потом съели треть всех яблок. Сколько осталось?",
    n_samples=5,
)
print(f"Ответ: {result['answer']} (уверенность: {result['confidence']:.0%})")
print(f"Голоса: {result['vote_counts']}")

import asyncio
import anthropic
from dataclasses import dataclass, field

client = anthropic.AsyncAnthropic()

@dataclass
class ThoughtNode:
    thought: str
    score: float = 0.0
    children: list["ThoughtNode"] = field(default_factory=list)
    depth: int = 0

async def generate_thoughts(
    problem: str,
    current_state: str,
    n_thoughts: int = 3,
) -> list[str]:
    """Генерируем несколько вариантов следующего шага."""
    response = await client.messages.create(
        model="claude-opus-4-6",
        system=(
            "Генерируй следующие шаги для решения задачи. "
            f"Предложи ровно {n_thoughts} разных варианта. "
            "Формат: каждый вариант с новой строки, начиная с номера."
        ),
        messages=[{
            "role": "user",
            "content": (
                f"Задача: {problem}\n\n"
                f"Текущее состояние решения:\n{current_state}\n\n"
                f"Предложи {n_thoughts} варианта следующего шага."
            ),
        }],
        max_tokens=512,
        temperature=0.8,   # нужна вариативность
    )
    text = response.content[0].text
    # Простой парсинг нумерованного списка
    lines = [l.strip() for l in text.split('\n') if l.strip()]
    thoughts = [re.sub(r'^\d+[.)]\s*', '', l) for l in lines if l[0].isdigit()]
    return thoughts[:n_thoughts]

async def evaluate_thought(
    problem: str,
    thought_path: str,
) -> float:
    """Оцениваем перспективность текущего пути (0.0 – 1.0)."""
    response = await client.messages.create(
        model="claude-opus-4-6",
        system=(
            "Оцени насколько текущий путь решения перспективен. "
            "Ответь одним числом от 0.0 до 1.0."
        ),
        messages=[{
            "role": "user",
            "content": f"Задача: {problem}\n\nПуть решения:\n{thought_path}",
        }],
        max_tokens=5,
        temperature=0,
    )
    try:
        return float(response.content[0].text.strip())
    except ValueError:
        return 0.5

async def tree_of_thoughts(
    problem: str,
    max_depth: int = 3,
    branching: int = 3,
    beam_width: int = 2,          # сколько лучших путей сохраняем
) -> str:
    """
    Beam search по дереву мыслей.
    На каждом уровне: генерируем branching вариантов,
    оцениваем, оставляем beam_width лучших.
    """
    # Начальные ветки
    current_paths = [""]  # каждый элемент — история шагов

    for depth in range(max_depth):
        candidates = []

        # Генерируем продолжения для всех текущих путей
        for path in current_paths:
            new_thoughts = await generate_thoughts(problem, path, branching)
            for thought in new_thoughts:
                new_path = f"{path}\nШаг {depth+1}: {thought}" if path else f"Шаг 1: {thought}"
                candidates.append(new_path)

        # Оцениваем всех кандидатов параллельно
        scores = await asyncio.gather(*[
            evaluate_thought(problem, c) for c in candidates
        ])

        # Берём beam_width лучших
        ranked = sorted(zip(scores, candidates), key=lambda x: -x[0])
        current_paths = [path for _, path in ranked[:beam_width]]

    # Финальный ответ на основе лучшего пути
    best_path = current_paths[0]
    response = await client.messages.create(
        model="claude-opus-4-6",
        messages=[{
            "role": "user",
            "content": (
                f"Задача: {problem}\n\n"
                f"Цепочка рассуждений:\n{best_path}\n\n"
                "Дай финальный ответ."
            ),
        }],
        max_tokens=256,
        temperature=0,
    )
    return response.content[0].text

import anthropic
import asyncio
import json
import re

client = anthropic.AsyncAnthropic()

REACT_SYSTEM = """
Ты — агент, решающий задачи с помощью инструментов.
Используй строго следующий формат:

Thought: [рассуждение о следующем шаге]
Action: [название_инструмента]
Action Input: [параметры в JSON]

После получения результата:
Observation: [результат инструмента — заполняется системой]
Thought: [следующее рассуждение]
...

Когда задача решена:
Thought: Теперь у меня достаточно информации для ответа.
Final Answer: [финальный ответ]

Доступные инструменты: calculator, search, get_weather
"""

# Инструменты
async def calculator(expression: str) -> str:
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f"Ошибка: {e}"

async def search(query: str) -> str:
    # В реальности — запрос к поисковику
    return f"Результаты поиска по '{query}': [заглушка данных]"

async def get_weather(city: str) -> str:
    return f"Погода в {city}: 15°C, облачно"

TOOLS = {
    "calculator": calculator,
    "search": search,
    "get_weather": get_weather,
}

def parse_react_output(text: str) -> dict:
    """Разбираем ReAct-ответ на компоненты."""
    result = {}
    for key in ["thought", "action", "action input", "final answer"]:
        pattern = rf'{key}:\s*(.+?)(?=\n(?:thought|action|observation|final answer):|$)'
        m = re.search(pattern, text, re.IGNORECASE | re.DOTALL)
        if m:
            result[key] = m.group(1).strip()
    return result

async def react_agent(task: str, max_steps: int = 8) -> str:
    """ReAct агент: думает → действует → наблюдает → повторяет."""
    messages = [{"role": "user", "content": task}]
    history = ""    # накапливаем историю шагов

    for step in range(max_steps):
        # Генерируем следующий шаг
        response = await client.messages.create(
            model="claude-opus-4-6",
            system=REACT_SYSTEM,
            messages=messages + ([
                {"role": "assistant", "content": history}
            ] if history else []),
            max_tokens=512,
            temperature=0,
        )
        output = response.content[0].text
        parsed = parse_react_output(output)

        print(f"\n[Шаг {step+1}]")
        if "thought" in parsed:
            print(f"  Thought: {parsed['thought'][:100]}")

        # Финальный ответ
        if "final answer" in parsed:
            print(f"  Final Answer: {parsed['final answer']}")
            return parsed["final answer"]

        # Вызов инструмента
        if "action" in parsed:
            tool_name  = parsed["action"].strip().lower()
            tool_input_str = parsed.get("action input", "{}")

            try:
                tool_input = json.loads(tool_input_str)
            except json.JSONDecodeError:
                tool_input = {"input": tool_input_str}

            print(f"  Action: {tool_name}({tool_input})")

            tool_fn = TOOLS.get(tool_name)
            if tool_fn:
                observation = await tool_fn(**tool_input)
            else:
                observation = f"Инструмент '{tool_name}' не найден"

            print(f"  Observation: {observation[:100]}")

            # Добавляем в историю
            history += (
                f"\nThought: {parsed.get('thought', '')}"
                f"\nAction: {tool_name}"
                f"\nAction Input: {tool_input_str}"
                f"\nObservation: {observation}"
            )
        else:
            break

    return "Не удалось завершить задачу за отведённые шаги"

# Запуск
answer = await react_agent("Сколько будет (15 * 23) + 47 / 2? Покажи рассуждения.")
print(f"\nИтог: {answer}")