AgentEvalTool/backend/agenteval/models.py

"""Shared Pydantic models for AgentEvalTool."""

from datetime import datetime
from enum import Enum
from typing import Any, Optional

from pydantic import BaseModel, Field, field_validator


class PlatformType(str, Enum):
    AI_DIGITAL_EMPLOYEE = "ai_digital_employee"
    AI_ASSISTANT = "ai_assistant"


class ChannelType(str, Enum):
    TUTU_API = "tutu-api"
    OPENCLAW = "openclaw"
    HTTP = "http"


class TargetStatus(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    ERROR = "error"


class CaseType(str, Enum):
    SINGLE = "single"
    MULTI_TURN = "multi_turn"
    DYNAMIC = "dynamic"


class EvalTarget(BaseModel):
    """Evaluation target (the agent being evaluated)."""

    id: Optional[str] = None
    name: str
    description: str = ""
    platform: PlatformType = PlatformType.AI_DIGITAL_EMPLOYEE
    channel_type: ChannelType = ChannelType.TUTU_API
    channel_config: dict[str, Any] = Field(default_factory=dict)
    status: TargetStatus = TargetStatus.ACTIVE
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None


class Expectation(BaseModel):
    """Expected behavior for a test case."""

    intent: Optional[str] = None
    keywords_include: list[str] = Field(default_factory=list)
    keywords_exclude: list[str] = Field(default_factory=list)
    response_time_max_ms: Optional[int] = None
    coherence_min_score: Optional[float] = None


class EvalRuleConfig(BaseModel):
    """Configuration for an evaluation rule."""

    type: str
    params: dict[str, Any] = Field(default_factory=dict)
    weight: float = 1.0  # used when rule_logic == "weighted"


class RuleLogic(str, Enum):
    """How to combine multiple rule results for a case."""

    ALL = "all"  # all rules must pass (default)
    ANY = "any"  # at least one rule must pass
    WEIGHTED = "weighted"  # weighted average score >= threshold


class Case(BaseModel):
    """A single evaluation case within a scenario."""

    id: str
    type: CaseType = CaseType.SINGLE
    messages: list[str] = Field(default_factory=list)
    prompt: Optional[str] = None
    turns: int = 3
    expectations: Expectation = Field(default_factory=Expectation)
    eval_rules: list[EvalRuleConfig] = Field(default_factory=list)
    rule_logic: RuleLogic = RuleLogic.ALL
    rule_pass_threshold: float = 0.6  # used when rule_logic == "weighted"

    @field_validator("messages")
    @classmethod
    def messages_not_empty(cls, v: list[str], info) -> list[str]:
        data = info.data
        case_type = data.get("type") if data else None
        if case_type and case_type != CaseType.DYNAMIC and not v:
            raise ValueError("messages must not be empty for non-dynamic cases")
        return v


class Scenario(BaseModel):
    """A collection of evaluation cases."""

    id: Optional[str] = None
    name: str
    description: str = ""
    tags: list[str] = Field(default_factory=list)
    cases: list[Case] = Field(default_factory=list)
    llm_config: Optional[dict[str, Any]] = None
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

    @field_validator("cases")
    @classmethod
    def cases_not_empty(cls, v: list[Case]) -> list[Case]:
        if not v:
            raise ValueError("scenario must contain at least one case")
        return v


class RunStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"


class EvalRun(BaseModel):
    """A single evaluation run."""

    id: Optional[str] = None
    target_id: str
    scenario_id: str
    status: RunStatus = RunStatus.PENDING
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    summary: Optional[dict[str, Any]] = None


class Turn(BaseModel):
    """A single turn in a conversation during evaluation."""

    id: Optional[str] = None
    run_id: str
    case_id: str
    round_index: int
    sent_message: dict[str, Any] = Field(default_factory=dict)
    sent_at: Optional[datetime] = None
    question_msg_id: Optional[str] = None
    reply: Optional[dict[str, Any]] = None
    received_at: Optional[datetime] = None
    latency_ms: Optional[int] = None


class EvalResult(BaseModel):
    """Result of applying one evaluation rule to one turn."""

    id: Optional[str] = None
    run_id: str
    case_id: str
    turn_id: str
    rule_type: str
    passed: bool
    score: Optional[float] = None
    reason: str = ""