diff --git a/adr/0016-pipeline-strategy-profile-sections.md b/adr/0016-pipeline-strategy-profile-sections.md new file mode 100644 index 0000000..4d7b663 --- /dev/null +++ b/adr/0016-pipeline-strategy-profile-sections.md @@ -0,0 +1,51 @@ +# ADR-0016:管线策略 profile 化(pipeline/retrieval/revise 三段) + +- 状态:proposed +- 日期:2026-08-22 +- 影响层:A 资产层(profiles/_schema.py + 两个 profile yaml);B 层仅消费 + +## 背景 + +SW-07(上游依赖卡):各 phase 重试次数、定向重生成策略、self-check 子步骤开关、 +检索注入条数(rerank n)此前是 `pipeline.py` / `p5_dialogue.py` / `cli.py` 里的 +代码常量(`attempts=2`、`range(3)`、`"lenient"`、`k=3`)。弱模型与强模型需要 +不同的重试预算,改常量就要动代码层,无法按 profile 分道调参。 + +## 决定 + +Profile 新增三段(缺省值 = 原代码常量,既有 profile 行为零变化): + +| 段.键 | 语义 | 缺省 | 消费点 | +|---|---|---|---| +| `pipeline.pass_attempts` | 单 Pass 输出波动重试(D13) | 2 | `pipeline._retry_pass` | +| `pipeline.phase_attempts` | p3/p4、p5、p6 相位级定向重生成 | 3 | `run_pipeline` 三个相位循环 | +| `retrieval.top_k` | 每命中的案例注入条数(rerank top-n) | 3 | `cli._make_retrieval` → `RetrievalService.k` | +| `revise.self_check` | p5 自检子步骤开关(T-31,既有键,入 schema 正名) | true | `p5_dialogue._self_check` | +| `revise.gate_mode` | 定向重生成采纳门槛(`nsc.revise.gate.MODES`) | lenient | `p5_dialogue._self_check` | + +`profiles/_schema.py` 相应新增 `PipelineSettings` / `RetrievalSettings` / +`ReviseSettings`(`extra="forbid"`,与既有段一致)。 + +## 被否决的替代 + +| 替代 | 为什么否决 | +|---|---| +| 环境变量(NSC_PASS_ATTEMPTS 等) | 与 SW-04 同理:策略被运行时环境稀释,且不可进 provenance/缓存键 | +| config/models.yaml 侧配置 | 那是模型路由的生成物配置;策略属于 profile 资产 | +| 每相位独立键(phase_p3_attempts...) | 现实中三个相位同预算;先给粗粒度,需要时再加 | + +## 对下游的约束 + +- 调参只改 profile yaml,不改 `src/`;`_retry_pass(attempts=...)` 显式参数保留 + (测试/特殊路径用),优先级:显式参数 > profile > 代码缺省。 +- `revise.gate_mode` 只允许 `nsc.revise.gate.MODES` 里的值(schema Literal 约束)。 + +## 迁移 + +非 breaking:新段全部有缺省,旧 profile(无三段)行为不变。两个在库 profile +(short_drama_v1 / short_video_v1)显式写入了缺省值以便发现。 + +## 验证 + +`tests/test_profile_strategy.py`:pass_attempts/phase_attempts 生效性与缺省回退、 +phase_attempts=1 时 BM-007 拦截即抛、gate_mode/top_k 读取;全量 `pytest -m "not llm"` 绿。 diff --git a/profiles/_schema.py b/profiles/_schema.py index 5cfc256..1e0089b 100644 --- a/profiles/_schema.py +++ b/profiles/_schema.py @@ -36,6 +36,31 @@ class BeatTemplate(BaseModel): note: str = "" +class PipelineSettings(BaseModel): + """SW-07 / ADR-0016:管线重试策略。缺省值 = 提案前的代码常量(零行为变化)。""" + + model_config = ConfigDict(extra="forbid") + pass_attempts: int = Field(default=2, ge=1, description="单 Pass 输出波动重试次数(D13)") + phase_attempts: int = Field(default=3, ge=1, description="p3/p4、p5、p6 相位级定向重生成次数") + + +class RetrievalSettings(BaseModel): + """SW-07 / ADR-0016:案例检索注入策略。""" + + model_config = ConfigDict(extra="forbid") + top_k: int = Field(default=3, ge=1, description="每 Pass 注入的命中案例条数(rerank top-n)") + + +class ReviseSettings(BaseModel): + """SW-07 / ADR-0016:定向重生成(self-check 修订)策略。""" + + model_config = ConfigDict(extra="forbid") + self_check: bool = Field(default=True, description="p5 自检子步骤开关(T-31)") + gate_mode: Literal["strict", "lenient", "always"] = Field( + default="lenient", description="修订采纳门槛(nsc.revise.gate.MODES)" + ) + + class Profile(BaseModel): model_config = ConfigDict(extra="forbid") schema_version: Literal["1.0"] = "1.0" @@ -63,6 +88,13 @@ class Profile(BaseModel): beat_templates: list[BeatTemplate] = Field(default_factory=list) novel: NovelSettings = Field(default_factory=NovelSettings) + #: SW-07 / ADR-0016:管线策略(重试与定向重生成次数)。缺省 = 原代码常量。 + pipeline: PipelineSettings = Field(default_factory=lambda: PipelineSettings()) + #: SW-07 / ADR-0016:检索注入条数(rerank top-n)。 + retrieval: RetrievalSettings = Field(default_factory=lambda: RetrievalSettings()) + #: SW-07 / ADR-0016:定向重生成(self-check 修订)策略。 + revise: ReviseSettings = Field(default_factory=lambda: ReviseSettings()) + render_targets: list[Slug] = Field(default_factory=lambda: ["novel_docx", "script_fountain"]) enabled_check_domains: list[str] = Field( default_factory=lambda: [ diff --git a/profiles/short_drama_v1.yaml b/profiles/short_drama_v1.yaml index affd2ab..07896bd 100644 --- a/profiles/short_drama_v1.yaml +++ b/profiles/short_drama_v1.yaml @@ -51,4 +51,8 @@ model_tiers: p3_beatsheet: tier_plan p4_scene: tier_draft p5_dialogue: tier_draft - p6_prose: tier_draft \ No newline at end of file + p6_prose: tier_draft +# SW-07 / ADR-0016:管线策略(缺省即原代码常量;弱模型/强模型 profile 可分道调参) +pipeline: {pass_attempts: 2, phase_attempts: 3} +retrieval: {top_k: 3} +revise: {self_check: true, gate_mode: lenient} diff --git a/profiles/short_video_v1.yaml b/profiles/short_video_v1.yaml index 330a5f5..3bb0acf 100644 --- a/profiles/short_video_v1.yaml +++ b/profiles/short_video_v1.yaml @@ -55,4 +55,8 @@ model_tiers: p3_beatsheet: tier_plan p4_scene: tier_draft p5_dialogue: tier_draft - p6_prose: tier_draft \ No newline at end of file + p6_prose: tier_draft +# SW-07 / ADR-0016:管线策略(缺省即原代码常量;弱模型/强模型 profile 可分道调参) +pipeline: {pass_attempts: 2, phase_attempts: 3} +retrieval: {top_k: 3} +revise: {self_check: true, gate_mode: lenient} diff --git a/src/nsc/cli.py b/src/nsc/cli.py index e9a4e92..262425d 100644 --- a/src/nsc/cli.py +++ b/src/nsc/cli.py @@ -39,6 +39,22 @@ def _make_ctx(brief: dict, out_dir: Path, router: Any = None) -> Any: ) +def _make_retrieval(ctx: Any) -> Any: + """SW-07:检索服务,注入条数 top_k 读 profile.retrieval(缺省 3=原常量)。""" + import typer + + from nsc.retrieval import RetrievalService + + raw = ctx.profile.get("retrieval", {}).get("top_k", 3) + try: + k = max(1, int(raw)) + except (TypeError, ValueError) as e: + raise typer.BadParameter( + f"profile 的 retrieval.top_k 必须是正整数,当前为 {raw!r}(review 修正:给可读报错)" + ) from e + return RetrievalService(db_path="cases/cases.db", k=k) + + # --- 编译 --- @app.command() def run( @@ -56,9 +72,7 @@ def run( brief_dict = yaml.safe_load(Path(brief).read_text("utf-8")) ctx = _make_ctx(brief_dict, Path(out)) if not no_retrieval: - from nsc.retrieval import RetrievalService - - ctx.retrieval = RetrievalService(db_path="cases/cases.db") + ctx.retrieval = _make_retrieval(ctx) try: ir = run_pipeline(ctx) except PassFailure as e: diff --git a/src/nsc/passes/p5_dialogue.py b/src/nsc/passes/p5_dialogue.py index b8bb2a1..d29a832 100644 --- a/src/nsc/passes/p5_dialogue.py +++ b/src/nsc/passes/p5_dialogue.py @@ -235,6 +235,23 @@ def _counts(findings: list[dict[str, Any]]) -> Counts: ) +def _gate_mode(ctx: PassContext) -> str: + """SW-07:定向重生成(self-check 修订)采纳策略(profile.revise.gate_mode,缺省 lenient)。 + + 非法值转 PassFailure(review 修正):裸 dict profile 无 schema 校验兜底, + 不能让 revise.gate.MODES 的 ValueError 直接击穿编排的 PassFailure 捕获链。 + """ + mode = str(ctx.profile.get("revise", {}).get("gate_mode", "lenient")) + from nsc.revise.gate import MODES + + if mode not in MODES: + raise PassFailure( + None, + f"profile.revise.gate_mode 必须是 {MODES} 之一,当前为 {mode!r};请修正 profile。", + ) + return mode + + def _self_check( ctx: PassContext, inputs: dict[str, Any], @@ -247,8 +264,8 @@ def _self_check( """自我修订(默认开,profile.revise.self_check=False 关闭)。 干净路径零 findings → 不调 LLM。有问题时把 revision_brief 五节文本注入重生成; - 修订经 revisionGate(lenient) 判定采纳,未达标或解析失败则回退原稿—— - 残留 findings 由 pipeline 的 check_stage(after_p5) 兜底拦截,不会静默丢失。 + 修订经 revisionGate(策略 profile.revise.gate_mode)判定采纳,未达标或解析失败 + 则回退原稿——残留 findings 由 pipeline 的 check_stage(after_p5) 兜底拦截,不会静默丢失。 """ if not ctx.profile.get("revise", {}).get("self_check", True): return lines, out @@ -270,6 +287,6 @@ def _self_check( except PassFailure: return lines, out findings2 = _scene_findings(ctx, scene, beats, lines2, characters) - if decide(_counts(findings), _counts(findings2), "lenient"): + if decide(_counts(findings), _counts(findings2), _gate_mode(ctx)): return lines2, out2 return lines, out diff --git a/src/nsc/passes/pipeline.py b/src/nsc/passes/pipeline.py index 964dce4..a123a7f 100644 --- a/src/nsc/passes/pipeline.py +++ b/src/nsc/passes/pipeline.py @@ -81,20 +81,46 @@ def _accum(diag: str, new: str) -> str: return f"{diag}\n---\n{new}" if diag else new -def _retry_pass(fn: Any, ctx: PassContext, fragment: dict[str, Any], *, attempts: int = 2) -> Any: +def _attempts_of(ctx: PassContext, key: str, default: int) -> int: + """SW-07:读 profile.pipeline. 的次数旋钮;坏值转 PassFailure(诊断句,review 修正)。""" + raw = ctx.profile.get("pipeline", {}).get(key, default) + try: + return max(1, int(raw)) + except (TypeError, ValueError) as e: + raise PassFailure( + None, + f"profile.pipeline.{key} 必须是正整数,当前为 {raw!r};请修正 profile 配置后重跑。", + ) from e + + +def _pass_attempts(ctx: PassContext) -> int: + """SW-07:单 Pass 输出波动重试次数(profile.pipeline.pass_attempts,缺省 2)。""" + return _attempts_of(ctx, "pass_attempts", 2) + + +def _phase_attempts(ctx: PassContext) -> int: + """SW-07:p3/p4、p5、p6 相位级定向重生成次数(profile.pipeline.phase_attempts,缺省 3)。""" + return _attempts_of(ctx, "phase_attempts", 3) + + +def _retry_pass( + fn: Any, ctx: PassContext, fragment: dict[str, Any], *, attempts: int | None = None +) -> Any: """生成型 Pass 的输出波动重试:把上次失败诊断注入重试输入(D13 反馈驱动再生成)。 LLM 输出有随机性(漏字段/数错个数),带诊断的重试能显著降低端到端失败率; 失败语义不变(全部失败照样抛 PassFailure,GEPA 反馈信号不受影响),缓存只存成功产物。 + attempts:SW-07 缺省读 profile.pipeline.pass_attempts(原常量 2)。 """ + total = attempts if attempts is not None else _pass_attempts(ctx) last_reason = "" - for i in range(attempts): + for i in range(total): frag = {**fragment, "_previous_failure": last_reason} if last_reason else fragment try: return fn(ctx, frag) except PassFailure as e: last_reason = str(e) - if i == attempts - 1: + if i == total - 1: raise @@ -325,7 +351,8 @@ def track() -> None: # p3(逐集)+ p4 + after_p3/p4 检查作为一个相位:L0 拦截(如 BM-002 植入间隔)时 # 带诊断整体重试(D13 反馈驱动再生成,诊断累积);重试前恢复相位前的状态。 diag = "" - for attempt in range(3): + phase_n = _phase_attempts(ctx) + for attempt in range(phase_n): snapshot = { k: list(st[k]) for k in ("beats", "setup_payoffs", "brand_moments", "scenes", "facts") } @@ -386,12 +413,13 @@ def track() -> None: for e_ in episodes: e_["state_changes"] = ep_state_snap[e_["id"]] diag = _accum(diag, str(e)) - if attempt == 2: + if attempt == phase_n - 1: raise # p5 相位:对白 + after_p5 检查(如 BM-007 必提台词);拦截时带累积诊断整体重试。 diag5 = "" - for attempt in range(3): + phase5_n = _phase_attempts(ctx) + for attempt in range(phase5_n): snapshot_lines = list(st["lines"]) _dbg(f"p5-phase attempt={attempt} scenes={len(st['scenes'])} diag={diag5[:120]!r}") try: @@ -413,14 +441,15 @@ def track() -> None: _dbg(f"p5-phase caught PassFailure: {str(e)[:160]!r}") st["lines"] = snapshot_lines diag5 = _accum(diag5, str(e)) - if attempt == 2: + if attempt == phase5_n - 1: raise if ctx.profile.get("novel", {}).get("enabled"): st["voice"] = _voice(ctx, bible) # p6 相位:小说 + after_p6 检查(如 NOV-001 锚点覆盖);同上带累积诊断重试。 diag6 = "" - for attempt in range(3): + phase6_n = _phase_attempts(ctx) + for attempt in range(phase6_n): snapshot_chapters = list(st["chapters"]) _dbg(f"p6-phase attempt={attempt} diag={diag6[:120]!r}") try: @@ -438,7 +467,7 @@ def track() -> None: _dbg(f"p6-phase caught PassFailure: {str(e)[:160]!r}") st["chapters"] = snapshot_chapters diag6 = _accum(diag6, str(e)) - if attempt == 2: + if attempt == phase6_n - 1: raise ir = cur() diff --git a/tests/test_profile_strategy.py b/tests/test_profile_strategy.py new file mode 100644 index 0000000..8f494e7 --- /dev/null +++ b/tests/test_profile_strategy.py @@ -0,0 +1,152 @@ +"""SW-07 pipeline 策略 profile 化:重试次数 / 定向重生成策略 / self-check 开关 / 检索 top-k。 + +这些原是 pipeline.py、p5_dialogue.py、cli.py 里的代码常量;现在全部从 +profile 的 pipeline.* / revise.* / retrieval.* 段读取,缺省值 = 原常量(零行为变化)。 +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import diskcache +import pytest +import yaml + +import nsc.runtime.cache as cache_mod +from tests.test_pipeline_stub import FullStubRouter + + +def _mini_ctx(tmp_path, profile: dict): + """策略读取测试用的最小 PassContext(真实类,防鸭子类型漂移)。""" + from nsc.passes import PassContext + from nsc.runtime.provenance import RunsStore + + return PassContext( + profile=profile, + brand={}, + router=None, + store=RunsStore(tmp_path / "runs.db"), + ruleset_ver="t", + spec_sha="t", + ) + + +def test_retry_pass_attempts_from_profile(tmp_path): + from nsc.passes import PassFailure + from nsc.passes.pipeline import _retry_pass + + calls = [] + + def flaky(ctx, frag): + calls.append(1) + raise PassFailure(None, "boom") + + # 显式 attempts 优先(既有调用方语义不变) + with pytest.raises(PassFailure): + _retry_pass(flaky, _mini_ctx(tmp_path, {}), {}, attempts=1) + assert len(calls) == 1 + + calls.clear() + with pytest.raises(PassFailure): + _retry_pass(flaky, _mini_ctx(tmp_path, {"pipeline": {"pass_attempts": 1}}), {}) + assert len(calls) == 1, "pass_attempts=1 时只允许一次尝试" + + calls.clear() + with pytest.raises(PassFailure): + _retry_pass(flaky, _mini_ctx(tmp_path, {}), {}) + assert len(calls) == 2, "缺省保持原常量 attempts=2" + + +class FlakyP5Router(FullStubRouter): + """首轮 p5 漏掉必提台词(触发 BM-007);输入带 _previous_failure 时修正。""" + + MUST_LINE = "不额外加蔗糖" + + def _p5(self, inputs): + payload = json.loads(super()._p5(inputs)["lines_json"]) + if "_previous_failure" not in inputs: + payload = [ln for ln in payload if self.MUST_LINE not in ln["text"]] + return {"lines_json": json.dumps(payload, ensure_ascii=False)} + + +def _ctx(tmp_path, monkeypatch, profile: dict): + monkeypatch.setenv("NSC_NO_CACHE", "1") + monkeypatch.setattr(cache_mod, "_cache", diskcache.Cache(str(tmp_path / "cache"))) + from nsc.passes import PassContext + from nsc.runtime.provenance import RunsStore + + brand = yaml.safe_load(Path("brands/demo_tea/brand.yaml").read_text("utf-8")) + brief = yaml.safe_load(Path("examples/demo_tea/brief.yaml").read_text("utf-8")) + return PassContext( + profile=profile, + brand=brand, + brief=brief, + router=FlakyP5Router(), + store=RunsStore(tmp_path / "runs.db"), + ruleset_ver="test-rules", + spec_sha="test-spec", + out_dir=tmp_path / "out", + ) + + +def test_phase_attempts_from_profile(tmp_path, monkeypatch): + """phase_attempts=1:p5 相位首轮被 BM-007 拦截后不得重试,整体抛 PassFailure。""" + from nsc.passes import PassFailure + from nsc.passes.pipeline import _phase_attempts, run_pipeline + + profile = yaml.safe_load(Path("profiles/short_drama_v1.yaml").read_text("utf-8")) + profile["pipeline"] = {"phase_attempts": 1} + assert _phase_attempts(_ctx(tmp_path, monkeypatch, profile)) == 1 + + ctx = _ctx(tmp_path, monkeypatch, profile) + with pytest.raises(PassFailure): + run_pipeline(ctx) + + profile2 = yaml.safe_load(Path("profiles/short_drama_v1.yaml").read_text("utf-8")) + assert _phase_attempts(_ctx(tmp_path, monkeypatch, profile2)) == 3, "缺省保持原常量 3" + + +def test_revise_gate_mode_read_from_profile(tmp_path): + from nsc.passes import p5_dialogue + from nsc.passes.pipeline import _pass_attempts + + assert p5_dialogue._gate_mode(_mini_ctx(tmp_path, {})) == "lenient", "缺省保持原常量 lenient" + assert ( + p5_dialogue._gate_mode(_mini_ctx(tmp_path, {"revise": {"gate_mode": "strict"}})) == "strict" + ), "revise.gate_mode 必须可从 profile 读" + assert _pass_attempts(_mini_ctx(tmp_path, {"pipeline": {"pass_attempts": 4}})) == 4 + + +def test_retrieval_top_k_from_profile(tmp_path): + from nsc.cli import _make_retrieval + + base = yaml.safe_load(Path("profiles/short_drama_v1.yaml").read_text("utf-8")) + svc = _make_retrieval(_mini_ctx(tmp_path, {**base, "retrieval": {"top_k": 5}})) + assert svc is not None and svc.k == 5, "retrieval.top_k 必须驱动检索条数" + + svc2 = _make_retrieval(_mini_ctx(tmp_path, dict(base))) + assert svc2 is not None and svc2.k == 3, "缺省保持原常量 k=3" + + +# ---------------------------------------------------------------- review 修正:坏配置的可诊断失败 +def test_bad_profile_values_fail_with_clear_errors(tmp_path): + import typer + + from nsc.cli import _make_retrieval + from nsc.passes import PassFailure, p5_dialogue + from nsc.passes.pipeline import _pass_attempts, _phase_attempts + + # 非整数重试次数 → PassFailure(走编排的失败/诊断链路,不是裸 TypeError) + with pytest.raises(PassFailure, match="pass_attempts"): + _pass_attempts(_mini_ctx(tmp_path, {"pipeline": {"pass_attempts": "many"}})) + with pytest.raises(PassFailure, match="phase_attempts"): + _phase_attempts(_mini_ctx(tmp_path, {"pipeline": {"phase_attempts": None}})) + + # 非法 gate_mode → PassFailure(不让 revise.gate 的 ValueError 击穿捕获链) + with pytest.raises(PassFailure, match="gate_mode"): + p5_dialogue._gate_mode(_mini_ctx(tmp_path, {"revise": {"gate_mode": "loose"}})) + + # 坏 top_k → typer.BadParameter(用户可读,不是裸堆栈) + with pytest.raises(typer.BadParameter, match="top_k"): + _make_retrieval(_mini_ctx(tmp_path, {"retrieval": {"top_k": "3条"}}))