1072 lines
36 KiB
Python
1072 lines
36 KiB
Python
"""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
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# DEPRECATED(v3): v2 生成落库，Phase 2 生成替换后删；过渡期与 v3 表并存
|
||
_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,
|
||
ended_at TEXT,
|
||
status TEXT NOT NULL DEFAULT 'running',
|
||
total_slots INTEGER,
|
||
accepted INTEGER,
|
||
rejected INTEGER,
|
||
heavy_sampled INTEGER
|
||
);
|
||
"""
|
||
|
||
# DEPRECATED(v3): v2 生成落库，Phase 2 生成替换后删；过渡期与 v3 表并存
|
||
_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,
|
||
sub_pattern TEXT,
|
||
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,
|
||
selector_scores TEXT,
|
||
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);",
|
||
]
|
||
|
||
# DEPRECATED(v3): v2 生成落库，Phase 2 生成替换后删；过渡期与 v3 表并存
|
||
_DDL_VERDICTS = """
|
||
CREATE TABLE IF NOT EXISTS adversarial_verdicts (
|
||
question_id TEXT NOT NULL,
|
||
question_hash TEXT NOT NULL,
|
||
stage TEXT NOT NULL,
|
||
round INTEGER NOT NULL,
|
||
agent_prediction TEXT,
|
||
agent_correct INTEGER,
|
||
verdict TEXT NOT NULL,
|
||
pair_id TEXT,
|
||
agent_config TEXT NOT NULL,
|
||
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||
PRIMARY KEY (question_id, question_hash, stage)
|
||
);
|
||
"""
|
||
|
||
_DDL_VERDICTS_INDEXES = [
|
||
"CREATE INDEX IF NOT EXISTS idx_av_qid ON adversarial_verdicts(question_id);",
|
||
"CREATE INDEX IF NOT EXISTS idx_av_verdict ON adversarial_verdicts(verdict);",
|
||
"CREATE INDEX IF NOT EXISTS idx_av_round ON adversarial_verdicts(round);",
|
||
]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# v3 出题领域观测表 DDL（facts/unit_verdict/collapse_metrics/quarantine/resume_state）
|
||
#
|
||
# 时间戳边界（Codex I-4）：v3 新表的 ts 列一律由 insert 方法外部传入，DDL 中不设
|
||
# DEFAULT (datetime('now'))、方法体不调 datetime.now()，以保证观测幂等可复现——
|
||
# 断点续跑重放时不因进程内时钟漂移产生脏数据。resume_state 无 ts 列（schema 未列）。
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_DDL_FACTS = """
|
||
CREATE TABLE IF NOT EXISTS facts (
|
||
fact_id TEXT PRIMARY KEY,
|
||
video_id TEXT,
|
||
segment_id TEXT,
|
||
subject TEXT,
|
||
action TEXT,
|
||
object TEXT,
|
||
frame_ids TEXT,
|
||
polarity TEXT,
|
||
fact_type TEXT,
|
||
difficulty_tier INTEGER,
|
||
verifier_refs TEXT,
|
||
cross_agree INTEGER CHECK(cross_agree IN (0, 1)),
|
||
negative_at_target TEXT,
|
||
session_id TEXT,
|
||
ts TEXT
|
||
);
|
||
"""
|
||
|
||
_DDL_UNIT_VERDICT = """
|
||
CREATE TABLE IF NOT EXISTS unit_verdict (
|
||
unit_id TEXT NOT NULL,
|
||
pair_id TEXT,
|
||
sub_pattern TEXT,
|
||
stage INTEGER NOT NULL CHECK(stage BETWEEN 1 AND 6),
|
||
verdict TEXT CHECK(verdict IN ('pass', 'fail', 'abstain')),
|
||
reason TEXT,
|
||
metric_value REAL,
|
||
model TEXT,
|
||
session_id TEXT,
|
||
ts TEXT,
|
||
PRIMARY KEY (unit_id, stage)
|
||
);
|
||
"""
|
||
|
||
_DDL_COLLAPSE_METRICS = """
|
||
CREATE TABLE IF NOT EXISTS collapse_metrics (
|
||
pair_id TEXT PRIMARY KEY,
|
||
text_only_acc REAL,
|
||
single_frame_acc REAL,
|
||
placebo_drop REAL,
|
||
majority_vote_hit REAL,
|
||
slot_chi2 REAL,
|
||
distractor_min_dist REAL,
|
||
multiformat_consistency REAL,
|
||
subtitle_answerability REAL,
|
||
ts TEXT
|
||
);
|
||
"""
|
||
|
||
_DDL_QUARANTINE = """
|
||
CREATE TABLE IF NOT EXISTS quarantine (
|
||
content_fingerprint TEXT PRIMARY KEY,
|
||
sub_pattern TEXT,
|
||
quarantine_reason TEXT,
|
||
round_no INTEGER,
|
||
ts TEXT
|
||
);
|
||
"""
|
||
|
||
_DDL_RESUME_STATE = """
|
||
CREATE TABLE IF NOT EXISTS resume_state (
|
||
unit_id TEXT PRIMARY KEY,
|
||
status TEXT CHECK(status IN ('pending', 'accepted', 'rejected')),
|
||
config_fingerprint TEXT,
|
||
seq_offset INTEGER
|
||
);
|
||
"""
|
||
|
||
_DDL_V3_TABLES = [
|
||
_DDL_FACTS,
|
||
_DDL_UNIT_VERDICT,
|
||
_DDL_COLLAPSE_METRICS,
|
||
_DDL_QUARANTINE,
|
||
_DDL_RESUME_STATE,
|
||
]
|
||
|
||
_DDL_V3_INDEXES = [
|
||
"CREATE INDEX IF NOT EXISTS idx_facts_video ON facts(video_id);",
|
||
"CREATE INDEX IF NOT EXISTS idx_facts_session ON facts(session_id);",
|
||
# unit_verdict(unit_id) 无需独立索引：主键 (unit_id, stage) 的左前缀已覆盖 unit_id 查询。
|
||
"CREATE INDEX IF NOT EXISTS idx_uv_sub_pattern ON unit_verdict(sub_pattern);",
|
||
"CREATE INDEX IF NOT EXISTS idx_quar_sub_pattern ON quarantine(sub_pattern);",
|
||
"CREATE INDEX IF NOT EXISTS idx_resume_status ON resume_state(status);",
|
||
]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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.execute(_DDL_VERDICTS)
|
||
for idx_sql in _DDL_VERDICTS_INDEXES:
|
||
self._conn.execute(idx_sql)
|
||
for ddl in _DDL_V3_TABLES:
|
||
self._conn.execute(ddl)
|
||
for idx_sql in _DDL_V3_INDEXES:
|
||
self._conn.execute(idx_sql)
|
||
self._conn.commit()
|
||
|
||
# 幂等迁移：为已有表加 sub_pattern / selector_scores 列
|
||
cols = {r[1] for r in self._conn.execute("PRAGMA table_info(question_gen_items)")}
|
||
if "sub_pattern" not in cols:
|
||
self._conn.execute("ALTER TABLE question_gen_items ADD COLUMN sub_pattern TEXT")
|
||
self._conn.commit()
|
||
if "selector_scores" not in cols:
|
||
self._conn.execute("ALTER TABLE question_gen_items ADD COLUMN selector_scores TEXT")
|
||
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")
|
||
cursor = self._conn.execute(
|
||
"""
|
||
UPDATE question_gen_runs
|
||
SET ended_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()
|
||
if cursor.rowcount == 0:
|
||
raise ValueError(f"run_id 不存在: {run_id}")
|
||
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,
|
||
sub_pattern: str | None = None,
|
||
) -> 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
|
||
题目文本。
|
||
sub_pattern : str | None
|
||
子模式标识（如有）。
|
||
"""
|
||
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, sub_pattern, created_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
item_id,
|
||
run_id,
|
||
slot_id,
|
||
video_id,
|
||
family,
|
||
task_type,
|
||
skill_target,
|
||
attempt,
|
||
question_text,
|
||
sub_pattern,
|
||
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"
|
||
cursor = 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()
|
||
if cursor.rowcount == 0:
|
||
raise ValueError(f"item_id 不存在: {item_id}")
|
||
|
||
def mark_item_rejected(self, item_id: str, reason: str) -> None:
|
||
"""将已记录的 item 标记为 rejected（用于门控外的拒绝场景，如去重）。
|
||
|
||
Parameters
|
||
----------
|
||
item_id : str
|
||
题目唯一 ID。
|
||
reason : str
|
||
拒绝原因描述。
|
||
"""
|
||
cursor = self._conn.execute(
|
||
"UPDATE question_gen_items SET final_status='rejected', gate_reject_reason=? "
|
||
"WHERE item_id=?",
|
||
(reason, item_id),
|
||
)
|
||
self._conn.commit()
|
||
if cursor.rowcount == 0:
|
||
raise ValueError(f"item_id 不存在: {item_id}")
|
||
|
||
def update_difficulty(self, item_id: str, difficulty_steps: int) -> None:
|
||
"""更新重量抽检产出的 Agent 步数。
|
||
|
||
Parameters
|
||
----------
|
||
item_id : str
|
||
题目唯一 ID。
|
||
difficulty_steps : int
|
||
Agent 完成该题所需步数。
|
||
"""
|
||
cursor = self._conn.execute(
|
||
"UPDATE question_gen_items SET difficulty_steps=? WHERE item_id=?",
|
||
(difficulty_steps, item_id),
|
||
)
|
||
self._conn.commit()
|
||
if cursor.rowcount == 0:
|
||
raise ValueError(f"item_id 不存在: {item_id}")
|
||
|
||
def record_verdict(
|
||
self,
|
||
*,
|
||
question_id: str,
|
||
question_hash: str,
|
||
stage: str,
|
||
round: int, # noqa: A002 — 与设计列名一致，仅 kwargs 传入无遮蔽风险
|
||
agent_prediction: str | None,
|
||
agent_correct: bool | None,
|
||
verdict: str,
|
||
pair_id: str | None,
|
||
agent_config: str,
|
||
) -> None:
|
||
"""写入一条 agent 门判定（同 (question_id, question_hash, stage) upsert）。
|
||
|
||
每次写入立即 commit，保证崩溃安全（进程中断最多丢失当前未提交的一条）。
|
||
|
||
Parameters
|
||
----------
|
||
question_id, question_hash, stage : str
|
||
续跑主键三元组（stage ∈ cheat|flip_original|flip_mirror）。
|
||
round : int
|
||
过滤轮次。
|
||
agent_prediction : str | None
|
||
agent 预测答案字母。
|
||
agent_correct : bool | None
|
||
作弊门是否答对（翻转门 stage 可为 None）。
|
||
verdict : str
|
||
passed | filtered_too_easy | filtered_no_flip | flip_skipped。
|
||
pair_id : str | None
|
||
关联原题与镜像题。
|
||
agent_config : str
|
||
agent 配置指纹（skill_mode/max_steps/model）。
|
||
"""
|
||
self._conn.execute(
|
||
"""
|
||
INSERT INTO adversarial_verdicts
|
||
(question_id, question_hash, stage, round, agent_prediction,
|
||
agent_correct, verdict, pair_id, agent_config)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(question_id, question_hash, stage) DO UPDATE SET
|
||
round=excluded.round,
|
||
agent_prediction=excluded.agent_prediction,
|
||
agent_correct=excluded.agent_correct,
|
||
verdict=excluded.verdict,
|
||
pair_id=excluded.pair_id,
|
||
agent_config=excluded.agent_config,
|
||
created_at=datetime('now')
|
||
""",
|
||
(
|
||
question_id,
|
||
question_hash,
|
||
stage,
|
||
round,
|
||
agent_prediction,
|
||
None if agent_correct is None else int(agent_correct),
|
||
verdict,
|
||
pair_id,
|
||
agent_config,
|
||
),
|
||
)
|
||
self._conn.commit()
|
||
|
||
def completed_stages(self, question_id: str, question_hash: str, agent_config: str) -> set[str]:
|
||
"""返回该题在当前 hash+config 下已完成的 stage 集合（续跑用）。
|
||
|
||
Parameters
|
||
----------
|
||
question_id : str
|
||
题目唯一标识。
|
||
question_hash : str
|
||
当前题面指纹；hash 不匹配的旧行视为未完成，需重跑试答。
|
||
agent_config : str
|
||
当前 agent 配置指纹。
|
||
|
||
Returns
|
||
-------
|
||
set[str]
|
||
已完成的 stage 名称集合。
|
||
"""
|
||
rows = self._conn.execute(
|
||
"SELECT stage FROM adversarial_verdicts "
|
||
"WHERE question_id=? AND question_hash=? AND agent_config=?",
|
||
(question_id, question_hash, agent_config),
|
||
).fetchall()
|
||
return {r[0] for r in rows}
|
||
|
||
def invalidate_stale_config(self, question_id: str, agent_config: str) -> None:
|
||
"""agent_config 变化时，删除该题所有非当前 config 的旧 verdict。
|
||
|
||
Parameters
|
||
----------
|
||
question_id : str
|
||
题目唯一标识。
|
||
agent_config : str
|
||
当前 agent 配置指纹；保留该 config 行，其余全部删除。
|
||
"""
|
||
self._conn.execute(
|
||
"DELETE FROM adversarial_verdicts WHERE question_id=? AND agent_config!=?",
|
||
(question_id, agent_config),
|
||
)
|
||
self._conn.commit()
|
||
|
||
def cheat_agent_accuracy(self, round_no: int) -> float:
|
||
"""某轮作弊门 agent 正确率（agent_correct 聚合），无数据返 0.0。
|
||
|
||
Parameters
|
||
----------
|
||
round_no : int
|
||
过滤轮次。
|
||
|
||
Returns
|
||
-------
|
||
float
|
||
该轮 stage='cheat' 的 agent_correct 平均值；无数据时返回 0.0。
|
||
"""
|
||
row = self._conn.execute(
|
||
"SELECT AVG(agent_correct) FROM adversarial_verdicts WHERE stage='cheat' AND round=?",
|
||
(round_no,),
|
||
).fetchone()
|
||
return float(row[0]) if row and row[0] is not None else 0.0
|
||
|
||
def final_passed_question_ids(self, hash_by_qid: dict[str, str], agent_config: str) -> set[str]:
|
||
"""在当前 hash+config 下通过两门的 question_id 集合（final JSON 全量重建用）。
|
||
|
||
终判规则（防 stale 泄漏）：仅当该题在 **当前 question_hash + 当前
|
||
agent_config** 下同时满足——存在 stage='cheat' 且 verdict='passed'
|
||
（agent 答错=不太简单），且不存在任何 stage 的 verdict='filtered_no_flip'
|
||
（未被翻转门剔除）——才计入 final-passed。stale hash / stale config 的旧行
|
||
因不匹配传入的 (qid, hash, config) 天然被排除，绝不泄漏进最终题库。
|
||
|
||
Parameters
|
||
----------
|
||
hash_by_qid : dict[str, str]
|
||
question_id → 当前 question_hash 映射（来自本轮 all_questions）。
|
||
agent_config : str
|
||
当前 agent 配置指纹。
|
||
|
||
Returns
|
||
-------
|
||
set[str]
|
||
终判 passed 的 question_id 集合。
|
||
"""
|
||
passed: set[str] = set()
|
||
for qid, qhash in hash_by_qid.items():
|
||
rows = self._conn.execute(
|
||
"SELECT stage, verdict FROM adversarial_verdicts "
|
||
"WHERE question_id=? AND question_hash=? AND agent_config=?",
|
||
(qid, qhash, agent_config),
|
||
).fetchall()
|
||
if not rows:
|
||
continue
|
||
cheat_passed = any(stage == "cheat" and verdict == "passed" for stage, verdict in rows)
|
||
no_flip = any(verdict == "filtered_no_flip" for _, verdict in rows)
|
||
if cheat_passed and not no_flip:
|
||
passed.add(qid)
|
||
return passed
|
||
|
||
def update_selector_scores(self, item_id: str, selector_scores_json: str) -> None:
|
||
"""写入 grounded selector 打分观测（JSON 字符串）。
|
||
|
||
Parameters
|
||
----------
|
||
item_id : str
|
||
题目唯一 ID。
|
||
selector_scores_json : str
|
||
观测 JSON：correct_score / chosen / pool_size / anneal_rounds / hard_fail。
|
||
|
||
Raises
|
||
------
|
||
ValueError
|
||
item_id 不存在时抛出。
|
||
"""
|
||
cursor = self._conn.execute(
|
||
"UPDATE question_gen_items SET selector_scores=? WHERE item_id=?",
|
||
(selector_scores_json, item_id),
|
||
)
|
||
self._conn.commit()
|
||
if cursor.rowcount == 0:
|
||
raise ValueError(f"item_id 不存在: {item_id}")
|
||
|
||
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 load_progress(self) -> dict[str, str]:
|
||
"""加载已接受 slot 的进度映射（用于断点续跑）。
|
||
|
||
从最近一次 running 状态的批次中，只读取 accepted 的 slot。
|
||
rejected 的 slot 不纳入 progress，以便重跑时重新尝试。
|
||
|
||
Returns
|
||
-------
|
||
dict[str, str]
|
||
{slot_id: "accepted"} 映射。无进度时返回空 dict。
|
||
"""
|
||
row = self._conn.execute(
|
||
"SELECT run_id FROM question_gen_runs WHERE status='running' "
|
||
"ORDER BY started_at DESC LIMIT 1",
|
||
).fetchone()
|
||
|
||
if row is None:
|
||
return {}
|
||
|
||
run_id = row[0]
|
||
rows = self._conn.execute(
|
||
"SELECT DISTINCT slot_id FROM question_gen_items "
|
||
"WHERE run_id=? AND final_status='accepted'",
|
||
(run_id,),
|
||
).fetchall()
|
||
|
||
return {row[0]: "accepted" for row in rows}
|
||
|
||
# -----------------------------------------------------------------------
|
||
# v3 出题领域观测写入（ts 一律外部传入，禁进程内 now，保幂等可复现）
|
||
# -----------------------------------------------------------------------
|
||
|
||
def insert_fact(
|
||
self,
|
||
*,
|
||
fact_id: str,
|
||
video_id: str,
|
||
segment_id: str,
|
||
subject: str,
|
||
action: str,
|
||
object: str, # noqa: A002 — 与设计列名 object 一致，仅 kwargs 传入无遮蔽风险
|
||
frame_ids: str,
|
||
polarity: str,
|
||
fact_type: str,
|
||
difficulty_tier: int,
|
||
verifier_refs: str,
|
||
cross_agree: int,
|
||
negative_at_target: str,
|
||
session_id: str,
|
||
ts: str,
|
||
) -> None:
|
||
"""写入一条帧感知抽取 Fact（facts 表，主键 fact_id）。
|
||
|
||
Parameters
|
||
----------
|
||
fact_id : str
|
||
Fact 唯一标识（UUID）。
|
||
video_id, segment_id : str
|
||
溯源：所属视频与片段。
|
||
subject, action, object : str
|
||
结构化绑定三元组。
|
||
frame_ids : str
|
||
感知所用帧 ID 的 JSON 数组字符串。
|
||
polarity : str
|
||
事实极性（真/假）。
|
||
fact_type : str
|
||
事实类型（binding/state/manner/order/evidence）。
|
||
difficulty_tier : int
|
||
感知难度分层。
|
||
verifier_refs : str
|
||
双 VLM（qwen/MiniMax）各自裁决的 JSON 字符串。
|
||
cross_agree : int
|
||
双 VLM 是否一致（0/1）。
|
||
negative_at_target : str
|
||
目标点为假的核实结果。
|
||
session_id : str
|
||
epoch/step 关联 ID。
|
||
ts : str
|
||
观测时间戳 ISO 字符串，由调用方外部传入（禁进程内 now）。
|
||
"""
|
||
self._conn.execute(
|
||
"""
|
||
INSERT INTO facts
|
||
(fact_id, video_id, segment_id, subject, action, object, frame_ids,
|
||
polarity, fact_type, difficulty_tier, verifier_refs, cross_agree,
|
||
negative_at_target, session_id, ts)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
fact_id,
|
||
video_id,
|
||
segment_id,
|
||
subject,
|
||
action,
|
||
object,
|
||
frame_ids,
|
||
polarity,
|
||
fact_type,
|
||
difficulty_tier,
|
||
verifier_refs,
|
||
cross_agree,
|
||
negative_at_target,
|
||
session_id,
|
||
ts,
|
||
),
|
||
)
|
||
self._conn.commit()
|
||
logger.debug("facts 已写入: fact_id={}", fact_id)
|
||
|
||
def insert_unit_verdict(
|
||
self,
|
||
*,
|
||
unit_id: str,
|
||
pair_id: str | None,
|
||
sub_pattern: str,
|
||
stage: int,
|
||
verdict: str,
|
||
reason: str,
|
||
metric_value: float | None,
|
||
model: str | None,
|
||
session_id: str,
|
||
ts: str,
|
||
) -> None:
|
||
"""写入六层验证某一层的裁决（unit_verdict 表，主键 (unit_id, stage)）。
|
||
|
||
同一 (unit_id, stage) 重跑时以 upsert 覆盖旧行——每 unit 每层仅保留最新一条，
|
||
便于断点续跑重放而不残留过期裁决。
|
||
|
||
覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*`` **整行覆盖**全部非键列
|
||
（含可空的 metric_value/model 及 ts）——只保留每层最新完整裁决，不支持部分字段
|
||
更新。调用方每次必须传完整行，否则会用 metric_value=None/model=None 误抹先前非空值。
|
||
|
||
Parameters
|
||
----------
|
||
unit_id : str
|
||
pair/single 单元标识。
|
||
pair_id : str | None
|
||
所属 pair 关联标识（single 单元可为 None）。
|
||
sub_pattern : str
|
||
6 子模式之一。
|
||
stage : int
|
||
验证层号（1-6）。
|
||
verdict : str
|
||
该层裁决（pass/fail/abstain）。
|
||
reason : str
|
||
拒因描述。
|
||
metric_value : float | None
|
||
该层量化值（无量化值时为 None）。
|
||
model : str | None
|
||
裁判模型名（无模型判定时为 None）。
|
||
session_id : str
|
||
epoch/step 关联 ID。
|
||
ts : str
|
||
观测时间戳 ISO 字符串，由调用方外部传入（禁进程内 now）。
|
||
"""
|
||
self._conn.execute(
|
||
"""
|
||
INSERT INTO unit_verdict
|
||
(unit_id, pair_id, sub_pattern, stage, verdict, reason,
|
||
metric_value, model, session_id, ts)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(unit_id, stage) DO UPDATE SET
|
||
pair_id=excluded.pair_id,
|
||
sub_pattern=excluded.sub_pattern,
|
||
verdict=excluded.verdict,
|
||
reason=excluded.reason,
|
||
metric_value=excluded.metric_value,
|
||
model=excluded.model,
|
||
session_id=excluded.session_id,
|
||
ts=excluded.ts
|
||
""",
|
||
(
|
||
unit_id,
|
||
pair_id,
|
||
sub_pattern,
|
||
stage,
|
||
verdict,
|
||
reason,
|
||
metric_value,
|
||
model,
|
||
session_id,
|
||
ts,
|
||
),
|
||
)
|
||
self._conn.commit()
|
||
logger.debug("unit_verdict 已写入: unit_id={}, stage={}", unit_id, stage)
|
||
|
||
def insert_collapse_metrics(
|
||
self,
|
||
*,
|
||
pair_id: str,
|
||
text_only_acc: float,
|
||
single_frame_acc: float,
|
||
placebo_drop: float,
|
||
majority_vote_hit: float,
|
||
slot_chi2: float,
|
||
distractor_min_dist: float,
|
||
multiformat_consistency: float,
|
||
subtitle_answerability: float,
|
||
ts: str,
|
||
) -> None:
|
||
"""写入配对坍缩度量（collapse_metrics 表，主键 pair_id）。
|
||
|
||
同一 pair_id 重算时以 upsert 覆盖旧行——每 pair 仅保留最新一组度量。
|
||
|
||
覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*`` **整行覆盖**全部非键列（含 ts），
|
||
不支持部分字段更新；调用方每次必须传完整度量行。
|
||
|
||
Parameters
|
||
----------
|
||
pair_id : str
|
||
配对唯一标识。
|
||
text_only_acc, single_frame_acc, placebo_drop : float
|
||
模型探针类度量。
|
||
majority_vote_hit, slot_chi2, distractor_min_dist : float
|
||
纯结构类度量。
|
||
multiformat_consistency, subtitle_answerability : float
|
||
探针类度量。
|
||
ts : str
|
||
观测时间戳 ISO 字符串，由调用方外部传入（禁进程内 now）。
|
||
"""
|
||
self._conn.execute(
|
||
"""
|
||
INSERT INTO collapse_metrics
|
||
(pair_id, text_only_acc, single_frame_acc, placebo_drop,
|
||
majority_vote_hit, slot_chi2, distractor_min_dist,
|
||
multiformat_consistency, subtitle_answerability, ts)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(pair_id) DO UPDATE SET
|
||
text_only_acc=excluded.text_only_acc,
|
||
single_frame_acc=excluded.single_frame_acc,
|
||
placebo_drop=excluded.placebo_drop,
|
||
majority_vote_hit=excluded.majority_vote_hit,
|
||
slot_chi2=excluded.slot_chi2,
|
||
distractor_min_dist=excluded.distractor_min_dist,
|
||
multiformat_consistency=excluded.multiformat_consistency,
|
||
subtitle_answerability=excluded.subtitle_answerability,
|
||
ts=excluded.ts
|
||
""",
|
||
(
|
||
pair_id,
|
||
text_only_acc,
|
||
single_frame_acc,
|
||
placebo_drop,
|
||
majority_vote_hit,
|
||
slot_chi2,
|
||
distractor_min_dist,
|
||
multiformat_consistency,
|
||
subtitle_answerability,
|
||
ts,
|
||
),
|
||
)
|
||
self._conn.commit()
|
||
logger.debug("collapse_metrics 已写入: pair_id={}", pair_id)
|
||
|
||
def quarantine(
|
||
self,
|
||
*,
|
||
content_fingerprint: str,
|
||
sub_pattern: str,
|
||
quarantine_reason: str,
|
||
round_no: int,
|
||
ts: str,
|
||
) -> None:
|
||
"""将失败题的内容指纹写入隔离区黑名单（quarantine 表，主键 content_fingerprint）。
|
||
|
||
同一 content_fingerprint 重复调用以 upsert 覆盖——保证语义相同题面只占一行，
|
||
补构造前查此表当黑名单去重。覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*``
|
||
**整行覆盖**全部非键列（含 ts），不支持部分字段更新。
|
||
|
||
Parameters
|
||
----------
|
||
content_fingerprint : str
|
||
题面语义内容指纹（去重键）。
|
||
sub_pattern : str
|
||
题目所属子模式。
|
||
quarantine_reason : str
|
||
隔离原因。
|
||
round_no : int
|
||
隔离发生的过滤轮次。
|
||
ts : str
|
||
观测时间戳 ISO 字符串，由调用方外部传入（禁进程内 now）。
|
||
"""
|
||
self._conn.execute(
|
||
"""
|
||
INSERT INTO quarantine
|
||
(content_fingerprint, sub_pattern, quarantine_reason, round_no, ts)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
ON CONFLICT(content_fingerprint) DO UPDATE SET
|
||
sub_pattern=excluded.sub_pattern,
|
||
quarantine_reason=excluded.quarantine_reason,
|
||
round_no=excluded.round_no,
|
||
ts=excluded.ts
|
||
""",
|
||
(content_fingerprint, sub_pattern, quarantine_reason, round_no, ts),
|
||
)
|
||
self._conn.commit()
|
||
logger.debug("quarantine 已写入: fingerprint={}", content_fingerprint)
|
||
|
||
def upsert_resume_state(
|
||
self,
|
||
*,
|
||
unit_id: str,
|
||
status: str,
|
||
config_fingerprint: str,
|
||
seq_offset: int,
|
||
) -> None:
|
||
"""写入/更新断点续跑状态（resume_state 表，主键 unit_id）。
|
||
|
||
同一 unit_id 覆盖更新——每单元仅保留最新续跑状态。config_fingerprint 变化时
|
||
由调用方据此判定旧进度作废并重跑。本表无 ts 列（schema 未定义），但方法体仍禁
|
||
进程内 now，保持 v3 观测一致的幂等可复现语义。
|
||
|
||
覆盖语义：ON CONFLICT DO UPDATE 用 ``excluded.*`` **整行覆盖**全部非键列，不支持
|
||
部分字段更新；调用方每次必须传完整状态行。
|
||
|
||
Parameters
|
||
----------
|
||
unit_id : str
|
||
单元唯一标识。
|
||
status : str
|
||
续跑状态（pending/accepted/rejected）。
|
||
config_fingerprint : str
|
||
求解器/裁判 config 指纹，变更时旧进度作废。
|
||
seq_offset : int
|
||
补构造续编偏移，防止编号相撞。
|
||
"""
|
||
self._conn.execute(
|
||
"""
|
||
INSERT INTO resume_state
|
||
(unit_id, status, config_fingerprint, seq_offset)
|
||
VALUES (?, ?, ?, ?)
|
||
ON CONFLICT(unit_id) DO UPDATE SET
|
||
status=excluded.status,
|
||
config_fingerprint=excluded.config_fingerprint,
|
||
seq_offset=excluded.seq_offset
|
||
""",
|
||
(unit_id, status, config_fingerprint, seq_offset),
|
||
)
|
||
self._conn.commit()
|
||
logger.debug("resume_state 已写入: unit_id={}, status={}", unit_id, status)
|
||
|
||
def close(self) -> None:
|
||
"""关闭数据库连接。"""
|
||
self._conn.close()
|
||
logger.debug("QuestionGenStore 已关闭")
|