feat: llm-filter 两级过滤（规则层+五类分类+去重+长文本保护），工作流单线程 v4

- 规则层（不调 LLM）：横线装饰/HTML 水印 token/URL/邮箱/单双 ASCII 字符直接删
- LLM 五类分类：garbage/overlay/noise 删，repeat/dialogue 留，未识别回退保留
- 按文本去重：相同文本只调一次 LLM（忽略空白/大小写），判定一致并省调用
- 长文本保护：≥min_keep_len 时 noise 不构成删除依据
- 真实任务 run_ac7f480a3ccb 验证：非规则误删 350→193（-45%），呻吟/对话保留
- ocr-subtitle 工作流 v4：pool 钉死单线程（1/1）
- 回归夹具 testdata/ocr_srt_run_ac7f480a3ccb.srt（真实 1666 条 OCR 输出）

nodes/llm_filter.py

@@ -1,9 +1,30 @@
 """LLM 字幕过滤节点。
 
-对 OCR 识别出的 SRT 字幕做二次过滤：为避免字幕上下文过长，把每条字幕连同其
-前后各 context_size 条字幕（纯文本，**不含时间戳**）分批提供给 LLM，模型仅判断
-目标字幕是否属于多余、无意义的字符（如重复、残缺、无实际语义的杂项）；
-判定无意义则删除该条（连同其时间戳），其余字幕保持原样并重新编号输出。
+对 OCR 识别出的 SRT 字幕做二次过滤，两级判断：
+
+1. **确定性规则层**（不调用 LLM）：横线装饰、HTML/水印 token、URL/邮箱、
+   单双 ASCII 字符等 OCR 噪声直接删除——这些模式是稳定可判的，走规则
+   既省 token 又保证结果确定（真实数据中占删除量的 65%+）。
+
+2. **LLM 五类分类层**：把目标字幕连同前后各 context_size 条纯文本分批
+   提供给 LLM，模型输出五个类别之一：
+   - garbage：垃圾字符（乱码、残缺、装饰性符号）→ 删除
+   - overlay：水印/网页/播放器等覆盖层文本 → 删除
+   - noise：与上下文无关、无实际语义的杂项 → 删除
+   - repeat：内容性重复（语气词、呻吟、重复感叹，属于内容本身）→ 保留
+   - dialogue：正常对话 → 保留
+   未识别输出一律回退 dialogue（宁滥勿缺，避免误删真实对话）。
+
+3. **文本去重**：相同文本（忽略全部空白与大小写差异）只调一次 LLM，
+   上下文取首次出现位置，结果缓存复用——修复"同一句字幕 5 留 6 删"的
+   判定不一致，同时把长视频的 LLM 调用量降到唯一文本数。
+
+4. **长文本保护**：长度 ≥ min_keep_len 的文本，仅 garbage/overlay 两个
+   明确垃圾类别可删，noise 不构成删除依据，防止长对话被误删。
+
+参考真实任务 run_ac7f480a3ccb（2026-08，OCR 1666 条）：旧实现把 394 条
+真实对话当噪声删掉（占删除 35%）、同文本判定不一致；新实现按上述机制
+回归测试已固化在 tests/test_llm_filter.py。
 """
 
 from __future__ import annotations
@@ -13,6 +34,7 @@ import os
 import re
 import urllib.request
 from pathlib import Path
+
 from nodes.adaptive_pool import AdaptiveThreadPool
 from wov_app.logging import get_logger
 from wov_sdk.models import InvokeRequest, InvokeResponse
@@ -31,6 +53,37 @@ TARGET_MARK = "【目标】"
 # 默认上下文窗口：目标字幕前后各取 10 条。
 DEFAULT_CONTEXT_SIZE = 10
 
+# LLM 输出类别：garbage/overlay/noise 删除；repeat/dialogue 保留。
+CATEGORY_GARBAGE = "garbage"
+CATEGORY_OVERLAY = "overlay"
+CATEGORY_NOISE = "noise"
+CATEGORY_REPEAT = "repeat"
+CATEGORY_DIALOGUE = "dialogue"
+_ALL_CATEGORIES = (
+    CATEGORY_GARBAGE,
+    CATEGORY_OVERLAY,
+    CATEGORY_NOISE,
+    CATEGORY_REPEAT,
+    CATEGORY_DIALOGUE,
+)
+DELETE_CATEGORIES = {CATEGORY_GARBAGE, CATEGORY_OVERLAY, CATEGORY_NOISE}
+
+# 规则层正则：横线装饰（含全角/半角横线、下划线、中点、句点等符号组合）。
+_DASH_RE = re.compile(r"^[\s\-—_~=•・。．.、]+$")
+# 规则层正则：URL / 邮箱。
+_URL_OR_MAIL_RE = re.compile(r"^(https?://|www\.)\S+$|^[\w.+-]+@[\w.-]+\.\w+$")
+# 默认水印/覆盖层 token（casefold 后比较，可经 overlay_tokens 参数覆盖）。
+DEFAULT_OVERLAY_TOKENS = frozenset(
+    {"html", "background", "___", "cleaning", "buffering", "loading"}
+)
+# 长文本保护阈值：≥ 该长度的文本，noise 类别不构成删除依据。
+DEFAULT_MIN_KEEP_LEN = 12
+
+
+def _has_cjk(text: str) -> bool:
+    """是否含 CJK 汉字：单字"嗯/好"等可能是内容，规则层不直接删。"""
+    return any("\u4e00" <= ch <= "\u9fff" for ch in text)
+
 
 def parse_srt(text: str) -> list[dict]:
     """解析 SRT 文本为条目列表：[{"start", "end", "text"}]。"""
