Evals 2.0: новое поколение оценки LLM-приложений

from dataclasses import dataclass
from enum import Enum


class StepType(Enum):
    THINK = "think"
    TOOL_CALL = "tool_call"
    TOOL_RESULT = "tool_result"
    RESPONSE = "response"
    ERROR = "error"


@dataclass
class AgentStep:
    type: StepType
    content: str
    tool_name: str | None = None
    tool_args: dict | None = None
    duration_ms: int = 0
    tokens_used: int = 0


@dataclass
class AgentTrajectory:
    task: str
    steps: list[AgentStep]
    final_result: str
    success: bool
    total_cost: float


def eval_trajectory(
    trajectory: AgentTrajectory,
    optimal_steps: int,
    required_tools: list[str],
    max_cost: float,
) -> dict:
    """Evaluate agent trajectory on multiple dimensions."""
    actual_steps = len([
        s for s in trajectory.steps
        if s.type == StepType.TOOL_CALL
    ])

    # Step efficiency: optimal / actual (capped at 1.0)
    step_efficiency = min(optimal_steps / max(actual_steps, 1), 1.0)

    # Tool accuracy: did the agent use the right tools?
    used_tools = {
        s.tool_name for s in trajectory.steps
        if s.type == StepType.TOOL_CALL and s.tool_name
    }
    tool_coverage = len(
        set(required_tools) & used_tools
    ) / max(len(required_tools), 1)

    # Recovery: errors that were handled
    errors = [s for s in trajectory.steps if s.type == StepType.ERROR]
    recoveries = 0
    for i, step in enumerate(trajectory.steps):
        if step.type == StepType.ERROR and i + 1 < len(trajectory.steps):
            if trajectory.steps[i + 1].type != StepType.ERROR:
                recoveries += 1
    recovery_rate = recoveries / max(len(errors), 1) if errors else 1.0

    # Cost efficiency
    cost_ok = trajectory.total_cost <= max_cost

    return {
        "task_completed": trajectory.success,
        "step_efficiency": round(step_efficiency, 2),
        "tool_coverage": round(tool_coverage, 2),
        "recovery_rate": round(recovery_rate, 2),
        "cost_within_budget": cost_ok,
        "actual_cost": trajectory.total_cost,
        "total_steps": len(trajectory.steps),
        "score": round(
            (
                (1.0 if trajectory.success else 0.0) * 0.4
                + step_efficiency * 0.2
                + tool_coverage * 0.2
                + recovery_rate * 0.2
            ),
            2,
        ),
    }

import anthropic
import json

client = anthropic.Anthropic()

MULTI_TURN_JUDGE_PROMPT = """You are evaluating a multi-turn conversation.
For each assistant turn, rate on a 1-5 scale:
- Relevance: does the response address the user's message?
- Consistency: does it contradict previous responses?
- Context retention: does it remember earlier context?
- Helpfulness: does it move toward resolving the user's need?

Conversation:
{conversation}

Return JSON array with one object per assistant turn:
[{{"turn": 1, "relevance": N, "consistency": N,
   "context_retention": N, "helpfulness": N, "notes": "..."}}]"""


@dataclass
class ConversationTurn:
    role: str  # "user" or "assistant"
    content: str
    expected_behavior: str | None = None  # for eval


def run_multi_turn_eval(
    system_prompt: str,
    scenario: list[ConversationTurn],
    model: str = "claude-sonnet-4-20250514",
    judge_model: str = "claude-opus-4-20250514",
) -> dict:
    """Run a multi-turn scenario and evaluate each turn."""
    messages = []
    actual_conversation = []

    for turn in scenario:
        if turn.role == "user":
            messages.append({"role": "user", "content": turn.content})
            actual_conversation.append(f"User: {turn.content}")

            # Get model response
            response = client.messages.create(
                model=model,
                max_tokens=1024,
                system=system_prompt,
                messages=messages,
            )
            assistant_msg = response.content[0].text
            messages.append({"role": "assistant", "content": assistant_msg})
            actual_conversation.append(f"Assistant: {assistant_msg}")

            if turn.expected_behavior:
                actual_conversation.append(
                    f"[Expected: {turn.expected_behavior}]"
                )

    # Judge the full conversation
    judge_response = client.messages.create(
        model=judge_model,
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": MULTI_TURN_JUDGE_PROMPT.format(
                conversation="\n".join(actual_conversation)
            ),
        }],
    )

    text = judge_response.content[0].text
    start = text.index("[")
    end = text.rindex("]") + 1
    scores = json.loads(text[start:end])

    avg_score = sum(
        sum(s[k] for k in ["relevance", "consistency",
                            "context_retention", "helpfulness"]) / 4
        for s in scores
    ) / len(scores)

    return {
        "scenario_score": round(avg_score, 2),
        "turn_scores": scores,
        "conversation": actual_conversation,
        "passed": avg_score >= 4.0,
    }

