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

feat(engine): expectations now additive with explicit rules (ticket 01)

Browse Source 
期望始终派生隐式判定并与显式规则叠加执行：rule_logic 只组合显式规则，
期望是叠加其上的硬约束，任一不满足即用例不通过。隐式判定以 EvalResult
同构落库，reason 前缀 [期望] 标明来源。连通用例（无规则无期望）行为不变。
This commit is contained in:
sinohqb 2026-07-29 10:24:01 +08:00
parent 2dcf415940
commit 5dd1bc8535
3 changed files with 169 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
.scratch/v0.5/issues/01-expectation-rule-additive.md
View File

@ -4,11 +4,13 @@
**Blocked by:** None — can start immediately.
**Status:** ready-for-agent
**Status:** done
- [ ] 同时配置期望与规则的用例，期望不满足时用例不通过（即使显式规则全部通过）
- [ ] 期望派生判定以 EvalResult 形式存储，reason 可辨识来源为期望
- [ ] rule_logic=any/weighted 时，隐式期望判定不参与组合计算，仍作为独立硬约束
- [ ] 纯期望用例（无显式规则）的判定行为与升级前一致
- [ ] 引擎单元测试覆盖叠加通过/失败矩阵（先例：现有引擎测试）
- [ ] 全部现有测试保持绿色
- [x] 同时配置期望与规则的用例，期望不满足时用例不通过（即使显式规则全部通过）
- [x] 期望派生判定以 EvalResult 形式存储，reason 可辨识来源为期望
- [x] rule_logic=any/weighted 时，隐式期望判定不参与组合计算，仍作为独立硬约束
- [x] 纯期望用例（无显式规则）的判定行为与升级前一致
- [x] 引擎单元测试覆盖叠加通过/失败矩阵（先例：现有引擎测试）
- [x] 全部现有测试保持绿色
> 实施备注：纯期望用例在默认 rule_logic=ALL 下行为与升级前完全一致；纯期望 + ANY/WEIGHTED 的病态组合下，期望现按硬约束全判（spec 决策优先于工单字面）。

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