@@ -55,13 +108,58 @@ def serialize_srt(entries: list[dict]) -> str:
     return "\n\n".join(blocks) + "\n"
 
 
-def _judge_target(
+def _rule_verdict(text: str, overlay_tokens: set[str]) -> bool | None:
+    """确定性规则层：返回 True（删除）/ None（交给 LLM 多维判断）。
+
+    规则覆盖 OCR 噪声的稳定模式：空文本、纯横线装饰、URL/邮箱、水印 token
+    （HTML/background 等）、单双 ASCII 字符。含 CJK 的短文本不算垃圾，
+    因为"嗯/好"等可能是内容；其余情况返回 None 交由 LLM 分类。
+    """
+    t = text.strip()
+    if not t:
+        return True
+    if _DASH_RE.match(t):
+        return True
+    if _URL_OR_MAIL_RE.match(t):
+        return True
+    if t.casefold() in overlay_tokens:
+        return True
+    if len(t) <= 2 and not _has_cjk(t):
+        return True
+    return None
+
+
+def _dedup_key(text: str) -> str:
+    """文本去重键：去掉全部空白并统一大小写。
+
+    OCR 同一句字幕常带/不带空格（"可没法胜任" vs "可 没法胜任"），
+    视为同一文本以保证判定一致；也用于把 LLM 调用量降到唯一文本数。
+    """
+    return "".join(text.split()).casefold()
+
+
+def _should_delete(category: str, text: str, min_keep_len: int) -> bool:
+    """按 LLM 类别与长文本保护决定是否删除。
+
+    repeat/dialogue 一律保留；garbage/overlay 一律删除（含长文本——
+    这两个是明确的垃圾信号）；noise 对短文本删除，但 ≥min_keep_len 的
+    长文本不删（noise 太模糊，不足以推翻一句完整台词）。
+    """
+    if category not in DELETE_CATEGORIES:
+        return False
+    if category == CATEGORY_NOISE and len(text) >= min_keep_len:
+        return False
+    return True
+
+
+def _judge_category(
     entries: list[dict], index: int, context_size: int, params: dict
-) -> bool:
-    """调用 LLM 判断目标字幕是否多余/无意义；返回 True 表示应删除。
+) -> str:
+    """调用 LLM 把目标字幕归入五类之一，返回类别词（未识别回退 dialogue）。
 
     请求体只含目标字幕及其前后各 context_size 条字幕的纯文本（无时间戳），
-    目标字幕用 TARGET_MARK 标记；模型只需回答"保留"或"删除"。
+    目标字幕用 TARGET_MARK 标记；模型只输出一个类别英文单词。
+    输出无法识别（空/乱码/旧式"保留"）时回退 dialogue，宁滥勿缺。
     """
     start = max(0, index - context_size)
     end = min(len(entries), index + context_size + 1)
@@ -81,9 +179,14 @@ def _judge_target(
     model = str(params.get("model") or os.getenv("LLM_MODEL", "Qwen/Qwen3.6-35B-A3B"))
     system_prompt = (
         "你是字幕质量过滤器。用户会提供一段字幕序列（纯文本，不含时间戳），"
-        f"其中用{TARGET_MARK}标记的字幕是需要判断的目标。请判断该字幕是否属于"
-        "多余、无意义的字符（如重复、残缺、无实际语义的杂项）。"
-        "只回答两个字：保留 或 删除，不要输出其他内容。"
+        f"其中用{TARGET_MARK}标记的字幕是需要判断的目标。"
+        "请把目标字幕归入以下五个类别之一：\n"
+        "garbage：垃圾字符（乱码、残缺、装饰性符号、横线）\n"
+        "overlay：水印、网页/播放器/字幕组等覆盖层文本，不是视频对白\n"
+        "noise：与上下文无关、无实际语义的杂项\n"
+        "repeat：内容性重复（如语气词、呻吟、重复的感叹或对话），属于内容本身\n"
+        "dialogue：正常对话\n"
+        "只输出一个类别英文单词，不要输出其他内容。"
     )
     body = {
         "model": model,
@@ -94,7 +197,7 @@ def _judge_target(
         # 关闭推理模式：Qwen3 等模型默认会把思考过程写入 reasoning_content，
         # 导致 content 为空或包含多余内容。
         "enable_thinking": False,
-        # 只输出"保留/删除"，输出上限给得很小即可。
+        # 类别词很短，输出上限给得很小即可。
         "max_tokens": 16,
     }
     headers = {"Content-Type": "application/json"}
@@ -109,13 +212,23 @@ def _judge_target(
     )
     with urllib.request.urlopen(request, timeout=request_timeout) as response:
         payload = json.loads(response.read().decode("utf-8"))
-    content = str(payload["choices"][0]["message"]["content"])
-    # 模型回答含"删除"即视为该条无意义；其余情况（保留/异常）一律保留，宁多勿删。
-    return "删除" in content
+    content = str(payload["choices"][0]["message"]["content"]).strip().lower()
+    # 精确匹配五个类别词（兼容"garbagexxx"这类多余输出）。
+    for category in _ALL_CATEGORIES:
+        if content == category or content.startswith(category):
+            return category
+    # 旧式"删除/保留"回答兼容：含"删除"视为垃圾，其余一律保留。
+    if "删除" in content:
+        return CATEGORY_GARBAGE
+    return CATEGORY_DIALOGUE
 
 
 def invoke(request: InvokeRequest) -> InvokeResponse:
-    """过滤 SRT 中多余/无意义的字幕，产物为 filtered.srt。"""
+    """过滤 SRT 中多余/无意义的字幕，产物为 filtered.srt。
+
+    流程：规则层（确定性删除）→ LLM 层（去重后按唯一文本多维分类）→
+    长文本保护 → 保留条重新编号输出。
+    """
     srt_uri = request.inputs.get("srt_uri")
     if not srt_uri:
         return InvokeResponse(status="failed", error="srt_uri is required")
@@ -125,17 +238,44 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
 
     entries = parse_srt(srt_path.read_text(encoding="utf-8"))
     context_size = int(request.params.get("context_size", DEFAULT_CONTEXT_SIZE))
-    # 单条判断的工作函数：返回 True 表示该条应删除。
-    def judge_one(index) -> bool:
-        return _judge_target(entries, index, context_size, request.params)
+    min_keep_len = int(request.params.get("min_keep_len", DEFAULT_MIN_KEEP_LEN))
+    # 水印 token 可经参数覆盖（JSON 数组字符串或列表），默认内置常见覆盖层词。
+    raw_tokens = request.params.get("overlay_tokens")
+    if isinstance(raw_tokens, str) and raw_tokens.strip():
+        raw_tokens = json.loads(raw_tokens)
+    overlay_tokens = {
+        str(t).casefold() for t in (raw_tokens or DEFAULT_OVERLAY_TOKENS)
+    }
+    # 去重开关：默认开；关掉时每个条目独立调用 LLM（不省调用，判定各自独立）。
+    dedupe = str(request.params.get("dedupe", "1")) not in ("0", "false", "False")
+
+    # 阶段 1：确定性规则层（不调 LLM）。
+    rule_verdicts = [_rule_verdict(entry["text"], overlay_tokens) for entry in entries]
+    llm_needed = [i for i, verdict in enumerate(rule_verdicts) if verdict is None]
+
+    # 阶段 2：LLM 分类层（去重：相同文本只判一次，上下文取首次出现）。
+    cat_by_index: dict[int, str] = {}
+    if llm_needed:
+        if dedupe:
+            first_of_key: dict[str, int] = {}
+            pool_indices: list[int] = []
+            for i in llm_needed:
+                key = _dedup_key(entries[i]["text"])
+                if key not in first_of_key:
+                    first_of_key[key] = i
+                    pool_indices.append(i)
+        else:
+            pool_indices = llm_needed
+
+        # 单条判断的工作函数：返回类别词。
+        def judge_one(index: int) -> str:
+            return _judge_category(entries, index, context_size, request.params)
 
         # 进度日志：打印已判定条数、总数与平均处理速度（条/s）。
         def log_progress(done: int, total: int, rate: float) -> None:
             logger.info("字幕判定进度 %d/%d 条 (%.1f 条/s)", done, total, rate)
 
-    # 自适应并发调用 LLM：10s 窗口内平均响应 < 0.3s 则加 1 线程（上限
-    # pool_max_workers），> pool_slow_threshold 则减 1 线程（下限 1），
-    # 按实测负载弹性伸缩，避免盲目并发压垮 LLM 接口。
+        # 自适应并发调用 LLM：按实测负载弹性伸缩，避免压垮 LLM 接口。
         pool = AdaptiveThreadPool(
             worker=judge_one,
             on_progress=log_progress,
@@ -145,18 +285,31 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
             fast_threshold=float(request.params.get("pool_fast_threshold", 0.3)),
             slow_threshold=float(request.params.get("pool_slow_threshold", 1.0)),
         )
-    verdicts = pool.map(range(len(entries)))
+        categories = pool.map(pool_indices)
 
-    kept: list[dict] = []
-    removed = 0
-    for index, (entry, verdict) in enumerate(zip(entries, verdicts)):
+        for index, category in zip(pool_indices, categories):
             # 并行下 LLM 异常被线程池隔离为异常结果：任一条失败即整体失败，
             # 避免静默输出未过滤结果。
-        if isinstance(verdict, Exception):
-            return InvokeResponse(status="failed", error=str(verdict))
-        if verdict:
+            if isinstance(category, Exception):
+                return InvokeResponse(status="failed", error=str(category))
+            cat_by_index[index] = category
+        if dedupe:
+            # 去重填充：与首次出现同键的条目复用同一类别，保证判定一致。
+            for i in llm_needed:
+                if i not in cat_by_index:
+                    cat_by_index[i] = cat_by_index[first_of_key[_dedup_key(entries[i]["text"])]]
+
+    # 阶段 3：合并规则与 LLM 判定，应用长文本保护并输出。
+    kept: list[dict] = []
+    removed = 0
+    for i, (entry, rule_verdict) in enumerate(zip(entries, rule_verdicts)):
+        # 规则层命中即删（True）；未命中则按 LLM 类别与长文本保护判定。
+        if rule_verdict is True:
             removed += 1
-            logger.info("删除无意义字幕 %d: %r", index + 1, entry["text"][:40])
+            continue
+        if _should_delete(cat_by_index[i], entry["text"], min_keep_len):
+            removed += 1
+            logger.info("删除无意义字幕 %d: %r", i + 1, entry["text"][:40])
         else:
             kept.append(entry)
 

scripts/regenerate_filter_ac7f480a3ccb.py

@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+"""用修复后的 llm_filter 对 run_ac7f480a3ccb 的真实 OCR 输出重新过滤。
+
+- 加载 .env（LLM Key/模型配置）
+- 复用节点 invoke 真实代码路径（与调度器执行完全一致）
+- 池参数按该 run 的 workflow v4 配置（单线程 pool 1/1）
+- 产物写入 run 的 filter 步骤目录（filtered.srt）
+"""
+import sys
+import time
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv(".env")
+
+from wov_sdk.models import InvokeRequest
+from nodes.llm_filter import invoke
+
+RUN = Path("data/storage/runs/run_ac7f480a3ccb")
+SRT_URI = RUN / "steps/ocr/subtitle.srt"
+OUT_DIR = RUN / "steps/filter"
+
+print(f"输入: {SRT_URI}", flush=True)
+print(f"输出目录: {OUT_DIR}", flush=True)
+
+started = time.monotonic()
+response = invoke(
+    InvokeRequest(
+        run_id="run_ac7f480a3ccb",
+        node_instance_id="",
+        inputs={"srt_uri": str(SRT_URI)},
+        # 与 workflow v4（单线程）保持一致。
+        params={"pool_min_workers": 1, "pool_max_workers": 1},
+        output_dir=str(OUT_DIR),
+    )
+)
+elapsed = time.monotonic() - started
+print(f"耗时 {elapsed:.0f}s", flush=True)
+if response.status != "completed":
+    print(f"FAILED: {response.error}", flush=True)
+    sys.exit(1)
+
+print(f"kept={response.outputs['kept']} removed={response.outputs['removed']}", flush=True)
+print(f"产物: {response.outputs['srt_uri']}", flush=True)

tests/test_llm_filter.py

@@ -1,14 +1,30 @@
 """LLM 字幕过滤节点测试。
 
-覆盖 SRT 解析/序列化、±N 上下文窗口组装（纯文本无时间戳、目标标记）、
-LLM 调用（按 I/O 边界 mock urlopen）与删除判定、invoke 全链路与异常路径。
+覆盖：
+- SRT 解析/序列化（多行、末条无空行、重新编号）；
+- 确定性规则层（横线装饰、HTML/水印 token、URL、单双 ASCII 字符直接删，不调 LLM）；
+- LLM 五类分类判断（garbage/overlay/noise 删，repeat/dialogue 留，未识别回退保留）；
+- 文本去重（相同文本只调一次 LLM，上下文取首次出现，判定结果一致）；
+- 长文本保护（≥min_keep_len 时 noise 类不删，需明确垃圾类别）；
+- 真实任务 run_ac7f480a3ccb 留存数据回归（OCR 1666 条：垃圾删除、对话保留、时间轴单调、去重后 LLM 调用数）；
+- invoke 全链路与异常路径。
 """
 
 import json
 import urllib.error
 from pathlib import Path
 
-from nodes.llm_filter import invoke, parse_srt, serialize_srt
+import pytest
+
+from nodes.llm_filter import (
+    _dedup_key,
+    _judge_category,
+    _rule_verdict,
+    _should_delete,
+    invoke,
+    parse_srt,
+    serialize_srt,
+)
 from wov_sdk.models import InvokeRequest
 
 # 4 条字幕的 SRT：第 3 条为"答："开头的无意义杂项，模拟 OCR 噪声。
@@ -19,6 +35,10 @@ _SRT = (
     "4\n00:00:13,000 --> 00:00:16,000\n第二句正常字幕\n"
 )
 
+# 真实任务 run_ac7f480a3ccb 的 OCR 输出（1666 条，2026-08 单线程 v4 跑完整 2 小时视频）。
+WORKSPACE = Path(__file__).resolve().parent.parent
+REAL_SRT = WORKSPACE / "testdata" / "ocr_srt_run_ac7f480a3ccb.srt"
+
 
 class FakeResponse:
     """模拟 urllib 响应：read() 返回 LLM 兼容接口的 JSON 载荷。"""
@@ -37,9 +57,9 @@ class FakeResponse:
 
 
 class FakeLLM:
-    """模拟 LLM 兼容接口：记录请求体，按策略返回"保留/删除"。
+    """模拟 LLM 兼容接口：记录请求体，按策略返回类别词。
 
-    支持两种策略：contents（按队列顺序，用于单次直调 _judge_target 的
+    支持两种策略：contents（按队列顺序，用于单次直调 _judge_category 的
     确定性测试）或 decision_fn（按请求体内容决策，用于并发 invoke 测试，
     保证任何线程执行顺序下判定结果都确定）。
     """
@@ -70,12 +90,17 @@ def _patch_llm(monkeypatch, contents: list[str] | None = None, decision_fn=None)
 
 
 def _decision_by_target(body) -> str:
-    """按目标字幕内容决策：含"答："判为删除，其余保留（与 _SRT 的噪声对应）。"""
+    """按目标字幕内容决策：含"答："判为 garbage，其余为 dialogue。"""
     target = next(
         line for line in body["messages"][1]["content"].splitlines()
         if line.startswith("【目标】")
     )
-    return "删除" if "答：" in target else "保留"
+    return "garbage" if "答：" in target else "dialogue"
+
+
+# ---------------------------------------------------------------------------
+# SRT 解析 / 序列化
+# ---------------------------------------------------------------------------
 
 
 def test_parse_srt_multiline_and_last_block() -> None:
@@ -105,14 +130,49 @@ def test_serialize_srt_renumbers() -> None:
     )
 
 
