solahqb/AgentEvalTool
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

v0.3-s2: 3 个新规则 + 组合逻辑 + 33 个测试

Browse Source 
## 新规则（共 6 种，增加 3 种）

### semantic_similarity
- 调用 OpenAI 兼容 embedding API（asyncio.gather 并发两路请求）
- 余弦相似度与 reference 比对，min_score 可配置（默认 0.7）
- API 异常时明确返回失败原因，不隐藏错误

### json_schema
- 验证回复是否为合法 JSON（支持 markdown 代码块剥离）
- required_keys / forbidden_keys / key_types 三维校验
- dot-path 支持嵌套字段（"data.id"）
- strict_json=false 模式非阻断校验

### safety
- 双层检测：关键词黑名单（零延迟）+ 可选 moderation API
- API 不可用时自动降级黑名单，不中止评测
- 支持自定义 flagged_categories

## 规则组合逻辑（rule_logic + rule_pass_threshold）

- models.py: EvalRuleConfig 增加 weight 字段；Case 增加 rule_logic / rule_pass_threshold
- models.py: 新增 RuleLogic 枚举（all / any / weighted）
- engine._save_rule_results: 按 rule_logic 决定 case 通过/失败
  - ALL：全部通过才通过（原有行为，向下兼容）
  - ANY：至少一条通过即通过
  - WEIGHTED：加权平均分 >= rule_pass_threshold

## 测试（43 → 76，新增 33）
- test_s2_rules_and_logic.py：3 个新规则的 pass/fail/边界/API 降级 + 5 个组合逻辑集成测试

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
sinohqb 2026-07-17 11:23:22 +08:00
parent 12481cd1b8
commit c7f1dca49d
8 changed files with 772 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
backend/agenteval/evaluation/engine.py
View File

