Тестирование промптов
варианты, A/B, регрессия

# tests/sentiment_cases.yaml
- id: sentiment_positive_001
  input: "Доставка пришла раньше срока, всё идеально!"
  expected: "positive"
  check: contains
  tags: [happy_path]

- id: sentiment_negative_002
  input: "Третий день жду ответа от поддержки. Кошмар."
  expected: "negative"
  check: contains
  tags: [happy_path]

- id: sentiment_sarcasm_003
  input: "Ну конечно, доставили через 3 недели. Прекрасно."
  expected: "negative"
  check: contains
  tags: [edge_case, regression]  # эта строка была багом раньше

- id: sentiment_neutral_004
  input: "Товар получил. Коробка без повреждений."
  expected: "neutral"
  check: contains
  tags: [happy_path]

- id: sentiment_mixed_005
  input: "Качество хорошее, но цена завышена."
  expected: "neutral"
  check: llm
  threshold: 0.7
  tags: [edge_case]

from __future__ import annotations
import re
import yaml
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Literal
import anthropic


CheckType = Literal["exact", "contains", "regex", "llm", "custom"]


@dataclass
class TestCase:
    id: str
    input: str
    expected: str
    check: CheckType = "contains"
    tags: list[str] = field(default_factory=list)
    context: str = ""
    threshold: float = 1.0


@dataclass
class TestResult:
    case_id: str
    passed: bool
    score: float
    actual: str
    expected: str
    error: str = ""


def load_cases(path: str | Path) -> list[TestCase]:
    data = yaml.safe_load(Path(path).read_text(encoding="utf-8"))
    return [TestCase(**item) for item in data]


class PromptRunner:
    """Запускает промпт против набора тест-кейсов."""

    def __init__(self, system_prompt: str, model: str = "claude-haiku-4-5-20251001"):
        self.system = system_prompt
        self.model = model
        self.client = anthropic.Anthropic()

    def run(self, case: TestCase) -> TestResult:
        try:
            resp = self.client.messages.create(
                model=self.model,
                max_tokens=256,
                system=self.system,
                messages=[{"role": "user", "content": case.input}],
            )
            actual = resp.content[0].text.strip().lower()
        except Exception as e:
            return TestResult(case.id, False, 0.0, "", case.expected, str(e))

        passed, score = self._check(actual, case)
        return TestResult(case.id, passed, score, actual, case.expected)

    def _check(self, actual: str, case: TestCase) -> tuple[bool, float]:
        match case.check:
            case "exact":
                ok = actual == case.expected.lower()
                return ok, 1.0 if ok else 0.0
            case "contains":
                ok = case.expected.lower() in actual
                return ok, 1.0 if ok else 0.0
            case "regex":
                ok = bool(re.search(case.expected, actual))
                return ok, 1.0 if ok else 0.0
            case _:
                return True, 1.0  # llm/custom — обрабатываются отдельно

    def run_suite(self, cases: list[TestCase]) -> list[TestResult]:
        return [self.run(c) for c in cases]


# Использование
SYSTEM = """
Классифицируй тональность отзыва.
Ответь одним словом: positive, negative или neutral.
"""

runner = PromptRunner(SYSTEM)
cases  = load_cases("tests/sentiment_cases.yaml")
results = runner.run_suite(cases)

passed = sum(r.passed for r in results)
print(f"Passed: {passed}/{len(results)} ({passed/len(results)*100:.0f}%)")
for r in results:
    icon = "✓" if r.passed else "✗"
    print(f"  {icon} {r.case_id}: got '{r.actual}', expected '{r.expected}'")

from dataclasses import dataclass
import statistics


@dataclass
class ABResult:
    variant: str
    passed: int
    total: int
    scores: list[float]

    @property
    def pass_rate(self) -> float:
        return self.passed / self.total if self.total else 0.0

    @property
    def avg_score(self) -> float:
        return statistics.mean(self.scores) if self.scores else 0.0