-def test_judge_target_window_and_keep(monkeypatch) -> None:
-    """窗口只含纯文本（无时间戳）、目标带标记；模型答"保留"则返回 False。"""
-    from nodes.llm_filter import _judge_target
+# ---------------------------------------------------------------------------
+# 确定性规则层
+# ---------------------------------------------------------------------------
 
+
+def test_rule_verdict_deletes_garbage_patterns() -> None:
+    """规则层直接删除：横线装饰、HTML/水印 token、URL/邮箱、单双 ASCII 字符。"""
+    tokens = {"html", "background", "___"}
+    for text in (
+        "---", "------", "------------------", "--- ---", "___", "= = =",
+        "…", "・", "HTML", "html", "Background", "background", "___",
+        "https://example.com/x", "www.example.com", "a@b.com",
+        "A", "V", "1", "DQ", "P4",
+    ):
+        assert _rule_verdict(text, tokens) is True, text
+    # 空文本/纯空白也删除。
+    assert _rule_verdict("   ", tokens) is True
+
+
+def test_rule_verdict_passes_cjk_and_dialogue() -> None:
+    """规则层不误伤：CJK 单字、正常对话、内容性短句交给 LLM 判断（None）。"""
+    tokens = {"html"}
+    for text in ("嗯", "好", "谢谢你 松井小姐", "不这么做的话 可没法胜任患者的对象", "好舒服"):
+        assert _rule_verdict(text, tokens) is None, text
+
+
+def test_rule_verdict_custom_overlay_tokens() -> None:
+    """overlay_tokens 参数化：自定义 token 同样直接删除。"""
+    assert _rule_verdict("Cleaning", {"cleaning"}) is True
+    assert _rule_verdict("Cleaning", set()) is None
+
+
+# ---------------------------------------------------------------------------
+# LLM 分类判断
+# ---------------------------------------------------------------------------
+
+
+def test_judge_category_window_and_dialogue(monkeypatch) -> None:
+    """窗口只含纯文本（无时间戳）、目标带标记；模型答 dialogue 则保留。"""
     entries = parse_srt(_SRT)
