src/eval/models.py - master - Watashicuvu/agentic-tools

src/eval/models.py
488 строк18 KB
Якуб
some fixes of api server
13 апр 2026, 11:24
13 апр 2026, 11:246827c5f
"""Pydantic/dataclass models for eval pipeline.

Contains:
- EvalTask: task definition
- EvalResult: single task execution result
- EvalReport: aggregated report from multiple results
- ClarificationEvent: IronUser clarification Q&A
- FinalCodeResponse: structured output for executor role
- FinalPlanResponse: structured output for strategist role
- FinalReviewResponse: structured output for reviewer role
- FinalClarificationResponse: structured output for clarification role
"""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

from pydantic import BaseModel, Field, field_validator


# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------


class EvalRole(str, Enum):
    """Роли для eval-сессий."""
    EXECUTOR = "executor"
    STRATEGIST = "strategist"
    REVIEWER = "reviewer"
    MULTI = "multi"  # Полный pipeline: strategist → executor → reviewer


class ReviewVerdict(str, Enum):
    """Вердикт ревьюера."""
    APPROVED = "APPROVED"
    WARN = "WARN"
    BLOCKER = "BLOCKER"


# ---------------------------------------------------------------------------
# EvalTask
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class EvalTask:
    """A single evaluation task.

    Attributes:
        id: Unique task identifier.
        description: Task description.
        expected_behavior: Expected behavior description.
        reference_code: Optional reference implementation.
        lint_rules: Lint rules to apply (default: ["luacheck"]).
        mode: Task mode ("batch" or "interactive").
        role: Eval role ("executor", "strategist", "reviewer", "multi").
        title: Human-readable title (from tasks.json).
        expected_lua: Expected Lua code (from tasks.json).
        context: Task context variables (from tasks.json).
        requires_clarification: Whether task needs clarification.
    """

    id: str
    description: str
    expected_behavior: str
    reference_code: Optional[str] = None
    lint_rules: list[str] = field(default_factory=lambda: ["luacheck"])
    mode: str = "batch"
    role: str = "executor"
    title: str = ""
    expected_lua: Optional[str] = None
    context: Optional[dict] = None
    requires_clarification: bool = False
    error_feedback: Optional[str] = None  # Feedback от предыдущих retry-попыток

    def __post_init__(self) -> None:
        """Validate mode and role."""
        if self.mode not in ("batch", "interactive"):
            raise ValueError(f"Invalid mode: {self.mode!r}. Must be 'batch' or 'interactive'.")
        valid_roles = {e.value for e in EvalRole}
        if self.role not in valid_roles:
            raise ValueError(f"Invalid role: {self.role!r}. Must be one of {valid_roles}")

    @classmethod
    def from_tasks_json_entry(cls, entry: dict) -> "EvalTask":
        """Parse from tasks.json entry.

        Supports two formats:
        1. Real tasks.json: title + description + expected_lua + context
        2. New format: description + expected_behavior + reference_code
        """
        role = entry.get("role", "executor")

        if "title" in entry or "expected_lua" in entry or "context" in entry:
            # Real tasks.json format
            return cls(
                id=entry["id"],
                title=entry.get("title", ""),
                description=entry.get("description", ""),
                expected_behavior=f"Task: {entry.get('title', '')}",
                expected_lua=entry.get("expected_lua"),
                reference_code=None,  # эталон НЕ в prompt — только для проверки
                context=entry.get("context"),
                lint_rules=entry.get("lint_rules", ["luacheck"]),
                mode=entry.get("mode", "batch"),
                role=role,
                requires_clarification=entry.get("requires_clarification", False),
            )
        else:
            # New format
            return cls(
                id=entry["id"],
                description=entry["description"],
                expected_behavior=entry["expected_behavior"],
                reference_code=entry.get("reference_code"),
                lint_rules=entry.get("lint_rules", ["luacheck"]),
                mode=entry.get("mode", "batch"),
                role=role,
                requires_clarification=entry.get("requires_clarification", False),
            )


# ---------------------------------------------------------------------------
# ClarificationEvent
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class ClarificationEvent:
    """A clarification Q&A event from IronUser.

    Attributes:
        question: The clarification question asked.
        answer: The answer provided.
        round_num: Round number in the conversation.
    """

    question: str
    answer: str
    round_num: int = 1