def ab_test(
    variants: dict[str, str],  # {"A": system_prompt_a, "B": system_prompt_b}
    cases: list[TestCase],
    model: str = "claude-haiku-4-5-20251001",
) -> dict[str, ABResult]:
    """
    Запускает несколько вариантов промпта на одном наборе тест-кейсов.
    Возвращает сводные метрики по каждому варианту.
    """
    results: dict[str, ABResult] = {}

    for name, system in variants.items():
        runner = PromptRunner(system, model)
        run_results = runner.run_suite(cases)

        results[name] = ABResult(
            variant=name,
            passed=sum(r.passed for r in run_results),
            total=len(run_results),
            scores=[r.score for r in run_results],
        )

    return results


def print_ab_report(results: dict[str, ABResult]) -> None:
    """Выводит сравнительный отчёт."""
    print(f"\n{'Вариант':<12} {'Pass rate':<12} {'Avg score':<12} {'Passed/Total'}")
    print("-" * 52)
    best = max(results.values(), key=lambda r: r.pass_rate)
    for name, res in sorted(results.items(), key=lambda x: -x[1].pass_rate):
        marker = " ← winner" if res is best else ""
        print(
            f"{name:<12} {res.pass_rate*100:>8.1f}%   "
            f"{res.avg_score:>8.3f}    "
            f"{res.passed}/{res.total}{marker}"
        )


# Использование
variants = {
    "A: краткий":    "Классифицируй тональность. Ответь: positive/negative/neutral.",
    "B: few-shot":   (
        "Классифицируй тональность отзыва.\n"
        "Примеры: 'отлично' → positive, 'кошмар' → negative, 'нормально' → neutral.\n"
        "Ответь одним словом: positive, negative или neutral."
    ),
    "C: verbose":    (
        "Ты — эксперт по анализу тональности текста. "
        "Используй продвинутые методы NLP для определения эмоциональной окраски. "
        "Учитывай саркасм, контекст и скрытые коннотации. "
        "Дай подробный анализ и в конце напиши итог: positive, negative или neutral."
    ),
}

cases = load_cases("tests/sentiment_cases.yaml")
ab_results = ab_test(variants, cases)
print_ab_report(ab_results)

import json
from pathlib import Path
from dataclasses import asdict


BASELINE_FILE = Path("tests/.baseline.json")


def save_baseline(results: list[TestResult]) -> None:
    """Сохраняет текущие результаты как baseline."""
    data = {r.case_id: {"score": r.score, "passed": r.passed} for r in results}
    BASELINE_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2))
    print(f"Baseline сохранён: {len(data)} тест-кейсов → {BASELINE_FILE}")


def load_baseline() -> dict[str, dict]:
    if not BASELINE_FILE.exists():
        return {}
    return json.loads(BASELINE_FILE.read_text())


def regression_check(
    new_results: list[TestResult],
    tolerance: float = 0.05,   # допустимое снижение score
) -> bool:
    """
    Сравнивает новые результаты с baseline.
    Возвращает True, если регрессий нет.
    tolerance: допустимое снижение avg score (например, 0.05 = 5%)
    """
    baseline = load_baseline()
    if not baseline:
        print("⚠ Baseline не найден, создаём первый.")
        save_baseline(new_results)
        return True

    regressions: list[str] = []
    improvements: list[str] = []

    for res in new_results:
        base = baseline.get(res.case_id)
        if base is None:
            continue  # новый тест — пропускаем

        delta = res.score - base["score"]
        if delta < -tolerance:
            regressions.append(
                f"  ✗ {res.case_id}: {base['score']:.2f} → {res.score:.2f} (Δ {delta:+.2f})"
            )
        elif delta > 0 and not base["passed"] and res.passed:
            improvements.append(f"  ↑ {res.case_id}: был FAIL, стал PASS")

    if improvements:
        print("Улучшения:")
        print("\n".join(improvements))

    if regressions:
        print(f"\n🔴 РЕГРЕССИИ ({len(regressions)}):")
        print("\n".join(regressions))
        return False

    new_count = sum(1 for r in new_results if r.case_id not in baseline)
    if new_count:
        print(f"ℹ {new_count} новых тест-кейсов добавлено в baseline.")

    print(f"\n🟢 Регрессий не обнаружено ({len(new_results)} тестов)")
    return True


# --- Workflow ---
# 1. Первый запуск: создаём baseline
runner_v1 = PromptRunner(SYSTEM_V1)
results_v1 = runner_v1.run_suite(load_cases("tests/sentiment_cases.yaml"))
save_baseline(results_v1)

