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

"""Fluency assessment rule using LLM to evaluate conversation naturalness."""

import json

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


@register_rule
class FluencyRule(EvalRule):
    """Use LLM to evaluate conversation fluency and naturalness.

    Evaluates:
    - Naturalness: Does the conversation flow naturally?
    - Repetition: Are there unnecessary repetitions?
    - Coherence: Is the conversation coherent and logical?

    Configuration params:
        min_score: Minimum fluency score to pass (0-10, default: 7)
        criteria: Custom evaluation criteria (optional)
    """

    name = "fluency"

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

        # Build conversation context
        conversation_parts = []
        for turn in dialog:
            # Extract sent message
            sent_text = ""
            if turn.sent_message:
                sent_text = self._extract_message_text(turn.sent_message)

            # Extract reply
            reply_text = extract_reply_text(turn.reply) if turn.reply else ""

            if sent_text:
                conversation_parts.append(f"用户: {sent_text}")
            if reply_text:
                conversation_parts.append(f"助手: {reply_text}")

        if not conversation_parts:
            return RuleResult(passed=False, reason="无法提取对话内容")

        conversation_text = "\n".join(conversation_parts)

        # Get evaluation criteria
        min_score = float(self.params.get("min_score", 7))
        custom_criteria = self.params.get("criteria", "")

        # Call LLM for fluency assessment
        if self.model_config and self.gateway:
            score, reason = await self._evaluate_fluency(conversation_text, custom_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="流畅度评估规则未绑定评估模型")
            score, reason = await self._call_llm(api_url, api_key, model, conversation_text, custom_criteria)

        if score is None:
            return RuleResult(passed=False, reason=f"流畅度评估失败: {reason}")

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

    def _extract_message_text(self, sent_message: dict) -> str:
        """Extract text from sent_message dict."""
        body = sent_message.get("msgBody", "")
        if isinstance(body, dict):
            return body.get("content", "")
        try:
            return json.loads(body).get("content", "")
        except Exception:
            return str(body)

    async def _evaluate_fluency(self, conversation: str, custom_criteria: str) -> tuple[float | None, str]:
        """Evaluate conversation fluency using the gateway."""
        system_prompt, user_prompt = self._build_prompts(conversation, custom_criteria)
        try:
            content = await self.gateway.chat(
                self.model_config,
                [
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt},
                ],
                temperature=0.2,
            )
            return self._parse_score(content)
        except Exception as exc:
            return None, str(exc)

    async def _call_llm(
        self,
        api_url: str,
        api_key: str | None,
        model: str,
        conversation: str,
        custom_criteria: str,
    ) -> tuple[float | None, str]:
        """Call LLM API directly for fluency assessment."""
        import httpx

        system_prompt, user_prompt = self._build_prompts(conversation, custom_criteria)

        headers = {"Content-Type": "application/json"}
        if api_key:
            headers["Authorization"] = f"Bearer {api_key}"

        payload = {
            "model": model,
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt},
            ],
            "temperature": 0.2,
        }

        try:
            async with httpx.AsyncClient(timeout=60) as client:
                resp = await client.post(api_url, headers=headers, json=payload)
            resp.raise_for_status()
            from agenteval.utils.llm import extract_content_from_llm_response

            content = extract_content_from_llm_response(resp.json())
            if not content:
                return None, "LLM 返回内容为空"
            return self._parse_score(content)
        except Exception as exc:
            return None, str(exc)

    @staticmethod
    def _build_prompts(conversation: str, custom_criteria: str) -> tuple[str, str]:
        """Build system and user prompts for fluency evaluation."""
        default_criteria = """评估对话的流畅度和自然性，考虑以下方面：
1. 自然性：对话是否流畅自然，像真人对话？
2. 重复性：是否有不必要的重复或冗余？
3. 连贯性：对话是否逻辑连贯，上下文一致？
4. 响应质量：助手的回复是否恰当、有帮助？"""

        criteria = custom_criteria if custom_criteria else default_criteria

        system_prompt = (
            "你是一位对话质量评估专家。请评估以下对话的流畅度和自然性。\n"
            f"评估标准：{criteria}\n"
            "打分范围：0-10分（10分最高）\n"
            '只输出一个 JSON 对象：{"score": number, "reason": "简短说明"}'
        )
        user_prompt = f"对话内容：\n{conversation}"
        return system_prompt, user_prompt

    @staticmethod
    def _parse_score(content: str) -> tuple[float, str]:
        """Parse LLM response to extract score and reason."""
        from agenteval.utils.llm import parse_json_from_llm_text

        parsed = parse_json_from_llm_text(content)
        score = float(parsed["score"])
        return max(0.0, min(10.0, score)), parsed.get("reason", "")