ai-video-fullstack/backend/services/test_runs/evaluator.py

"""Deterministic and LLM-backed assertions for text test results."""

from __future__ import annotations

import json
import re
from dataclasses import dataclass
from typing import Any

import httpx
from models import RuntimeModelResource

from services.test_runs.errors import TestExecutionError
from services.test_runs.text_runner import RawToolCall, RawTurnResult, TextCaseResult
from test_schemas import (
    BatchEvaluationResult,
    BatchTurnResult,
    FixedInputTurn,
    OverallCriterion,
    ReplyExpectedBehavior,
    TestCaseDefinition,
    ToolCallExpectedBehavior,
    ToolParamAssertion,
)


@dataclass(frozen=True)
class EvaluatedCase:
    turns: list[BatchTurnResult]
    overall_criteria: list[BatchEvaluationResult]

    @property
    def passed(self) -> bool:
        evaluations = [
            item
            for turn in self.turns
            for item in turn.evaluations
        ] + self.overall_criteria
        return all(item.status == "pass" for item in evaluations)


class LLMJudge:
    """Small OpenAI-compatible JSON judge using a dedicated model resource."""

    def __init__(
        self,
        resource: RuntimeModelResource,
        *,
        timeout_seconds: float = 60,
    ):
        self._resource = resource
        self._timeout_seconds = timeout_seconds

    async def evaluate(
        self,
        *,
        criteria: str,
        subject: str,
        context: str,
    ) -> tuple[bool, str, str]:
        values = self._resource.values or {}
        secrets = self._resource.secrets or {}
        base_url = str(values.get("apiUrl") or "").rstrip("/")
        api_key = str(secrets.get("apiKey") or "")
        model = str(values.get("modelId") or "")
        if not base_url or not api_key or not model:
            raise TestExecutionError(
                code="EVALUATOR_NOT_CONFIGURED",
                message="LLM 评估所需的模型资源未完整配置",
                stage="evaluation",
            )
        prompt = (
            "你是严格的自动化测试评估器。只根据提供的内容判断标准是否满足。\n"
            "必须只输出一个 JSON 对象，不要输出 Markdown：\n"
            '{"passed":true或false,"reason":"简短中文理由","actual":"实际表现摘要"}\n\n'
            f"评估对象：{subject}\n"
            f"评估标准：{criteria}\n\n"
            f"待评估内容：\n{context}"
        )
        payload: dict[str, Any] = {
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0,
        }
        extra_body = values.get("extraBody")
        if isinstance(extra_body, dict):
            payload.update(
                {
                    key: value
                    for key, value in extra_body.items()
                    if key not in {"model", "messages"}
                }
            )
        try:
            async with httpx.AsyncClient(timeout=self._timeout_seconds) as client:
                response = await client.post(
                    f"{base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {api_key}",
                        "Content-Type": "application/json",
                    },
                    json=payload,
                )
            response.raise_for_status()
            body = response.json()
            content = body["choices"][0]["message"]["content"]
            if isinstance(content, list):
                content = "".join(
                    str(part.get("text") or "")
                    for part in content
                    if isinstance(part, dict)
                )
            parsed = self._parse_json(str(content or ""))
            return (
                bool(parsed.get("passed")),
                str(parsed.get("reason") or "LLM 未提供判断理由"),
                str(parsed.get("actual") or ""),
            )
        except TestExecutionError:
            raise
        except (httpx.HTTPError, KeyError, IndexError, TypeError, ValueError) as exc:
            raise TestExecutionError(
                code="EVALUATOR_REQUEST_FAILED",
                message=f"LLM 评估失败: {exc}",
                stage="evaluation",
                retryable=True,
            ) from exc

    @staticmethod
    def _parse_json(value: str) -> dict[str, Any]:
        text = value.strip()
        if text.startswith("```"):
            text = re.sub(r"^```(?:json)?\s*", "", text)
            text = re.sub(r"\s*```$", "", text)
        start = text.find("{")
        end = text.rfind("}")
        if start < 0 or end < start:
            raise ValueError("评估模型没有返回 JSON")
        parsed = json.loads(text[start : end + 1])
        if (
            not isinstance(parsed, dict)
            or not isinstance(parsed.get("passed"), bool)
        ):
            raise ValueError("评估模型返回格式不正确")
        return parsed


