v0.9: parallelize phase4 and add model/search playbooks

2026-04-24 10:19:48 +08:00
parent c444007f04
commit c93fce5078
13 changed files with 626 additions and 100 deletions
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""根据 glossary 的核查结果，在 final_zh_polished.md 上做精确的文本替换。
+"""根据 glossary 的核查结果，在 final_zh.md 上做精确的文本替换。

 原理：
 - build_glossary.py 会给每条术语标 `zh`（正确中文）和 `issue`（有发现问题）
@@ -178,7 +178,7 @@ def main() -> int:
    parser = argparse.ArgumentParser(description="用 glossary 修正正文术语")
    parser.add_argument("project", help="项目 slug 或路径")
    parser.add_argument(
-        "--input", default="phase4/final_zh_polished.md",
+        "--input", default="phase4/final_zh.md",
        help="待修正的 Markdown",
    )
    parser.add_argument(
@@ -22,7 +22,7 @@

 用法：
    uv run python scripts/build_glossary.py <project_slug>
-    uv run python scripts/build_glossary.py <project_slug> --workers 6
+    uv run python scripts/build_glossary.py <project_slug> --workers 4
    uv run python scripts/build_glossary.py <project_slug> --only "Mabwell,Maywavee"
    uv run python scripts/build_glossary.py <project_slug> --force

@@ -184,7 +184,7 @@ def main() -> int:
    )
    parser.add_argument("--model", default=DEFAULT_MODEL, help="LLM 模型")
    parser.add_argument(
-        "--workers", type=int, default=6, help="并发度（默认 6，Exa 限速 5 QPS）"
+        "--workers", type=int, default=4, help="并发度（默认 4；网络不稳时建议降到 3）"
    )
    parser.add_argument(
        "--force", action="store_true",
@@ -15,6 +15,7 @@ from __future__ import annotations
 import json
 import os
 import sys
+import threading
 import time
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -111,6 +112,8 @@ class ZenMuxClient:
        self.log_file = log_file
        self.usage = UsageStats()
        self._client = httpx.Client(timeout=timeout)
+        self._log_lock = threading.Lock()
+        self._usage_lock = threading.Lock()

    def close(self) -> None:
        self._client.close()
@@ -125,8 +128,9 @@ class ZenMuxClient:
        if not self.log_file:
            return
        self.log_file.parent.mkdir(parents=True, exist_ok=True)
-        with self.log_file.open("a", encoding="utf-8") as f:
-            f.write(json.dumps(payload, ensure_ascii=False) + "\n")
+        with self._log_lock:
+            with self.log_file.open("a", encoding="utf-8") as f:
+                f.write(json.dumps(payload, ensure_ascii=False) + "\n")

    def chat_complete(
        self,
@@ -191,7 +195,8 @@ class ZenMuxClient:
                except Exception as e:
                    raise ZenMuxError(f"invalid JSON from zenmux: {e}; body={resp.text[:500]}")
                usage = data.get("usage", {}) or {}
-                self.usage.add(model, usage)
+                with self._usage_lock:
+                    self.usage.add(model, usage)
                content = ""
                choices = data.get("choices") or []
                if choices:
@@ -19,6 +19,7 @@
 from __future__ import annotations

 import argparse
+import concurrent.futures
 import json
 import sys
 import time
@@ -141,6 +142,12 @@ def main() -> int:
        default=None,
        help="最多润色前 N 个未缓存的块（调试用）",
    )
+    parser.add_argument(
+        "--workers",
+        type=int,
+        default=4,
+        help="并发润色 worker 数（默认 4；设为 1 回退串行）",
+    )
    args = parser.parse_args()

    load_secrets()
@@ -167,50 +174,82 @@ def main() -> int:
    total_cn = sum(count_chinese_chars(b.content) for b in blocks)
    print(f"Source: {src_path.relative_to(project_root)}")
    print(f"Blocks: {len(blocks)}  | total Chinese chars: {total_cn:,}")
-    print(f"Model: {args.model}  | temperature: {args.temperature}")
+    workers = max(1, args.workers)
+    print(f"Model: {args.model}  | temperature: {args.temperature}  | workers: {workers}")
    print()

    start = time.time()
-    translated_this_run = 0
+    todo: list[MarkdownBlock] = []
+    cached_count = 0
    notes_records: list[dict] = []
+    for b in blocks:
+        chunk_path = chunks_dir / f"{b.order:03d}-{b.anchor}.md"
+        if only_orders is not None and b.order not in only_orders:
+            continue
+        if chunk_path.exists() and not args.force:
+            cached_count += 1
+            print(f"  [ok  ] #{b.order:03d} {b.short_title}  (cached)")
+            continue
+        if args.limit is not None and len(todo) >= args.limit:
+            continue
+        todo.append(b)
+
+    if todo:
+        print(f"To polish this run: {len(todo)} blocks  | cached: {cached_count}")
+        print()
+
    with ZenMuxClient(log_file=log_file) as client:
-        for b in blocks:
+        def run_one(b: MarkdownBlock) -> tuple[MarkdownBlock, str, str, float]:
            chunk_path = chunks_dir / f"{b.order:03d}-{b.anchor}.md"
-            if only_orders is not None and b.order not in only_orders:
-                continue
-            if chunk_path.exists() and not args.force:
-                print(f"  [ok  ] #{b.order:03d} {b.short_title}  (cached)")
-                continue
-            if args.limit is not None and translated_this_run >= args.limit:
-                continue
-
-            label = f"#{b.order:03d} L{b.level} {count_chinese_chars(b.content):>4}字 {b.short_title}"
-            print(f"  [... ] {label}  ", end="", flush=True)
            t0 = time.time()
-            try:
-                polished, notes = polish_block(
-                    client,
-                    b,
-                    model=args.model,
-                    system_prompt=system_prompt,
-                    temperature=args.temperature,
-                )
-            except (ZenMuxError, RuntimeError) as e:
-                print(f"\n  [FAIL] {label}\n         {e}")
-                continue
-            elapsed = time.time() - t0
-
+            polished, notes = polish_block(
+                client,
+                b,
+                model=args.model,
+                system_prompt=system_prompt,
+                temperature=args.temperature,
+            )
            chunk_path.write_text(polished + "\n", encoding="utf-8")
-            cn = count_chinese_chars(polished)
-            before_cn = count_chinese_chars(b.content)
-            delta = cn - before_cn
-            sign = "+" if delta >= 0 else ""
-            translated_this_run += 1
-            if notes:
-                notes_records.append(
-                    {"order": b.order, "anchor": b.anchor, "title": b.short_title, "notes": notes}
-                )
-            print(f"\r  [done] {label}  → {cn}字 ({sign}{delta}, {elapsed:4.1f}s)")
+            return b, polished, notes, time.time() - t0
+
+        if workers == 1:
+            for b in todo:
+                label = f"#{b.order:03d} L{b.level} {count_chinese_chars(b.content):>4}字 {b.short_title}"
+                print(f"  [... ] {label}  ", end="", flush=True)
+                try:
+                    block, polished, notes, elapsed = run_one(b)
+                except (ZenMuxError, RuntimeError) as e:
+                    print(f"\n  [FAIL] {label}\n         {e}")
+                    continue
+                cn = count_chinese_chars(polished)
+                before_cn = count_chinese_chars(block.content)
+                delta = cn - before_cn
+                sign = "+" if delta >= 0 else ""
+                if notes:
+                    notes_records.append(
+                        {"order": block.order, "anchor": block.anchor, "title": block.short_title, "notes": notes}
+                    )
+                print(f"\r  [done] {label}  → {cn}字 ({sign}{delta}, {elapsed:4.1f}s)")
+        else:
+            with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
+                futures = {pool.submit(run_one, b): b for b in todo}
+                for fut in concurrent.futures.as_completed(futures):
+                    b = futures[fut]
+                    label = f"#{b.order:03d} L{b.level} {count_chinese_chars(b.content):>4}字 {b.short_title}"
+                    try:
+                        block, polished, notes, elapsed = fut.result()
+                    except (ZenMuxError, RuntimeError) as e:
+                        print(f"  [FAIL] {label}\n         {e}")
+                        continue
+                    cn = count_chinese_chars(polished)
+                    before_cn = count_chinese_chars(block.content)
+                    delta = cn - before_cn
+                    sign = "+" if delta >= 0 else ""
+                    if notes:
+                        notes_records.append(
+                            {"order": block.order, "anchor": block.anchor, "title": block.short_title, "notes": notes}
+                        )
+                    print(f"  [done] {label}  → {cn}字 ({sign}{delta}, {elapsed:4.1f}s)")

    # 汇总
    merged: list[str] = []
@@ -6,12 +6,12 @@
    # 或：
    uv run python scripts/translate.py projects/dual-target-rnai-pipeline-2026

-断点续传：每块翻译完立即写入 `phase4/zh_chunks/<anchor>.md` 和术语表 patch。
+断点续传：每块翻译完立即写入 `phase4/zh_chunks/<anchor>.md`，术语表 patch 在本轮结束后统一合并。
 重跑时已存在的块直接跳过，只译缺的。

 设计要点：
 1. 切块按 H2 粒度，单块一般 <600 英文词，单次 API 调用远低于 Sonnet output token 上限
-2. 术语表累积式更新：每块调用传入当前已知术语，译完回写 patch，保证全文一致
+2. 并发翻译使用稳定术语表快照，译完后统一合并 glossary patch，避免多线程写冲突
 3. 失败不会污染最终产物：块级文件独立，可重跑；汇总步骤独立
 4. 日志完整：每次 API 调用写 `phase4/logs/translate.jsonl`
 """
