iomgaa/Video-Tree-TRM5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(question_gen): add lightweight 4-gate quality check

Browse Source 
Implement 4 concurrent LLM-based quality gates for generated questions:
- key_verify: validates answer evidence in source material
- blind_answer: rejects questions answerable without video context
- multi_true: detects ambiguous multi-correct options
- leak_test: per-family shortcut detection (5 probe templates)

Includes run_gates orchestrator with verbatim_ratio short-circuit,
JSON response parsing with fallback, and 9 unit tests (all passing).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
iomgaa
2026-07-11 23:30:13 -04:00
parent 9627ac9cf9
commit 271d1682c9
10 changed files with 1030 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_gates.py
+312
View File
@@ -0,0 +1,312 @@
"""轻量四门质量检查单元测试。"""
from __future__ import annotations
import json
from typing import Any
import pytest
from app.question_gen.families import RETRIEVAL_FAMILY
from app.question_gen.gates import (
CandidateQuestion,
GateVerdict,
_gate_blind_answer,
_gate_key_verify,
_gate_leak_test,
_gate_multi_true,
run_gates,
)
from app.question_gen.postprocess import PostprocessResult
from app.tree.index import (
IndexMeta,
L1Card,
L1Node,
L2Card,
L2Node,
L3Card,
L3Node,
TreeIndex,
)
from core.types import LLMResponse
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _make_llm_response(verdict: str, reason: str) -> LLMResponse:
"""构造一个 LLM 返回值，content 为标准 JSON 格式。"""
return LLMResponse(
content=json.dumps({"verdict": verdict, "reason": reason}),
thinking="",
model="mock-model",
provider="mock",
prompt_tokens=10,
completion_tokens=5,
latency_ms=100,
ttft_ms=None,
max_inter_token_ms=None,
cache_hit=False,
call_id="mock-call-001",
)
class MockLLM:
"""可配置的 LLM mock — 按调用顺序返回预设响应。"""
def __init__(self, responses: list[LLMResponse]) -> None:
self._responses = list(responses)
self._call_count = 0
self.calls: list[dict[str, Any]] = []
async def chat(
self,
messages: list[dict[str, Any]],
*,
session_id: str | None = None,
parent_call_id: str | None = None,
) -> LLMResponse:
"""记录调用并返回预设响应。"""
self.calls.append({"messages": messages, "session_id": session_id})
idx = self._call_count
self._call_count += 1
if idx < len(self._responses):
return self._responses[idx]
# 默认返回 PASS
return _make_llm_response("pass", "default")
@pytest.fixture()
def mock_llm_pass() -> MockLLM:
"""返回始终 PASS 的 mock LLM。"""
return MockLLM([_make_llm_response("pass", "evidence found")])
@pytest.fixture()
def mock_llm_fail() -> MockLLM:
"""返回始终 FAIL 的 mock LLM。"""
return MockLLM([_make_llm_response("fail", "no evidence")])
def _make_candidate() -> CandidateQuestion:
"""构造测试用候选题目。"""
return CandidateQuestion(
question_id="q-001",
video_id="v-001",
task_type="Action Reasoning",
skill_target="M1",
question="What does the person do after picking up the book?",
options=(
"A. Reads it",
"B. Puts it back",
"C. Throws it away",
"D. Gives it to someone",
),
answer="A",
source_nodes=("L2_001", "L3_001"),
difficulty="medium",
)
def _make_tree() -> TreeIndex:
"""构造最小测试树。"""
l3_card = L3Card(
frame_summary="Person picks up a book from the shelf and starts reading it.",
visible_entities=["person", "book", "shelf"],
ongoing_actions=["picking up book", "reading"],
visible_text=[],
spatial_layout="person in center, shelf on left",
visual_attributes={},
subtitle="He picks up the book and reads.",
)
l3 = L3Node(id="L3_001", card=l3_card)
l2_card = L2Card(
event_description="A person picks up a book from the shelf and begins reading it attentively.",
entities=["person", "book"],
actions=["pick up", "read"],
action_subjects=["person"],
visible_text=[],
spatial_relations="person near shelf",
state_changes="book moves from shelf to hands",
subtitle="He picks up the book and reads.",
)
l2 = L2Node(id="L2_001", card=l2_card, children=[l3])
l1_card = L1Card(
scene_summary="Library scene with a person browsing and reading books.",
main_setting="library",
key_entities=["person", "books", "shelf"],
main_actions=["browsing", "reading"],
topic_keywords=["library", "reading"],
visible_text=[],
temporal_flow="enter → browse → pick up → read",
)
l1 = L1Node(id="L1_001", card=l1_card, children=[l2])
meta = IndexMeta(source_path="test.mp4", modality="video")
return TreeIndex(metadata=meta, roots=[l1])
def _make_postprocess(verbatim_ratio: float = 0.1) -> PostprocessResult:
"""构造测试用后处理结果。"""
return PostprocessResult(
options=(
"A. Reads it",
"B. Puts it back",
"C. Throws it away",
"D. Gives it to someone",
),
answer="A",
referent_violations=[],
verbatim_ratio=verbatim_ratio,
has_time_anchor=False,
)
# ---------------------------------------------------------------------------
# TestGateKeyVerify
# ---------------------------------------------------------------------------
class TestGateKeyVerify:
"""key_verify 门：验证答案在来源素材中有证据支撑。"""
@pytest.mark.asyncio()
async def test_pass_evidence(self, mock_llm_pass: MockLLM) -> None:
"""LLM 确认有证据 → PASS。"""
candidate = _make_candidate()
tree = _make_tree()
result = await _gate_key_verify(candidate, tree, mock_llm_pass, session_id="test-session")
assert result.verdict == GateVerdict.PASS
assert mock_llm_pass.calls # 确认调用了 LLM
@pytest.mark.asyncio()
async def test_fail_no_evidence(self, mock_llm_fail: MockLLM) -> None:
"""LLM 判断无证据 → FAIL。"""
candidate = _make_candidate()
tree = _make_tree()
result = await _gate_key_verify(candidate, tree, mock_llm_fail, session_id="test-session")
assert result.verdict == GateVerdict.FAIL
assert "no evidence" in result.reason
# ---------------------------------------------------------------------------
# TestGateBlindAnswer
# ---------------------------------------------------------------------------
class TestGateBlindAnswer:
"""blind_answer 门：无上下文答对 → FAIL（题目太简单/泄漏）。"""
@pytest.mark.asyncio()
async def test_pass_wrong(self) -> None:
"""LLM 在无上下文时答错 → PASS（题目确实需要视频信息）。"""
# LLM 返回 "pass" 表示它无法正确回答
llm = MockLLM([_make_llm_response("pass", "cannot determine without context")])
candidate = _make_candidate()
result = await _gate_blind_answer(candidate, llm, session_id="test-session")
assert result.verdict == GateVerdict.PASS
@pytest.mark.asyncio()
async def test_fail_correct(self) -> None:
"""LLM 在无上下文时答对 → FAIL（题目泄漏）。"""
llm = MockLLM([_make_llm_response("fail", "answer is obvious from options")])
candidate = _make_candidate()
result = await _gate_blind_answer(candidate, llm, session_id="test-session")
assert result.verdict == GateVerdict.FAIL
# ---------------------------------------------------------------------------
# TestGateMultiTrue
# ---------------------------------------------------------------------------
class TestGateMultiTrue:
"""multi_true 门：多选项正确 → FAIL。"""
@pytest.mark.asyncio()
async def test_pass_single(self) -> None:
"""只有一个正确选项 → PASS。"""
llm = MockLLM([_make_llm_response("pass", "only one correct answer")])
candidate = _make_candidate()
tree = _make_tree()
result = await _gate_multi_true(candidate, tree, llm, session_id="test-session")
assert result.verdict == GateVerdict.PASS
@pytest.mark.asyncio()
async def test_fail_multi(self) -> None:
"""多个选项可被视为正确 → FAIL。"""
llm = MockLLM([_make_llm_response("fail", "options A and B are both plausible")])
candidate = _make_candidate()
tree = _make_tree()
result = await _gate_multi_true(candidate, tree, llm, session_id="test-session")
assert result.verdict == GateVerdict.FAIL
# ---------------------------------------------------------------------------
# TestGateLeakTest
# ---------------------------------------------------------------------------
class TestGateLeakTest:
"""leak_test 门：按家族模板执行泄漏探测。"""
@pytest.mark.asyncio()
async def test_per_family_template(self) -> None:
"""使用家族特定的 probe_template 调用 LLM。"""
llm = MockLLM([_make_llm_response("pass", "no shortcut detected")])
candidate = _make_candidate()
result = await _gate_leak_test(candidate, RETRIEVAL_FAMILY, llm, session_id="test-session")
assert result.verdict == GateVerdict.PASS
# 验证 session_id 被正确传递
assert llm.calls[0]["session_id"] == "test-session"
# ---------------------------------------------------------------------------
# TestRunGates
# ---------------------------------------------------------------------------
class TestRunGates:
"""run_gates 编排：并发四门 + verbatim 短路。"""
@pytest.mark.asyncio()
async def test_all_pass(self) -> None:
"""四门全 PASS → GateReport.passed=True。"""
llm = MockLLM([_make_llm_response("pass", f"gate {i} ok") for i in range(4)])
candidate = _make_candidate()
tree = _make_tree()
postprocess = _make_postprocess(verbatim_ratio=0.1)
report = await run_gates(
candidate=candidate,
tree=tree,
llm=llm,
family_spec=RETRIEVAL_FAMILY,
postprocess=postprocess,
session_id="test-session",
)
assert report.passed is True
assert report.reject_reason is None
@pytest.mark.asyncio()
async def test_high_verbatim_shortcircuits(self) -> None:
"""verbatim_ratio > 0.5 → key_verify 直接 FAIL，不调用 LLM。"""
llm = MockLLM([_make_llm_response("pass", "should not be called")])
candidate = _make_candidate()
tree = _make_tree()
postprocess = _make_postprocess(verbatim_ratio=0.8)
report = await run_gates(
candidate=candidate,
tree=tree,
llm=llm,
family_spec=RETRIEVAL_FAMILY,
postprocess=postprocess,
session_id="test-session",
)
assert report.passed is False
assert report.key_verify.verdict == GateVerdict.FAIL
assert "verbatim" in report.key_verify.reason.lower()