Zero-shot и Few-shot Prompting

import anthropic

client = anthropic.AsyncAnthropic()

# ── Базовый zero-shot ──
async def classify_sentiment_zero(review: str) -> str:
    response = await client.messages.create(
        model="claude-opus-4-6",
        system=(
            "Ты — классификатор тональности текста. "
            "Отвечай строго одним словом: positive, negative или neutral."
        ),
        messages=[{
            "role": "user",
            "content": f"Отзыв: {review}"
        }],
        max_tokens=10,
        temperature=0,   # детерминированный ответ для классификации
    )
    return response.content[0].text.strip().lower()

# Тест
reviews = [
    "Отличный продукт, всем рекомендую!",
    "Полный провал, деньги на ветер.",
    "Нормально, ничего особенного.",
]
for r in reviews:
    label = await classify_sentiment_zero(r)
    print(f"{label:10} | {r}")

EXTRACTION_SYSTEM = """
Ты — система извлечения структурированной информации из текста.

## Задача
Из произвольного текста извлеки контактные данные.

## Формат ответа
Отвечай строго в JSON без дополнительного текста:
{
  "name": "...",        // полное имя или null
  "email": "...",       // email или null
  "phone": "...",       // телефон в формате +7XXXXXXXXXX или null
  "company": "..."      // название компании или null
}

## Правила
- Если поле не найдено — верни null, не угадывай
- Телефон нормализуй: убери пробелы, скобки, дефисы
- Email приведи к нижнему регистру
"""

async def extract_contacts(text: str) -> dict:
    import json
    response = await client.messages.create(
        model="claude-opus-4-6",
        system=EXTRACTION_SYSTEM,
        messages=[{"role": "user", "content": text}],
        max_tokens=256,
        temperature=0,
    )
    return json.loads(response.content[0].text)

result = await extract_contacts(
    "Привет! Я Иван Петров из СберТех, пишите на ivan@sbertech.ru "
    "или звоните: +7 (916) 123-45-67"
)
# → {"name": "Иван Петров", "email": "ivan@sbertech.ru",
#    "phone": "+79161234567", "company": "СберТех"}

import anthropic

client = anthropic.AsyncAnthropic()

# Примеры: (входной текст, ожидаемый ответ)
TICKET_EXAMPLES = [
    ("Не могу войти, пишет «неверный пароль»",
     "КАТЕГОРИЯ: auth | ПРИОРИТЕТ: medium"),

    ("Сайт лежит, недоступен уже 3 часа!!!",
     "КАТЕГОРИЯ: outage | ПРИОРИТЕТ: critical"),

    ("Как экспортировать данные в Excel?",
     "КАТЕГОРИЯ: question | ПРИОРИТЕТ: low"),

    ("Ошибка 500 при оформлении заказа",
     "КАТЕГОРИЯ: bug | ПРИОРИТЕТ: high"),

    ("Верните деньги за подписку",
     "КАТЕГОРИЯ: billing | ПРИОРИТЕТ: high"),
]

def build_few_shot_messages(
    examples: list[tuple[str, str]],
    user_input: str,
) -> list[dict]:
    """Строим messages с примерами + реальный запрос."""
    messages = []
    for user_text, assistant_text in examples:
        messages.append({"role": "user",      "content": user_text})
        messages.append({"role": "assistant", "content": assistant_text})
    messages.append({"role": "user", "content": user_input})
    return messages

async def classify_ticket(text: str) -> str:
    messages = build_few_shot_messages(TICKET_EXAMPLES, text)
    response = await client.messages.create(
        model="claude-opus-4-6",
        system="Классифицируй тикет поддержки строго в формате из примеров.",
        messages=messages,
        max_tokens=32,
        temperature=0,
    )
    return response.content[0].text.strip()

# Тесты
tickets = [
    "Оплата прошла, но заказ не создался",
    "Можно ли подключить API к нашей системе?",
    "Приложение вылетает при открытии профиля",
]
for t in tickets:
    result = await classify_ticket(t)
    print(f"{result:<40} | {t}")

