feat: add adversarial round loop with backfill iteration and difficulty report

This commit is contained in:
2026-07-14 16:40:05 -04:00
parent c1565a01c2
commit 73d0bb9190
2 changed files with 305 additions and 1 deletions
+140 -1
View File
@@ -12,6 +12,7 @@ from __future__ import annotations
import enum
import hashlib
import json
import os
from pathlib import Path
from typing import TYPE_CHECKING, Protocol, overload
@@ -602,6 +603,9 @@ def write_final_bank(
终判集合来自 `final_passed_question_ids`(当前 hash+config 下 cheat=passed 且
无 filtered_no_flip 行)。镜像题不在 questions_by_id 中,天然被排除。
原子性:先写同目录 `.tmp`,再 `os.replace` 覆盖目标,保证任何时刻读到的
final JSON 都是完整的(迭代循环每轮全量重写,崩溃可从 verdicts 表重建)。
参数:
out_path: 输出 JSON 路径。
store: verdict 来源。
@@ -616,5 +620,140 @@ def write_final_bank(
records = [
_question_to_record(questions_by_id[qid]) for qid in questions_by_id if qid in passed
]
out_path.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
out_path.parent.mkdir(parents=True, exist_ok=True)
tmp = out_path.with_suffix(".tmp")
tmp.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
os.replace(str(tmp), str(out_path))
logger.info("final 题库全量重写: {} 题 → {}", len(records), out_path)
return records
def _report_difficulty(store: QuestionGenStore, *, round_no: int, threshold: float) -> float:
"""记录并按阈值告警本轮 agent 正确率(作弊门聚合)。
正确率来自 `cheat_agent_accuracy`stage='cheat' 的 agent_correct 平均),高于
阈值说明 agent 太容易答对 = 出题偏简单,据此告警提示提高难度。
参数:
store: verdict 来源。
round_no: 当前轮次。
threshold: 正确率告警阈值(超过即 WARNING)。
返回:
本轮 agent 正确率。
"""
acc = store.cheat_agent_accuracy(round_no)
logger.info("难度报告 round={}: agent 正确率={:.2%}", round_no, acc)
if acc > threshold:
logger.warning(
"出题太简单: round={} agent 正确率={:.2%} > 阈值 {:.2%}",
round_no,
acc,
threshold,
)
return acc
class BackfillFn(Protocol):
"""补生成回调 — 缺额驱动,返回新增 AR 题。
真实实现由 Task 10 用 run_pipeline_v2 组装(传 initial_used_node_ids /
initial_embed_pool / seq_offset 防与已有题碰撞);单测可注入返回定制新题的假实现。
"""
async def __call__(
self,
deficit: int,
round_no: int,
existing: dict[str, GeneratedQuestion],
) -> list[GeneratedQuestion]:
"""按缺额补生成新题。
参数:
deficit: 目标缺额(target - 当前终判 passed 数)。
round_no: 当前轮次。
existing: 已有全部原题(供实现计算 used_node_ids / seq_offset)。
返回:
新增 AR 题列表(长度应等于 deficit)。
"""
...
async def run_adversarial_rounds(
initial_questions: list[GeneratedQuestion],
*,
agent: AgentRunner,
vlm: VLMProvider,
store: QuestionGenStore,
trees: dict[str, TreeIndex],
config: AdversarialFilterConfig,
final_path: Path,
target: int,
backfill: BackfillFn,
session_id: str,
) -> None:
"""两门 + 补生成迭代主循环,每轮全量重写 final 并做难度报告。
每轮:作弊门 → 翻转门 → 全量重写 `accepted_questions_final.json`(原子)→ 难度报告。
缺额 deficit = target - 终判 passed 数;deficit≤0 或已达轮次上限即停止(不无限循环),
否则调 `backfill` 补生成 deficit 题并只对新题重跑两门(已判题走续跑,不重跑)。
参数:
initial_questions: 首轮 AR 题(来自 accepted_questions.json 过滤)。
agent: 完整 agent 试答端口。
vlm: 镜像题生成 VLM 端口。
store: verdict 持久化。
trees: video_id → 三层树索引(重建镜像素材用)。
config: 过滤配置(轮次上限 / max_steps / 难度阈值)。
final_path: 最终题库 JSON 路径(每轮全量原子重写)。
target: 目标 passed 题数(缺额 = target - passed,锁定为首轮 AR 题数)。
backfill: 补生成回调 (deficit, round_no, existing) → 新增题;Task 10 用
run_pipeline_v2 实现,单测可 mock。
session_id: 遥测会话 ID(派生各门 run_id)。
"""
cfg_fp = agent_config_fingerprint(
skill_mode=agent.skill_mode,
max_steps=config.adversarial_agent_max_steps,
model=agent.model,
)
all_questions: dict[str, GeneratedQuestion] = {
q.question_id: q for q in initial_questions
}
pending = list(initial_questions)
passed_now = 0
for round_no in range(config.adversarial_max_rounds):
survivors = await run_cheater_gate(
pending,
agent=agent,
store=store,
config=config,
round_no=round_no,
run_id=f"{session_id}_cheat_{round_no}",
)
await run_flip_gate(
survivors,
agent=agent,
vlm=vlm,
store=store,
trees=trees,
config=config,
round_no=round_no,
run_id=f"{session_id}_flip_{round_no}",
session_id=session_id,
)
passed_now = len(write_final_bank(final_path, store, all_questions, cfg_fp))
_report_difficulty(
store, round_no=round_no, threshold=config.difficulty_warn_threshold
)
deficit = target - passed_now
if deficit <= 0 or round_no + 1 >= config.adversarial_max_rounds:
break
new_qs = await backfill(deficit, round_no, all_questions)
assert len(new_qs) == deficit, (
f"backfill 应产出 {deficit} 题,实得 {len(new_qs)}(缺额驱动契约)"
)
for q in new_qs:
all_questions[q.question_id] = q
pending = new_qs # 只对新补的题重新过滤(已判题走续跑)
logger.info("对抗过滤结束: final={} 题(target={}", passed_now, target)
+165
View File
@@ -0,0 +1,165 @@
"""Task 9:final 全量重写 + 补生成迭代循环 + 难度报告。
覆盖:
- write_final_bank 仅含当前 hash+config 过两门的题、stale-config 旧行被排除、原子重写。
- run_adversarial_rounds 缺额驱动:deficit≤0 或 round≥max → 停止(fake backfill 计数)。
- _report_difficultyagent 正确率 > 阈值 → 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()