feat: wire grounded selector into AR slot processing

将 Task 5 的 grounded selector 织入 AR 出题路径(Phase 3.5,位于
record_item 与 postprocess 之间),仅在 strategy.uses_grounded_selector
为真时进入。observation 始终落库(含 hard-fail),硬失败走重出。
PipelineConfig 新增 candidate_pool_size/selector_delta_low/
selector_delta_high 三参,YAML 与 CLI seed override 同步。
This commit is contained in:
2026-07-14 14:15:50 -04:00
parent d0194f5840
commit b13eab0659
5 changed files with 171 additions and 0 deletions
+7
View File
@@ -194,8 +194,15 @@ class MockVLM:
parent_call_id: str | None = None,
) -> LLMResponse:
prompt_text = str(messages)
system_text = messages[0].get("content", "") if messages else ""
if "verdict" in prompt_text.lower():
return _make_llm_response(self._gate_response)
if "distractor" in system_text.lower() and "grader" not in system_text.lower():
# 候选池请求:返回 4 个 grounded 干扰项
return _make_llm_response('{"distractors": ["", "", "", ""]}')
if "grader" in system_text.lower():
# 打分请求:正解高分、3 个落区间、1 个负空间
return _make_llm_response('{"scores": [0.90, 0.80, 0.70, 0.60, 0.20]}')
idx = min(self._gen_count, len(self._responses) - 1)
self._gen_count += 1
return _make_llm_response(self._responses[idx])
@@ -0,0 +1,42 @@
"""selector 织入辅助:正解文本提取 + 分流。"""
import pytest
from app.question_gen.pipeline_v2 import _extract_correct_text
def test_extract_correct_text_strips_prefix():
options = ("A. 蒸", "B. 炒", "C. 煮", "D. 炸")
assert _extract_correct_text(options, "C") == ""
def test_extract_correct_text_handles_lowercase_answer():
options = ("A. run", "B. walk", "C. jump", "D. sit")
assert _extract_correct_text(options, "b") == "walk"
def test_extract_correct_text_out_of_range_raises():
options = ("A. a", "B. b", "C. c", "D. d")
with pytest.raises(ValueError):
_extract_correct_text(options, "E")
def test_replace_candidate_options():
from app.question_gen.generator_v2 import CandidateQuestion
from app.question_gen.pipeline_v2 import _replace_candidate_options
c = CandidateQuestion(
question_id="q",
video_id="v",
task_type="Action Recognition",
skill_target="M1_AR",
question="?",
options=("A. a", "B. b", "C. c", "D. d"),
answer="A",
source_nodes=("n1",),
difficulty="hard",
)
new = _replace_candidate_options(c, ("A. 蒸", "B. 炒", "C. 煮", "D. 炸"), "A")
assert new.options == ("A. 蒸", "B. 炒", "C. 煮", "D. 炸")
assert new.question == "?" # 其余字段不变
assert new.source_nodes == ("n1",)