From 8b9e8aa19f0b561be18078c2d3c11820d34846f8 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 14 Jul 2026 15:39:57 -0400 Subject: [PATCH] feat: add optional backfill params to run_pipeline_v2 --- app/question_gen/pipeline_v2.py | 17 +++++++++--- tests/unit/test_pipeline_v2_resume_params.py | 27 ++++++++++++++++++++ 2 files changed, 40 insertions(+), 4 deletions(-) create mode 100644 tests/unit/test_pipeline_v2_resume_params.py diff --git a/app/question_gen/pipeline_v2.py b/app/question_gen/pipeline_v2.py index 2f07491..7ff2669 100644 --- a/app/question_gen/pipeline_v2.py +++ b/app/question_gen/pipeline_v2.py @@ -174,6 +174,7 @@ def _assign_slots( video_ids: list[str], task_types: list[str], per_type: int, + seq_offset: int = 0, ) -> list[SlotAssignment]: """将出题目标分配为具体 slot 列表。 @@ -184,12 +185,13 @@ def _assign_slots( video_ids: 视频 ID 列表。 task_types: 任务类型列表。 per_type: 每种 task_type 的目标题数。 + seq_offset: 全局序号起始偏移（补生成续编，默认 0）。 返回: SlotAssignment 列表。 """ slots: list[SlotAssignment] = [] - global_seq = 0 + global_seq = seq_offset for task_type in task_types: for i in range(per_type): @@ -907,6 +909,9 @@ async def run_pipeline_v2( task_types: list[str] | None = None, progress: dict[str, str] | None = None, on_accept: Callable[[GeneratedQuestion], None] | None = None, + initial_used_node_ids: set[str] | None = None, + initial_embed_pool: list[np.ndarray] | None = None, + seq_offset: int = 0, ) -> PipelineResult: """v2 出题管线主入口 — 编排全部 slot 的生成、检查与抽检。 @@ -929,6 +934,9 @@ async def run_pipeline_v2( task_types: 任务类型列表（默认使用 12 类标准集）。 progress: 已完成 slot 映射 {slot_id → "accepted"|"rejected"}。 on_accept: 每接受一题时的回调（用于实时持久化，防崩溃丢数据）。 + initial_used_node_ids: 补生成时继承的已用节点集合（避开已用节点，默认空）。 + initial_embed_pool: 补生成时继承的已接受题 embedding 池（跨 run 去重，默认空）。 + seq_offset: 全局序号起始偏移（补生成续编 seq，防撞 question_id，默认 0）。 返回: PipelineResult 实例。 @@ -938,7 +946,7 @@ async def run_pipeline_v2( rng = random.Random(config.seed) # Phase 1: 分配 slot + 创建 run 记录 - slots = _assign_slots(video_ids, task_types, config.per_type) + slots = _assign_slots(video_ids, task_types, config.per_type, seq_offset=seq_offset) logger.info( "管线启动: {} slots, {} 视频, {} 任务类型", len(slots), len(video_ids), len(task_types) ) @@ -951,8 +959,9 @@ async def run_pipeline_v2( # Phase 3: 并发处理 sem = asyncio.Semaphore(config.concurrency) - embed_pool: list[np.ndarray] = [] - used_node_ids: set[str] = set() + # 复制传入容器，避免补生成 run 就地改动调用方对象（不传=空初始化=现状） + embed_pool: list[np.ndarray] = list(initial_embed_pool) if initial_embed_pool else [] + used_node_ids: set[str] = set(initial_used_node_ids) if initial_used_node_ids else set() session_id = f"pipeline_v2_{run_id[:8]}" async def _process_wrapper(slot: SlotAssignment) -> GeneratedQuestion | None: diff --git a/tests/unit/test_pipeline_v2_resume_params.py b/tests/unit/test_pipeline_v2_resume_params.py new file mode 100644 index 0000000..3862f5f --- /dev/null +++ b/tests/unit/test_pipeline_v2_resume_params.py @@ -0,0 +1,27 @@ +"""补生成参数：_assign_slots seq_offset 续编 + run_pipeline_v2 默认签名兼容。""" + +import inspect + +from app.question_gen.pipeline_v2 import _assign_slots, run_pipeline_v2 + + +def test_assign_slots_seq_offset_continues_numbering(): + slots = _assign_slots(["v1"], ["Action Recognition"], 2, seq_offset=10) + assert [s.seq for s in slots] == [11, 12] + assert slots[0].slot_id == "Action Recognition_0011" + + +def test_assign_slots_default_offset_unchanged(): + slots = _assign_slots(["v1"], ["Action Recognition"], 2) + assert [s.seq for s in slots] == [1, 2] + assert slots[0].slot_id == "Action Recognition_0001" + + +def test_run_pipeline_v2_new_optional_params_default_none(): + sig = inspect.signature(run_pipeline_v2) + for name in ("initial_used_node_ids", "initial_embed_pool", "seq_offset"): + assert name in sig.parameters, name + assert sig.parameters[name].kind == inspect.Parameter.KEYWORD_ONLY + assert sig.parameters["initial_used_node_ids"].default is None + assert sig.parameters["initial_embed_pool"].default is None + assert sig.parameters["seq_offset"].default == 0