EVAL_GEN_PROMPT = """You are generating eval test cases for an AI application.

Application description: {app_description}

Generate {count} diverse test cases. For each test case provide:
1. input: the user's message
2. expected: what a correct response should contain
3. category: type of test (typical, edge_case, adversarial, out_of_scope)
4. difficulty: easy, medium, hard

Rules:
- 40% typical cases, 30% edge cases, 20% adversarial, 10% out of scope
- Include multilingual inputs if the app supports it
- Include ambiguous inputs that require clarification
- Include inputs that try to manipulate the system

Return JSON array:
[{{"input": "...", "expected": "...",
   "category": "...", "difficulty": "..."}}]"""


def generate_eval_dataset(
    app_description: str,
    count: int = 50,
    model: str = "claude-opus-4-20250514",
) -> list[dict]:
    """Generate eval dataset from app description."""
    response = client.messages.create(
        model=model,
        max_tokens=8192,
        messages=[{
            "role": "user",
            "content": EVAL_GEN_PROMPT.format(
                app_description=app_description,
                count=count,
            ),
        }],
    )

    text = response.content[0].text
    start = text.index("[")
    end = text.rindex("]") + 1
    dataset = json.loads(text[start:end])

    return dataset


def generate_from_logs(
    production_logs: list[dict],
    model: str = "claude-opus-4-20250514",
) -> list[dict]:
    """Generate eval cases from production logs."""
    prompt = f"""Analyze these production logs and generate eval test cases.

Focus on:
- Queries that received low user ratings
- Queries where the model refused to answer
- Queries with unusual patterns
- Common query types that should always work

Logs (sample):
{json.dumps(production_logs[:20], ensure_ascii=False, indent=2)}

Generate 20 eval test cases based on patterns you see.
Return JSON array with input, expected, category, difficulty."""

    response = client.messages.create(
        model=model,
        max_tokens=8192,
        messages=[{"role": "user", "content": prompt}],
    )

    text = response.content[0].text
    start = text.index("[")
    end = text.rindex("]") + 1
    return json.loads(text[start:end])

# .github/workflows/eval.yml
name: LLM Eval

on:
  pull_request:
    paths:
      - 'prompts/**'
      - 'src/llm/**'
      - 'src/agents/**'

jobs:
  quick-eval:
    name: Quick Eval
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

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

      - run: pip install -r requirements.txt

      - name: Run quick eval
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: python scripts/run_eval.py --suite quick --output results.json

      - name: Generate eval diff
        run: python scripts/eval_diff.py --current results.json --baseline baseline.json --output diff.md

      - name: Post eval diff to PR
        uses: marocchino/sticky-pull-request-comment@v2
        with:
          path: diff.md

      - name: Check thresholds
        run: |
          python scripts/check_thresholds.py results.json \
            --min-score 4.0 \
            --max-regression 0.05 \
            --max-cost-increase 0.20

# scripts/eval_diff.py
import json
import sys


def generate_diff(current_file: str, baseline_file: str, output_file: str):
    with open(current_file) as f:
        current = json.load(f)
    with open(baseline_file) as f:
        baseline = json.load(f)

    lines = ["## Eval Results\n"]

    # Overall metrics
    curr_score = current["avg_score"]
    base_score = baseline["avg_score"]
    delta = curr_score - base_score
    sign = "+" if delta >= 0 else ""

    lines.append(f"| Metric | Baseline | Current | Delta |")
    lines.append(f"|--------|----------|---------|-------|")
    lines.append(
        f"| Avg Score | {base_score:.2f} | {curr_score:.2f} "
        f"| {sign}{delta:.2f} |"
    )
    lines.append(
        f"| Cost | ${baseline['total_cost']:.3f} "
        f"| ${current['total_cost']:.3f} "
        f"| {sign}${current['total_cost'] - baseline['total_cost']:.3f} |"
    )

    # Regressions
    regressions = []
    improvements = []
    for c_case in current.get("details", []):
        case_id = c_case["input"]["question"]
        b_case = next(
            (b for b in baseline.get("details", [])
             if b["input"]["question"] == case_id),
            None,
        )
        if b_case:
            c_score = c_case["scores"]["overall"]
            b_score = b_case["scores"]["overall"]
            if c_score < b_score - 0.5:
                regressions.append((case_id, b_score, c_score))
            elif c_score > b_score + 0.5:
                improvements.append((case_id, b_score, c_score))

    if regressions:
        lines.append("\n### Regressions\n")
        for q, b, c in regressions:
            lines.append(f"- **{q}**: {b:.1f} -> {c:.1f}")

    if improvements:
        lines.append("\n### Improvements\n")
        for q, b, c in improvements:
            lines.append(f"- **{q}**: {b:.1f} -> {c:.1f}")

    with open(output_file, "w") as f:
        f.write("\n".join(lines))