import random
from collections import defaultdict

# Пул размеченных примеров (из вашей базы)
EXAMPLE_POOL = [
    ("Не могу войти",           "auth",    "medium"),
    ("Забыл пароль",            "auth",    "low"),
    ("Аккаунт заблокирован",    "auth",    "high"),
    ("Сайт недоступен",         "outage",  "critical"),
    ("Страница 404",            "outage",  "medium"),
    ("Как изменить email?",     "question","low"),
    ("Есть ли API?",            "question","low"),
    ("Верните деньги",          "billing", "high"),
    ("Ошибка при оплате",       "billing", "high"),
    ("Ошибка 500",              "bug",     "high"),
    ("Приложение вылетает",     "bug",     "high"),
]

def select_balanced_examples(
    pool: list[tuple],
    n_per_class: int = 1,
    seed: int = 42,
) -> list[tuple[str, str]]:
    """
    Выбираем по n_per_class примеров на каждый класс.
    Возвращаем как (user_text, assistant_text).
    """
    rng = random.Random(seed)
    by_class: dict[str, list] = defaultdict(list)
    for text, category, priority in pool:
        by_class[category].append((text, category, priority))

    selected = []
    for category, items in by_class.items():
        chosen = rng.sample(items, min(n_per_class, len(items)))
        for text, cat, pri in chosen:
            answer = f"КАТЕГОРИЯ: {cat} | ПРИОРИТЕТ: {pri}"
            selected.append((text, answer))

    rng.shuffle(selected)
    return selected

# Используем сбалансированные примеры
examples = select_balanced_examples(EXAMPLE_POOL, n_per_class=1)
messages = build_few_shot_messages(examples, "Ошибка при сохранении файла")
print(f"Используем {len(examples)} примеров из {len(set(c for _,c,_ in EXAMPLE_POOL))} классов")

import numpy as np
from openai import AsyncOpenAI

embed_client = AsyncOpenAI()

async def embed(text: str) -> np.ndarray:
    """Получаем эмбеддинг текста через OpenAI."""
    r = await embed_client.embeddings.create(
        model="text-embedding-3-small",
        input=text,
    )
    return np.array(r.data[0].embedding)

def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))


class DynamicFewShotSelector:
    """
    Хранит пул примеров с эмбеддингами.
    При запросе возвращает k наиболее похожих.
    """

    def __init__(self):
        self.examples: list[dict] = []   # {text, answer, embedding}

    async def add(self, text: str, answer: str) -> None:
        embedding = await embed(text)
        self.examples.append({
            "text": text,
            "answer": answer,
            "embedding": embedding,
        })

    async def select(self, query: str, k: int = 3) -> list[tuple[str, str]]:
        """Возвращает k примеров, наиболее похожих на query."""
        if not self.examples:
            return []

        query_emb = await embed(query)
        scored = [
            (cosine_similarity(query_emb, ex["embedding"]), ex)
            for ex in self.examples
        ]
        scored.sort(key=lambda x: -x[0])
        return [(ex["text"], ex["answer"]) for _, ex in scored[:k]]


# ── Использование ──
selector = DynamicFewShotSelector()

# Наполняем пул
for text, category, priority in EXAMPLE_POOL:
    await selector.add(text, f"КАТЕГОРИЯ: {category} | ПРИОРИТЕТ: {priority}")

# При запросе — выбираем похожие
user_query = "Не удаётся авторизоваться через Google"
best_examples = await selector.select(user_query, k=3)
# Вернёт примеры про auth, т.к. они ближайшие по смыслу

messages = build_few_shot_messages(best_examples, user_query)
print(f"Выбраны примеры:")
for ex_text, ex_ans in best_examples:
    print(f"  '{ex_text}' → {ex_ans}")

import anthropic
import re

client = anthropic.AsyncAnthropic()