-    fake = _patch_llm(monkeypatch, ["保留"])
+    fake = _patch_llm(monkeypatch, ["dialogue"])
     # context_size=1，目标为第 2 条（index=1）：窗口 0..2 共 3 行，目标在中间。
-    assert _judge_target(entries, 1, context_size=1, params={}) is False
+    assert _judge_category(entries, 1, context_size=1, params={}) == "dialogue"
     body = fake.bodies[0]
     lines = body["messages"][1]["content"].splitlines()
     assert len(lines) == 3
@@ -125,32 +185,71 @@ def test_judge_target_window_and_keep(monkeypatch) -> None:
     assert body["max_tokens"] == 16
 
 
-def test_judge_target_delete(monkeypatch) -> None:
-    """模型答"删除"时返回 True（判定该条无意义）。"""
-    from nodes.llm_filter import _judge_target
-
+def test_judge_category_delete_classes(monkeypatch) -> None:
+    """模型答 garbage/overlay/noise 时返回对应类别（删除类）。"""
     entries = parse_srt(_SRT)
+    for cat in ("garbage", "overlay", "noise"):
+        _patch_llm(monkeypatch, [cat])
+        assert _judge_category(entries, 2, context_size=10, params={}) == cat
+
+
+def test_judge_category_repeat_and_unknown_kept(monkeypatch) -> None:
+    """repeat/dialogue 返回保留类；未识别输出（空/乱码/旧式"保留"）回退 dialogue。"""
+    entries = parse_srt(_SRT)
+    for answer in ("repeat", "dialogue", "", "???", "保留"):
+        _patch_llm(monkeypatch, [answer])
+        got = _judge_category(entries, 2, context_size=10, params={})
+        assert got in ("repeat", "dialogue"), (answer, got)
+    # 旧式"删除"回答兼容为垃圾类。
     _patch_llm(monkeypatch, ["删除"])
