192 lines
7.2 KiB
Python
192 lines
7.2 KiB
Python
"""Task 9：final 全量重写 + 补生成迭代循环 + 难度报告。
|
||
|
||
覆盖：
|
||
- write_final_bank 仅含当前 hash+config 过两门的题、stale-config 旧行被排除、原子重写。
|
||
- run_adversarial_rounds 缺额驱动：deficit≤0 或 round≥max → 停止（fake backfill 计数）。
|
||
- _report_difficulty：agent 正确率 > 阈值 → WARNING。
|
||
"""
|
||
|
||
import json
|
||
|
||
import pytest
|
||
from loguru import logger
|
||
|
||
from app.question_gen.adversarial_config import AdversarialFilterConfig
|
||
from app.question_gen.adversarial_filter import (
|
||
_report_difficulty,
|
||
question_hash,
|
||
run_adversarial_rounds,
|
||
write_final_bank,
|
||
)
|
||
from app.question_gen.run_store import QuestionGenStore
|
||
from core.types import GeneratedQuestion
|
||
|
||
|
||
def _q(qid, sub=None):
|
||
"""构造一条 AR 题；sub=None → 不支持 flip，翻转门直接放行（无需 VLM/树）。"""
|
||
return GeneratedQuestion(
|
||
question_id=qid,
|
||
video_id="v1",
|
||
task_type="Action Recognition",
|
||
question=f"{qid} 之前做了什么？",
|
||
options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"),
|
||
answer="A",
|
||
source_nodes=("n1",),
|
||
difficulty="hard",
|
||
sub_pattern=sub,
|
||
)
|
||
|
||
|
||
class _FakeAgent:
|
||
"""完整 agent 试答桩：对每题返回固定预测（AgentRunner Protocol 要求 model/skill_mode）。"""
|
||
|
||
def __init__(self, pred="B", model="m1", skill_mode="auto"):
|
||
self._pred = pred
|
||
self.model = model
|
||
self.skill_mode = skill_mode
|
||
self.calls: list[str] = []
|
||
|
||
async def predict(self, questions, *, max_steps, run_id):
|
||
self.calls.extend(q.question_id for q in questions)
|
||
return {q.question_id: self._pred for q in questions}
|
||
|
||
|
||
class _CountingBackfill:
|
||
"""补生成回调桩：记录调用次数，按工厂返回新题。"""
|
||
|
||
def __init__(self, factory):
|
||
self.calls = 0
|
||
self._factory = factory
|
||
|
||
async def __call__(self, deficit, round_no, existing):
|
||
self.calls += 1
|
||
return self._factory(deficit, round_no)
|
||
|
||
|
||
def test_write_final_bank_only_passed(tmp_path):
|
||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||
q1, q2 = _q("q1"), _q("q2")
|
||
store.record_verdict(
|
||
question_id="q1", question_hash=question_hash(q1), stage="cheat",
|
||
round=0, agent_prediction="B", agent_correct=False,
|
||
verdict="passed", pair_id=None, agent_config="c",
|
||
)
|
||
store.record_verdict(
|
||
question_id="q2", question_hash=question_hash(q2), stage="cheat",
|
||
round=0, agent_prediction="A", agent_correct=True,
|
||
verdict="filtered_too_easy", pair_id=None, agent_config="c",
|
||
)
|
||
out = tmp_path / "accepted_questions_final.json"
|
||
write_final_bank(out, store, {"q1": q1, "q2": q2}, "c")
|
||
data = json.loads(out.read_text(encoding="utf-8"))
|
||
assert [d["question_id"] for d in data] == ["q1"]
|
||
store.close()
|
||
|
||
|
||
def test_write_final_bank_excludes_stale_config(tmp_path):
|
||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||
q1 = _q("q1")
|
||
store.record_verdict(
|
||
question_id="q1", question_hash=question_hash(q1), stage="cheat",
|
||
round=0, agent_prediction="B", agent_correct=False,
|
||
verdict="passed", pair_id=None, agent_config="OLD",
|
||
)
|
||
out = tmp_path / "accepted_questions_final.json"
|
||
write_final_bank(out, store, {"q1": q1}, "NEW")
|
||
assert json.loads(out.read_text(encoding="utf-8")) == []
|
||
store.close()
|
||
|
||
|
||
def test_difficulty_warns_above_threshold(tmp_path):
|
||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||
for i in range(4): # 4 题全对 = 正确率 1.0 > 阈值 0.85 → 必触发告警
|
||
store.record_verdict(
|
||
question_id=f"q{i}", question_hash=str(i), stage="cheat",
|
||
round=0, agent_prediction="A", agent_correct=True,
|
||
verdict="filtered_too_easy", pair_id=None, agent_config="c",
|
||
)
|
||
# loguru 不走标准 logging，用项目既定 sink 捕获模式（见 test_pool_strategy）。
|
||
captured: list[str] = []
|
||
sink_id = logger.add(lambda msg: captured.append(str(msg)), level="WARNING")
|
||
try:
|
||
_report_difficulty(store, round_no=0, threshold=0.85)
|
||
finally:
|
||
logger.remove(sink_id)
|
||
assert any("太简单" in m or "简单" in m for m in captured), f"未捕获告警: {captured}"
|
||
store.close()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_rounds_stop_when_deficit_met(tmp_path):
|
||
"""首轮即达标（passed≥target）→ 不调 backfill，迭代立即停止。"""
|
||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||
agent = _FakeAgent(pred="B") # 答错 → 过作弊门；sub=None → 过翻转门
|
||
backfill = _CountingBackfill(lambda d, r: [])
|
||
final_path = tmp_path / "accepted_questions_final.json"
|
||
await run_adversarial_rounds(
|
||
[_q("q0")],
|
||
agent=agent,
|
||
vlm=object(),
|
||
store=store,
|
||
trees={},
|
||
config=AdversarialFilterConfig(adversarial_max_rounds=5),
|
||
final_path=final_path,
|
||
target=1,
|
||
backfill=backfill,
|
||
session_id="s",
|
||
)
|
||
assert backfill.calls == 0
|
||
assert [d["question_id"] for d in json.loads(final_path.read_text(encoding="utf-8"))] == ["q0"]
|
||
store.close()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_rounds_backfill_then_stop_at_max(tmp_path):
|
||
"""缺额 > 0 → 调 backfill 补生成；达轮次上限即停（不无限循环）。"""
|
||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||
agent = _FakeAgent(pred="B")
|
||
backfill = _CountingBackfill(lambda d, r: [_q(f"bf{r}_{i}") for i in range(d)])
|
||
final_path = tmp_path / "accepted_questions_final.json"
|
||
await run_adversarial_rounds(
|
||
[_q("q0")],
|
||
agent=agent,
|
||
vlm=object(),
|
||
store=store,
|
||
trees={},
|
||
config=AdversarialFilterConfig(adversarial_max_rounds=2),
|
||
final_path=final_path,
|
||
target=99, # 永远达不到 → 靠 max_rounds 终止
|
||
backfill=backfill,
|
||
session_id="s",
|
||
)
|
||
assert backfill.calls == 1 # round0 补生成一次；round1 达上限 break，不再补
|
||
ids = {d["question_id"] for d in json.loads(final_path.read_text(encoding="utf-8"))}
|
||
assert "q0" in ids and any(x.startswith("bf0_") for x in ids)
|
||
store.close()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_rounds_tolerate_backfill_under_delivery(tmp_path):
|
||
"""backfill 欠产（实得 < deficit，被 gate 拒常态）→ 不崩，并入实得题，达上限停。"""
|
||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||
agent = _FakeAgent(pred="B")
|
||
# 每轮只产出 deficit-1 道（模拟部分题被 gate 拒的真实欠产）
|
||
backfill = _CountingBackfill(lambda d, r: [_q(f"bf{r}_{i}") for i in range(max(d - 1, 0))])
|
||
final_path = tmp_path / "accepted_questions_final.json"
|
||
await run_adversarial_rounds(
|
||
[_q("q0")],
|
||
agent=agent,
|
||
vlm=object(),
|
||
store=store,
|
||
trees={},
|
||
config=AdversarialFilterConfig(adversarial_max_rounds=3),
|
||
final_path=final_path,
|
||
target=99, # 永远达不到 → 靠 max_rounds 终止，不因欠产崩溃或死循环
|
||
backfill=backfill,
|
||
session_id="s",
|
||
)
|
||
assert backfill.calls == 2 # round0/round1 各补一次；round2 达上限 break
|
||
ids = {d["question_id"] for d in json.loads(final_path.read_text(encoding="utf-8"))}
|
||
assert "q0" in ids and any(x.startswith("bf0_") for x in ids) # 实得欠产题已并入
|
||
store.close()
|