# ---------------------------------------------------------------------------
# EvalResult
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class EvalResult:
    """Result of executing a single eval task.

    Attributes:
        task_id: Task identifier.
        role: Role that was tested ("executor", "strategist", "reviewer", "multi").
        status: Overall status ("passed", "failed", "partial").
        execution_time_ms: Total execution time in milliseconds.
        code_syntax_ok: Syntax check passed.
        code_lint_ok: Lint check passed.
        code_run_ok: Execution passed.
        code_lint_errors: List of lint errors.
        code_run_output: Output from execution.
        llm_score: LLM quality score (0-10).
        llm_feedback: LLM feedback text.
        tools_called: List of tools that were called.
        tool_call_count: Total number of tool calls.
        questions_asked: Number of clarification questions asked.
        questions_answered: Number of questions answered.
        ambiguity_detected: Whether ambiguity was detected.
        clarifications: List of clarification events.
        plan_tasks: List of planned tasks (strategist role).
        plan_risks: List of identified risks (strategist role).
        review_findings: List of review findings (reviewer role).
        review_verdict: Review verdict (reviewer role).
        clarification_qas: Clarification Q&A pairs (clarification role).
        error: Error message if any.
    """

    task_id: str
    status: str
    execution_time_ms: float
    role: str = "executor"
    code_syntax_ok: bool = True
    code_lint_ok: bool = True
    code_run_ok: bool = True
    code_lint_errors: list[str] = field(default_factory=list)
    code_run_output: str = ""
    llm_score: Optional[float] = None
    llm_feedback: Optional[str] = None
    tools_called: list[str] = field(default_factory=list)
    tool_call_count: int = 0
    questions_asked: int = 0
    questions_answered: int = 0
    ambiguity_detected: bool = False
    clarifications: list[ClarificationEvent] = field(default_factory=list)
    plan_tasks: list[dict] = field(default_factory=list)
    plan_risks: list[str] = field(default_factory=list)
    review_findings: list[dict] = field(default_factory=list)
    review_verdict: Optional[str] = None
    clarification_qas: list[dict] = field(default_factory=list)
    error: Optional[str] = None
    error_analysis: Optional[dict] = field(default=None)
    """LLM-анализ ошибки: root_cause, suggested_fix, confidence, error_category."""

    def __post_init__(self) -> None:
        """Validate fields."""
        valid_statuses = ("passed", "failed", "partial")
        if self.status not in valid_statuses:
            raise ValueError(f"Invalid status: {self.status!r}. Must be one of {valid_statuses}")

        if self.llm_score is not None and not (0.0 <= self.llm_score <= 10.0):
            raise ValueError(f"llm_score must be 0-10, got {self.llm_score}")


# ---------------------------------------------------------------------------
# EvalReport
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class EvalReport:
    """Aggregated report from multiple eval results.

    Attributes:
        total: Total number of tasks.
        passed: Number of passed tasks.
        failed: Number of failed tasks.
        partial: Number of partial tasks.
        total_tool_calls: Total number of tool calls across all tasks.
        avg_execution_time_ms: Average execution time.
        timestamp: Report generation timestamp.
        total_questions_asked: Total clarification questions asked.
        total_questions_answered: Total questions answered.
        tasks_with_ambiguity: Number of tasks with detected ambiguity.
        results: Individual EvalResult instances.
    """

    total: int = 0
    passed: int = 0
    failed: int = 0
    partial: int = 0
    total_tool_calls: int = 0
    avg_execution_time_ms: float = 0.0
    timestamp: str = ""
    total_questions_asked: int = 0
    total_questions_answered: int = 0
    tasks_with_ambiguity: int = 0
    results: list[EvalResult] = field(default_factory=list)

    @property
    def tasks(self) -> list[EvalResult]:
        """Alias for results (compatibility with report_generator)."""
        return self.results

    @classmethod
    def from_results(cls, results: list[EvalResult], timestamp: str = "") -> "EvalReport":
        """Create report from a list of EvalResult instances."""
        total = len(results)
        passed = sum(1 for r in results if r.status == "passed")
        failed = sum(1 for r in results if r.status == "failed")
        partial = sum(1 for r in results if r.status == "partial")
        total_tool_calls = sum(r.tool_call_count for r in results)
        avg_time = (sum(r.execution_time_ms for r in results) / total) if total > 0 else 0.0
        total_questions = sum(r.questions_asked for r in results)
        total_answered = sum(r.questions_answered for r in results)
        ambiguity_count = sum(1 for r in results if r.ambiguity_detected)

        return cls(
            total=total,
            passed=passed,
            failed=failed,
            partial=partial,
            total_tool_calls=total_tool_calls,
            avg_execution_time_ms=round(avg_time, 2),
            timestamp=timestamp,
            total_questions_asked=total_questions,
            total_questions_answered=total_answered,
            tasks_with_ambiguity=ambiguity_count,
            results=results,
        )


# ---------------------------------------------------------------------------
# FinalCodeResponse (Pydantic for structured output)
# ---------------------------------------------------------------------------


class FinalCodeResponse(BaseModel):
    """Final code response from LLM after tool-calling loop.

    This model is used to extract the FINAL code from the LLM after all
    tool calls are complete. Using structured output ensures we always
    get the actual code, not intermediate tool call responses.

    Attributes:
        code: The final generated code (Lua, Python, etc.).
        reasoning: Brief explanation of the solution approach.
        tools_used: List of tools that were used during the session.
        confidence: Confidence level in the solution (0.0-1.0).
    """

    reasoning: str = Field(
        description="Brief explanation (1-2 sentences) of the solution approach."
    )
    code: str = Field(
        description="The final generated code. Must be complete and ready to use."
    )
    tools_used: list[str] = Field(
        default_factory=list,
        description="List of tool names that were used during the session."
    )
    confidence: float = Field(
        default=1.0,
        description="Confidence level in the solution (0.0-1.0). Will be clamped."
    )

    @field_validator("confidence", mode="before")
    @classmethod
    def clamp_confidence(cls, v):
        """Clamp confidence to [0.0, 1.0] range — qwen ignores constraints."""
        if isinstance(v, (int, float)):
            return max(0.0, min(1.0, float(v)))
        return v


