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

feat(question_gen): add SQLite run store for generation telemetry

Browse Source 
Implements QuestionGenStore with:
- Idempotent schema initialization (question_gen_runs + question_gen_items)
- Run lifecycle: record_run_start / record_run_end / get_run_stats
- Per-item recording: record_item / update_gates / update_difficulty
- GateReportLike Protocol for duck-type gate report compatibility
- WAL mode + foreign keys + check_same_thread=False

DDL aligns with research-wiki/schemas/question-gen-{runs,items}.md.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
iomgaa
2026-07-11 23:15:53 -04:00
parent c83d771923
commit 6a5424a618
2 changed files with 624 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/question_gen/run_store.py
+380
View File
@@ -0,0 +1,380 @@
"""SQLite 出题管线日志记录器 — 记录批次运行与逐题门判定。
写入频率低（每批次 ~240 题），使用同步 sqlite3 即可。
schema 定义参照 research-wiki/schemas/question-gen-runs.md 和 question-gen-items.md。
"""
from __future__ import annotations
import sqlite3
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Protocol, runtime_checkable
from loguru import logger
if TYPE_CHECKING:
from pathlib import Path
# ---------------------------------------------------------------------------
# GateReport Protocol（Task 6 尚未实现，此处声明 duck-type 接口）
# ---------------------------------------------------------------------------
@runtime_checkable
class _GateResultLike(Protocol):
"""单门判定结果的最小接口。"""
@property
def verdict(self) -> object:
"""PASS / FAIL / SKIP 枚举值，.value 为小写字符串。"""
...
@property
def reason(self) -> str:
"""判定理由。"""
...
@runtime_checkable
class GateReportLike(Protocol):
"""GateReport 的最小 duck-type 接口。"""
@property
def key_verify(self) -> _GateResultLike: ...
@property
def blind_answer(self) -> _GateResultLike: ...
@property
def multi_true(self) -> _GateResultLike: ...
@property
def leak_test(self) -> _GateResultLike: ...
@property
def passed(self) -> bool: ...
@property
def reject_reason(self) -> str | None: ...
# ---------------------------------------------------------------------------
# 数据类
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class RunStats:
"""批次运行的统计摘要。
Attributes
----------
total_slots : int
目标题数（slot 总数）。
accepted : int
最终通过门判定的题数。
rejected : int
最终被拒绝的题数。
heavy_sampled : int
进行重量抽检的题数。
"""
total_slots: int
accepted: int
rejected: int
heavy_sampled: int
# ---------------------------------------------------------------------------
# DDL
# ---------------------------------------------------------------------------
_DDL_RUNS = """
CREATE TABLE IF NOT EXISTS question_gen_runs (
run_id TEXT PRIMARY KEY,
git_sha TEXT NOT NULL,
config_snapshot TEXT,
started_at TEXT NOT NULL,
finished_at TEXT,
status TEXT NOT NULL DEFAULT 'running',
total_slots INTEGER,
accepted INTEGER,
rejected INTEGER,
heavy_sampled INTEGER
);
"""
_DDL_ITEMS = """
CREATE TABLE IF NOT EXISTS question_gen_items (
item_id TEXT PRIMARY KEY,
run_id TEXT NOT NULL REFERENCES question_gen_runs(run_id),
slot_id TEXT NOT NULL,
video_id TEXT NOT NULL,
family TEXT NOT NULL,
task_type TEXT NOT NULL,
skill_target TEXT NOT NULL,
attempt INTEGER NOT NULL,
question_text TEXT NOT NULL,
gate_key_verify TEXT,
gate_blind_answer TEXT,
gate_multi_true TEXT,
gate_leak_test TEXT,
gate_reject_reason TEXT,
final_status TEXT NOT NULL DEFAULT 'pending',
difficulty_steps INTEGER,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
"""
_DDL_INDEXES = [
"CREATE INDEX IF NOT EXISTS idx_qgr_status ON question_gen_runs(status);",
"CREATE INDEX IF NOT EXISTS idx_qgi_run ON question_gen_items(run_id);",
"CREATE INDEX IF NOT EXISTS idx_qgi_slot ON question_gen_items(slot_id);",
"CREATE INDEX IF NOT EXISTS idx_qgi_status ON question_gen_items(final_status);",
"CREATE INDEX IF NOT EXISTS idx_qgi_family ON question_gen_items(family);",
"CREATE INDEX IF NOT EXISTS idx_qgi_task_type ON question_gen_items(task_type);",
]
# ---------------------------------------------------------------------------
# Store 实现
# ---------------------------------------------------------------------------
class QuestionGenStore:
"""SQLite 出题管线日志记录器。
记录每次出题批次的元数据（run）和每题的门判定结果（item）。
使用同步 sqlite3，写入频率低无需异步。
Parameters
----------
db_path : Path
SQLite 数据库文件路径。父目录必须存在。
"""
def __init__(self, db_path: Path) -> None:
self._db_path = db_path
self._conn = sqlite3.connect(
str(db_path),
check_same_thread=False,
timeout=10.0,
)
self._conn.execute("PRAGMA journal_mode=WAL")
self._conn.execute("PRAGMA busy_timeout=5000")
self._conn.execute("PRAGMA foreign_keys=ON")
self._init_schema()
logger.debug("QuestionGenStore 已初始化: {}", db_path)
def _init_schema(self) -> None:
"""幂等创建表和索引。多次调用安全。"""
self._conn.execute(_DDL_RUNS)
self._conn.execute(_DDL_ITEMS)
for idx_sql in _DDL_INDEXES:
self._conn.execute(idx_sql)
self._conn.commit()
def record_run_start(self, run_id: str, git_sha: str, config_snapshot: str) -> None:
"""记录批次开始。
Parameters
----------
run_id : str
批次唯一标识（UUID）。
git_sha : str
当前代码版本 HEAD commit。
config_snapshot : str
科研配置快照（JSON 序列化字符串）。
"""
now = datetime.now(tz=UTC).isoformat(timespec="seconds")
self._conn.execute(
"""
INSERT INTO question_gen_runs (run_id, git_sha, config_snapshot, started_at, status)
VALUES (?, ?, ?, ?, 'running')
""",
(run_id, git_sha, config_snapshot, now),
)
self._conn.commit()
logger.info("出题批次已开始: run_id={}", run_id)
def record_run_end(self, run_id: str, status: str, stats: RunStats) -> None:
"""记录批次结束，更新状态与统计。
Parameters
----------
run_id : str
批次唯一标识。
status : str
最终状态（completed / failed）。
stats : RunStats
批次统计摘要。
"""
now = datetime.now(tz=UTC).isoformat(timespec="seconds")
self._conn.execute(
"""
UPDATE question_gen_runs
SET finished_at=?, status=?, total_slots=?, accepted=?, rejected=?, heavy_sampled=?
WHERE run_id=?
""",
(
now,
status,
stats.total_slots,
stats.accepted,
stats.rejected,
stats.heavy_sampled,
run_id,
),
)
self._conn.commit()
logger.info(
"出题批次已结束: run_id={}, status={}, accepted={}/{}",
run_id,
status,
stats.accepted,
stats.total_slots,
)
def record_item(
self,
item_id: str,
run_id: str,
slot_id: str,
video_id: str,
family: str,
task_type: str,
skill_target: str,
attempt: int,
question_text: str,
) -> None:
"""记录一道新生成的题目（初始状态 pending）。
Parameters
----------
item_id : str
题目唯一 ID（每轮独立）。
run_id : str
关联的批次 ID。
slot_id : str
逻辑 slot 标识（同 slot 多次重出共享）。
video_id : str
视频 ID。
family : str
题族（retrieval/reasoning/enumeration/visual/spatial）。
task_type : str
Video-MME 12 类主标签。
skill_target : str
M1-M5 + 题族子标签。
attempt : int
当前重出轮次（1-based）。
question_text : str
题目文本。
"""
now = datetime.now(tz=UTC).isoformat(timespec="seconds")
self._conn.execute(
"""
INSERT INTO question_gen_items
(item_id, run_id, slot_id, video_id, family, task_type,
skill_target, attempt, question_text, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
item_id,
run_id,
slot_id,
video_id,
family,
task_type,
skill_target,
attempt,
question_text,
now,
),
)
self._conn.commit()
def update_gates(self, item_id: str, report: GateReportLike) -> None:
"""更新门判定结果及 final_status。
Parameters
----------
item_id : str
题目唯一 ID。
report : GateReportLike
门判定报告（duck-type，需有 key_verify/blind_answer/multi_true/leak_test
属性，每个属性具有 .verdict.value 和 .reason；以及 passed/reject_reason 属性）。
"""
final_status = "accepted" if report.passed else "rejected"
self._conn.execute(
"""
UPDATE question_gen_items
SET gate_key_verify=?, gate_blind_answer=?, gate_multi_true=?,
gate_leak_test=?, gate_reject_reason=?, final_status=?
WHERE item_id=?
""",
(
report.key_verify.verdict.value,
report.blind_answer.verdict.value,
report.multi_true.verdict.value,
report.leak_test.verdict.value,
report.reject_reason,
final_status,
item_id,
),
)
self._conn.commit()
def update_difficulty(self, item_id: str, difficulty_steps: int) -> None:
"""更新重量抽检产出的 Agent 步数。
Parameters
----------
item_id : str
题目唯一 ID。
difficulty_steps : int
Agent 完成该题所需步数。
"""
self._conn.execute(
"UPDATE question_gen_items SET difficulty_steps=? WHERE item_id=?",
(difficulty_steps, item_id),
)
self._conn.commit()
def get_run_stats(self, run_id: str) -> RunStats:
"""查询批次统计摘要。
Parameters
----------
run_id : str
批次唯一标识。
Returns
-------
RunStats
该批次的统计数据。
Raises
------
ValueError
run_id 不存在时抛出。
"""
row = self._conn.execute(
"SELECT total_slots, accepted, rejected, heavy_sampled "
"FROM question_gen_runs WHERE run_id=?",
(run_id,),
).fetchone()
if row is None:
raise ValueError(f"run_id 不存在: {run_id}")
return RunStats(
total_slots=row[0] or 0,
accepted=row[1] or 0,
rejected=row[2] or 0,
heavy_sampled=row[3] or 0,
)
def close(self) -> None:
"""关闭数据库连接。"""
self._conn.close()
logger.debug("QuestionGenStore 已关闭")
tests/unit/test_run_store.py
+244
View File
@@ -0,0 +1,244 @@
"""QuestionGenStore 单元测试 — SQLite 出题管线日志记录器。
覆盖 schema 幂等性、run 生命周期、item 写入与门判定更新、难度更新。
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from typing import TYPE_CHECKING
import pytest
if TYPE_CHECKING:
from pathlib import Path
from app.question_gen.run_store import QuestionGenStore, RunStats
# ---------------------------------------------------------------------------
# GateReport 测试替身（Task 6 尚未实现，此处构造兼容 duck-type）
# ---------------------------------------------------------------------------
class _Verdict(Enum):
"""门判定枚举替身。"""
PASS = "pass"
FAIL = "fail"
SKIP = "skip"
@dataclass(frozen=True)
class _GateResult:
"""单门判定结果替身。"""
verdict: _Verdict
reason: str
@dataclass
class _MockGateReport:
"""GateReport 测试替身，duck-type 兼容。"""
key_verify: _GateResult
blind_answer: _GateResult
multi_true: _GateResult
leak_test: _GateResult
@property
def passed(self) -> bool:
"""所有门均 PASS 或 SKIP 时视为通过。"""
results = [self.key_verify, self.blind_answer, self.multi_true, self.leak_test]
return all(r.verdict != _Verdict.FAIL for r in results)
@property
def reject_reason(self) -> str | None:
"""返回首个 FAIL 门的 reason，若全部通过则 None。"""
for r in [self.key_verify, self.blind_answer, self.multi_true, self.leak_test]:
if r.verdict == _Verdict.FAIL:
return r.reason
return None
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture()
def store(tmp_path: Path) -> QuestionGenStore:
"""创建临时 SQLite 上的 QuestionGenStore 实例。"""
db_path = tmp_path / "question_gen.db"
s = QuestionGenStore(db_path=db_path)
yield s
s.close()
# ---------------------------------------------------------------------------
# TestQuestionGenStore
# ---------------------------------------------------------------------------
class TestQuestionGenStore:
"""QuestionGenStore 全功能验证。"""
def test_schema_idempotent(self, tmp_path: Path) -> None:
"""连续两次初始化同一 db 不报错，表结构不变。"""
db_path = tmp_path / "idem.db"
s1 = QuestionGenStore(db_path=db_path)
s2 = QuestionGenStore(db_path=db_path)
# 第二次初始化不应抛异常
s2.record_run_start(run_id="r-idem", git_sha="abc123", config_snapshot="{}")
s1.close()
s2.close()
def test_record_run_lifecycle(self, store: QuestionGenStore) -> None:
"""start -> record items -> end -> get_stats 完整生命周期。"""
run_id = "run-001"
store.record_run_start(
run_id=run_id,
git_sha="deadbeef",
config_snapshot='{"n_slots": 240}',
)
# 写入几个 item 来模拟统计
for i in range(5):
store.record_item(
item_id=f"item-{i}",
run_id=run_id,
slot_id=f"slot-{i}",
video_id="v001",
family="retrieval",
task_type="Object Recognition",
skill_target="M1",
attempt=1,
question_text=f"问题{i}",
)
# 结束 run
stats = RunStats(total_slots=10, accepted=5, rejected=3, heavy_sampled=2)
store.record_run_end(run_id=run_id, status="completed", stats=stats)
# 查询 stats
retrieved = store.get_run_stats(run_id)
assert retrieved.total_slots == 10
assert retrieved.accepted == 5
assert retrieved.rejected == 3
assert retrieved.heavy_sampled == 2
def test_record_item_and_gates(self, store: QuestionGenStore) -> None:
"""record_item 后 update_gates 正确更新门判定列与 final_status。"""
run_id = "run-gates"
store.record_run_start(run_id=run_id, git_sha="aaa111", config_snapshot="{}")
item_id = "item-gate-1"
store.record_item(
item_id=item_id,
run_id=run_id,
slot_id="slot-0",
video_id="v002",
family="reasoning",
task_type="Action Sequence",
skill_target="M2",
attempt=1,
question_text="为什么这样做？",
)
# 所有门通过
report_pass = _MockGateReport(
key_verify=_GateResult(_Verdict.PASS, "ok"),
blind_answer=_GateResult(_Verdict.PASS, "ok"),
multi_true=_GateResult(_Verdict.SKIP, "n/a"),
leak_test=_GateResult(_Verdict.PASS, "ok"),
)
store.update_gates(item_id=item_id, report=report_pass)
# 验证数据库中的值
import sqlite3
conn = sqlite3.connect(str(store._db_path))
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT * FROM question_gen_items WHERE item_id=?", (item_id,)
).fetchone()
conn.close()
assert row["gate_key_verify"] == "pass"
assert row["gate_blind_answer"] == "pass"
assert row["gate_multi_true"] == "skip"
assert row["gate_leak_test"] == "pass"
assert row["gate_reject_reason"] is None
assert row["final_status"] == "accepted"
def test_update_gates_rejected(self, store: QuestionGenStore) -> None:
"""门判定失败时 final_status 为 rejected 且记录 reject_reason。"""
run_id = "run-rej"
store.record_run_start(run_id=run_id, git_sha="bbb222", config_snapshot="{}")
item_id = "item-rej-1"
store.record_item(
item_id=item_id,
run_id=run_id,
slot_id="slot-1",
video_id="v003",
family="visual",
task_type="Scene Transition",
skill_target="M3",
attempt=2,
question_text="画面中有什么？",
)
report_fail = _MockGateReport(
key_verify=_GateResult(_Verdict.PASS, "ok"),
blind_answer=_GateResult(_Verdict.FAIL, "答案可由常识推断"),
multi_true=_GateResult(_Verdict.SKIP, "n/a"),
leak_test=_GateResult(_Verdict.SKIP, "n/a"),
)
store.update_gates(item_id=item_id, report=report_fail)
import sqlite3
conn = sqlite3.connect(str(store._db_path))
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT * FROM question_gen_items WHERE item_id=?", (item_id,)
).fetchone()
conn.close()
assert row["final_status"] == "rejected"
assert row["gate_reject_reason"] == "答案可由常识推断"
def test_update_difficulty(self, store: QuestionGenStore) -> None:
"""update_difficulty 正确写入 difficulty_steps。"""
run_id = "run-diff"
store.record_run_start(run_id=run_id, git_sha="ccc333", config_snapshot="{}")
item_id = "item-diff-1"
store.record_item(
item_id=item_id,
run_id=run_id,
slot_id="slot-2",
video_id="v004",
family="enumeration",
task_type="Counting",
skill_target="M4",
attempt=1,
question_text="有多少个？",
)
store.update_difficulty(item_id=item_id, difficulty_steps=7)
import sqlite3
conn = sqlite3.connect(str(store._db_path))
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT difficulty_steps FROM question_gen_items WHERE item_id=?",
(item_id,),
).fetchone()
conn.close()
assert row["difficulty_steps"] == 7
def test_get_run_stats_not_found(self, store: QuestionGenStore) -> None:
"""查询不存在的 run_id 应报错。"""
with pytest.raises(ValueError, match="run_id"):
store.get_run_stats("nonexistent-run")