# Ваш внутренний формат лога
LOG_FORMAT_EXAMPLES = [
    (
        # вход: произвольное описание события
        "Пользователь user_123 купил подписку Pro за 990р",
        # выход: ваш формат
        "[EVENT:purchase][USER:user_123][PLAN:pro][AMOUNT:990][CURRENCY:RUB]"
    ),
    (
        "Аня из отдела маркетинга загрузила файл report_q1.xlsx (2.4MB)",
        "[EVENT:upload][USER:аня][DEPT:marketing][FILE:report_q1.xlsx][SIZE:2.4MB]"
    ),
    (
        "Сервис payments упал, ошибка connection timeout после 30 секунд",
        "[EVENT:error][SERVICE:payments][TYPE:connection_timeout][DURATION:30s]"
    ),
]

async def parse_to_log_format(description: str) -> str:
    """Конвертируем произвольное описание в структурированный лог."""
    messages = build_few_shot_messages(LOG_FORMAT_EXAMPLES, description)
    response = await client.messages.create(
        model="claude-opus-4-6",
        system=(
            "Конвертируй описание события в структурированный формат лога. "
            "Следуй строго формату из примеров. "
            "Ключи пиши заглавными, значения — в нижнем регистре."
        ),
        messages=messages,
        max_tokens=128,
        temperature=0,
    )
    return response.content[0].text.strip()

# Тест
events = [
    "Максим из команды backend задеплоил версию 2.3.1 на прод",
    "База данных postgres недоступна 5 минут из-за перезагрузки",
    "Новый пользователь user_456 зарегистрировался через GitHub",
]
for event in events:
    log = await parse_to_log_format(event)
    print(log)

import json
import re
from typing import TypeVar, Type
from pydantic import BaseModel, ValidationError

T = TypeVar("T", bound=BaseModel)

# ── Подход 1: строгий однострочный ответ ──
async def classify_strict(text: str, valid_labels: list[str]) -> str:
    """Гарантируем, что ответ — одна из допустимых меток."""
    labels_str = " / ".join(valid_labels)
    response = await client.messages.create(
        model="claude-opus-4-6",
        system=f"Классифицируй текст. Отвечай одним словом из: {labels_str}",
        messages=[{"role": "user", "content": text}],
        max_tokens=10,
        temperature=0,
    )
    result = response.content[0].text.strip().lower()
    if result not in valid_labels:
        # Fallback: ищем метку в ответе
        for label in valid_labels:
            if label in result:
                return label
        return valid_labels[0]  # дефолт
    return result


# ── Подход 2: JSON через prefill ──
async def extract_to_model(text: str, schema: Type[T]) -> T:
    """Few-shot + prefill для гарантированного JSON → Pydantic-модель."""
    schema_str = json.dumps(schema.model_json_schema(), ensure_ascii=False, indent=2)

    response = await client.messages.create(
        model="claude-opus-4-6",
        system=f"Извлеки данные и верни строго в JSON по схеме:\n{schema_str}",
        messages=[
            {"role": "user",      "content": text},
            {"role": "assistant", "content": "{"},  # prefill → только JSON
        ],
        max_tokens=512,
        temperature=0,
    )
    raw = "{" + response.content[0].text
    try:
        return schema.model_validate_json(raw)
    except (json.JSONDecodeError, ValidationError):
        # Второй шанс: ищем JSON-блок в ответе
        match = re.search(r'\{.*\}', raw, re.DOTALL)
        if match:
            return schema.model_validate_json(match.group())
        raise


# Пример
from pydantic import BaseModel

class ContactInfo(BaseModel):
    name: str | None = None
    email: str | None = None
    company: str | None = None

info = await extract_to_model(
    "Привет, я Сергей Иванов, сергей@примерсайт.ру, ООО «Рога и Копыта»",
    ContactInfo,
)
print(info.model_dump())

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