@@ -19,6 +19,7 @@
 from __future__ import annotations

 import argparse
+import concurrent.futures
 import json
 import sys
 import time
@@ -29,7 +30,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 from scripts.lib.markdown_chunker import (
    MarkdownBlock,
    count_chinese_chars,
-    merge_blocks,
    split_by_headers,
 )
 from scripts.lib.zenmux_client import ZenMuxClient, ZenMuxError, load_secrets
@@ -112,12 +112,19 @@ def save_glossary(path: Path, glossary: dict[str, str]) -> None:
    )


+def _glossary_value(value) -> str:
+    if isinstance(value, dict):
+        return str(value.get("zh") or value.get("中文") or "")
+    return str(value)
+
+
 def build_user_prompt(block: MarkdownBlock, glossary: dict[str, str]) -> str:
-    glossary_hint = (
-        "\n".join(f"{en} || {zh}" for en, zh in sorted(glossary.items()))
-        if glossary
-        else "(none yet)"
+    glossary_lines = (
+        f"{en} || {_glossary_value(zh)}"
+        for en, zh in sorted(glossary.items())
+        if _glossary_value(zh)
    )
+    glossary_hint = "\n".join(glossary_lines) or "(none yet)"
    level_hint = (
        f"H{block.level}" if block.level >= 1 else "frontmatter (no heading)"
    )
@@ -195,6 +202,12 @@ def main() -> int:
        default=None,
        help="最多翻译前 N 个未缓存的块（调试用）",
    )