class EvaluationEngine:
    def __init__(self, judge_resource: RuntimeModelResource):
        self._judge = LLMJudge(judge_resource)

    async def evaluate(
        self,
        definition: TestCaseDefinition,
        execution: TextCaseResult,
    ) -> EvaluatedCase:
        raw_by_id = {turn.id: turn for turn in execution.turns}
        turns: list[BatchTurnResult] = []
        for index, expected_turn in enumerate(definition.turns):
            raw = raw_by_id.get(expected_turn.id) or RawTurnResult(
                id=expected_turn.id,
                index=index,
                user_input=expected_turn.user_input,
                image=expected_turn.image,
                assistant_reply="",
                tool_calls=[],
                runtime_records=[],
            )
            evaluations = [
                await self._evaluate_behavior(behavior, raw)
                for behavior in expected_turn.behaviors
            ]
            turns.append(
                BatchTurnResult(
                    id=raw.id,
                    index=index,
                    user_input=raw.user_input,
                    image=raw.image,
                    assistant_reply=raw.assistant_reply,
                    tool_calls=[item.as_result() for item in raw.tool_calls],
                    runtime_records=list(raw.runtime_records),
                    evaluations=evaluations,
                )
            )

        transcript = self._format_transcript(execution, definition)
        overall = [
            await self._evaluate_overall(criterion, transcript)
            for criterion in definition.overall_criteria
        ]
        return EvaluatedCase(turns=turns, overall_criteria=overall)

    async def _evaluate_behavior(
        self,
        behavior: ReplyExpectedBehavior | ToolCallExpectedBehavior,
        turn: RawTurnResult,
    ) -> BatchEvaluationResult:
        if behavior.type == "reply":
            return await self._evaluate_reply(behavior, turn.assistant_reply)
        return await self._evaluate_tool(behavior, turn.tool_calls)

    async def _evaluate_reply(
        self,
        behavior: ReplyExpectedBehavior,
        reply: str,
    ) -> BatchEvaluationResult:
        if behavior.assertion_type == "llm":
            passed, reason, actual = await self._judge.evaluate(
                criteria=behavior.llm_criteria,
                subject="当前轮助手回复",
                context=reply or "（助手没有生成文字回复）",
            )
            return BatchEvaluationResult(
                id=behavior.id,
                label="回复 · LLM 判断",
                kind="reply",
                status="pass" if passed else "fail",
                expected=behavior.llm_criteria,
                actual=actual or reply,
                reason="" if passed else reason,
            )

        normalized_reply = reply.casefold()
        matches = [keyword.casefold() in normalized_reply for keyword in behavior.keywords]
        matched = all(matches) if behavior.keyword_match_mode == "all" else any(matches)
        passed = not matched if behavior.negate_keywords else matched
        relation = (
            "不应包含"
            if behavior.negate_keywords
            else "应包含全部"
            if behavior.keyword_match_mode == "all"
            else "应至少包含其一"
        )
        return BatchEvaluationResult(
            id=behavior.id,
            label="回复 · 关键词",
            kind="reply",
            status="pass" if passed else "fail",
            expected=f"{relation}：{'、'.join(behavior.keywords)}",
            actual=reply or "（无文字回复）",
            reason="" if passed else "实际回复未满足关键词规则。",
        )

    async def _evaluate_tool(
        self,
        behavior: ToolCallExpectedBehavior,
        calls: list[RawToolCall],
    ) -> BatchEvaluationResult:
        matching_calls = [
            call for call in calls if call.function_name == behavior.function_name
        ]
        count = len(matching_calls)
        if behavior.expectation == "not_called":
            passed = count == 0
            return BatchEvaluationResult(
                id=behavior.id,
                label=f"工具 · {behavior.function_name}",
                kind="tool_call",
                status="pass" if passed else "fail",
                expected=f"不应调用 {behavior.function_name}",
                actual=f"实际调用 {count} 次 {behavior.function_name}",
                reason="" if passed else "检测到本轮不应发生的工具调用。",
            )

        count_passed = count >= behavior.min_calls and (
            behavior.max_calls is None or count <= behavior.max_calls
        )
        params_passed = True
        parameter_reason = ""
        if count_passed and behavior.param_assertions:
            params_passed = False
            reasons: list[str] = []
            for call in matching_calls:
                call_passed, call_reason = await self._call_matches_parameters(
                    call, behavior.param_assertions
                )
                if call_passed:
                    params_passed = True
                    break
                reasons.append(call_reason)
            parameter_reason = "；".join(item for item in reasons if item)

        passed = count_passed and params_passed
        count_text = (
            f"{behavior.min_calls} 次以上"
            if behavior.max_calls is None
            else f"{behavior.min_calls} 次"
            if behavior.min_calls == behavior.max_calls
            else f"{behavior.min_calls}–{behavior.max_calls} 次"
        )
        reason = ""
        if not count_passed:
            reason = "工具调用次数不符合要求。"
        elif not params_passed:
            reason = parameter_reason or "没有一次工具调用满足全部参数断言。"
        return BatchEvaluationResult(
            id=behavior.id,
            label=f"工具 · {behavior.function_name}",
            kind="tool_call",
            status="pass" if passed else "fail",
            expected=f"应调用 {behavior.function_name} {count_text}",
            actual=f"实际调用 {count} 次 {behavior.function_name}",
            reason=reason,
        )

    async def _call_matches_parameters(
        self,
        call: RawToolCall,
        assertions: list[ToolParamAssertion],
    ) -> tuple[bool, str]:
        arguments = call.arguments if isinstance(call.arguments, dict) else {}
        for assertion in assertions:
            if assertion.name not in arguments:
                return False, f"缺少参数 {assertion.name}"
            actual = arguments[assertion.name]
            if assertion.match_mode == "exact":
                try:
                    expected: Any = json.loads(assertion.value)
                except json.JSONDecodeError:
                    expected = assertion.value
                if actual != expected and str(actual) != assertion.value:
                    return False, f"参数 {assertion.name} 与期望值不一致"
            elif assertion.match_mode == "regex":
                actual_text = (
                    actual
                    if isinstance(actual, str)
                    else json.dumps(actual, ensure_ascii=False, default=str)
                )
                if re.search(assertion.value, actual_text) is None:
                    return False, f"参数 {assertion.name} 未匹配正则表达式"
            else:
                passed, reason, _actual = await self._judge.evaluate(
                    criteria=assertion.value,
                    subject=f"工具参数 {assertion.name}",
                    context=json.dumps(actual, ensure_ascii=False, default=str),
                )
                if not passed:
                    return False, reason
        return True, ""

    async def _evaluate_overall(
        self,
        criterion: OverallCriterion,
        transcript: str,
    ) -> BatchEvaluationResult:
        passed, reason, actual = await self._judge.evaluate(
            criteria=criterion.criteria,
            subject="完整测试对话",
            context=transcript,
        )
        return BatchEvaluationResult(
            id=criterion.id,
            label=criterion.name,
            kind="overall",
            status="pass" if passed else "fail",
            expected=criterion.criteria,
            actual=actual or "已评估完整对话",
            reason="" if passed else reason,
        )

    @staticmethod
    def _format_transcript(
        execution: TextCaseResult,
        definition: TestCaseDefinition,
    ) -> str:
        if execution.transcript:
            labels = {"user": "User", "assistant": "Agent"}
            return "\n".join(
                f"{labels.get(item['role'], item['role'])}: {item['content']}"
                for item in execution.transcript
            )
        raw_by_id = {turn.id: turn for turn in execution.turns}
        lines: list[str] = []
        for turn in definition.turns:
            raw = raw_by_id.get(turn.id)
            lines.append(f"User: {turn.user_input}")
            lines.append(f"Agent: {raw.assistant_reply if raw else ''}")
        return "\n".join(lines)