-    assert _judge_target(entries, 2, context_size=10, params={}) is True
+    assert _judge_category(entries, 2, context_size=10, params={}) == "garbage"
 
 
-def test_judge_target_model_and_auth(monkeypatch) -> None:
+def test_judge_category_model_and_auth(monkeypatch) -> None:
     """模型名从参数取；配置 API Key 时附带 Bearer 鉴权头。"""
-    from nodes.llm_filter import _judge_target
-
     entries = parse_srt(_SRT)
     monkeypatch.setenv("LLM_API_KEY", "sk-test")
-    fake = _patch_llm(monkeypatch, ["保留"])
-    assert _judge_target(entries, 0, context_size=10, params={"model": "m/1"}) is False
+    fake = _patch_llm(monkeypatch, ["dialogue"])
+    assert _judge_category(entries, 0, context_size=10, params={"model": "m/1"}) == "dialogue"
     assert fake.bodies[0]["model"] == "m/1"
     assert fake.headers[0]["Authorization"] == "Bearer sk-test"
 
 
+# ---------------------------------------------------------------------------
+# 去重键与删除判定
+# ---------------------------------------------------------------------------
+
+
+def test_dedup_key_normalizes_whitespace_and_case() -> None:
+    """去重键：去掉全部空白并统一大小写，空格差异视为同一文本。"""
+    assert _dedup_key("不这么做的话 可没法胜任患者的对象") == _dedup_key(
+        "不这么做的话可没法胜任患者的对象"
+    )
+    assert _dedup_key("HTML") == _dedup_key("html")
+    assert _dedup_key("你好 世界") != _dedup_key("你好世界2")
+
+
+def test_should_delete_long_text_noise_protected() -> None:
+    """长文本保护：≥min_keep_len 时 noise 不删，garbage/overlay 仍删；短文本三类都删。"""
+    long_text = "不这么做的话 可没法胜任患者的对象"
+    assert _should_delete("noise", long_text, min_keep_len=12) is False
+    assert _should_delete("garbage", long_text, min_keep_len=12) is True
+    assert _should_delete("overlay", long_text, min_keep_len=12) is True
+    assert _should_delete("noise", "答：无意义杂项", min_keep_len=12) is True
+    assert _should_delete("repeat", long_text, min_keep_len=12) is False
+    assert _should_delete("dialogue", long_text, min_keep_len=12) is False
+
+
+# ---------------------------------------------------------------------------
+# invoke 全链路
+# ---------------------------------------------------------------------------
+
+
 def test_invoke_filters_and_renumbers(monkeypatch, tmp_path) -> None:
-    """全链路（并发）：按 LLM 判定删除无意义条，保留条重新编号输出。"""
+    """全链路（并发）：按 LLM 类别判定删除无意义条，保留条重新编号输出。"""
     srt = tmp_path / "in.srt"
     srt.write_text(_SRT, encoding="utf-8")
-    # 内容决策：目标字幕含"答："判删除，其余保留（任何线程顺序下结果确定）。
+    # 内容决策：目标字幕含"答："判 garbage，其余 dialogue（任何线程顺序下结果确定）。
     _patch_llm(monkeypatch, decision_fn=_decision_by_target)
     response = invoke(
         InvokeRequest(
@@ -172,6 +271,79 @@ def test_invoke_filters_and_renumbers(monkeypatch, tmp_path) -> None:
     assert "00:00:13,000 --> 00:00:16,000\n第二句正常字幕\n" in out
 
 
+def test_invoke_rules_skip_llm(monkeypatch, tmp_path) -> None:
+    """规则层命中时直接删除且不调用 LLM：横线/HTML/单字符全部清理。"""
+    srt = tmp_path / "in.srt"
+    srt.write_text(
+        "1\n00:00:01,000 --> 00:00:04,000\n---\n\n"
+        "2\n00:00:05,000 --> 00:00:08,000\nHTML\n\n"
+        "3\n00:00:09,000 --> 00:00:12,000\nV\n\n"
+        "4\n00:00:13,000 --> 00:00:16,000\n正常对话\n",
+        encoding="utf-8",
+    )
+    fake = _patch_llm(monkeypatch, ["dialogue"])
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+    assert response.outputs["kept"] == 1
+    assert response.outputs["removed"] == 3
+    assert len(fake.bodies) == 1  # 仅"正常对话"需 LLM，其余全走规则。
+
+
+def test_invoke_dedup_single_llm_call(monkeypatch, tmp_path) -> None:
+    """去重：相同文本（含空格变体）只调一次 LLM，判定结果一致。"""
+    srt = tmp_path / "in.srt"
+    srt.write_text(
+        "1\n00:00:01,000 --> 00:00:04,000\n好舒服\n\n"
+        "2\n00:00:05,000 --> 00:00:08,000\n好 舒 服\n\n"
+        "3\n00:00:09,000 --> 00:00:12,000\n好舒服\n\n"
+        "4\n00:00:13,000 --> 00:00:16,000\n别的台词\n",
+        encoding="utf-8",
+    )
+    fake = _patch_llm(monkeypatch, decision_fn=lambda body: "repeat")
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+    # 2 个唯一文本（好舒服/别的台词）→ 恰好 2 次 LLM 调用。
+    assert len(fake.bodies) == 2
+    # repeat 类保留 → 4 条全部保留。
+    assert response.outputs["kept"] == 4
+    assert response.outputs["removed"] == 0
+
+
+def test_invoke_dedupe_disabled(monkeypatch, tmp_path) -> None:
+    """dedupe=0 关闭去重：每个条目都调一次 LLM。"""
+    srt = tmp_path / "in.srt"
+    srt.write_text(
+        "1\n00:00:01,000 --> 00:00:04,000\n同一句\n\n"
+        "2\n00:00:05,000 --> 00:00:08,000\n同一句\n",
+        encoding="utf-8",
+    )
+    fake = _patch_llm(monkeypatch, decision_fn=lambda body: "dialogue")
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={"dedupe": "0"},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+    assert len(fake.bodies) == 2
+
+
 def test_invoke_context_size_param(monkeypatch, tmp_path) -> None:
     """context_size 参数生效：窗口大小=2×context_size+1（两端截断除外）。"""
     srt = tmp_path / "in.srt"
@@ -195,6 +367,52 @@ def test_invoke_context_size_param(monkeypatch, tmp_path) -> None:
     assert len(window) == 3
 
 
+def test_invoke_overlay_tokens_param(monkeypatch, tmp_path) -> None:
+    """overlay_tokens 参数：自定义 token 走规则层删除，不调用 LLM。"""
+    srt = tmp_path / "in.srt"
+    srt.write_text(
+        "1\n00:00:01,000 --> 00:00:04,000\nCleaning\n\n"
+        "2\n00:00:05,000 --> 00:00:08,000\n正常对话\n",
+        encoding="utf-8",
+    )
+    fake = _patch_llm(monkeypatch, decision_fn=lambda body: "dialogue")
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={"overlay_tokens": ["cleaning"]},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+    assert response.outputs["removed"] == 1
+    assert response.outputs["kept"] == 1
+    assert len(fake.bodies) == 1  # 只判正常对话。
+
+
+def test_invoke_overlay_tokens_json_string(monkeypatch, tmp_path) -> None:
+    """overlay_tokens 以 JSON 字符串形式传入（工作流参数常见形态）同样生效。"""
+    srt = tmp_path / "in.srt"
+    srt.write_text(
+        "1\n00:00:01,000 --> 00:00:04,000\nXXLogo\n\n"
+        "2\n00:00:05,000 --> 00:00:08,000\n正常对话\n",
+        encoding="utf-8",
+    )
+    fake = _patch_llm(monkeypatch, ["dialogue"])
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={"overlay_tokens": '["xxlogo", "xx"]'},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+    assert response.outputs["removed"] == 1
+    assert response.outputs["kept"] == 1
+    assert len(fake.bodies) == 1
+
+
 def test_invoke_missing_input(tmp_path) -> None:
     """缺少 srt_uri 时返回失败。"""
     response = invoke(
@@ -255,3 +473,79 @@ def test_invoke_empty_srt(monkeypatch, tmp_path) -> None:
     assert response.outputs["kept"] == 0
     assert response.outputs["removed"] == 0
     assert fake.bodies == []
+
+
+# ---------------------------------------------------------------------------
+# 真实任务留存数据回归（testdata/ocr_srt_run_ac7f480a3ccb.srt）
+# ---------------------------------------------------------------------------
+
+
+def test_real_run_rules_and_dialogue_regression(monkeypatch, tmp_path) -> None:
+    """真实数据回归：规则层清理垃圾、对话保留、时间轴单调、去重生效。
+
+    夹具为 run_ac7f480a3ccb（单线程 v4 跑完整 2 小时视频）的 OCR 输出 1666 条。
+    假 LLM 对非规则条目一律答 dialogue：验证规则层删掉全部垃圾（横线/水印/
+    单字符），真实对话全部保留，且相同文本只调一次 LLM（去重）。
+    """
+    srt = REAL_SRT
+    if not srt.is_file():
+        pytest.skip("缺少 testdata/ocr_srt_run_ac7f480a3ccb.srt，跳过回归测试")
+    fake = _patch_llm(monkeypatch, decision_fn=lambda body: "dialogue")
+    response = invoke(
+        InvokeRequest(
+            run_id="r", node_instance_id="",
+            inputs={"srt_uri": str(srt)},
+            params={},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed", response.error
+
+    entries = parse_srt(srt.read_text(encoding="utf-8"))
+    out = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8")
+
+    # 规则层删除量 = 夹具中直接命中规则文本的条数（真实数据动态计算）。
+    # 规则层删除量 = 夹具中直接命中规则文本的条数（与 invoke 默认 token 集一致）。
+    from nodes.llm_filter import DEFAULT_OVERLAY_TOKENS
+
+    tokens = set(DEFAULT_OVERLAY_TOKENS)
+    rule_removed = sum(
+        1 for e in entries if _rule_verdict(e["text"], tokens) is True
+    )
+    assert response.outputs["removed"] == rule_removed
+    assert response.outputs["kept"] == len(entries) - rule_removed
+
+    # 垃圾全部清出输出（横线、HTML、单字符都不再出现；只检查文本字段，
+    # 排除 SRT 的序号行与时间轴行）。
+    for entry in parse_srt(out):
+        assert _rule_verdict(entry["text"], tokens) is not True, entry["text"]
+    # 真实对话保留（修复前被误删的自我介绍、请求对话、呻吟内容行）。
+    # 注意：片头标题卡（淫魔病院…）真实 LLM 判为 overlay（烧录标题），
+    # 不在必保留列表内；本测试假 LLM 一律答 dialogue，仅验证机制正确。
+    for kept_line in (
+        "谢谢你 松井小姐",
+        "我姓泷本 请多多指教",
+        "不这么做的话 可没法胜任患者的对象",
+        "你要看着北冈小姐的脸 你们俩相互看看嘛",
+        "好舒服",
+    ):
+        assert kept_line in out, kept_line
+
+    # 时间轴单调递增（输出顺序即时间顺序）。
+    times = []
+    for m in __import__("re").finditer(
+        r"(\d{2}:\d{2}:\d{2},\d{3})\s*-->", out
+    ):
+        t = m.group(1).replace(",", ".")
+        h, mi, s = t.split(":")
+        times.append(int(h) * 3600 + int(mi) * 60 + float(s))
+    assert all(a < b for a, b in zip(times, times[1:]))
+
+    # 去重：LLM 调用数 == 非规则条目中的唯一文本数（远小于条目数）。
+    unique_llm = len({
+        _dedup_key(e["text"])
+        for e in entries
+        if _rule_verdict(e["text"], tokens) is None
+    })
+    assert len(fake.bodies) == unique_llm
+    assert len(fake.bodies) < len(entries)  # 去重确实省调用。

workflows/ocr-subtitle.json

@@ -2,10 +2,10 @@
   "id": "ocr-subtitle",
   "name": "字幕OCR提取",
   "description": "抽帧并 OCR 提取视频烧录字幕，经 LLM 过滤无意义内容后生成带时间轴的 SRT 基准数据。",
-  "version": 3,
+  "version": 4,
   "definition": {
     "name": "字幕OCR提取",
-    "version": 3,
+    "version": 4,
     "nodes": [
       {
         "id": "extract",
@@ -27,7 +27,9 @@
         "id": "ocr",
         "node_type": "subtitle-ocr",
         "params": {
-          "prompt": "提取图像中的文字，不要描述图片中的内容"
+          "prompt": "提取图像中的文字，不要描述图片中的内容",
+          "pool_min_workers": 1,
+          "pool_max_workers": 1
         },
         "inputs": {
           "frames_manifest": "extract.frames_manifest"
@@ -36,7 +38,10 @@
       {
         "id": "filter",
         "node_type": "llm-filter",
-        "params": {},
+        "params": {
+          "pool_min_workers": 1,
+          "pool_max_workers": 1
+        },
         "inputs": {
           "srt_uri": "ocr.srt_uri"
         }