@dataclass
class FewShotPrompt:
    """
    Переиспользуемый few-shot промпт с динамическими примерами.
    """
    system: str
    examples: list[tuple[str, str]] = field(default_factory=list)
    model: str = "claude-opus-4-6"
    temperature: float = 0
    max_tokens: int = 256
    # Опциональный постпроцессор ответа
    postprocess: Callable[[str], str] = lambda x: x.strip()

    def add_example(self, user: str, assistant: str) -> "FewShotPrompt":
        """Fluent API для добавления примеров."""
        self.examples.append((user, assistant))
        return self

    def build_messages(self, query: str) -> list[dict]:
        messages = []
        for u, a in self.examples:
            messages.append({"role": "user",      "content": u})
            messages.append({"role": "assistant", "content": a})
        messages.append({"role": "user", "content": query})
        return messages

    async def run(self, query: str, client: anthropic.AsyncAnthropic | None = None) -> str:
        if client is None:
            client = anthropic.AsyncAnthropic()

        response = await client.messages.create(
            model=self.model,
            system=self.system,
            messages=self.build_messages(query),
            max_tokens=self.max_tokens,
            temperature=self.temperature,
        )
        raw = response.content[0].text
        return self.postprocess(raw)

    @property
    def n_examples(self) -> int:
        return len(self.examples)

    @property
    def example_tokens_estimate(self) -> int:
        """Грубая оценка токенов на примеры (~4 символа = 1 токен)."""
        total = sum(len(u) + len(a) for u, a in self.examples)
        return total // 4


# ── Пример использования ──
sentiment = (
    FewShotPrompt(
        system="Определи тональность. Отвечай одним словом: positive / negative / neutral.",
        temperature=0,
        max_tokens=5,
        postprocess=lambda x: x.strip().lower(),
    )
    .add_example("Отличный сервис, спасибо!", "positive")
    .add_example("Ужасно, больше не приду", "negative")
    .add_example("Обычный магазин", "neutral")
    .add_example("Быстро и удобно, рекомендую", "positive")
    .add_example("Долго ждал, но в целом нормально", "neutral")
)

print(f"Промпт: {sentiment.n_examples} примеров, ~{sentiment.example_tokens_estimate} токенов")

results = await asyncio.gather(
    sentiment.run("Лучший продукт года!"),
    sentiment.run("Полный мусор, выбросил деньги"),
    sentiment.run("Пришло вовремя"),
)
print(results)  # ['positive', 'negative', 'neutral']

import asyncio
from dataclasses import dataclass

@dataclass
class PromptTestCase:
    input: str
    expected: str
    description: str = ""

async def evaluate_prompt(
    prompt: FewShotPrompt,
    test_cases: list[PromptTestCase],
    match_fn: Callable[[str, str], bool] = lambda a, b: a == b,
) -> dict:
    """
    Запускаем тест-кейсы и считаем accuracy.
    match_fn — функция сравнения (exact match или contains или regex).
    """
    results = []

    async def run_one(tc: PromptTestCase) -> dict:
        actual = await prompt.run(tc.input)
        passed = match_fn(actual, tc.expected)
        return {
            "input": tc.input,
            "expected": tc.expected,
            "actual": actual,
            "passed": passed,
            "description": tc.description,
        }

    results = await asyncio.gather(*[run_one(tc) for tc in test_cases])

    passed = sum(1 for r in results if r["passed"])
    total = len(results)
    accuracy = passed / total if total > 0 else 0

    # Выводим провальные тесты
    failures = [r for r in results if not r["passed"]]
    if failures:
        print(f"\n❌ Failed {len(failures)}/{total} tests:")
        for f in failures:
            print(f"  Input:    {f['input'][:60]}")
            print(f"  Expected: {f['expected']}")
            print(f"  Actual:   {f['actual']}")
            print()

    return {
        "accuracy": accuracy,
        "passed": passed,
        "total": total,
        "results": list(results),
    }

# Тест-кейсы для нашего классификатора
TEST_CASES = [
    PromptTestCase("Отличный продукт!",         "positive"),
    PromptTestCase("Ужасное качество",           "negative"),
    PromptTestCase("Нормально, ничего особого",  "neutral"),
    PromptTestCase("Советую всем!",              "positive"),
    PromptTestCase("Деньги выброшены на ветер",  "negative"),
]

metrics = await evaluate_prompt(sentiment, TEST_CASES)
print(f"Accuracy: {metrics['accuracy']:.0%} ({metrics['passed']}/{metrics['total']})")