SW-07 pipeline 策略 profile 化：重试/定向重生成/自检开关/检索 top-k（ADR-0016）

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"` 绿。

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: [

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
+  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}

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
+  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}

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:

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

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.<key> 的次数旋钮；坏值转 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()