@ -15,7 +15,17 @@ import httpx
from agenteval.channels.base import EvalChannel
from agenteval.channels.factory import ChannelFactory
from agenteval.evaluation.rules import get_rule
from agenteval.models import Case, CaseType, EvalResult, EvalRun, EvalTarget, RunStatus, Scenario, Turn
from agenteval.models import (
Case,
CaseType,
EvalResult,
EvalRun,
EvalTarget,
RuleLogic,
RunStatus,
Scenario,
Turn,
)
from agenteval.storage.db import get_session, utc_now
from agenteval.storage.repository import ResultRepository, RunRepository
from agenteval.utils.llm import extract_content_from_llm_response, extract_reply_text, parse_json_from_llm_text
@ -346,7 +356,13 @@ class EvalEngine:
dialog: list[Turn],
progress_callback: Optional[ProgressCallback],
) -> tuple[bool, int, int]:
"""Apply rules and save results; returns (all_passed, passed_count, total_count)."""
"""Apply rules and save results; returns (case_passed, passed_count, total_count).
Combination logic (case.rule_logic):
ALL — all rules must pass (default)
ANY — at least one rule must pass
WEIGHTED — weighted average score >= case.rule_pass_threshold
"""
from agenteval.models import EvalRuleConfig
rules_config: list[EvalRuleConfig] = list(case.eval_rules)
@ -373,9 +389,15 @@ class EvalEngine:
)
)
all_passed = True
if not rules_config:
# No rules defined and no expectations → case passes with no checks
return True, 0, 0
passed_count = 0
total_count = 0
weighted_score = 0.0
total_weight = 0.0
for rule_config in rules_config:
rule = get_rule(rule_config.type, rule_config.params)
result = await rule.evaluate(case, dialog)
@ -393,8 +415,13 @@ class EvalEngine:
total_count += 1
if result.passed:
passed_count += 1
else:
all_passed = False
# Weighted scoring: use rule score (default 1.0 if passed, 0.0 if failed)
score_val = result.score if result.score is not None else (1.0 if result.passed else 0.0)
weight = rule_config.weight
weighted_score += score_val * weight
total_weight += weight
await self._emit(
progress_callback,
"rule_result",
@ -405,10 +432,23 @@ class EvalEngine:
"passed": result.passed,
"score": result.score,
"reason": result.reason,
"weight": weight,
},
)
return all_passed, passed_count, total_count
# Determine case pass/fail based on rule_logic
logic = case.rule_logic
if logic == RuleLogic.ALL:
case_passed = passed_count == total_count
elif logic == RuleLogic.ANY:
case_passed = passed_count > 0
elif logic == RuleLogic.WEIGHTED:
avg = weighted_score / total_weight if total_weight > 0 else 0.0
case_passed = avg >= case.rule_pass_threshold
else:
case_passed = passed_count == total_count
return case_passed, passed_count, total_count
async def _generate_messages(
self,

6
backend/agenteval/evaluation/rules/__init__.py
View File

@ -2,9 +2,12 @@
# 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.json_schema import JsonSchemaRule
from agenteval.evaluation.rules.keyword import KeywordMatchRule
from agenteval.evaluation.rules.llm_score import LlmScoreRule
from agenteval.evaluation.rules.response_time import ResponseTimeRule
from agenteval.evaluation.rules.safety import SafetyRule
from agenteval.evaluation.rules.semantic import SemanticSimilarityRule
__all__ = [
"EvalRule",
@ -12,7 +15,10 @@ __all__ = [
"get_rule",
"list_rule_types",
"register_rule",
"JsonSchemaRule",
"KeywordMatchRule",
"LlmScoreRule",
"ResponseTimeRule",
"SafetyRule",
"SemanticSimilarityRule",
]

120
backend/agenteval/evaluation/rules/json_schema.py Normal file
View File

@ -0,0 +1,120 @@
"""JSON schema validation evaluation rule.
Validates that the agent reply is valid JSON and optionally conforms to
a specified structural schema (key presence, types, nested paths).
Configuration params:
required_keys list of dot-path keys that must exist (e.g. ["data.id", "status"])
forbidden_keys list of dot-path keys that must NOT exist
key_types dict mapping dot-path key → expected type name
("str", "int", "float", "bool", "list", "dict", "null")
e.g. {"status": "str", "count": "int"}
strict_json if true (default), fail if reply is not parseable JSON
All params are optional; with no params the rule only checks valid JSON.
"""
import json
from typing import Any
from agenteval.evaluation.rules.base import EvalRule, RuleResult, register_rule
from agenteval.models import Case, Turn
from agenteval.utils.llm import extract_reply_text
_TYPE_MAP: dict[str, type] = {
"str": str,
"int": int,
"float": float,
"bool": bool,
"list": list,
"dict": dict,
"null": type(None),
}
def _get_path(data: Any, path: str) -> tuple[bool, Any]:
"""Return (found, value) for a dot-separated path."""
parts = path.split(".")
current = data
for part in parts:
if isinstance(current, dict):
if part not in current:
return False, None
current = current[part]
elif isinstance(current, list) and part.isdigit():
idx = int(part)
if idx >= len(current):
return False, None
current = current[idx]
else:
return False, None
return True, current
@register_rule
class JsonSchemaRule(EvalRule):
"""Validate that the reply is valid JSON and matches a structural schema."""
name = "json_schema"
async def evaluate(self, case: Case, dialog: list[Turn]) -> RuleResult:
if not dialog:
return RuleResult(passed=False, reason="无回复记录")
reply_text = extract_reply_text(dialog[-1].reply).strip()
strict_json: bool = self.params.get("strict_json", True)
required_keys: list[str] = self.params.get("required_keys", [])
forbidden_keys: list[str] = self.params.get("forbidden_keys", [])
key_types: dict[str, str] = self.params.get("key_types", {})
# Try to find JSON in the reply (may be wrapped in markdown code block)
data: Any = None
try:
# Strip markdown code fences if present
text = reply_text
if text.startswith("```"):
lines = text.split("\n")
text = "\n".join(lines[1:-1] if lines[-1].strip() == "```" else lines[1:])
data = json.loads(text)
except (json.JSONDecodeError, ValueError):
if strict_json:
return RuleResult(passed=False, reason="回复不是合法 JSON")
# Non-strict: proceed with None data, required_keys will catch it
errors: list[str] = []
if data is not None:
# Check required keys
for key in required_keys:
found, _ = _get_path(data, key)
if not found:
errors.append(f"缺少字段: {key}")
# Check forbidden keys
for key in forbidden_keys:
found, _ = _get_path(data, key)
if found:
errors.append(f"存在禁止字段: {key}")
# Check key types
for key, expected_type_name in key_types.items():
found, value = _get_path(data, key)
if not found:
errors.append(f"类型检查字段缺失: {key}")
continue
expected_type = _TYPE_MAP.get(expected_type_name)
if expected_type is None:
continue # unknown type name — skip
if not isinstance(value, expected_type):
actual = type(value).__name__
errors.append(f"{key} 类型错误: 期望 {expected_type_name}，实际 {actual}")
if errors:
return RuleResult(passed=False, reason="; ".join(errors))
total_checks = len(required_keys) + len(forbidden_keys) + len(key_types)
return RuleResult(
passed=True,
score=1.0,
reason=f"JSON 结构校验通过（{total_checks} 项检查）" if total_checks else "合法 JSON",
)

115
backend/agenteval/evaluation/rules/safety.py Normal file
View File

@ -0,0 +1,115 @@
"""Safety evaluation rule.
Checks the agent reply for unsafe / sensitive content using two layers:
1. Keyword blacklist (always active, zero-latency)
2. Optional moderation API (OpenAI-compatible /v1/moderations endpoint)
- Falls back silently to blacklist-only if API is unavailable
Configuration params:
blacklist list of forbidden keywords/phrases (case-insensitive)
use_moderation_api bool, whether to call moderation API (default: false)
api_url moderation endpoint URL (required if use_moderation_api)
api_key Bearer token for moderation API (optional)
flagged_categories list of OpenAI moderation categories to flag
default: ["hate", "harassment", "self-harm", "sexual", "violence"]
"""
import httpx
from agenteval.evaluation.rules.base import EvalRule, RuleResult, register_rule
from agenteval.models import Case, Turn
from agenteval.utils.llm import extract_reply_text
_DEFAULT_BLACKLIST: list[str] = []
_DEFAULT_FLAGGED_CATEGORIES: list[str] = [
"hate",
"harassment",
"self-harm",
"self_harm",
"sexual",
"violence",
]
@register_rule
class SafetyRule(EvalRule):
"""Check reply for unsafe content via keyword blacklist and optional moderation API."""
name = "safety"
async def evaluate(self, case: Case, dialog: list[Turn]) -> RuleResult:
if not dialog:
return RuleResult(passed=False, reason="无回复记录")
reply_text = extract_reply_text(dialog[-1].reply)
if not reply_text:
return RuleResult(passed=True, score=1.0, reason="空回复，安全检查通过")
blacklist: list[str] = self.params.get("blacklist", _DEFAULT_BLACKLIST)
use_api: bool = bool(self.params.get("use_moderation_api", False))
api_url: str | None = self.params.get("api_url")
api_key: str | None = self.params.get("api_key")
flagged_cats: list[str] = self.params.get("flagged_categories", _DEFAULT_FLAGGED_CATEGORIES)
# Layer 1: keyword blacklist
reply_lower = reply_text.lower()
hit_words = [w for w in blacklist if w.lower() in reply_lower]
if hit_words:
return RuleResult(
passed=False,
score=0.0,
reason=f"包含违禁词: {hit_words}",
)
# Layer 2: moderation API (optional, degrades gracefully)
if use_api and api_url:
try:
flagged, categories_hit = await self._call_moderation(api_url, api_key, reply_text, flagged_cats)
if flagged:
return RuleResult(
passed=False,
score=0.0,
reason=f"moderation API 标记: {categories_hit}",
)
except Exception as exc:
# Degrade gracefully: log in reason but don't fail
return RuleResult(
passed=True,
score=1.0,
reason=f"安全检查通过（moderation API 不可用，已降级: {exc}）",
)
return RuleResult(passed=True, score=1.0, reason="安全检查通过")
async def _call_moderation(
self,
api_url: str,
api_key: str | None,
text: str,
flagged_categories: list[str],
) -> tuple[bool, list[str]]:
headers = {"Content-Type": "application/json"}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(
api_url,
headers=headers,
json={"input": text},
)
resp.raise_for_status()
data = resp.json()
# Standard OpenAI moderation response shape
results = data.get("results", [])
if not results:
return False, []
result = results[0]
cats: dict[str, bool] = result.get("categories", {})
# Normalize category names (API uses "/" separator in some versions)
hit = [c for c in flagged_categories if cats.get(c) or cats.get(c.replace("-", "/"))]
flagged = bool(hit) or result.get("flagged", False)
return flagged, hit

94
backend/agenteval/evaluation/rules/semantic.py Normal file
View File

@ -0,0 +1,94 @@
"""Semantic similarity evaluation rule.
Uses an external embedding API to compute cosine similarity between the
agent reply and a reference answer. Requires an OpenAI-compatible
embeddings endpoint (POST /v1/embeddings or equivalent).
Configuration params:
api_url Embeddings API endpoint (required)
api_key Bearer token (optional)
model Embedding model name (default: text-embedding-3-small)
reference Reference text to compare against (required)
min_score Minimum cosine similarity to pass, 0-1 (default: 0.7)
"""
import asyncio
import math
import httpx
from agenteval.evaluation.rules.base import EvalRule, RuleResult, register_rule
from agenteval.models import Case, Turn
from agenteval.utils.llm import extract_reply_text
def _cosine(a: list[float], b: list[float]) -> float:
dot = sum(x * y for x, y in zip(a, b))
norm_a = math.sqrt(sum(x * x for x in a))
norm_b = math.sqrt(sum(x * x for x in b))
if norm_a == 0 or norm_b == 0:
return 0.0
return dot / (norm_a * norm_b)
async def _embed(
client: httpx.AsyncClient,
api_url: str,
api_key: str | None,
model: str,
text: str,
) -> list[float]:
headers = {"Content-Type": "application/json"}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
resp = await client.post(
api_url,
headers=headers,
json={"model": model, "input": text},
timeout=30,
)
resp.raise_for_status()
data = resp.json()
return data["data"][0]["embedding"]
@register_rule
class SemanticSimilarityRule(EvalRule):
"""Score reply by cosine similarity to a reference answer via embedding API."""
name = "semantic_similarity"
async def evaluate(self, case: Case, dialog: list[Turn]) -> RuleResult:
if not dialog:
return RuleResult(passed=False, reason="无回复记录")
reply_text = extract_reply_text(dialog[-1].reply)
if not reply_text:
return RuleResult(passed=False, reason="回复内容为空")
api_url: str | None = self.params.get("api_url")
api_key: str | None = self.params.get("api_key")
model: str = self.params.get("model", "text-embedding-3-small")
reference: str | None = self.params.get("reference")
min_score: float = float(self.params.get("min_score", 0.7))
if not api_url:
return RuleResult(passed=False, reason="semantic_similarity 未配置 api_url")
if not reference:
return RuleResult(passed=False, reason="semantic_similarity 未配置 reference")
try:
async with httpx.AsyncClient() as client:
reply_vec, ref_vec = await asyncio.gather(
_embed(client, api_url, api_key, model, reply_text),
_embed(client, api_url, api_key, model, reference),
)
similarity = _cosine(reply_vec, ref_vec)
passed = similarity >= min_score
return RuleResult(
passed=passed,
score=round(similarity, 4),
reason=f"语义相似度 {similarity:.3f}（阈值 {min_score}）",
)
except Exception as exc:
return RuleResult(passed=False, reason=f"embedding 调用失败: {exc}")

11
backend/agenteval/models.py
View File

@ -59,6 +59,15 @@ class EvalRuleConfig(BaseModel):
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):
@ -71,6 +80,8 @@ class Case(BaseModel):
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

4
backend/agenteval/web/routers/runs.py
View File

@ -172,7 +172,9 @@ async def get_run_logs(run_id: str, session: Session = Depends(get_db)) -> dict:
"response_time_max_ms": case.expectations.response_time_max_ms,
"coherence_min_score": case.expectations.coherence_min_score,
},
"eval_rules": [{"type": r.type, "params": dict(r.params)} for r in case.eval_rules],
"eval_rules": [{"type": r.type, "params": dict(r.params), "weight": r.weight} for r in case.eval_rules],
"rule_logic": case.rule_logic.value if hasattr(case.rule_logic, "value") else str(case.rule_logic),
"rule_pass_threshold": case.rule_pass_threshold,
}
return {"turns": turns_data, "results": results_data, "scenario_snapshot": scenario_snapshot}

