AgentEvalTool/backend/agenteval/evaluation/rules/llm_score.py

"""LLM-based scoring evaluation rule."""

import asyncio
import json

from agenteval.evaluation.rules.base import EvalRule, RuleResult, register_rule
from agenteval.evaluation.rules.scored_llm import call_scored_llm, parse_scored_content
from agenteval.models import Case, Turn
from agenteval.utils.llm import extract_reply_text


@register_rule
class LlmScoreRule(EvalRule):
    """Use an external LLM to score reply quality against criteria."""

    name = "llm_score"

    async def evaluate(self, case: Case, dialog: list[Turn]) -> RuleResult:
        if not dialog:
            return RuleResult(passed=False, reason="无回复记录")

        last_turn = dialog[-1]
        reply_text = extract_reply_text(last_turn.reply)

        question_text = ""
        if last_turn.sent_message:
            body = last_turn.sent_message.get("msgBody", "")
            if isinstance(body, dict):
                question_text = body.get("content", "")
            else:
                try:
                    question_text = json.loads(body).get("content", "")
                except Exception:
                    question_text = str(body)

        dimensions = self.params.get("dimensions")
        if dimensions:
            return await self._evaluate_dimensions(question_text, reply_text, dimensions)

        criteria = self.params.get("criteria", "")
        min_score = float(self.params.get("min_score", 7))
        if self.model_config and self.gateway:
            score, reason = await self._call_gateway(question_text, reply_text, criteria)
        else:
            api_url = self.params.get("api_url")
            api_key = self.params.get("api_key")
            model = self.params.get("model", "gpt-4o-mini")
            if not api_url:
                return RuleResult(passed=False, reason="LLM 评分规则未绑定评估模型（兼容配置缺少 api_url）")
            score, reason = await self._call_llm(api_url, api_key, model, question_text, reply_text, criteria)
        if score is None:
            return RuleResult(passed=False, reason=f"LLM 评分失败: {reason}")

        passed = score >= min_score
        verdict = "通过" if passed else "未通过"
        detail = f"；{reason}" if reason else ""
        return RuleResult(
            passed=passed,
            score=score / 10.0,
            reason=f"LLM 评分 {score}/10，{verdict} (阈值 {min_score}){detail}",
        )

    _MAX_CONCURRENT_DIMENSIONS = 5

    async def _evaluate_dimensions(
        self, question: str, reply: str, dimensions: list[dict]
    ) -> RuleResult:
        semaphore = asyncio.Semaphore(self._MAX_CONCURRENT_DIMENSIONS)

        async def bounded(dim: dict) -> tuple[float | None, str]:
            async with semaphore:
                return await self._evaluate_one_dimension(question, reply, dim)

        results = await asyncio.gather(*(bounded(dim) for dim in dimensions))

        dimension_scores = {}
        dimension_reasons = []
        all_passed = True

        for dim, (score, reason) in zip(dimensions, results):
            dim_name = dim.get("name", "unknown")
            min_score = float(dim.get("min_score", 7))
            if score is None:
                all_passed = False
                dimension_scores[dim_name] = None
                dimension_reasons.append(f"{dim_name}: 评分失败 ({reason})")
            else:
                passed = score >= min_score
                if not passed:
                    all_passed = False
                dimension_scores[dim_name] = score
                dimension_reasons.append(f"{dim_name}: {score}/10")

        valid_scores = [s for s in dimension_scores.values() if s is not None]
        avg_score = sum(valid_scores) / len(valid_scores) if valid_scores else 0

        verdict = "通过" if all_passed else "未通过"
        reasons_str = "，".join(dimension_reasons)
        return RuleResult(
            passed=all_passed,
            score=avg_score / 10.0,
            reason=f"多维度 LLM 评分 {avg_score:.1f}/10，{verdict}；{reasons_str}",
            details={"dimensions": dimension_scores},
        )

    async def _evaluate_one_dimension(
        self, question: str, reply: str, dimension: dict
    ) -> tuple[float | None, str]:
        criteria = dimension.get("criteria", "")
        if self.model_config and self.gateway:
            return await self._call_gateway(question, reply, criteria)
        api_url = self.params.get("api_url")
        api_key = self.params.get("api_key")
        model = self.params.get("model", "gpt-4o-mini")
        if not api_url:
            return None, "未绑定评估模型"
        return await self._call_llm(api_url, api_key, model, question, reply, criteria)

    async def _call_gateway(self, question: str, reply: str, criteria: str) -> tuple[float | None, str]:
        system_prompt, user_prompt = self._prompts(question, reply, criteria)
        try:
            content, usage = await self.gateway.chat_with_usage(
                self.model_config,
                [
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt},
                ],
                temperature=0.2,
            )
            self._record_llm_usage(usage)
            return parse_scored_content(content)
        except Exception as exc:
            return None, str(exc)

    @staticmethod
    def _prompts(question: str, reply: str, criteria: str) -> tuple[str, str]:
        system_prompt = (
            "你是一位严格的智能客服质量评估专家。请根据用户问题和智能体回复，"
            f"按照以下标准打分（0-10分，10分最高）：{criteria}\n"
            '只输出一个 JSON 对象：{"score": number, "reason": "简短说明"}'
        )
        return system_prompt, f"用户问题：{question}\n智能体回复：{reply}"

    async def _call_llm(
        self,
        api_url: str,
        api_key: str | None,
        model: str,
        question: str,
        reply: str,
        criteria: str,
    ) -> tuple[float | None, str]:
        """Call the configured LLM API and parse a numeric score between 0 and 10."""
        system_prompt, user_prompt = self._prompts(question, reply, criteria)
        return await call_scored_llm(api_url, api_key, model, system_prompt, user_prompt)