if __name__ == "__main__":
    generate_diff(sys.argv[1], sys.argv[2], sys.argv[3])

from dataclasses import dataclass


@dataclass
class ModelConfig:
    name: str
    cost_per_1k_input: float
    cost_per_1k_output: float
    avg_latency_ms: int


MODELS = [
    ModelConfig("gpt-4.1-nano", 0.0001, 0.0004, 200),
    ModelConfig("claude-haiku", 0.00025, 0.00125, 300),
    ModelConfig("claude-sonnet", 0.003, 0.015, 800),
    ModelConfig("gpt-4.1", 0.002, 0.008, 600),
    ModelConfig("claude-opus", 0.015, 0.075, 2000),
]


def cost_quality_eval(
    eval_dataset: list[dict],
    models: list[ModelConfig],
    run_fn,  # (model_name, input) -> (output, tokens_in, tokens_out)
    judge_fn,  # (input, expected, actual) -> score
) -> list[dict]:
    """Compare models on cost, quality, and latency."""
    results = []

    for model in models:
        scores = []
        total_cost = 0.0

        for case in eval_dataset:
            output, tok_in, tok_out = run_fn(model.name, case["input"])
            score = judge_fn(case["input"], case["expected"], output)
            cost = (
                tok_in / 1000 * model.cost_per_1k_input
                + tok_out / 1000 * model.cost_per_1k_output
            )
            scores.append(score)
            total_cost += cost

        avg_score = sum(scores) / len(scores)
        results.append({
            "model": model.name,
            "avg_score": round(avg_score, 2),
            "total_cost": round(total_cost, 4),
            "cost_per_case": round(total_cost / len(eval_dataset), 5),
            "avg_latency_ms": model.avg_latency_ms,
            "quality_per_dollar": round(
                avg_score / max(total_cost, 0.0001), 1
            ),
        })

    return sorted(
        results, key=lambda x: x["quality_per_dollar"], reverse=True
    )

# promptfooconfig.yaml — agent eval
description: "Agent eval: customer support with tools"

providers:
  - id: anthropic:messages:claude-sonnet-4-20250514
    config:
      max_tokens: 2048
      tools:
        - name: search_orders
          description: "Search customer orders by email or order ID"
          parameters:
            type: object
            properties:
              query:
                type: string
            required: [query]
        - name: create_refund
          description: "Create a refund for an order"
          parameters:
            type: object
            properties:
              order_id:
                type: string
              amount:
                type: number
              reason:
                type: string
            required: [order_id, reason]

prompts:
  - id: agent-v2
    raw: |
      You are a customer support agent. Use tools to look up orders
      and process refunds. Always verify the order exists before
      processing a refund. If the customer is upset, acknowledge
      their frustration before proceeding.

tests:
  # Outcome eval
  - vars:
      question: "I want a refund for order #12345"
    assert:
      - type: is-valid-openai-tools-call
      - type: llm-rubric
        value: "Agent searched for the order before processing refund"
      - type: llm-rubric
        value: "Agent asked for refund reason"
      - type: cost
        threshold: 0.05

  # Recovery eval
  - vars:
      question: "Refund my order #99999"
    provider:
      config:
        toolOverrides:
          search_orders:
            output: '{"error": "Order not found"}'
    assert:
      - type: llm-rubric
        value: "Agent gracefully handled the missing order"
      - type: not-contains
        value: "error"

  # Trajectory eval
  - vars:
      question: "I bought a laptop last week and it's broken"
    assert:
      - type: javascript
        value: |
          // Check that agent used tools efficiently
          const toolCalls = output.tool_calls || [];
          return toolCalls.length <= 3;
      - type: llm-rubric
        value: "Agent gathered necessary info before suggesting solution"