# ---------------------------------------------------------------------------
# FinalPlanResponse (strategist role)
# ---------------------------------------------------------------------------


class PlanTaskItem(BaseModel):
    """Single task in a plan (strategist output)."""

    task_id: str = Field(description="Unique task identifier")
    role: str = Field(description="Role responsible: executor, reviewer, devops, indexator")
    description: str = Field(description="Task description")
    target_files: list[str] = Field(
        default_factory=list,
        description="Files to modify or create"
    )
    dependencies: list[str] = Field(
        default_factory=list,
        description="Task IDs this task depends on"
    )
    priority: int = Field(
        default=1,
        ge=1,
        le=5,
        description="Priority (1=highest, 5=lowest)"
    )


class FinalPlanResponse(BaseModel):
    """Final plan response from LLM after strategist tool-calling loop.

    This model is used to extract the FINAL plan from the LLM after all
    tool calls (architecture analysis, impact analysis, etc.) are complete.

    Attributes:
        tasks: List of planned tasks with roles, files, dependencies.
        reasoning: Brief explanation of the planning approach.
        tools_used: List of tools that were used during the session.
        confidence: Confidence level in the plan (0.0-1.0).
        risks: Identified risks and concerns.
    """

    reasoning: str = Field(
        description="Brief explanation (2-3 sentences) of the planning approach."
    )
    tasks: list[PlanTaskItem] = Field(
        description="List of planned tasks with roles, files, dependencies."
    )
    tools_used: list[str] = Field(
        default_factory=list,
        description="List of tool names that were used during the session."
    )
    confidence: float = Field(
        default=1.0,
        ge=0.0,
        le=1.0,
        description="Confidence level in the plan (0.0-1.0)."
    )
    risks: list[str] = Field(
        default_factory=list,
        description="Identified risks and architectural concerns."
    )


# ---------------------------------------------------------------------------
# FinalReviewResponse (reviewer role)
# ---------------------------------------------------------------------------


class ReviewFinding(BaseModel):
    """Single finding from code review."""

    severity: str = Field(
        description="Severity: CRITICAL, WARNING, INFO"
    )
    category: str = Field(
        description="Category: type_safety, architecture, duplication, performance, security, test_coverage"
    )
    description: str = Field(description="Description of the issue")
    file_path: str = Field(default="", description="Affected file path")
    suggestion: str = Field(default="", description="Suggested fix")


class FinalReviewResponse(BaseModel):
    """Final review response from LLM after reviewer tool-calling loop.

    This model is used to extract the FINAL review from the LLM after all
    tool calls (static analysis, impact analysis, etc.) are complete.

    Attributes:
        findings: List of review findings with severity and category.
        verdict: Overall verdict: APPROVED, WARN, or BLOCKER.
        reasoning: Brief explanation of the review conclusions.
        tools_used: List of tools that were used during the session.
        confidence: Confidence level in the review (0.0-1.0).
    """

    reasoning: str = Field(
        description="Brief explanation (2-3 sentences) of the review conclusions."
    )
    findings: list[ReviewFinding] = Field(
        description="List of review findings with severity and category."
    )
    verdict: ReviewVerdict = Field(
        description="Overall verdict: APPROVED, WARN, or BLOCKER."
    )
    tools_used: list[str] = Field(
        default_factory=list,
        description="List of tool names that were used during the session."
    )
    confidence: float = Field(
        default=1.0,
        ge=0.0,
        le=1.0,
        description="Confidence level in the review (0.0-1.0)."
    )


# ---------------------------------------------------------------------------
# FinalClarificationResponse (iron user / clarification role)
# ---------------------------------------------------------------------------


class ClarificationQA(BaseModel):
    """Single clarification question and answer pair."""

    question: str = Field(description="The clarification question")
    answer: str = Field(description="The answer provided")
    category: str = Field(
        default="",
        description="Category: input_format, edge_cases, error_handling, expected_output"
    )


class FinalClarificationResponse(BaseModel):
    """Final clarification response from LLM after clarification tool-calling loop.

    This model is used to extract the FINAL clarification Q&A from the LLM.

    Attributes:
        qa_pairs: List of question-answer pairs.
        resolution: Summary of how ambiguity was resolved.
        tools_used: List of tools that were used during the session.
        confidence: Confidence level that ambiguity is resolved (0.0-1.0).
    """

    qa_pairs: list[ClarificationQA] = Field(
        description="List of question-answer pairs from clarification."
    )
    resolution: str = Field(
        description="Summary of how ambiguity was resolved."
    )
    tools_used: list[str] = Field(
        default_factory=list,
        description="List of tool names that were used during the session."
    )
    confidence: float = Field(
        default=1.0,
        ge=0.0,
        le=1.0,
        description="Confidence level that ambiguity is resolved (0.0-1.0)."
    )