feat: add cheater gate with resume-safe survivor recovery
This commit is contained in:
@@ -12,6 +12,14 @@ from __future__ import annotations
|
||||
import enum
|
||||
import hashlib
|
||||
import json
|
||||
from typing import TYPE_CHECKING, Protocol
|
||||
|
||||
from loguru import logger
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from app.question_gen.adversarial_config import AdversarialFilterConfig
|
||||
from app.question_gen.run_store import QuestionGenStore
|
||||
from core.types import GeneratedQuestion
|
||||
|
||||
_VALID_LETTERS = ("A", "B", "C", "D")
|
||||
|
||||
@@ -94,3 +102,114 @@ def judge_flip(*, p_text: str | None, q_text: str | None) -> FlipDecision:
|
||||
if p_text.strip() != q_text.strip():
|
||||
return FlipDecision.PASSED
|
||||
return FlipDecision.FILTERED_NO_FLIP
|
||||
|
||||
|
||||
class AgentRunner(Protocol):
|
||||
"""完整 inference agent 试答端口 — Phase B 只依赖此接口（便于 mock）。
|
||||
|
||||
实现见 Task 10 的 _RealAgentRunner（复用 run_inference + RunLogImpl）。
|
||||
"""
|
||||
|
||||
model: str
|
||||
skill_mode: str # 从首次定义即入 Protocol，保证 Task 6/8/10 指纹口径一致
|
||||
|
||||
async def predict(
|
||||
self,
|
||||
questions: list[GeneratedQuestion],
|
||||
*,
|
||||
max_steps: int,
|
||||
run_id: str,
|
||||
) -> dict[str, str | None]:
|
||||
"""跑完整 agent，返回 question_id → 预测答案字母（无预测为 None）。"""
|
||||
...
|
||||
|
||||
|
||||
def _cheat_hash(question: GeneratedQuestion) -> str:
|
||||
"""按散参数签名计算作弊门题面 hash（question/options/answer）。"""
|
||||
return question_hash(question.question, question.options, question.answer)
|
||||
|
||||
|
||||
def _recover_survivors(
|
||||
questions: list[GeneratedQuestion],
|
||||
store: QuestionGenStore,
|
||||
cfg_fp: str,
|
||||
) -> list[GeneratedQuestion]:
|
||||
"""从已落 cheat verdict 恢复"agent 答错"的题（续跑，不重跑 agent）。"""
|
||||
survivors: list[GeneratedQuestion] = []
|
||||
for q in questions:
|
||||
rows = store._conn.execute(
|
||||
"SELECT agent_correct FROM adversarial_verdicts "
|
||||
"WHERE question_id=? AND question_hash=? AND stage='cheat' AND agent_config=?",
|
||||
(q.question_id, _cheat_hash(q), cfg_fp),
|
||||
).fetchall()
|
||||
if rows and rows[0][0] == 0:
|
||||
survivors.append(q)
|
||||
return survivors
|
||||
|
||||
|
||||
async def run_cheater_gate(
|
||||
questions: list[GeneratedQuestion],
|
||||
*,
|
||||
agent: AgentRunner,
|
||||
store: QuestionGenStore,
|
||||
config: AdversarialFilterConfig,
|
||||
round_no: int,
|
||||
run_id: str,
|
||||
) -> list[GeneratedQuestion]:
|
||||
"""作弊门：完整 agent 试答；答对→filtered_too_easy，答错→cheat passed 待翻转。
|
||||
|
||||
续跑：已在当前 hash+config 有 cheat verdict 的题跳过重跑。agent_config 变
|
||||
化时先作废该题旧 verdict。预测立即落表（崩溃不丢）。
|
||||
|
||||
参数:
|
||||
questions: 待判定的 AR 题列表。
|
||||
agent: 完整 agent 试答端口。
|
||||
store: verdict 持久化。
|
||||
config: 过滤配置（提供 max_steps）。
|
||||
round_no: 当前轮次。
|
||||
run_id: agent 推理 run 标识。
|
||||
|
||||
返回:
|
||||
agent 答错的题（进翻转门）——含"已完成续跑恢复的存活者"与"本轮新判答错者"
|
||||
两部分合并。答错题的 cheat 预测字母已落 adversarial_verdicts 表
|
||||
（stage='cheat'），翻转门经 `_read_cheat_prediction` 从表读取复用（不重跑）。
|
||||
"""
|
||||
cfg_fp = agent_config_fingerprint(
|
||||
skill_mode=agent.skill_mode,
|
||||
max_steps=config.adversarial_agent_max_steps,
|
||||
model=agent.model,
|
||||
)
|
||||
todo: list[GeneratedQuestion] = []
|
||||
completed: list[GeneratedQuestion] = []
|
||||
for q in questions:
|
||||
h = _cheat_hash(q)
|
||||
store.invalidate_stale_config(q.question_id, cfg_fp)
|
||||
if "cheat" in store.completed_stages(q.question_id, h, cfg_fp):
|
||||
completed.append(q)
|
||||
else:
|
||||
todo.append(q)
|
||||
|
||||
# C1: 无条件先从已完成题恢复存活者（agent 答错），再对未判题跑 agent 追加。
|
||||
# 两者都流向翻转门——绝不因 todo 非空而丢掉已完成的存活者（混合续跑正确性）。
|
||||
survivors: list[GeneratedQuestion] = _recover_survivors(completed, store, cfg_fp)
|
||||
|
||||
if todo:
|
||||
preds = await agent.predict(
|
||||
todo, max_steps=config.adversarial_agent_max_steps, run_id=run_id
|
||||
)
|
||||
for q in todo:
|
||||
pred = preds.get(q.question_id)
|
||||
correct = pred is not None and pred.strip().upper() == q.answer.strip().upper()
|
||||
verdict = "filtered_too_easy" if correct else "passed"
|
||||
store.record_verdict(
|
||||
question_id=q.question_id, question_hash=_cheat_hash(q), stage="cheat",
|
||||
round=round_no, agent_prediction=pred, agent_correct=correct,
|
||||
verdict=verdict, pair_id=None, agent_config=cfg_fp,
|
||||
)
|
||||
if not correct:
|
||||
survivors.append(q)
|
||||
logger.info(
|
||||
"作弊门: {} 题（续跑复用 {}，新判 {}）→ 存活 {}",
|
||||
len(questions), len(completed), len(todo), len(survivors),
|
||||
)
|
||||
return survivors
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
"""作弊门：agent 答对=filtered_too_easy 并落表；答错=cheat verdict=passed 待翻转。"""
|
||||
|
||||
import pytest
|
||||
|
||||
from app.question_gen.adversarial_config import AdversarialFilterConfig
|
||||
from app.question_gen.adversarial_filter import run_cheater_gate
|
||||
from app.question_gen.run_store import QuestionGenStore
|
||||
from core.types import GeneratedQuestion
|
||||
|
||||
|
||||
class _FakeAgent:
|
||||
"""按 question_id → 预测字母返回的 mock AgentRunner。"""
|
||||
|
||||
def __init__(self, preds: dict[str, str], model: str = "m1", skill_mode: str = "auto"):
|
||||
self._preds = preds
|
||||
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._preds.get(q.question_id) for q in questions}
|
||||
|
||||
|
||||
def _q(qid, answer="A"):
|
||||
return GeneratedQuestion(
|
||||
question_id=qid, video_id="v1", task_type="Action Recognition",
|
||||
question="?", options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), answer=answer,
|
||||
source_nodes=("n1",), difficulty="hard", sub_pattern="temporal_reasoning_failure",
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cheater_gate_filters_too_easy_and_keeps_hard(tmp_path):
|
||||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||||
agent = _FakeAgent({"easy": "A", "hard": "B"}) # easy 答对(A), hard 答错
|
||||
cfg = AdversarialFilterConfig()
|
||||
survivors = await run_cheater_gate(
|
||||
[_q("easy"), _q("hard")], agent=agent, store=store,
|
||||
config=cfg, round_no=0, run_id="r0",
|
||||
)
|
||||
ids = {q.question_id for q in survivors}
|
||||
assert ids == {"hard"} # 只有答错的进翻转门
|
||||
verdicts = {
|
||||
r[0]: r[1] for r in store._conn.execute(
|
||||
"SELECT question_id, verdict FROM adversarial_verdicts WHERE stage='cheat'"
|
||||
)
|
||||
}
|
||||
assert verdicts["easy"] == "filtered_too_easy"
|
||||
# hard 在 cheat 阶段先记 passed（待翻转门可能改写；不支持翻转的题即终判 passed）
|
||||
assert verdicts["hard"] == "passed"
|
||||
store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cheater_gate_resume_skips_completed(tmp_path):
|
||||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||||
agent = _FakeAgent({"hard": "B"})
|
||||
cfg = AdversarialFilterConfig()
|
||||
await run_cheater_gate([_q("hard")], agent=agent, store=store,
|
||||
config=cfg, round_no=0, run_id="r0")
|
||||
first = list(agent.calls)
|
||||
await run_cheater_gate([_q("hard")], agent=agent, store=store,
|
||||
config=cfg, round_no=0, run_id="r1")
|
||||
assert agent.calls == first # 第二次不重跑（已有 cheat verdict）
|
||||
store.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cheater_gate_resume_mixed_keeps_all_survivors(tmp_path):
|
||||
"""混合续跑：部分题已有 cheat verdict、部分未判——已完成的存活者不得被丢。"""
|
||||
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||||
cfg = AdversarialFilterConfig()
|
||||
# 第一轮：先只判 done_hard（答错=存活），落表
|
||||
agent1 = _FakeAgent({"done_hard": "B"}) # 答错(正解 A)
|
||||
await run_cheater_gate([_q("done_hard")], agent=agent1, store=store,
|
||||
config=cfg, round_no=0, run_id="r0")
|
||||
# 第二轮：done_hard 已判 + 新题 new_hard 未判混在一起
|
||||
agent2 = _FakeAgent({"new_hard": "C"}) # 新题答错(正解 A)=存活
|
||||
survivors = await run_cheater_gate(
|
||||
[_q("done_hard"), _q("new_hard")], agent=agent2, store=store,
|
||||
config=cfg, round_no=0, run_id="r1",
|
||||
)
|
||||
ids = {q.question_id for q in survivors}
|
||||
assert ids == {"done_hard", "new_hard"} # 已完成存活者 done_hard 未被丢
|
||||
assert agent2.calls == ["new_hard"] # 只对未判题跑 agent
|
||||
store.close()
|
||||
Reference in New Issue
Block a user