# 2. После изменения промпта: проверяем регрессии
runner_v2 = PromptRunner(SYSTEM_V2)
results_v2 = runner_v2.run_suite(load_cases("tests/sentiment_cases.yaml"))
ok = regression_check(results_v2)

if not ok:
    raise SystemExit("❌ Регрессионный тест не пройден. Промпт не принят.")

import statistics
from collections import Counter


def accuracy(results: list[TestResult]) -> float:
    """Доля успешных тестов (pass rate)."""
    if not results:
        return 0.0
    return sum(r.passed for r in results) / len(results)


def avg_score(results: list[TestResult]) -> float:
    """Средний score по всем тестам (для нечётких метрик)."""
    scores = [r.score for r in results]
    return statistics.mean(scores) if scores else 0.0


def fail_analysis(results: list[TestResult]) -> dict[str, list[str]]:
    """
    Группирует провалившиеся тесты по тегам.
    Помогает найти, какие категории страдают больше всего.
    """
    # Загружаем оригинальные кейсы для доступа к тегам
    failed_ids = {r.case_id for r in results if not r.passed}
    # (предполагается, что у вас есть доступ к тест-кейсам)
    return {"failed_ids": list(failed_ids)}


def consistency_score(
    runner: PromptRunner,
    case: TestCase,
    n_runs: int = 5,
) -> float:
    """
    Запускает один кейс N раз и измеряет стабильность ответа.
    Score = 1.0 если все ответы одинаковые, <1.0 если разброс есть.
    Важно для задач с temperature > 0.
    """
    run_results = [runner.run(case) for _ in range(n_runs)]
    answers = [r.actual for r in run_results]
    most_common_count = Counter(answers).most_common(1)[0][1]
    return most_common_count / n_runs


def latency_percentiles(latencies: list[float]) -> dict[str, float]:
    """P50/P95/P99 латентности в секундах."""
    s = sorted(latencies)
    n = len(s)
    return {
        "p50":  s[int(n * 0.50)],
        "p95":  s[int(n * 0.95)],
        "p99":  s[int(n * 0.99)],
        "mean": statistics.mean(s),
    }


# Полный отчёт
def full_report(results: list[TestResult]) -> None:
    total = len(results)
    n_pass = sum(r.passed for r in results)
    print(f"{'='*40}")
    print(f"Всего тестов:  {total}")
    print(f"Passed:        {n_pass} ({accuracy(results)*100:.1f}%)")
    print(f"Failed:        {total - n_pass}")
    print(f"Avg score:     {avg_score(results):.3f}")
    print(f"{'='*40}")
    failed = [r for r in results if not r.passed]
    if failed:
        print("Провалились:")
        for r in failed:
            print(f"  ✗ {r.case_id}")

import json
import anthropic


JUDGE_SYSTEM = """
Ты — строгий эксперт по оценке качества ответов языковых моделей.
Оценивай объективно, игнорируй стиль — только содержание и соответствие критериям.
Всегда отвечай в JSON: {"score": 0.0-1.0, "passed": true/false, "reason": "..."}.
"""


def llm_judge(
    question: str,
    actual_answer: str,
    expected_criteria: str,
    judge_model: str = "claude-opus-4-6",
) -> tuple[float, bool, str]:
    """
    Оценивает ответ модели с помощью другой модели-судьи.

    question:           исходный вопрос/запрос
    actual_answer:      ответ тестируемой модели
    expected_criteria:  критерии хорошего ответа (что должно быть)
    judge_model:        сильная модель для оценки

    Returns: (score 0-1, passed, reason)
    """
    client = anthropic.Anthropic()

    judge_prompt = f"""

{question}



{actual_answer}



{expected_criteria}


Оцени ответ по критериям. Выставь score от 0.0 до 1.0.
Считай passed=true если score >= 0.7.
Ответь строго в JSON: {{"score": ..., "passed": ..., "reason": "..."}}.
"""

    resp = client.messages.create(
        model=judge_model,
        max_tokens=256,
        system=JUDGE_SYSTEM,
        messages=[{"role": "user", "content": judge_prompt}],
    )

    raw = resp.content[0].text.strip()
    # Извлекаем JSON из ответа
    import re
    match = re.search(r'\{.*\}', raw, re.DOTALL)
    if not match:
        return 0.0, False, f"Не удалось распарсить: {raw[:100]}"

    data = json.loads(match.group())
    return float(data["score"]), bool(data["passed"]), data.get("reason", "")


