feat: wire real agent runner, backfill assembly and adversarial-filter CLI

This commit is contained in:
2026-07-14 16:58:32 -04:00
parent 73d0bb9190
commit 1d222d9f18
3 changed files with 734 additions and 1 deletions
+282 -1
View File
@@ -9,6 +9,7 @@ shortcut:作弊门(agent 秒杀=太简单,剔除)+ 翻转门(agent 答
from __future__ import annotations from __future__ import annotations
import dataclasses
import enum import enum
import hashlib import hashlib
import json import json
@@ -22,11 +23,16 @@ from loguru import logger
from core.types import GeneratedQuestion from core.types import GeneratedQuestion
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Callable
import numpy as np
from app.question_gen.adversarial_config import AdversarialFilterConfig from app.question_gen.adversarial_config import AdversarialFilterConfig
from app.question_gen.pipeline_v2 import PipelineConfig
from app.question_gen.run_store import QuestionGenStore from app.question_gen.run_store import QuestionGenStore
from app.question_gen.sampler_v2 import MaterialContext from app.question_gen.sampler_v2 import MaterialContext
from app.tree.index import TreeIndex from app.tree.index import TreeIndex
from core.protocols import VLMProvider from core.protocols import LLMProvider, VLMProvider
_VALID_LETTERS = ("A", "B", "C", "D") _VALID_LETTERS = ("A", "B", "C", "D")
@@ -757,3 +763,278 @@ async def run_adversarial_rounds(
all_questions[q.question_id] = q all_questions[q.question_id] = q
pending = new_qs # 只对新补的题重新过滤(已判题走续跑) pending = new_qs # 只对新补的题重新过滤(已判题走续跑)
logger.info("对抗过滤结束: final={} 题(target={}", passed_now, target) logger.info("对抗过滤结束: final={} 题(target={}", passed_now, target)
# ---------------------------------------------------------------------------
# Task 10: 真实 agent 装配 — _RealAgentRunnerrun_inference + RunLogImpl 回读)
# ---------------------------------------------------------------------------
class _RealAgentRunner:
"""AgentRunner 实现 — 复用 run_inference + RunLogImpl 读回预测。
每次 predict 开一次 HarnessLog(按 run_id 幂等 upsert),跑完整 agent 推理把
prediction 落 predictions 表,再用只读 RunLogImpl 按 (run_id, question_ids) 读回。
prediction 是 submit_answer 的答案字母(A/B/C/D);agent 报错/未提交时该行
prediction 为 None,据此保守处理(不误判)。
参数:
llm: 推理 LLMProvider(共享注入)。
tool_dispatch_fn: InferenceDepsRouter.create_dispatch() 返回的调度闭包。
prompt_builder: InferenceDepsRouter.create_prompt_builder() 返回的构建闭包。
db_path: HarnessLog / RunLogImpl 的 sqlite 路径。
concurrency: run_inference 并发数(asyncio.Semaphore 容量)。
skill_mode: skill 模式("auto"/"manual"/"none")——同时是指纹来源,须与
router 的 skill_mode 一致,保证续跑/失效口径稳定。
model: 推理模型名——指纹来源之一。
"""
def __init__(
self,
*,
llm: LLMProvider,
tool_dispatch_fn: Callable[..., object],
prompt_builder: Callable[[GeneratedQuestion], tuple[str, str]],
db_path: str,
concurrency: int,
skill_mode: str,
model: str,
) -> None:
self._llm = llm
self._dispatch = tool_dispatch_fn
self._builder = prompt_builder
self._db_path = db_path
self._concurrency = concurrency
self.skill_mode = skill_mode
self.model = model
async def predict(
self,
questions: list[GeneratedQuestion],
*,
max_steps: int,
run_id: str,
) -> dict[str, str | None]:
"""跑完整 agent,回读 predictions 表,返回 question_id → 预测字母(无预测 None)。
参数:
questions: 待推理题目列表。
max_steps: AgentLoop 单题最大步数。
run_id: 本次推理 run 标识(predictions 表按此过滤回读)。
返回:
question_id → 预测答案字母(缺失/未提交为 None)。
"""
from app.harness.inference import run_inference
from app.harness.log import HarnessLog, RunLogImpl
with HarnessLog(self._db_path, run_id) as log:
await run_inference(
questions,
llm=self._llm,
tool_dispatch_fn=self._dispatch,
prompt_builder=self._builder,
log=log,
run_id=run_id,
concurrency=self._concurrency,
max_steps=max_steps,
skill_mode=self.skill_mode,
)
rows = await RunLogImpl(self._db_path).get_predictions(
run_id, question_ids=[q.question_id for q in questions]
)
return {r["question_id"]: r["prediction"] for r in rows}
# ---------------------------------------------------------------------------
# Task 10: 真实 backfill 装配 — 缺额驱动 run_pipeline_v2
# ---------------------------------------------------------------------------
def _max_existing_seq(existing: dict[str, GeneratedQuestion]) -> int:
"""从已有题 question_id(格式 "{video_id}_{task_type}_{seq:04d}")取最大 seq。
只解析末段为纯数字的 id;无可解析 id → 0(新 run 从 seq_offset+1 起编,防撞。)
"""
max_seq = 0
for qid in existing:
tail = qid.rsplit("_", 1)[-1]
if tail.isdigit():
max_seq = max(max_seq, int(tail))
return max_seq
def build_backfill(
*,
trees: dict[str, TreeIndex],
vlm: VLMProvider,
llm: LLMProvider,
embed_fn: Callable[[str], np.ndarray],
store: QuestionGenStore,
pipeline_config: PipelineConfig,
filter_task_types: tuple[str, ...],
session_id: str,
) -> BackfillFn:
"""组装真实 backfill 回调 — 缺额驱动 run_pipeline_v2,返回新增 AR 题。
闭包捕获 run_pipeline_v2 全部依赖。每轮按缺额 deficit 用 dataclasses.replace 生成
per_type=deficit 的新 configPipelineConfig frozen,绝不原地改),并传:
- initial_used_node_ids = 已有题 source_nodes 并集(避开已用节点);
- initial_embed_pool = 已有题 embedding(跨 run 去重);
- seq_offset = 已有题最大 seq(新题续编,防撞 question_id)。
仅补 filter_task_types(当前锁定为 AR 单类),返回 PipelineResult.accepted。
参数:
trees: video_id → 三层树索引(生成素材来源)。
vlm: VLM 端口。
llm: LLM 端口(门控用)。
embed_fn: 文本嵌入函数。
store: 出题持久化。
pipeline_config: ar30 原配置(除 per_type 外全部继承)。
filter_task_types: 补生成的题型(仅这些)。
session_id: 遥测会话 ID(透传给管线子调用)。
返回:
BackfillFn 闭包。
"""
from app.question_gen.pipeline_v2 import run_pipeline_v2
video_ids = list(trees.keys())
async def _backfill(
deficit: int,
round_no: int,
existing: dict[str, GeneratedQuestion],
) -> list[GeneratedQuestion]:
"""按缺额补生成 deficit 道 AR 题。"""
run_cfg = dataclasses.replace(pipeline_config, per_type=deficit)
used_node_ids: set[str] = set()
for q in existing.values():
used_node_ids.update(q.source_nodes)
embed_pool = [embed_fn(q.question).flatten() for q in existing.values()]
seq_offset = _max_existing_seq(existing)
logger.info(
"backfill round={}: deficit={}, seq_offset={}, used_nodes={}",
round_no,
deficit,
seq_offset,
len(used_node_ids),
)
result = await run_pipeline_v2(
video_ids=video_ids,
trees=trees,
vlm=vlm,
llm=llm,
embed_fn=embed_fn,
store=store,
config=run_cfg,
task_types=list(filter_task_types),
initial_used_node_ids=used_node_ids,
initial_embed_pool=embed_pool,
seq_offset=seq_offset,
)
return result.accepted
return _backfill
# ---------------------------------------------------------------------------
# Task 10: 顶层入口 — run_adversarial_filter
# ---------------------------------------------------------------------------
def _load_filter_questions(
accepted_path: Path, filter_task_types: tuple[str, ...]
) -> list[GeneratedQuestion]:
"""从 accepted_questions.json 读题并过滤到 filter_task_types(只读,不改 Phase A)。
accepted_questions.json 是 Phase A 产物(JSON 列表);仅 filter_task_types 的题
进 agent 门,其余原样留在 Phase A 文件中(本层不触碰)。
参数:
accepted_path: accepted_questions.json 路径。
filter_task_types: 需过滤的题型集合。
返回:
过滤后的题目列表(保持文件内顺序)。
异常:
FileNotFoundError: 文件不存在。
ValueError: JSON 结构非列表。
"""
if not accepted_path.exists():
raise FileNotFoundError(f"accepted_questions.json 不存在: {accepted_path}")
raw = json.loads(accepted_path.read_text(encoding="utf-8"))
if not isinstance(raw, list):
raise ValueError(f"accepted_questions.json 顶层结构应为列表: {accepted_path}")
types = set(filter_task_types)
questions: list[GeneratedQuestion] = []
for item in raw:
if item.get("task_type") not in types:
continue
questions.append(
GeneratedQuestion(
question_id=item["question_id"],
video_id=item["video_id"],
task_type=item["task_type"],
question=item["question"],
options=tuple(item["options"]),
answer=item["answer"],
source_nodes=tuple(item.get("source_nodes", ())),
difficulty=item.get("difficulty", "medium"),
family=item.get("family"),
skill_target=item.get("skill_target"),
difficulty_steps=item.get("difficulty_steps"),
sub_pattern=item.get("sub_pattern"),
)
)
return questions
async def run_adversarial_filter(
*,
accepted_path: Path,
final_path: Path,
agent: AgentRunner,
vlm: VLMProvider,
trees: dict[str, TreeIndex],
store: QuestionGenStore,
filter_config: AdversarialFilterConfig,
backfill: BackfillFn,
session_id: str,
) -> None:
"""Phase B 顶层入口:读 accepted_questions.json 过滤 AR,跑两门 + 补生成迭代。
路径隔离:只处理 filter_task_types 的题,accepted_questions.json 只读,Phase A
状态机不受影响。target 锁定为首轮过滤出的 AR 题数(缺额 = target - passed)。
参数:
accepted_path: Phase A 产物 accepted_questions.json 路径(只读)。
final_path: 最终题库 JSON 路径(每轮全量原子重写)。
agent: 完整 agent 试答端口(Task 10 的 _RealAgentRunner,单测可 mock)。
vlm: 镜像题生成 VLM 端口。
trees: video_id → 三层树索引(重建镜像素材 / 补生成用)。
store: verdict 持久化。
filter_config: 后置对抗过滤配置(题型 / 轮次 / max_steps / 难度阈值)。
backfill: 补生成回调(CLI 用 build_backfill 组装真实实现,单测可 mock)。
session_id: 遥测会话 ID(派生各门 run_id)。
"""
initial_questions = _load_filter_questions(accepted_path, filter_config.filter_task_types)
target = len(initial_questions)
logger.info(
"对抗过滤启动: 过滤题型={}, 首轮 AR 题数={}",
filter_config.filter_task_types,
target,
)
await run_adversarial_rounds(
initial_questions,
agent=agent,
vlm=vlm,
store=store,
trees=trees,
config=filter_config,
final_path=final_path,
target=target,
backfill=backfill,
session_id=session_id,
)
@@ -0,0 +1,277 @@
"""Task 10run_adversarial_filter 顶层入口 + _RealAgentRunner 真实装配 e2e。
覆盖两条路径:
- run_adversarial_filter 编排(mock agent + mock backfill):过滤 filter_task_types、
非 AR 题不进 agent 门、final 仅含 passed、断点续跑不重跑已判题。
- _RealAgentRunner 真实装配 smokeI6):predict 经 run_inference 落 predictions 表再
读回(LLM mock,路径真穿过 HarnessLog / RunLogImpl,非假 runner 短路)。
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, MagicMock
import numpy as np
import pytest
if TYPE_CHECKING:
from pathlib import Path
from app.harness.deps_router import InferenceDepsRouter
from app.question_gen.adversarial_config import AdversarialFilterConfig
from app.question_gen.adversarial_filter import _RealAgentRunner, run_adversarial_filter
from app.question_gen.run_store import QuestionGenStore
from core.types import GeneratedQuestion, LLMResponse
def _ar_q(qid: str, video_id: str = "v1") -> dict:
"""构造一条 AR 题 JSON 记录;sub_pattern 缺省 → 不支持 flip,翻转门直接放行。"""
return {
"question_id": qid,
"video_id": video_id,
"task_type": "Action Recognition",
"question": f"{qid} 之前做了什么?",
"options": ["A. 蒸", "B. 炒", "C. 煮", "D. 炸"],
"answer": "A",
"source_nodes": ["n1"],
"difficulty": "hard",
}
def _non_ar_q(qid: str, video_id: str = "v1") -> dict:
"""构造一条非 AR 题 JSON 记录(不应进 agent 门)。"""
return {
"question_id": qid,
"video_id": video_id,
"task_type": "Object Recognition",
"question": f"{qid} 里的物体是什么?",
"options": ["A. 锅", "B. 碗", "C. 盘", "D. 勺"],
"answer": "A",
"source_nodes": ["n2"],
"difficulty": "hard",
}
class _FakeAgent:
"""完整 agent 试答桩:对每题返回固定预测,记录被调用的 question_id。"""
def __init__(self, pred: str = "B", model: str = "m1", skill_mode: str = "auto") -> None:
self._pred = pred
self.model = model
self.skill_mode = skill_mode
self.calls: list[str] = []
async def predict(
self, questions: list[GeneratedQuestion], *, max_steps: int, run_id: str
) -> dict[str, str]:
self.calls.extend(q.question_id for q in questions)
return {q.question_id: self._pred for q in questions}
class _NoBackfill:
"""补生成回调桩:记录调用次数,永远返回空(首轮即达标时不应被调用)。"""
def __init__(self) -> None:
self.calls = 0
async def __call__(
self, deficit: int, round_no: int, existing: dict[str, GeneratedQuestion]
) -> list[GeneratedQuestion]:
self.calls += 1
return []
def _write_accepted(path: Path, records: list[dict]) -> None:
"""写 accepted_questions.jsonPhase A 产物形态:JSON 列表)。"""
path.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
@pytest.mark.asyncio
async def test_filter_only_processes_ar_and_writes_passed(tmp_path: Path) -> None:
"""非 AR 题不进 agent 门;final 仅含 passed AR 题。"""
accepted = tmp_path / "accepted_questions.json"
_write_accepted(accepted, [_ar_q("q1"), _ar_q("q2"), _non_ar_q("obj1")])
final_path = tmp_path / "accepted_questions_final.json"
store = QuestionGenStore(str(tmp_path / "q.db"))
agent = _FakeAgent(pred="B") # 答错 → 过作弊门;sub_pattern=None → 过翻转门
backfill = _NoBackfill()
await run_adversarial_filter(
accepted_path=accepted,
final_path=final_path,
agent=agent,
vlm=object(),
trees={},
store=store,
filter_config=AdversarialFilterConfig(adversarial_max_rounds=3),
backfill=backfill,
session_id="s",
)
# 只有 AR 题进 agent 门
assert set(agent.calls) == {"q1", "q2"}
# 非 AR 题不出现在 verdicts 表
rows = store._conn.execute(
"SELECT question_id FROM adversarial_verdicts WHERE question_id=?", ("obj1",)
).fetchall()
assert rows == []
# final 仅含 passed AR 题
data = json.loads(final_path.read_text(encoding="utf-8"))
assert {d["question_id"] for d in data} == {"q1", "q2"}
assert backfill.calls == 0 # 首轮即达标(target=2, passed=2
store.close()
@pytest.mark.asyncio
async def test_filter_resume_does_not_rerun_judged(tmp_path: Path) -> None:
"""断点续跑:第二次调用不重跑已判题(agent 调用计数不变)。"""
accepted = tmp_path / "accepted_questions.json"
_write_accepted(accepted, [_ar_q("q1"), _ar_q("q2")])
final_path = tmp_path / "accepted_questions_final.json"
store = QuestionGenStore(str(tmp_path / "q.db"))
agent = _FakeAgent(pred="B")
backfill = _NoBackfill()
async def _run() -> None:
await run_adversarial_filter(
accepted_path=accepted,
final_path=final_path,
agent=agent,
vlm=object(),
trees={},
store=store,
filter_config=AdversarialFilterConfig(adversarial_max_rounds=3),
backfill=backfill,
session_id="s",
)
await _run()
first_calls = list(agent.calls)
await _run()
assert agent.calls == first_calls # 第二次未新增 agent 调用
store.close()
# ---------------------------------------------------------------------------
# I6_RealAgentRunner 真实装配 smokeLLM mock,路径真穿过 predictions 表)
# ---------------------------------------------------------------------------
class _MockLLM:
"""最小 LLM 桩:一步即产出 submit_answer,让真实 AgentLoop 稳定收敛。"""
def __init__(self, answer: str) -> None:
self._answer = answer
async def chat(
self,
messages: list[dict[str, Any]],
*,
session_id: str | None = None,
parent_call_id: str | None = None,
) -> LLMResponse:
content = json.dumps(
{"action": {"tool": "submit_answer", "args": {"answer": self._answer}}}
)
return LLMResponse(
content=content,
thinking="",
model="mock",
provider="mock",
prompt_tokens=1,
completion_tokens=1,
latency_ms=1,
ttft_ms=None,
max_inter_token_ms=None,
cache_hit=False,
call_id="mock-call",
)
def _build_real_router(tmp_path: Path) -> InferenceDepsRouter:
"""构建真实 InferenceDepsRouter(仅 embed/vlm 打桩,router/deps 装配全真实)。"""
vid = "smoke_vid"
vid_dir = tmp_path / "videos" / vid
(vid_dir / "frames").mkdir(parents=True)
minimal_tree = {
"metadata": {"source_path": "test", "modality": "video"},
"roots": [
{
"id": "L1_000",
"card": {
"scene_summary": "s",
"main_setting": "s",
"key_entities": [],
"main_actions": [],
"topic_keywords": [],
"visible_text": [],
"temporal_flow": "s",
},
"time_range": [0, 10],
"children": [],
}
],
}
(vid_dir / "tree.json").write_text(json.dumps(minimal_tree))
prompts_dir = tmp_path / "prompts"
prompts_dir.mkdir()
(prompts_dir / "system.md").write_text("You are a search agent.")
fake_embed = MagicMock()
fake_embed.dim = 4
fake_embed.embed = lambda t: np.zeros((1, 4), dtype=np.float32)
return InferenceDepsRouter(
store_dir=tmp_path,
embed_provider=fake_embed,
llm=_MockLLM(answer="B"),
vlm=AsyncMock(),
ocr=None,
default_prompts_dir=prompts_dir,
default_skills_dir=None,
skill_mode="none",
verify_vision=False,
anchor=False,
assemble_mode="ids",
)
def _smoke_q(qid: str) -> GeneratedQuestion:
"""构造 smoke 题(video_id 对应真实 router 的树 fixture)。"""
return GeneratedQuestion(
question_id=qid,
video_id="smoke_vid",
task_type="Action Recognition",
question="他之前做了什么?",
options=("A. 蒸", "B. 炒", "C. 煮", "D. 炸"),
answer="A",
source_nodes=("L1_000",),
difficulty="hard",
)
@pytest.mark.asyncio
async def test_real_agent_runner_predict_roundtrips_predictions(tmp_path: Path) -> None:
"""真实装配 smokepredict 经 run_inference 落 predictions 表再读回(LLM mock)。"""
router = _build_real_router(tmp_path)
runner = _RealAgentRunner(
llm=_MockLLM(answer="B"),
tool_dispatch_fn=router.create_dispatch(),
prompt_builder=router.create_prompt_builder(),
db_path=str(tmp_path / "harness.db"),
concurrency=1,
skill_mode="none",
model="mock",
)
preds = await runner.predict([_smoke_q("smoke")], max_steps=2, run_id="smoke_r0")
assert preds["smoke"] == "B" # 真的从 predictions 表读回,非 mock 直返
# 断言确实写进了 predictions 表(穿过 HarnessLog / RunLogImpl
from app.harness.log import RunLogImpl
rows = await RunLogImpl(str(tmp_path / "harness.db")).get_predictions(
"smoke_r0", question_ids=["smoke"]
)
assert rows and rows[0]["prediction"] == "B"
+175
View File
@@ -1173,6 +1173,176 @@ async def _run_generate_v2(args: argparse.Namespace) -> None:
logger.warning("超过 50% 的 slot 被拒绝,建议检查 VLM/门控配置") logger.warning("超过 50% 的 slot 被拒绝,建议检查 VLM/门控配置")
def _load_trees_abs(videos_dir: Path, video_ids: list[str]) -> dict:
"""加载视频树并将相对帧路径解析为绝对路径(复用 generate-v2 逻辑)。
参数:
videos_dir: store/videos 目录。
video_ids: 待加载的 video_id 列表。
返回:
video_id → TreeIndex 映射(加载失败的视频被跳过)。
"""
from app.tree.index import TreeIndex
trees: dict = {}
for vid in video_ids:
tree_path = videos_dir / vid / "tree.json"
try:
tree = TreeIndex.load_json(str(tree_path))
except (OSError, ValueError, KeyError) as exc:
logger.warning("加载树 {} 失败,跳过: {}", tree_path, exc)
continue
video_dir = videos_dir / vid
for l1 in tree.roots:
for l2 in l1.children:
for l3 in l2.children:
if l3.frame_path and not Path(l3.frame_path).is_absolute():
l3.frame_path = str(video_dir / l3.frame_path)
trees[vid] = tree
return trees
def _add_adversarial_filter_parser(subparsers: argparse._SubParsersAction) -> None:
"""注册 adversarial-filter 子命令(Phase B 后置对抗过滤 CLI 入口)。
参数:
subparsers: argparse 子命令注册器。
"""
p = subparsers.add_parser(
"adversarial-filter",
help="Phase B 后置对抗过滤(作弊门 + 翻转门 + 缺额补生成)",
)
p.add_argument("--config", type=Path, required=True, help="YAML 配置(含 question_gen_v2 / adversarial_filter / embed 段)")
p.add_argument("--store-dir", type=Path, required=True, help="store 根目录(含 videos/ prompts/ skills/")
p.add_argument("--accepted-path", type=Path, required=True, help="Phase A 产物 accepted_questions.json 路径(只读)")
p.add_argument("--final-path", type=Path, default=None, help="最终题库输出路径(默认 accepted 同目录 accepted_questions_final.json")
p.add_argument("--db-path", type=Path, default=Path("logs/question_gen.db"), help="QuestionGenStore SQLite 路径")
p.add_argument("--harness-db", type=Path, default=Path("logs/adversarial_harness.db"), help="agent 推理 HarnessLog SQLite 路径")
p.add_argument("--prompts-version", type=str, default="v1", help="推理 prompt 版本目录名(store/prompts/<version>")
p.add_argument("--skills-version", type=str, default="v1", help="推理 skill 版本目录名(store/skills/<version>")
p.add_argument("--skill-mode", type=str, choices=["auto", "manual", "none"], default="auto", help="skill 模式")
p.add_argument("--concurrency", type=int, default=4, help="agent 推理并发数")
p.add_argument("--session-id", type=str, default="adversarial", help="遥测会话 ID(派生各门 run_id)")
async def _run_adversarial_filter(args: argparse.Namespace) -> None:
"""adversarial-filter 子命令主流程。
装配 adapters(同 main._build_adapters)、InferenceDepsRouter(同 main.py 参数)、
QuestionGenStore、视频树(帧路径绝对化)、真实 _RealAgentRunner 与真实 backfill
调 run_adversarial_filter 跑两门 + 缺额补生成迭代。
参数:
args: CLI 参数(config, store_dir, accepted_path, final_path, db_path,
harness_db, prompts_version, skills_version, skill_mode, concurrency,
session_id)。
"""
import yaml
from app.harness.deps_router import InferenceDepsRouter
from app.question_gen.adversarial_config import load_adversarial_config
from app.question_gen.adversarial_filter import (
_RealAgentRunner,
build_backfill,
run_adversarial_filter,
)
from app.question_gen.pipeline_v2 import load_pipeline_config
from app.question_gen.run_store import QuestionGenStore
from main import InfraSettings, _build_adapters
config_path = args.config.resolve()
store_dir = args.store_dir.resolve()
# Phase 1: 加载配置(对抗过滤 + 出题管线 + embed 段)
with config_path.open(encoding="utf-8") as f:
raw_yaml = yaml.safe_load(f) or {}
embed_cfg = raw_yaml.get("embed", {})
filter_config = load_adversarial_config(config_path)
pipeline_config = load_pipeline_config(config_path)
# Phase 2: 装配 adapters
settings = InfraSettings()
adapters = _build_adapters(settings, embed_cfg)
# Phase 3: 加载视频树(帧路径绝对化)
videos_dir = store_dir / "videos"
if not videos_dir.exists():
logger.error("视频目录不存在: {}", videos_dir)
sys.exit(1)
video_ids = sorted(
d.name for d in videos_dir.iterdir() if d.is_dir() and (d / "tree.json").exists()
)
trees = _load_trees_abs(videos_dir, video_ids)
if not trees:
logger.error("所有视频树加载失败,无法继续")
sys.exit(1)
logger.info("成功加载 {} / {} 棵视频树", len(trees), len(video_ids))
# Phase 4: InferenceDepsRouter(同 main.py 参数)
router = InferenceDepsRouter(
store_dir=store_dir,
embed_provider=adapters.embed,
llm=adapters.llm,
vlm=adapters.vlm,
ocr=adapters.ocr,
default_prompts_dir=store_dir / "prompts" / args.prompts_version,
default_skills_dir=store_dir / "skills" / args.skills_version,
skill_mode=args.skill_mode,
verify_vision=True,
anchor=True,
assemble_mode="ids_expand",
)
# Phase 5: QuestionGenStore + 真实 agent + 真实 backfill
db_path = args.db_path.resolve()
db_path.parent.mkdir(parents=True, exist_ok=True)
store = QuestionGenStore(str(db_path))
harness_db = args.harness_db.resolve()
harness_db.parent.mkdir(parents=True, exist_ok=True)
agent = _RealAgentRunner(
llm=adapters.llm,
tool_dispatch_fn=router.create_dispatch(),
prompt_builder=router.create_prompt_builder(),
db_path=str(harness_db),
concurrency=args.concurrency,
skill_mode=args.skill_mode,
model=settings.search_llm_model,
)
backfill = build_backfill(
trees=trees,
vlm=adapters.vlm,
llm=adapters.llm,
embed_fn=adapters.embed.embed,
store=store,
pipeline_config=pipeline_config,
filter_task_types=filter_config.filter_task_types,
session_id=args.session_id,
)
# Phase 6: 运行对抗过滤
accepted_path = args.accepted_path.resolve()
final_path = (
args.final_path.resolve()
if args.final_path is not None
else accepted_path.parent / "accepted_questions_final.json"
)
await run_adversarial_filter(
accepted_path=accepted_path,
final_path=final_path,
agent=agent,
vlm=adapters.vlm,
trees=trees,
store=store,
filter_config=filter_config,
backfill=backfill,
session_id=args.session_id,
)
store.close()
logger.info("对抗过滤完成,final 已写入: {}", final_path)
def _parse_args() -> argparse.Namespace: def _parse_args() -> argparse.Namespace:
"""解析命令行参数。""" """解析命令行参数。"""
parser = argparse.ArgumentParser(description="赛题生成工具:generate + calibrate + generate-v2") parser = argparse.ArgumentParser(description="赛题生成工具:generate + calibrate + generate-v2")
@@ -1181,6 +1351,9 @@ def _parse_args() -> argparse.Namespace:
# generate-v2 子命令 # generate-v2 子命令
_add_generate_v2_parser(subparsers) _add_generate_v2_parser(subparsers)
# adversarial-filter 子命令(Phase B 后置对抗过滤)
_add_adversarial_filter_parser(subparsers)
# generate 子命令 # generate 子命令
gen_parser = subparsers.add_parser("generate", help="生成新题目(v1 传统模式)") gen_parser = subparsers.add_parser("generate", help="生成新题目(v1 传统模式)")
gen_parser.add_argument( gen_parser.add_argument(
@@ -1280,6 +1453,8 @@ def main() -> None:
asyncio.run(_run_generate(args)) asyncio.run(_run_generate(args))
elif args.command == "generate-v2": elif args.command == "generate-v2":
asyncio.run(_run_generate_v2(args)) asyncio.run(_run_generate_v2(args))
elif args.command == "adversarial-filter":
asyncio.run(_run_adversarial_filter(args))
elif args.command == "calibrate": elif args.command == "calibrate":
_run_calibrate(args) _run_calibrate(args)