377
tests/unit/test_s2_rules_and_logic.py Normal file
View File

@ -0,0 +1,377 @@
"""Tests for S2 new rules: json_schema, safety, semantic_similarity,
and rule combination logic (all/any/weighted)."""
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from agenteval.evaluation.rules.json_schema import JsonSchemaRule, _get_path
from agenteval.evaluation.rules.safety import SafetyRule
from agenteval.evaluation.rules.semantic import SemanticSimilarityRule, _cosine
from agenteval.models import Case, CaseType, EvalRuleConfig, Expectation, RuleLogic, Turn
# ── helpers ──────────────────────────────────────────────────────────────
def _turn(reply_text: str, latency_ms: int = 100) -> Turn:
return Turn(
id="t1", run_id="r1", case_id="c1", round_index=1,
reply={"msgBody": {"content": reply_text}},
latency_ms=latency_ms,
)
def _case(
*,
rules: list[dict] | None = None,
rule_logic: RuleLogic = RuleLogic.ALL,
rule_pass_threshold: float = 0.6,
) -> Case:
eval_rules = [EvalRuleConfig(**r) for r in (rules or [])]
return Case(
id="c1", type=CaseType.SINGLE, messages=["hi"],
eval_rules=eval_rules,
rule_logic=rule_logic,
rule_pass_threshold=rule_pass_threshold,
)
# ── _get_path ─────────────────────────────────────────────────────────────
def test_json_schema_get_path_nested():
assert _get_path({"a": {"b": 1}}, "a.b") == (True, 1)
def test_json_schema_get_path_missing():
assert _get_path({"a": 1}, "a.b") == (False, None)
# ── JsonSchemaRule ────────────────────────────────────────────────────────
async def test_json_schema_valid_json_no_constraints():
rule = JsonSchemaRule({})
result = await rule.evaluate(_case(), [_turn('{"key": "value"}')])
assert result.passed is True
async def test_json_schema_invalid_json_strict():
rule = JsonSchemaRule({"strict_json": True})
result = await rule.evaluate(_case(), [_turn("not json at all")])
assert result.passed is False
assert "合法 JSON" in result.reason
async def test_json_schema_invalid_json_nonstrict():
rule = JsonSchemaRule({"strict_json": False})
result = await rule.evaluate(_case(), [_turn("not json")])
assert result.passed is True
async def test_json_schema_required_key_present():
rule = JsonSchemaRule({"required_keys": ["status", "data.id"]})
result = await rule.evaluate(_case(), [_turn('{"status": "ok", "data": {"id": 42}}')])
assert result.passed is True
async def test_json_schema_required_key_missing():
rule = JsonSchemaRule({"required_keys": ["missing_key"]})
result = await rule.evaluate(_case(), [_turn('{"status": "ok"}')])
assert result.passed is False
assert "missing_key" in result.reason
async def test_json_schema_forbidden_key_present():
rule = JsonSchemaRule({"forbidden_keys": ["error"]})
result = await rule.evaluate(_case(), [_turn('{"status": "ok", "error": "oops"}')])
assert result.passed is False
assert "error" in result.reason
async def test_json_schema_type_check_pass():
rule = JsonSchemaRule({"key_types": {"count": "int", "name": "str"}})
result = await rule.evaluate(_case(), [_turn('{"count": 5, "name": "hello"}')])
assert result.passed is True
async def test_json_schema_type_check_fail():
rule = JsonSchemaRule({"key_types": {"count": "int"}})
result = await rule.evaluate(_case(), [_turn('{"count": "five"}')])
assert result.passed is False
assert "count" in result.reason
async def test_json_schema_strips_markdown_fence():
rule = JsonSchemaRule({"required_keys": ["id"]})
reply = '```json\n{"id": 123}\n```'
result = await rule.evaluate(_case(), [_turn(reply)])
assert result.passed is True
async def test_json_schema_empty_dialog():
rule = JsonSchemaRule({})
result = await rule.evaluate(_case(), [])
assert result.passed is False
# ── SafetyRule ────────────────────────────────────────────────────────────
async def test_safety_clean_reply():
rule = SafetyRule({"blacklist": ["kill", "harm"]})
result = await rule.evaluate(_case(), [_turn("This is a helpful response.")])
assert result.passed is True
async def test_safety_blacklist_hit():
rule = SafetyRule({"blacklist": ["kill"]})
result = await rule.evaluate(_case(), [_turn("I will kill you!")])
assert result.passed is False
assert "kill" in result.reason
async def test_safety_blacklist_case_insensitive():
rule = SafetyRule({"blacklist": ["HARM"]})
result = await rule.evaluate(_case(), [_turn("this causes harm")])
assert result.passed is False
async def test_safety_no_blacklist():
rule = SafetyRule({})
result = await rule.evaluate(_case(), [_turn("any text")])
assert result.passed is True
async def test_safety_moderation_api_flagged():
rule = SafetyRule({
"use_moderation_api": True,
"api_url": "http://mock/v1/moderations",
"api_key": "test",
})
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_response.json = MagicMock(return_value={
"results": [{"flagged": True, "categories": {"hate": True, "violence": False}}]
})
with patch("agenteval.evaluation.rules.safety.httpx.AsyncClient") as MockClient:
instance = MockClient.return_value.__aenter__.return_value
instance.post = AsyncMock(return_value=mock_response)
result = await rule.evaluate(_case(), [_turn("hateful content here")])
assert result.passed is False
assert "hate" in result.reason
async def test_safety_moderation_api_unavailable_degrades():
rule = SafetyRule({
"use_moderation_api": True,
"api_url": "http://unavailable/v1/moderations",
})
with patch("agenteval.evaluation.rules.safety.httpx.AsyncClient") as MockClient:
instance = MockClient.return_value.__aenter__.return_value
instance.post = AsyncMock(side_effect=Exception("connection refused"))
result = await rule.evaluate(_case(), [_turn("normal text")])
assert result.passed is True
assert "降级" in result.reason
async def test_safety_empty_dialog():
rule = SafetyRule({})
result = await rule.evaluate(_case(), [])
assert result.passed is False
# ── SemanticSimilarityRule ────────────────────────────────────────────────
def test_cosine_identical():
v = [1.0, 0.0, 0.0]
assert _cosine(v, v) == pytest.approx(1.0)
def test_cosine_orthogonal():
assert _cosine([1.0, 0.0], [0.0, 1.0]) == pytest.approx(0.0)
def test_cosine_zero_vector():
assert _cosine([0.0, 0.0], [1.0, 0.0]) == 0.0
async def test_semantic_missing_api_url():
rule = SemanticSimilarityRule({"reference": "hello world"})
result = await rule.evaluate(_case(), [_turn("hello")])
assert result.passed is False
assert "api_url" in result.reason
async def test_semantic_missing_reference():
rule = SemanticSimilarityRule({"api_url": "http://mock/embed"})
result = await rule.evaluate(_case(), [_turn("hello")])
assert result.passed is False
assert "reference" in result.reason
async def test_semantic_high_similarity_passes():
rule = SemanticSimilarityRule({
"api_url": "http://mock/embed",
"reference": "hello world",
"min_score": 0.8,
})
vec = [1.0, 0.0, 0.0]
mock_resp = MagicMock()
mock_resp.raise_for_status = MagicMock()
mock_resp.json = MagicMock(return_value={"data": [{"embedding": vec}]})
with patch("agenteval.evaluation.rules.semantic.httpx.AsyncClient") as MockClient:
instance = MockClient.return_value.__aenter__.return_value
instance.post = AsyncMock(return_value=mock_resp)
result = await rule.evaluate(_case(), [_turn("hello world")])
assert result.passed is True
assert result.score == pytest.approx(1.0)
async def test_semantic_low_similarity_fails():
rule = SemanticSimilarityRule({
"api_url": "http://mock/embed",
"reference": "hello world",
"min_score": 0.9,
})
call_count = {"n": 0}
async def mock_post(url, **kwargs):
mock_resp = MagicMock()
mock_resp.raise_for_status = MagicMock()
# First call (reply): orthogonal to reference
vecs = [[1.0, 0.0], [0.0, 1.0]]
mock_resp.json = MagicMock(return_value={"data": [{"embedding": vecs[call_count["n"]]}]})
call_count["n"] += 1
return mock_resp
with patch("agenteval.evaluation.rules.semantic.httpx.AsyncClient") as MockClient:
instance = MockClient.return_value.__aenter__.return_value
instance.post = AsyncMock(side_effect=mock_post)
result = await rule.evaluate(_case(), [_turn("completely different")])
assert result.passed is False
async def test_semantic_api_error_fails_gracefully():
rule = SemanticSimilarityRule({
"api_url": "http://mock/embed",
"reference": "ref",
})
with patch("agenteval.evaluation.rules.semantic.httpx.AsyncClient") as MockClient:
instance = MockClient.return_value.__aenter__.return_value
instance.post = AsyncMock(side_effect=Exception("timeout"))
result = await rule.evaluate(_case(), [_turn("reply")])
assert result.passed is False
assert "embedding 调用失败" in result.reason
async def test_semantic_empty_dialog():
rule = SemanticSimilarityRule({"api_url": "http://x", "reference": "ref"})
result = await rule.evaluate(_case(), [])
assert result.passed is False
# ── Rule combination logic (engine integration) ───────────────────────────
from agenteval.evaluation.engine import EvalEngine, TimeoutConfig
from agenteval.models import EvalTarget, PlatformType, RunStatus, Scenario, TargetStatus
from tests.unit.mock_channel import MockChannel
def _make_target() -> EvalTarget:
return EvalTarget(
id="t-1", name="t", platform=PlatformType.AI_DIGITAL_EMPLOYEE,
channel_type=__import__("agenteval.models", fromlist=["ChannelType"]).ChannelType.TUTU_API,
channel_config={"base_url": "http://x", "token": "x", "tenant": "t",
"chat_channel_id": "c", "chat_contact_id": "u"},
status=TargetStatus.ACTIVE,
)
def _build_engine(scenario, session) -> EvalEngine:
engine = EvalEngine(target=_make_target(), scenario=scenario, session=session)
engine.channel = MockChannel()
return engine
async def test_rule_logic_all_passes_when_all_pass(db_session):
scenario = Scenario(id="s1", name="s", cases=[Case(
id="c1", type=CaseType.SINGLE, messages=["hi"],
eval_rules=[
EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]}), # MockChannel replies "echo: q-1"
EvalRuleConfig(type="response_time", params={"max_ms": 99999}),
],
rule_logic=RuleLogic.ALL,
)])
engine = _build_engine(scenario, db_session)
run = await engine.run()
assert run.status == RunStatus.COMPLETED
assert run.summary["passed_cases"] == 1
async def test_rule_logic_all_fails_when_one_fails(db_session):
scenario = Scenario(id="s1", name="s", cases=[Case(
id="c1", type=CaseType.SINGLE, messages=["hi"],
eval_rules=[
EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]}), # passes
EvalRuleConfig(type="keyword_match", params={"keywords": ["__IMPOSSIBLE__"]}), # fails
],
rule_logic=RuleLogic.ALL,
)])
engine = _build_engine(scenario, db_session)
run = await engine.run()
assert run.status == RunStatus.COMPLETED
assert run.summary["failed_cases"] == 1
async def test_rule_logic_any_passes_when_one_passes(db_session):
scenario = Scenario(id="s1", name="s", cases=[Case(
id="c1", type=CaseType.SINGLE, messages=["hi"],
eval_rules=[
EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]}), # passes
EvalRuleConfig(type="keyword_match", params={"keywords": ["__IMPOSSIBLE__"]}), # fails
],
rule_logic=RuleLogic.ANY,
)])
engine = _build_engine(scenario, db_session)
run = await engine.run()
assert run.status == RunStatus.COMPLETED
assert run.summary["passed_cases"] == 1
async def test_rule_logic_weighted_passes_above_threshold(db_session):
scenario = Scenario(id="s1", name="s", cases=[Case(
id="c1", type=CaseType.SINGLE, messages=["hi"],
eval_rules=[
EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]}, weight=0.8), # passes
EvalRuleConfig(type="keyword_match", params={"keywords": ["__IMPOSSIBLE__"]}, weight=0.2), # fails
],
rule_logic=RuleLogic.WEIGHTED,
rule_pass_threshold=0.6, # weighted score = 0.8/(0.8+0.2)=0.8 >= 0.6 → pass
)])
engine = _build_engine(scenario, db_session)
run = await engine.run()
assert run.status == RunStatus.COMPLETED
assert run.summary["passed_cases"] == 1
async def test_rule_logic_weighted_fails_below_threshold(db_session):
scenario = Scenario(id="s1", name="s", cases=[Case(
id="c1", type=CaseType.SINGLE, messages=["hi"],
eval_rules=[
EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]}, weight=0.2), # passes
EvalRuleConfig(type="keyword_match", params={"keywords": ["__IMPOSSIBLE__"]}, weight=0.8), # fails
],
rule_logic=RuleLogic.WEIGHTED,
rule_pass_threshold=0.6, # weighted score = 0.2/(0.2+0.8)=0.2 < 0.6 → fail
)])
engine = _build_engine(scenario, db_session)
run = await engine.run()
assert run.status == RunStatus.COMPLETED
assert run.summary["failed_cases"] == 1