# Пример: проверяем суммаризатор
question = "Сделай краткое резюме текста о компании."
actual   = "Компания основана в 2019 году в Москве, специализируется на AI-решениях."
criteria = """
- Содержит год основания (2019)
- Содержит город (Москва)
- Содержит направление деятельности
- Не более 2 предложений
"""

score, passed, reason = llm_judge(question, actual, criteria)
print(f"Score: {score:.2f} | Passed: {passed} | {reason}")

# tests/test_prompts.py
import pytest
import yaml
from pathlib import Path
from prompt_runner import PromptRunner, TestCase, load_cases, regression_check

# Загружаем промпт из файла (не хардкодим в тестах)
SYSTEM_PROMPT = Path("prompts/sentiment_v2.txt").read_text()

# Фильтр: в быстром режиме (CI) только happy_path тесты
def pytest_configure(config):
    config.addinivalue_line("markers", "slow: медленные тесты (edge cases, llm judge)")


@pytest.fixture(scope="module")
def runner():
    return PromptRunner(SYSTEM_PROMPT, model="claude-haiku-4-5-20251001")


@pytest.fixture(scope="module")
def all_cases():
    return load_cases("tests/sentiment_cases.yaml")


# Параметризованные тесты — по одному на каждый кейс
def pytest_generate_tests(metafunc):
    if "test_case" in metafunc.fixturenames:
        cases = load_cases("tests/sentiment_cases.yaml")
        metafunc.parametrize("test_case", cases, ids=[c.id for c in cases])


def test_prompt_case(runner, test_case):
    """Каждый тест-кейс запускается отдельно — удобно видеть, что именно упало."""
    if "slow" in test_case.tags:
        pytest.mark.slow
    result = runner.run(test_case)
    assert result.passed, (
        f"Case '{test_case.id}' failed: "
        f"expected '{test_case.expected}', got '{result.actual}'"
    )


def test_no_regression(runner, all_cases):
    """Регрессионный тест: avg score не должен падать ниже baseline."""
    results = runner.run_suite(all_cases)
    ok = regression_check(results, tolerance=0.05)
    assert ok, "Обнаружена регрессия по сравнению с baseline!"


def test_pass_rate_above_threshold(runner, all_cases):
    """Общий pass rate должен быть не ниже 90%."""
    results = runner.run_suite(all_cases)
    pass_rate = sum(r.passed for r in results) / len(results)
    assert pass_rate >= 0.90, f"Pass rate {pass_rate:.1%} < 90%"

# .github/workflows/prompt-tests.yml
name: Prompt regression tests

on:
  pull_request:
    paths:
      - 'prompts/**'       # запускаем только при изменении промптов
      - 'tests/**'

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install dependencies
        run: pip install anthropic pytest pyyaml

      - name: Run prompt tests (fast suite only)
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          pytest tests/test_prompts.py -m "not slow" \
            --tb=short -q \
            --junit-xml=test-results.xml

      - name: Upload results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: prompt-test-results
          path: test-results.xml

# promptfooconfig.yaml
description: "Sentiment classifier A/B test"

prompts:
  - id: prompt-a
    raw: "Классифицируй тональность. Ответь: positive/negative/neutral."
  - id: prompt-b
    raw: |
      Классифицируй тональность отзыва.
      Примеры: "отлично" → positive, "кошмар" → negative.
      Ответь одним словом: positive, negative или neutral.

providers:
  - anthropic:messages:claude-haiku-4-5-20251001
  - anthropic:messages:claude-sonnet-4-6

tests:
  - vars:
      text: "Доставка пришла раньше срока, всё идеально!"
    assert:
      - type: contains
        value: "positive"

  - vars:
      text: "Третий день жду ответа от поддержки. Кошмар."
    assert:
      - type: contains
        value: "negative"

  - vars:
      text: "Ну конечно, доставили через 3 недели. Прекрасно."
    assert:
      - type: llm-rubric
        value: "Ответ должен быть 'negative' — это явный сарказм"

  - vars:
      text: "Качество хорошее, но цена завышена."
    assert:
      - type: contains
        value: "neutral"
      - type: latency
        threshold: 3000  # ms