iomgaa
bdcc93d7de
PairPendingBuffer collects pair members by pair_id and emits units; write_accepted does tmp+os.replace atomic write; read_accepted rebuilds units and drops disk-dangling orphans. Pure helper; on_accept wiring is Phase 2 (real write point: adversarial_filter.write_final_bank).
200 lines
8.7 KiB
Python
200 lines
8.7 KiB
Python
"""accepted 题库的 pair 原子成对落盘 helper（纯件，不依赖 pipeline）。
|
||
|
||
三件可复用纯件供 Phase 2 新 pipeline 的 on_accept 回调复用：``PairPendingBuffer``
|
||
（按 pair_id 收齐才 emit 单元）、``write_accepted``（tmp + os.replace 原子写）、
|
||
``read_accepted``（聚合成单元并剔除磁盘悬挂孤儿）。
|
||
|
||
wiring 归属：本模块**只是纯件**，不接任何生成侧回调；真正的 on_accept wiring
|
||
**见 Phase 2**。当前项目真实 accepted 写入点是 ``adversarial_filter.write_final_bank``
|
||
（Phase 2 待重建的旧代码），本模块沿用其 tmp + os.replace 模式但不 import/不改动它。
|
||
|
||
unit_hash 校验：``QuestionUnit.unit_hash`` 目前恒为 ""（填充是 Phase 2 的事），且
|
||
``GeneratedQuestion`` 不携带 unit_hash 字段。故把"unit_hash 不一致→拒"落地为**同
|
||
pair_id 两成员的绑定一致性校验**（video_id / task_type / flip_axis），不一致即
|
||
fail-fast raise，语义等价——只有 payload 绑定一致的孪生对才允许聚合。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
from collections import defaultdict
|
||
from typing import TYPE_CHECKING
|
||
|
||
from loguru import logger
|
||
|
||
from core.types import GeneratedQuestion, QuestionUnit
|
||
|
||
if TYPE_CHECKING:
|
||
from pathlib import Path
|
||
|
||
|
||
def _check_pair_binding(first: GeneratedQuestion, second: GeneratedQuestion) -> None:
|
||
"""校验同 pair_id 两成员的绑定一致性，不一致 fail-fast（unit_hash 语义代偿）。
|
||
|
||
参数 first/second 为先后到达的孪生对成员。video_id / task_type / flip_axis 任一
|
||
不一致即 raise ValueError——绑定不一致的两条题目不构成同一 payload 的孪生对，
|
||
拒绝聚合而非静默兜底。
|
||
"""
|
||
mismatches: list[str] = []
|
||
if first.video_id != second.video_id:
|
||
mismatches.append(f"video_id: {first.video_id} != {second.video_id}")
|
||
if first.task_type != second.task_type:
|
||
mismatches.append(f"task_type: {first.task_type} != {second.task_type}")
|
||
if first.flip_axis != second.flip_axis:
|
||
mismatches.append(f"flip_axis: {first.flip_axis} != {second.flip_axis}")
|
||
if mismatches:
|
||
raise ValueError(
|
||
f"pair {first.pair_id} 两成员绑定不一致（" + "；".join(mismatches) + "），拒绝聚合"
|
||
)
|
||
|
||
|
||
class PairPendingBuffer:
|
||
"""按 pair_id 收齐孪生对才 emit 单元的有状态缓冲器。
|
||
|
||
喂题接口 ``add`` 逐条消费题目：single 立即 emit ``QuestionUnit.from_single``；
|
||
pair 成员先缓存，等同一 pair_id 的 original+mirror 都到齐才 emit 一个 pair 单元
|
||
（复用 ``QuestionUnit.from_pair`` 走 fail-fast 校验）。批次末尾用
|
||
``pending_orphans`` 检测"只落 P 未落 Q"的悬挂项。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
"""初始化空缓冲器。pending 以 pair_id 索引首个到达的孪生对成员。"""
|
||
self._pending: dict[str, GeneratedQuestion] = {}
|
||
|
||
def add(self, q: GeneratedQuestion) -> QuestionUnit | None:
|
||
"""喂入一条题目 q，返回本次凑齐的单元或 None（pair 尚未配齐）。
|
||
|
||
single 立即返回 kind="single" 单元；pair 首个成员缓存并返回 None，第二个
|
||
成员到齐后返回 kind="pair" 单元。同 pair_id 两成员绑定不一致或角色非法（如
|
||
两个 original）即 raise ValueError。
|
||
|
||
关键实现:配齐后先 ``_check_pair_binding`` 显式绑定校验（把不一致落成清晰
|
||
ValueError，不依赖会被 ``-O`` 剥除的 assert），再 ``QuestionUnit.from_pair``
|
||
组装。
|
||
"""
|
||
if not q.pair_id:
|
||
return QuestionUnit.from_single(q)
|
||
|
||
partner = self._pending.get(q.pair_id)
|
||
if partner is None:
|
||
self._pending[q.pair_id] = q
|
||
return None
|
||
|
||
del self._pending[q.pair_id]
|
||
_check_pair_binding(partner, q)
|
||
original, mirror = _order_pair(partner, q)
|
||
return QuestionUnit.from_pair(original, mirror)
|
||
|
||
def pending_orphans(self) -> list[GeneratedQuestion]:
|
||
"""返回当前仍未配齐的孤儿成员（供批次末尾检测悬挂项）。
|
||
|
||
返回:
|
||
尚在缓冲、未等到伙伴的孪生对成员列表（按 pair_id 插入顺序）。
|
||
"""
|
||
return list(self._pending.values())
|
||
|
||
|
||
def _order_pair(
|
||
a: GeneratedQuestion, b: GeneratedQuestion
|
||
) -> tuple[GeneratedQuestion, GeneratedQuestion]:
|
||
"""按 question_role 把两成员 a/b 定序为 (original, mirror)。
|
||
|
||
两成员不构成恰好 1 original + 1 mirror（角色缺失或重复）即 raise ValueError，
|
||
防 next(...) 静默 StopIteration。
|
||
"""
|
||
originals = [q for q in (a, b) if q.question_role == "pair_original"]
|
||
mirrors = [q for q in (a, b) if q.question_role == "pair_mirror"]
|
||
if len(originals) != 1 or len(mirrors) != 1:
|
||
raise ValueError(
|
||
f"pair {a.pair_id} 角色非法：original={len(originals)} mirror={len(mirrors)}，"
|
||
"需各恰好 1 条"
|
||
)
|
||
return originals[0], mirrors[0]
|
||
|
||
|
||
def _is_intact_pair(pair_id: str, group: list[GeneratedQuestion]) -> bool:
|
||
"""判定同 pair_id 分组 group 是否为完整合法孪生对（恰好 1 original + 1 mirror）。
|
||
|
||
完整为 True；否则 warn（含成员构成）并返回 False，由调用方剔除该悬挂 unit。
|
||
"""
|
||
originals = sum(1 for q in group if q.question_role == "pair_original")
|
||
mirrors = sum(1 for q in group if q.question_role == "pair_mirror")
|
||
if len(group) == 2 and originals == 1 and mirrors == 1:
|
||
return True
|
||
logger.warning(
|
||
"磁盘悬挂孤儿 pair {}：成员数={}（original={} mirror={}），剔除该 unit",
|
||
pair_id,
|
||
len(group),
|
||
originals,
|
||
mirrors,
|
||
)
|
||
return False
|
||
|
||
|
||
def _drop_dangling_pairs(questions: list[GeneratedQuestion]) -> list[GeneratedQuestion]:
|
||
"""剔除磁盘上收不齐 2 条 / 角色非法的悬挂孤儿 pair，告警不静默。
|
||
|
||
与 T6 inference 的 ``_drop_orphan_pairs`` 语义一致：磁盘上"只落 P 未落 Q"是
|
||
业务上合法的悬挂项（非结构损坏），warn+drop 而非 raise，使后续 build_units 只
|
||
面对合法孪生对。single 全保留。
|
||
|
||
参数:
|
||
questions: 从磁盘读回的题目列表（可混含 single 与孪生对成员）。
|
||
|
||
返回:
|
||
可安全交给 build_units 的题目列表（single 全保留，pair 仅保留合法成对者）。
|
||
"""
|
||
by_pair: dict[str, list[GeneratedQuestion]] = defaultdict(list)
|
||
singles: list[GeneratedQuestion] = []
|
||
for q in questions:
|
||
if q.pair_id:
|
||
by_pair[q.pair_id].append(q)
|
||
else:
|
||
singles.append(q)
|
||
|
||
kept_pairs = [q for pid, grp in by_pair.items() if _is_intact_pair(pid, grp) for q in grp]
|
||
return singles + kept_pairs
|
||
|
||
|
||
def write_accepted(path: Path, units: list[QuestionUnit]) -> None:
|
||
"""把 units 全量原子写到 path（tmp + os.replace），孪生对两题相邻落盘。
|
||
|
||
parent 不存在则自动创建；任一 pair 单元结构非法（size≠2）落盘前 fail-fast raise。
|
||
|
||
关键实现:沿用 ``write_final_bank`` 的原子写模式（先写同目录 ``.tmp`` 再
|
||
``os.replace`` 覆盖，保证读到的 JSON 恒完整）。序列化复用 T9 pools.py 的
|
||
``_q_to_dict``（唯一 GeneratedQuestion↔dict schema，含 pair 四字段），函数内
|
||
import 规避 app.question_gen↔app.harness 循环依赖。
|
||
"""
|
||
from app.harness.pools import _q_to_dict
|
||
from app.harness.question_units import flatten_units, validate_units
|
||
|
||
validate_units(units)
|
||
records = [_q_to_dict(q) for q in flatten_units(units)]
|
||
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = path.with_suffix(".tmp")
|
||
tmp.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
os.replace(str(tmp), str(path))
|
||
logger.info("accepted 题库全量原子写: {} 单元 / {} 题 → {}", len(units), len(records), path)
|
||
|
||
|
||
def read_accepted(path: Path) -> list[QuestionUnit]:
|
||
"""读回 path 的 accepted JSON → 聚合为单元列表，剔除磁盘悬挂孤儿。
|
||
|
||
返回校验通过的单元列表（single + 合法 pair）；磁盘上"只落 P 未落 Q"的悬挂孤儿
|
||
被 warn+drop，不进结果、不 raise。
|
||
|
||
关键实现:反序列化复用 T9 pools.py 的 ``_dict_to_q``（pair 四字段 .get 兼容），
|
||
函数内 import 规避循环依赖。先 ``_drop_dangling_pairs`` 剔除悬挂孤儿，再
|
||
``build_units`` 聚合、``validate_units`` 二次防御闸门。
|
||
"""
|
||
from app.harness.pools import _dict_to_q
|
||
from app.harness.question_units import build_units, validate_units
|
||
|
||
raw = json.loads(path.read_text(encoding="utf-8"))
|
||
questions = [_dict_to_q(d) for d in raw]
|
||
kept = _drop_dangling_pairs(questions)
|
||
return validate_units(build_units(kept))
|