fix: tolerate backfill under-delivery, fix predict None-fill, drop dead session_id
This commit is contained in:
@@ -756,9 +756,18 @@ async def run_adversarial_rounds(
|
|||||||
if deficit <= 0 or round_no + 1 >= config.adversarial_max_rounds:
|
if deficit <= 0 or round_no + 1 >= config.adversarial_max_rounds:
|
||||||
break
|
break
|
||||||
new_qs = await backfill(deficit, round_no, all_questions)
|
new_qs = await backfill(deficit, round_no, all_questions)
|
||||||
assert len(new_qs) == deficit, (
|
# 真实 backfill 经 run_pipeline_v2 只返回过门的题:部分题被 gate 拒 = 常态欠产。
|
||||||
f"backfill 应产出 {deficit} 题,实得 {len(new_qs)}(缺额驱动契约)"
|
# 容忍 len(new_qs) <= deficit——欠产先并入,deficit 下轮重算继续补,max_rounds 兜底
|
||||||
|
# 防死循环。多产(> deficit)才是契约被破坏的 bug,保留 sanity 断言。
|
||||||
|
assert len(new_qs) <= deficit, (
|
||||||
|
f"backfill 多产: 需 {deficit} 实得 {len(new_qs)}(超额 = 契约被破坏)"
|
||||||
)
|
)
|
||||||
|
if len(new_qs) < deficit:
|
||||||
|
logger.warning(
|
||||||
|
"backfill 欠产: 需 {} 实得 {},本轮先并入,后续轮次继续补",
|
||||||
|
deficit,
|
||||||
|
len(new_qs),
|
||||||
|
)
|
||||||
for q in new_qs:
|
for q in new_qs:
|
||||||
all_questions[q.question_id] = q
|
all_questions[q.question_id] = q
|
||||||
pending = new_qs # 只对新补的题重新过滤(已判题走续跑)
|
pending = new_qs # 只对新补的题重新过滤(已判题走续跑)
|
||||||
@@ -843,7 +852,12 @@ class _RealAgentRunner:
|
|||||||
rows = await RunLogImpl(self._db_path).get_predictions(
|
rows = await RunLogImpl(self._db_path).get_predictions(
|
||||||
run_id, question_ids=[q.question_id for q in questions]
|
run_id, question_ids=[q.question_id for q in questions]
|
||||||
)
|
)
|
||||||
return {r["question_id"]: r["prediction"] for r in rows}
|
# 契约:每个入参 qid 都必须有键。先全填 None,再用回读行覆盖——回读缺行
|
||||||
|
# (agent 未落库/异常)的 qid 保持 None,不静默丢键。
|
||||||
|
preds: dict[str, str | None] = {q.question_id: None for q in questions}
|
||||||
|
for r in rows:
|
||||||
|
preds[r["question_id"]] = r["prediction"]
|
||||||
|
return preds
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
@@ -873,7 +887,6 @@ def build_backfill(
|
|||||||
store: QuestionGenStore,
|
store: QuestionGenStore,
|
||||||
pipeline_config: PipelineConfig,
|
pipeline_config: PipelineConfig,
|
||||||
filter_task_types: tuple[str, ...],
|
filter_task_types: tuple[str, ...],
|
||||||
session_id: str,
|
|
||||||
) -> BackfillFn:
|
) -> BackfillFn:
|
||||||
"""组装真实 backfill 回调 — 缺额驱动 run_pipeline_v2,返回新增 AR 题。
|
"""组装真实 backfill 回调 — 缺额驱动 run_pipeline_v2,返回新增 AR 题。
|
||||||
|
|
||||||
@@ -883,6 +896,7 @@ def build_backfill(
|
|||||||
- initial_embed_pool = 已有题 embedding(跨 run 去重);
|
- initial_embed_pool = 已有题 embedding(跨 run 去重);
|
||||||
- seq_offset = 已有题最大 seq(新题续编,防撞 question_id)。
|
- seq_offset = 已有题最大 seq(新题续编,防撞 question_id)。
|
||||||
仅补 filter_task_types(当前锁定为 AR 单类),返回 PipelineResult.accepted。
|
仅补 filter_task_types(当前锁定为 AR 单类),返回 PipelineResult.accepted。
|
||||||
|
session/run_id 由 run_pipeline_v2 内部自生成,故本层不接受 session_id。
|
||||||
|
|
||||||
参数:
|
参数:
|
||||||
trees: video_id → 三层树索引(生成素材来源)。
|
trees: video_id → 三层树索引(生成素材来源)。
|
||||||
@@ -892,7 +906,6 @@ def build_backfill(
|
|||||||
store: 出题持久化。
|
store: 出题持久化。
|
||||||
pipeline_config: ar30 原配置(除 per_type 外全部继承)。
|
pipeline_config: ar30 原配置(除 per_type 外全部继承)。
|
||||||
filter_task_types: 补生成的题型(仅这些)。
|
filter_task_types: 补生成的题型(仅这些)。
|
||||||
session_id: 遥测会话 ID(透传给管线子调用)。
|
|
||||||
|
|
||||||
返回:
|
返回:
|
||||||
BackfillFn 闭包。
|
BackfillFn 闭包。
|
||||||
|
|||||||
@@ -275,3 +275,27 @@ async def test_real_agent_runner_predict_roundtrips_predictions(tmp_path: Path)
|
|||||||
"smoke_r0", question_ids=["smoke"]
|
"smoke_r0", question_ids=["smoke"]
|
||||||
)
|
)
|
||||||
assert rows and rows[0]["prediction"] == "B"
|
assert rows and rows[0]["prediction"] == "B"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_real_agent_runner_predict_fills_none_for_missing(
|
||||||
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||||
|
) -> None:
|
||||||
|
"""predict 对 predictions 表缺失的 qid 返回 None(契约:每个入参 qid 都有键)。"""
|
||||||
|
|
||||||
|
async def _noop_run_inference(*args: Any, **kwargs: Any) -> None:
|
||||||
|
"""空跑:不写 predictions 表,模拟回读缺行。"""
|
||||||
|
return None
|
||||||
|
|
||||||
|
monkeypatch.setattr("app.harness.inference.run_inference", _noop_run_inference)
|
||||||
|
runner = _RealAgentRunner(
|
||||||
|
llm=_MockLLM(answer="B"),
|
||||||
|
tool_dispatch_fn=lambda *a, **k: None,
|
||||||
|
prompt_builder=lambda q: ("s", "u"),
|
||||||
|
db_path=str(tmp_path / "harness.db"),
|
||||||
|
concurrency=1,
|
||||||
|
skill_mode="none",
|
||||||
|
model="mock",
|
||||||
|
)
|
||||||
|
preds = await runner.predict([_smoke_q("missing")], max_steps=1, run_id="r_none")
|
||||||
|
assert preds == {"missing": None} # 缺回读行 → 该 qid 键存在且为 None
|
||||||
|
|||||||
@@ -163,3 +163,29 @@ async def test_rounds_backfill_then_stop_at_max(tmp_path):
|
|||||||
ids = {d["question_id"] for d in json.loads(final_path.read_text(encoding="utf-8"))}
|
ids = {d["question_id"] for d in json.loads(final_path.read_text(encoding="utf-8"))}
|
||||||
assert "q0" in ids and any(x.startswith("bf0_") for x in ids)
|
assert "q0" in ids and any(x.startswith("bf0_") for x in ids)
|
||||||
store.close()
|
store.close()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_rounds_tolerate_backfill_under_delivery(tmp_path):
|
||||||
|
"""backfill 欠产(实得 < deficit,被 gate 拒常态)→ 不崩,并入实得题,达上限停。"""
|
||||||
|
store = QuestionGenStore(str(tmp_path / "q.db"))
|
||||||
|
agent = _FakeAgent(pred="B")
|
||||||
|
# 每轮只产出 deficit-1 道(模拟部分题被 gate 拒的真实欠产)
|
||||||
|
backfill = _CountingBackfill(lambda d, r: [_q(f"bf{r}_{i}") for i in range(max(d - 1, 0))])
|
||||||
|
final_path = tmp_path / "accepted_questions_final.json"
|
||||||
|
await run_adversarial_rounds(
|
||||||
|
[_q("q0")],
|
||||||
|
agent=agent,
|
||||||
|
vlm=object(),
|
||||||
|
store=store,
|
||||||
|
trees={},
|
||||||
|
config=AdversarialFilterConfig(adversarial_max_rounds=3),
|
||||||
|
final_path=final_path,
|
||||||
|
target=99, # 永远达不到 → 靠 max_rounds 终止,不因欠产崩溃或死循环
|
||||||
|
backfill=backfill,
|
||||||
|
session_id="s",
|
||||||
|
)
|
||||||
|
assert backfill.calls == 2 # round0/round1 各补一次;round2 达上限 break
|
||||||
|
ids = {d["question_id"] for d in json.loads(final_path.read_text(encoding="utf-8"))}
|
||||||
|
assert "q0" in ids and any(x.startswith("bf0_") for x in ids) # 实得欠产题已并入
|
||||||
|
store.close()
|
||||||
|
|||||||
@@ -1318,7 +1318,6 @@ async def _run_adversarial_filter(args: argparse.Namespace) -> None:
|
|||||||
store=store,
|
store=store,
|
||||||
pipeline_config=pipeline_config,
|
pipeline_config=pipeline_config,
|
||||||
filter_task_types=filter_config.filter_task_types,
|
filter_task_types=filter_config.filter_task_types,
|
||||||
session_id=args.session_id,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
# Phase 6: 运行对抗过滤
|
# Phase 6: 运行对抗过滤
|
||||||
|
|||||||
Reference in New Issue
Block a user