+    parser.add_argument(
+        "--workers",
+        type=int,
+        default=4,
+        help="并发翻译 worker 数（默认 4；设为 1 回退串行）",
+    )
    args = parser.parse_args()

    load_secrets()
@@ -223,47 +236,83 @@ def main() -> int:
    print(f"Source: {src_path.relative_to(project_root)}")
    print(f"Blocks: {len(blocks)}  | total English words: {total_en_words:,}")
    print(f"Glossary loaded: {len(glossary)} terms")
-    print(f"Model: {args.model}  | temperature: {args.temperature}")
+    workers = max(1, args.workers)
+    print(f"Model: {args.model}  | temperature: {args.temperature}  | workers: {workers}")
    print()

    start = time.time()
-    translated_this_run = 0
+    todo: list[MarkdownBlock] = []
+    cached_count = 0
+    for b in blocks:
+        chunk_path = chunks_dir / f"{b.order:03d}-{b.anchor}.md"
+        if only_orders is not None and b.order not in only_orders:
+            continue
+        if chunk_path.exists() and not args.force:
+            cached_count += 1
+            print(f"  [ok  ] #{b.order:03d} {b.short_title}  (cached)")
+            continue
+        if args.limit is not None and len(todo) >= args.limit:
+            continue
+        todo.append(b)
+
+    if todo:
+        print(f"To translate this run: {len(todo)} blocks  | cached: {cached_count}")
+        print()
+
+    successful_patches: list[dict[str, str]] = []
    with ZenMuxClient(log_file=log_file) as client:
-        for b in blocks:
+        def run_one(b: MarkdownBlock) -> tuple[MarkdownBlock, str, dict[str, str], float]:
            chunk_path = chunks_dir / f"{b.order:03d}-{b.anchor}.md"
-            if only_orders is not None and b.order not in only_orders:
-                continue
-            if chunk_path.exists() and not args.force:
-                print(f"  [ok  ] #{b.order:03d} {b.short_title}  (cached)")
-                continue
-            if args.limit is not None and translated_this_run >= args.limit:
-                continue
-
-            label = f"#{b.order:03d} L{b.level} {b.word_count:>4}w  {b.short_title}"
-            print(f"  [... ] {label}  ", end="", flush=True)
            t0 = time.time()
-            try:
-                translation, patch = translate_block(
-                    client,
-                    b,
-                    glossary,
-                    model=args.model,
-                    system_prompt=system_prompt,
-                    temperature=args.temperature,
-                )
-            except (ZenMuxError, RuntimeError) as e:
-                print(f"\n  [FAIL] {label}\n         {e}")
-                continue
-            elapsed = time.time() - t0
-
+            translation, patch = translate_block(
+                client,
+                b,
+                glossary,
+                model=args.model,
+                system_prompt=system_prompt,
+                temperature=args.temperature,
+            )
            chunk_path.write_text(translation + "\n", encoding="utf-8")
-            if patch:
-                for k, v in patch.items():
-                    glossary.setdefault(k, v)
-                save_glossary(glossary_path, glossary)
-            cn = count_chinese_chars(translation)
-            translated_this_run += 1
-            print(f"\r  [done] {label}  → {cn:>4}字  ({elapsed:4.1f}s, +{len(patch)} terms)")
+            return b, translation, patch, time.time() - t0
+
+        if workers == 1:
+            for b in todo:
+                label = f"#{b.order:03d} L{b.level} {b.word_count:>4}w  {b.short_title}"
+                print(f"  [... ] {label}  ", end="", flush=True)
+                try:
+                    _block, translation, patch, elapsed = run_one(b)
+                except (ZenMuxError, RuntimeError) as e:
+                    print(f"\n  [FAIL] {label}\n         {e}")
+                    continue
+                successful_patches.append(patch)
+                cn = count_chinese_chars(translation)
+                print(f"\r  [done] {label}  → {cn:>4}字  ({elapsed:4.1f}s, +{len(patch)} terms)")
+        else:
+            with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
+                futures = {pool.submit(run_one, b): b for b in todo}
+                for fut in concurrent.futures.as_completed(futures):
+                    b = futures[fut]
+                    label = f"#{b.order:03d} L{b.level} {b.word_count:>4}w  {b.short_title}"
+                    try:
+                        _block, translation, patch, elapsed = fut.result()
+                    except (ZenMuxError, RuntimeError) as e:
+                        print(f"  [FAIL] {label}\n         {e}")
+                        continue
+                    successful_patches.append(patch)
+                    cn = count_chinese_chars(translation)
+                    print(f"  [done] {label}  → {cn:>4}字  ({elapsed:4.1f}s, +{len(patch)} terms)")
+
+        glossary_conflicts: list[tuple[str, str, str]] = []
+        for patch in successful_patches:
+            for k, v in patch.items():
+                if k not in glossary:
+                    glossary[k] = v
+                elif _glossary_value(glossary[k]) != v:
+                    glossary_conflicts.append((k, _glossary_value(glossary[k]), v))
+        if successful_patches:
+            save_glossary(glossary_path, glossary)
+        if glossary_conflicts:
+            print(f"  [WARN] glossary patch 冲突 {len(glossary_conflicts)} 条，保留既有译法")

    # 汇总：按 order 拼接所有 chunk
    merged: list[str] = []