@ -379,47 +379,51 @@ class EvalEngine:
) -> tuple[bool, int, int]:
"""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
Judgement semantics (spec v0.5 / CONTEXT.md):
- Explicit rules are combined by case.rule_logic (ALL / ANY / WEIGHTED).
- Expectations always derive implicit checks, additive to explicit
rules. They are hard constraints: they never join the rule_logic
combination, and any implicit failure fails the case.
"""
from agenteval.models import EvalRuleConfig
rules_config: list[EvalRuleConfig] = list(case.eval_rules)
# If no explicit rules, derive implicit rules from expectations.
if not rules_config:
if case.expectations.response_time_max_ms:
rules_config.append(
EvalRuleConfig(
type="response_time",
params={
"max_ms": case.expectations.response_time_max_ms,
},
)
implicit_config: list[EvalRuleConfig] = []
if case.expectations.response_time_max_ms:
implicit_config.append(
EvalRuleConfig(
type="response_time",
params={
"max_ms": case.expectations.response_time_max_ms,
},
)
if case.expectations.keywords_include or case.expectations.keywords_exclude:
rules_config.append(
EvalRuleConfig(
type="keyword_match",
params={
"keywords": case.expectations.keywords_include,
"exclude_keywords": case.expectations.keywords_exclude,
},
)
)
if case.expectations.keywords_include or case.expectations.keywords_exclude:
implicit_config.append(
EvalRuleConfig(
type="keyword_match",
params={
"keywords": case.expectations.keywords_include,
"exclude_keywords": case.expectations.keywords_exclude,
},
)
)
if not rules_config:
# No rules defined and no expectations → case passes with no checks
if not rules_config and not implicit_config:
# 连通用例：无任何判定标准，收到回复即通过
return True, 0, 0
passed_count = 0
total_count = 0
explicit_passed = 0
explicit_total = 0
weighted_score = 0.0
total_weight = 0.0
implicit_all_passed = True
for rule_config in rules_config:
all_rules = [(cfg, False) for cfg in rules_config] + [(cfg, True) for cfg in implicit_config]
for rule_config, is_implicit in all_rules:
purpose = {
"llm_score": ModelPurpose.JUDGE,
"semantic_similarity": ModelPurpose.EMBEDDING,
@ -436,6 +440,7 @@ class EvalEngine:
result = await rule.evaluate(case, dialog)
except Exception as exc:
result = RuleResult(passed=False, reason=f"模型配置解析失败: {exc}")
reason = f"[期望] {result.reason}" if is_implicit else result.reason
eval_result = EvalResult(
id=str(uuid.uuid4()),
run_id=run.id,
@ -444,18 +449,25 @@ class EvalEngine:
rule_type=rule_config.type,
passed=result.passed,
score=result.score,
reason=result.reason,
reason=reason,
)
self.result_repo.save_result(eval_result)
total_count += 1
if result.passed:
passed_count += 1
# 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
if is_implicit:
if not result.passed:
implicit_all_passed = False
else:
explicit_total += 1
if result.passed:
explicit_passed += 1
# 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,
@ -466,22 +478,24 @@ class EvalEngine:
"rule_type": rule_config.type,
"passed": result.passed,
"score": result.score,
"reason": result.reason,
"weight": weight,
"reason": reason,
"weight": rule_config.weight,
},
)
# Determine case pass/fail based on rule_logic
# Combine explicit rules by rule_logic; no explicit rules → vacuously true.
logic = case.rule_logic
if logic == RuleLogic.ALL:
case_passed = passed_count == total_count
if not rules_config:
explicit_ok = True
elif logic == RuleLogic.ANY:
case_passed = passed_count > 0
explicit_ok = explicit_passed > 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
explicit_ok = avg >= case.rule_pass_threshold
else: # ALL and fallback
explicit_ok = explicit_passed == explicit_total
case_passed = explicit_ok and implicit_all_passed
return case_passed, passed_count, total_count

105
tests/unit/test_engine.py
View File

@ -332,3 +332,108 @@ async def test_dynamic_generation_failure_records_case_error(db_session):
# 被测通道不应被调用（生成阶段就失败了）
assert channel.send_calls == 0
# ── expectation + explicit rules are additive (ticket 01) ────────────────
# MockChannel replies "echo: q-1", so keyword "echo" passes, "__NOPE__" fails.
from agenteval.models import EvalRuleConfig, RuleLogic # noqa: E402
def _case_with(
*,
rules: list[EvalRuleConfig] | None = None,
expectations: Expectation | None = None,
rule_logic: RuleLogic = RuleLogic.ALL,
rule_pass_threshold: float = 0.6,
) -> Case:
return Case(
id="c1", type=CaseType.SINGLE, messages=["hi"],
eval_rules=rules or [],
expectations=expectations or Expectation(),
rule_logic=rule_logic,
rule_pass_threshold=rule_pass_threshold,
)
async def test_expectation_fails_case_even_when_rules_pass(db_session):
"""期望不满足 → 用例不通过，即使显式规则全部通过。"""
scenario = Scenario(id="s1", name="s", cases=[_case_with(
rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]})],
expectations=Expectation(keywords_include=["__NOPE__"]),
)])
channel = MockChannel()
engine = _build_engine(scenario, channel, session=db_session)
run = await engine.run()
assert run.status == RunStatus.COMPLETED
assert run.summary["failed_cases"] == 1
async def test_expectation_and_rules_both_pass(db_session):
"""期望与规则都满足 → 通过，且期望派生判定同构落库、reason 可辨识来源。"""
scenario = Scenario(id="s1", name="s", cases=[_case_with(
rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]})],
expectations=Expectation(keywords_include=["echo"], response_time_max_ms=99999),
)])
channel = MockChannel()
engine = _build_engine(scenario, channel, session=db_session)
run = await engine.run()
assert run.summary["passed_cases"] == 1
results = RunRepository(db_session).get_results(run.id)
# 1 显式规则 + 2 期望派生（keyword + response_time）
assert len(results) == 3
implicit = [r for r in results if "期望" in r.reason]
assert len(implicit) == 2
assert all(r.passed for r in results)
async def test_implicit_expectation_not_in_any_combination(db_session):
"""rule_logic=ANY 只组合显式规则：期望通过不能救活全败的显式规则组。"""
scenario = Scenario(id="s1", name="s", cases=[_case_with(
rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["__NOPE__"]})],
expectations=Expectation(keywords_include=["echo"]), # 通过
rule_logic=RuleLogic.ANY,
)])
channel = MockChannel()
engine = _build_engine(scenario, channel, session=db_session)
run = await engine.run()
assert run.summary["failed_cases"] == 1
async def test_implicit_expectation_is_hard_constraint_over_weighted(db_session):
"""rule_logic=WEIGHTED 达标但期望不满足 → 仍不通过（期望是硬约束）。"""
scenario = Scenario(id="s1", name="s", cases=[_case_with(
rules=[EvalRuleConfig(type="keyword_match", params={"keywords": ["echo"]}, weight=1.0)],
expectations=Expectation(keywords_include=["__NOPE__"]),
rule_logic=RuleLogic.WEIGHTED,
rule_pass_threshold=0.5, # 显式加权得分 1.0 ≥ 0.5
)])
channel = MockChannel()
engine = _build_engine(scenario, channel, session=db_session)
run = await engine.run()
assert run.summary["failed_cases"] == 1
async def test_pure_expectation_case_behavior_unchanged(db_session):
"""纯期望用例（无显式规则）：满足通过、不满足失败，与升级前一致。"""
scenario = Scenario(id="s1", name="s", cases=[
Case(id="ok", type=CaseType.SINGLE, messages=["hi"],
expectations=Expectation(keywords_include=["echo"])),
Case(id="bad", type=CaseType.SINGLE, messages=["hi"],
expectations=Expectation(keywords_include=["__NOPE__"])),
])
channel = MockChannel()
engine = _build_engine(scenario, channel, session=db_session)
run = await engine.run()
assert run.summary["passed_cases"] == 1
assert run.summary["failed_cases"] == 1