Merge pull request 'feat: 用户体验指标（放弃率+流畅度评估）' (#32) from feat/user-experience-metrics into main

backend/agenteval/evaluation/rules/__init__.py

@@ -2,6 +2,7 @@
 
 # Import all rules to populate the registry.
 from agenteval.evaluation.rules.base import EvalRule, RuleResult, get_rule, list_rule_types, register_rule
+from agenteval.evaluation.rules.fluency import FluencyRule
 from agenteval.evaluation.rules.json_schema import JsonSchemaRule
 from agenteval.evaluation.rules.keyword import KeywordMatchRule
 from agenteval.evaluation.rules.llm_score import LlmScoreRule
@@ -15,6 +16,7 @@ __all__ = [
     "get_rule",
     "list_rule_types",
     "register_rule",
+    "FluencyRule",
     "JsonSchemaRule",
     "KeywordMatchRule",
     "LlmScoreRule",

backend/agenteval/evaluation/rules/fluency.py

@@ -0,0 +1,169 @@
+"""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", "")

backend/agenteval/models.py

@@ -184,12 +184,15 @@ class RunSummary(BaseModel):
     total_cases: int = 0
     passed_cases: int = 0
     failed_cases: int = 0
+    abandoned_cases: int = 0  # Cases abandoned by user before completion
     total_rules: int = 0
     passed_rules: int = 0
     # 用例级通过率，含执行失败（ADR-0002）；失败/取消的 run 无此值
     pass_rate: Optional[float] = None
     # 判定型通过率：连通用例从分子分母双双剔除；无判定型用例时为空
     judged_pass_rate: Optional[float] = None
+    # 用户放弃率：abandoned_cases / total_cases
+    abandonment_rate: Optional[float] = None
     avg_latency_ms: Optional[float] = None
     case_outcomes: dict[str, CaseOutcomeSummary] = Field(default_factory=dict)
     case_errors: list[dict[str, str]] = Field(default_factory=list)

tests/unit/test_user_experience.py

@@ -0,0 +1,110 @@
+"""Tests for fluency assessment rule and abandonment tracking."""
+
+import pytest
+
+from agenteval.evaluation.rules.base import get_rule
+from agenteval.models import Case, RunSummary, Turn
+
+
+def test_run_summary_abandonment_fields():
+    """RunSummary should have abandonment tracking fields."""
+    summary = RunSummary(
+        total_cases=10,
+        passed_cases=7,
+        failed_cases=2,
+        abandoned_cases=1,
+    )
+    assert summary.total_cases == 10
+    assert summary.abandoned_cases == 1
+    assert summary.abandonment_rate is None  # Not calculated yet
+
+
+def test_run_summary_abandonment_rate_calculation():
+    """Abandonment rate should be calculable from summary fields."""
+    summary = RunSummary(
+        total_cases=10,
+        abandoned_cases=2,
+    )
+    # Calculate abandonment rate
+    if summary.total_cases > 0:
+        rate = summary.abandoned_cases / summary.total_cases
+        assert rate == 0.2
+
+
+def test_fluency_rule_registered():
+    """Fluency rule should be registered in the rule registry."""
+    from agenteval.evaluation.rules import list_rule_types
+
+    assert "fluency" in list_rule_types()
+
+
+def test_fluency_rule_config():
+    """Fluency rule should accept configuration parameters."""
+    rule = get_rule(
+        "fluency",
+        {"min_score": 8, "criteria": "评估对话是否自然流畅"},
+    )
+    assert rule.params["min_score"] == 8
+    assert rule.params["criteria"] == "评估对话是否自然流畅"
+
+
+def test_fluency_rule_empty_dialog():
+    """Fluency rule should fail on empty dialog."""
+    rule = get_rule("fluency", {"min_score": 7})
+    case = Case(id="c1", messages=["hello"])
+    import asyncio
+
+    result = asyncio.run(rule.evaluate(case, []))
+    assert result.passed is False
+    assert "无回复记录" in result.reason
+
+
+@pytest.mark.asyncio
+async def test_fluency_rule_no_model_config():
+    """Fluency rule should fail without model configuration."""
+    rule = get_rule("fluency", {"min_score": 7})
+    case = Case(id="c1", messages=["hello"])
+    dialog = [
+        Turn(
+            id="t1",
+            run_id="r1",
+            case_id="c1",
+            round_index=1,
+            sent_message={"msgBody": "你好"},
+            reply={"msgBody": "您好，有什么可以帮助您的？"},
+        )
+    ]
+    result = await rule.evaluate(case, dialog)
+    assert result.passed is False
+    assert "未绑定评估模型" in result.reason
+
+
+def test_fluency_rule_default_min_score():
+    """Fluency rule should have default min_score of 7."""
+    rule = get_rule("fluency", {})
+    assert rule.params.get("min_score", 7) == 7
+
+
+def test_abandonment_rate_zero_total():
+    """Abandonment rate should handle zero total cases."""
+    summary = RunSummary(total_cases=0, abandoned_cases=0)
+    # Should not raise division by zero
+    if summary.total_cases > 0:
+        rate = summary.abandoned_cases / summary.total_cases
+    else:
+        rate = None
+    assert rate is None
+
+
+def test_abandonment_rate_all_abandoned():
+    """Abandonment rate should be 1.0 when all cases are abandoned."""
+    summary = RunSummary(total_cases=5, abandoned_cases=5)
+    rate = summary.abandoned_cases / summary.total_cases
+    assert rate == 1.0
+
+
+def test_abandonment_rate_none_abandoned():
+    """Abandonment rate should be 0.0 when no cases are abandoned."""
+    summary = RunSummary(total_cases=5, abandoned_cases=0)
+    rate = summary.abandoned_cases / summary.total_cases
+    assert rate == 0.0