book-translation

第10章 · 多 Agent 协作 · 配套项目 chapter10/book-translation

项目说明

实验 10-3：书籍翻译 Agent —— 管理者模式（Orchestration）

配套代码，演示如何用管理者模式把长文档翻译拆给多个专职 Agent。核心是 上下文隔离与控制 Manager 上下文膨胀：Manager 只保存任务、计划、各 Agent 调用记录和文件索引，完整译文全部落盘到文件系统，因此无论书有多长， Manager 的上下文都基本恒定。

目的

对比「单 Agent 一条对话翻完整本书」与「管理者模式多 Agent 协作」两种方案，用 真实 token 数说明后者如何控制主/Manager 上下文膨胀，并用共享术语表保证 全书术语一致。

架构：四种 Agent

Agent	输入（独立上下文）	产出	上下文特点
Glossary Agent	全书内容	结构化术语表 glossary.json	读全书，产出后即释放
Translation Agent	当前章节 + 术语表 + 翻译指南	chapterN_zh.md	每章一个独立实例，只看到自己这一章
Proofreading Agent	所有译文 + 术语表	审校报告 proofreading_report.json	做一致性 / 流畅性检查
Manager Agent	任务 + 文件索引 + 报告摘要	调度决策（是否发回修订）	只存元信息，不存正文

数据流：Manager 调度 Glossary → 逐章 Translation（共享同一份术语表文件）→ Proofreading → Manager 依据报告决定是否把个别章节发回 Translation 修订。译文与 术语表都通过文件系统传递，Manager 只在上下文里保存文件路径。

关键设计：Manager 把「编辑部指定术语」（house style，如 token→词元、 prompt→提示词、latency→时延）强制写入共享术语表，下发给每个 Translation Agent， 从而把指定译法贯彻到全书。单 Agent 看不到术语表，只能用自己的默认译法。

目录

book-translation/
├── agents.py          # 四种 Agent + 两种运行方式 + token 追踪
├── consistency.py     # 术语一致性 / 术语表遵从率（确定性字符串匹配）
├── demo.py            # 一键演示：跑管理者模式 + 单 Agent 对照，打印对比表
├── sample_book/       # 自带英文技术小书（4 个短章节，含术语与代码）
│   ├── chapter1.md ... chapter4.md
├── output/            # 运行时生成：术语表 / 各章译文 / 审校报告（已 gitignore）
├── requirements.txt
└── env.example

运行

pip install -r requirements.txt
cp env.example .env      # 填入 OPENAI_API_KEY
python demo.py

python demo.py 会先打印四 Agent 协作的实时轨迹（Manager 制定计划 → 调度 Glossary → 逐章 Translation → Proofreading → 依报告决定修订），再打印各 Agent 的 token 消耗与管理者模式 vs 单 Agent 的核心对比表。

• 模型默认 gpt-5.6-luna（当前便宜旗舰），可用 OPENAI_MODEL 覆盖；如需自建/代理端点，设 OPENAI_BASE_URL。
• Key 与通用回退：优先用 OPENAI_API_KEY 直连 OpenAI；若未设置该变量但设了 OPENROUTER_API_KEY，则自动改走 OpenRouter，并把模型名映射到其命名空间 （gpt-5.6-luna → openai/gpt-5.6-luna）。提示：gpt-5.6 系列直连 OpenAI 需组织验证， 只填 OPENROUTER_API_KEY（不填 OPENAI_API_KEY）即可强制走 OpenRouter，更省事。
• 任务规模刻意很小（4 个短章节），一次运行成本约几百分之一美元。
• 不带任何参数运行与旧版行为完全一致。

命令行参数（python demo.py --help）

参数	作用	默认
--dry-run	离线预演：只画四 Agent 协作图、Manager 计划、编辑部术语与各 Agent 的 token 预算，不调用任何 API、无需 Key	关闭
--sample-dir DIR	待翻译书籍目录（读取其中 *.md，按文件名排序）	sample_book/
--out-dir DIR	产物根目录（其下再分 orchestration/、single_agent/）	output/
--source-lang LANG / --target-lang LANG	源 / 目标语言（仅影响提示词措辞）	英文 / 中文
--no-glossary	关闭 Glossary Agent（仅保留编辑部指定术语）	启用
--no-proofreading	关闭 Proofreading Agent 与 Manager 修订闭环	启用
--model MODEL	临时覆盖模型（等价于设 OPENAI_MODEL）	gpt-5.6-luna
--skip-single	只跑管理者模式，跳过单 Agent 对照组	关闭

注意：内置的术语一致性 / 遵从率统计（consistency.py）针对 英文→中文 调校； 改翻译方向仍可正常翻译，但该统计表意义有限。

无 Key / 离线快速查看架构：

python demo.py --dry-run     # 打印四 Agent 协作图 + Manager 计划 + token 预算，不联网

该模式用 tiktoken 离线估算各 Agent 会读到的上下文规模，直观印证「Manager 上下文 只随章节数加几行记录、与每章正文长度无关」，而单 Agent 的累积上下文随书长线性膨胀。

token 统计口径

• 子 Agent / 单 Agent 的输入、输出 token 取 OpenAI 返回的真实 usage。
• 「上下文峰值」= 某 Agent 所有调用中，单次输入上下文（prompt tokens）的最大值， 用来衡量上下文膨胀。
• Manager 上下文峰值：Manager 状态（任务/计划/调用记录/文件索引）序列化后用 tiktoken 统计的 token 数峰值 —— 它从不包含完整译文。

结论（真实运行结果，gpt-5.6-luna，4 章）

指标	管理者模式	单 Agent
主/Manager 上下文峰值 (tokens)	697	2320
Manager LLM 决策调用上下文 (tokens)	783	—
全流程总 token	11849	6886
术语内部一致率	100%	89%
指定术语遵从率	100%	53%
参与 Agent 种类数	4	1

1. 控制上下文膨胀：单 Agent 的主上下文随章节累积，峰值达 2320 tokens；管理者 模式下 Manager 上下文峰值仅 697 tokens（约 3.3 倍差距）。更重要的是，Manager 上下文与书的长度基本无关（只加一行调用记录/文件索引），而单 Agent 的累积 上下文会随章节线性增长——书越长，差距越大。子 Agent 的上下文各自隔离、互不污染 （每个 Translation 实例峰值仅约 547 tokens）。
2. 术语一致性：管理者模式把编辑部指定术语写入共享术语表并强制下发，4 个指定 术语在全书的遵从率 100%；单 Agent 看不到术语表，遵从率仅 53%。换用更强 的 gpt-5.6-luna 后，单 Agent 会自发采用部分「常识译法」（token→词元、 prompt→提示词都命中了指定译法），但对没有唯一标准的术语仍各行其是（latency 全书 译成「延迟」而非规定的「时延」，embedding 译成「嵌入」而非「嵌入向量」，各 0/4、 0/3 遵从）。更关键的是，单 Agent 即便同一个术语也会跨章漂移——token 在部分章 译成「词元」、另一些章直接留「token」，术语内部一致率因此掉到 89%；管理者模式靠 共享术语表把这两类问题一起消除（内部一致率 100%、遵从率 100%）。
3. 代价：管理者模式花了明显更多 token（11849 vs 6886，额外的术语表抽取、审校、 调度调用，且推理模型输出更长），换来的是主上下文可控与术语可强制统一—— 这正是长文档翻译真正需要的性质。

说明：术语一致性用确定性字符串匹配统计（见 consistency.py），不是让模型自评。 具体数字每次运行会有小幅波动，但上述量级与结论稳定复现。

局限

• 上表在 gpt-5.6-luna 上验证；换更强/更弱的模型，两种模式的差距会变化——越强的 单 Agent 越容易自发命中部分常识译法（遵从率从更弱模型的近 0% 升到本次的 53%）， 但仍无法覆盖没有唯一标准的术语，也仍会跨章漂移，管理者模式的共享术语表始终 100%。
• 样例书刻意做得很小（4 个短章节），目的是清晰暴露机制，不代表大规模真实书籍的 绝对 token 数值。
• 术语表遵从率、术语一致性都用确定性字符串匹配（consistency.py），不是模型自评， 可能漏判措辞更灵活的译法变体。
• 每次运行的具体数字会因模型输出的随机性小幅波动（上表为最近一次真实运行结果）， 但量级与结论稳定复现。

源代码

agents.py

"""
实验 10-3：书籍翻译 Agent —— 管理者模式（Orchestration）

本模块实现四种专职 Agent，以及两种运行方式：
  1) 管理者模式（orchestrate）：Manager 只保存任务/计划/调用记录/文件索引，
     不保存完整译文；各子 Agent 拥有独立、隔离的上下文。
  2) 单 Agent 模式（single_agent）：一个 Agent 在同一条不断增长的对话里
     依次读全书、逐章翻译，用于对照“上下文膨胀”与“术语漂移”。

核心验证点：
  - 记录每个 Agent / Manager 的上下文 token 消耗；
  - 证明管理者模式下 Manager 的上下文明显小于单 Agent 的累积上下文；
  - 证明共享术语表能让术语在各章保持一致。
"""

import os
import json

import tiktoken
from openai import OpenAI


# ----------------------------------------------------------------------------
# 配置：model / base_url 可通过环境变量覆盖，默认当前便宜旗舰 gpt-5.6-luna
# ----------------------------------------------------------------------------
MODEL = os.environ.get("OPENAI_MODEL", "gpt-5.6-luna")
BASE_URL = os.environ.get("OPENAI_BASE_URL")  # 可选，兼容自建/代理端点


def _report_issues(report: dict) -> list:
    """JSON null issues must behave like omit ([])."""
    issues = report.get("issues")
    return issues if issues is not None else []


def _to_openrouter_model(model: str) -> str:
    """把模型名映射到 OpenRouter 命名空间（用于无 OPENAI_API_KEY 的回退路径）。"""
    if "/" in model:
        return model                      # 已是 OpenRouter 命名空间，原样使用
    if model.startswith("gpt-"):
        return "openai/" + model          # gpt-* -> openai/gpt-*
    if model.startswith("claude-"):
        return "anthropic/claude-opus-4.8"
    return "openai/gpt-5.6-luna"          # 兜底：当前便宜旗舰


def get_client() -> OpenAI:
    """创建 LLM 客户端。

    通用回退策略：
      1) 有 OPENAI_API_KEY -> 直连 OpenAI（尊重可选的 OPENAI_BASE_URL）；
      2) 否则有 OPENROUTER_API_KEY -> 自动改走 OpenRouter 网关，并把 MODEL
         映射到 OpenRouter 命名空间（如 gpt-5.6-luna -> openai/gpt-5.6-luna）；
      3) 都没有则报清晰错误。
    """
    global MODEL
    api_key = os.environ.get("OPENAI_API_KEY")
    if api_key:
        kwargs = {"api_key": api_key}
        if BASE_URL:
            kwargs["base_url"] = BASE_URL
        return OpenAI(**kwargs)
    or_key = os.environ.get("OPENROUTER_API_KEY")
    if or_key:
        MODEL = _to_openrouter_model(MODEL)
        return OpenAI(api_key=or_key, base_url="https://openrouter.ai/api/v1")
    raise RuntimeError(
        "未设置 OPENAI_API_KEY 或 OPENROUTER_API_KEY，请参考 env.example 配置。"
    )


# tiktoken 编码器：用于统计“未真正发给模型”的上下文（如 Manager 状态）token 数
try:
    _ENC = tiktoken.encoding_for_model(MODEL)
except Exception:
    _ENC = tiktoken.get_encoding("o200k_base")


def _slug(name: str) -> str:
    """把章节名转成干净的文件名前缀，如 'Chapter 1: ...' -> 'chapter1'。"""
    import re
    m = re.search(r"chapter\s*0*(\d+)", name, re.IGNORECASE)
    if m:
        return f"chapter{m.group(1)}"
    return re.sub(r"[^0-9a-zA-Z]+", "_", name).strip("_").lower() or "chapter"


def _loads_lenient(content: str):
    """容错解析 JSON：兼容个别模型把 JSON 包在 ```json ... ``` 代码围栏里的情况。"""
    s = (content or "").strip()
    if s.startswith("```"):
        s = s.split("\n", 1)[-1] if "\n" in s else s
        s = s.rsplit("```", 1)[0].strip()
        if s.lower().startswith("json"):
            s = s[4:].strip()
    return json.loads(s)


def count_tokens(text: str) -> int:
    """统计一段文本的 token 数。"""
    return len(_ENC.encode(text or ""))


def count_messages_tokens(messages) -> int:
    """统计一组 chat messages 的 token 数（近似：内容 + 每条消息固定开销）。"""
    total = 0
    for m in messages:
        total += count_tokens(m.get("content", "")) + 4  # 每条消息约 4 token 结构开销
    return total


# ----------------------------------------------------------------------------
# Token 追踪器：记录每一次 LLM 调用的上下文规模，并按 Agent 聚合
# ----------------------------------------------------------------------------
class TokenTracker:
    """
    记录每个 Agent 每次调用的上下文 token 消耗。

    - prompt_tokens：本次调用发送给模型的“上下文”大小（真实 API usage）。
      这是衡量“上下文膨胀”的关键指标。
    - peak：某个 Agent 在其所有调用中，单次上下文的最大值（上下文峰值）。
    """

    def __init__(self):
        self.calls = []  # 每次调用一条记录

    def record(self, agent, prompt_tokens, completion_tokens, note=""):
        self.calls.append(
            {
                "agent": agent,
                "prompt_tokens": prompt_tokens,
                "completion_tokens": completion_tokens,
                "note": note,
            }
        )

    def by_agent(self):
        """按 Agent 聚合：调用次数、输入/输出总量、上下文峰值。"""
        agg = {}
        for c in self.calls:
            a = agg.setdefault(
                c["agent"],
                {"calls": 0, "in": 0, "out": 0, "peak_context": 0},
            )
            a["calls"] += 1
            a["in"] += c["prompt_tokens"]
            a["out"] += c["completion_tokens"]
            a["peak_context"] = max(a["peak_context"], c["prompt_tokens"])
        return agg

    def total_tokens(self):
        return sum(c["prompt_tokens"] + c["completion_tokens"] for c in self.calls)


# ----------------------------------------------------------------------------
# LLM 调用封装：每次调用都带上 agent 名字，便于按 Agent 记账
# ----------------------------------------------------------------------------
def llm_chat(client, tracker, agent, messages, json_mode=False, note=""):
    """
    发起一次 chat completion，并把真实 token usage 记入 tracker。

    注意：messages 是本次调用的“独立上下文”。子 Agent 每次都从零构造 messages，
    因此各 Agent 的上下文天然隔离，互不污染。
    """
    kwargs = {"model": MODEL, "messages": messages, "temperature": 0.2}
    if json_mode:
        kwargs["response_format"] = {"type": "json_object"}

    try:
        resp = client.chat.completions.create(**kwargs)
    except Exception as e:
        # 推理模型（如 gpt-5.x）只接受默认 temperature，会拒绝自定义值；
        # 此时去掉 temperature 重试一次，保持其余参数不变。
        if "temperature" in str(e).lower():
            kwargs.pop("temperature", None)
            resp = client.chat.completions.create(**kwargs)
        else:
            raise
    usage = resp.usage
    tracker.record(agent, usage.prompt_tokens, usage.completion_tokens, note)
    return resp.choices[0].message.content


# ============================================================================
# 四种专职 Agent
# ============================================================================

# 编辑部指定术语（house style）：Manager 会把这些译法强制写入共享术语表，
# 让所有 Translation Agent 全书统一采用。单 Agent 看不到术语表，无法贯彻。
EDITORIAL_MANDATE = {
    "token": "词元",
    "prompt": "提示词",
    "latency": "时延",
    "embedding": "嵌入向量",
}


def translation_guide(target_lang="中文"):
    """按目标语言生成翻译指南。默认中文，保持与旧行为一致。"""
    return (
        f"翻译指南：面向{target_lang}技术读者，语言流畅自然；保留 Markdown 结构；"
        "代码块内的代码原样保留、不翻译（可保留英文注释）；"
        "术语表中出现的术语必须严格使用规定译法；遇到术语表之外的新术语，"
        "先给出你推断的译法，并在其后紧跟标记 [待审] 提示人工复核。"
    )


# 向后兼容：模块级默认（英文→中文）翻译指南，供 Manager 上下文展示等引用。
TRANSLATION_GUIDE = translation_guide("中文")


# Manager 的固定执行计划（供实际运行与 --dry-run 的 Agent 图共用，避免两处漂移）。
ORCHESTRATION_PLAN = [
    "1. 调用 Glossary Agent 生成术语表并落盘",
    "2. 逐章调用 Translation Agent（各自独立上下文，共享术语表文件）",
    "3. 调用 Proofreading Agent 做一致性审校并落盘报告",
    "4. 依据报告决定是否发回个别章节修订",
]


def glossary_agent(client, tracker, book_text, source_lang="英文", target_lang="中文"):
    """
    Glossary Agent：读全书内容，识别反复出现的专业术语，
    输出结构化术语对照表（JSON）。独立上下文，产出后即可释放。
    """
    system = (
        f"你是术语抽取专家。阅读整本{source_lang}技术书，找出反复出现的专业术语，"
        f"为每个术语给出统一的{target_lang}译法。只输出 JSON。"
    )
    user = (
        "请阅读下面全书内容，抽取 6-10 个反复出现的核心专业术语，"
        "输出 JSON，格式为："
        f'{{"glossary": [{{"en": "{source_lang}术语", "zh": "{target_lang}译法", '
        '"pos": "词性", "context": "该术语在书中的语境说明"}]}。\n\n'
        "全书内容如下：\n\n" + book_text
    )
    messages = [
        {"role": "system", "content": system},
        {"role": "user", "content": user},
    ]
    content = llm_chat(
        client, tracker, "Glossary", messages, json_mode=True, note="抽取术语表"
    )
    data = _loads_lenient(content)
    # 模型偶尔输出 JSON 数组等合法但非对象的 JSON；此时无法取 glossary，按空表处理。
    if not isinstance(data, dict):
        return []
    return data.get("glossary", [])


def translation_agent(client, tracker, chapter_text, glossary, chapter_name,
                      feedback=None, source_lang="英文", target_lang="中文"):
    """
    Translation Agent：接收「当前章节 + 术语表 + 翻译指南」，翻成流畅译文。
    每个实例都是独立上下文（只看到自己这一章 + 术语表，不看到别的章节译文）。

    feedback：可选，Manager 依据审校报告发回的针对本章的修订意见。
    """
    glossary_lines = "\n".join(
        f'- {g["en"]} → {g["zh"]}（{g.get("pos","")}）' for g in glossary
    )
    system = f"你是专业技术翻译。把{source_lang}章节翻译为流畅、准确的{target_lang}。"
    user = (
        f"{translation_guide(target_lang)}\n\n"
        f"【术语表（必须严格遵守）】\n{glossary_lines}\n\n"
    )
    if feedback:
        user += f"【本章修订意见（请据此修改）】\n{feedback}\n\n"
    user += (
        f"【待翻译章节：{chapter_name}】\n{chapter_text}\n\n"
        f"请直接输出该章节的{target_lang}译文（Markdown），不要额外解释。"
    )
    messages = [
        {"role": "system", "content": system},
        {"role": "user", "content": user},
    ]
    note = f"翻译 {chapter_name}" + ("（修订）" if feedback else "")
    return llm_chat(client, tracker, "Translation", messages, note=note)


def proofreading_agent(client, tracker, translations, glossary, target_lang="中文"):
    """
    Proofreading Agent：接收所有译文 + 术语表，做一致性检查
    （术语是否统一、前后是否矛盾、是否流畅），输出结构化审校报告（JSON）。

    translations：{chapter_name: 译文文本}
    """
    glossary_lines = "\n".join(f'- {g["en"]} → {g["zh"]}' for g in glossary)
    joined = "\n\n".join(
        f"===== {name} =====\n{text}" for name, text in translations.items()
    )
    system = (
        f"你是资深审校。检查多章{target_lang}译文的术语一致性、前后一致性与流畅性。"
        "只输出 JSON。"
    )
    user = (
        f"【术语表】\n{glossary_lines}\n\n"
        f"【全部译文】\n{joined}\n\n"
        "请输出 JSON："
        '{"issues": [{"chapter": "章节名", "type": "术语不一致/前后矛盾/流畅性", '
        '"detail": "问题描述"}], "chapters_need_revision": ["需要修订的章节名"], '
        '"summary": "总体评价"}'
    )
    messages = [
        {"role": "system", "content": system},
        {"role": "user", "content": user},
    ]
    content = llm_chat(
        client, tracker, "Proofreading", messages, json_mode=True, note="一致性审校"
    )
    return _loads_lenient(content)


def manager_decision(client, tracker, task, file_index, report):
    """
    Manager Agent 的一次真实 LLM 决策调用。

    关键点：Manager 只把「任务 + 文件索引 + 审校报告摘要」这类很小的上下文
    发给模型，用来决定「哪些章节需要发回 Translation Agent 修订」。
    它从不把完整译文放进自己的上下文 —— 这正是控制 Manager 上下文膨胀的做法。
    """
    system = "你是翻译项目的管理者，只做调度决策，输出 JSON。"
    user = (
        f"任务：{task}\n"
        f"文件索引（只存路径，不存正文）：{json.dumps(file_index, ensure_ascii=False)}\n"
        f"审校报告摘要：{json.dumps(report, ensure_ascii=False)}\n\n"
        "根据审校报告，决定需要修订的章节。输出 JSON："
        '{"revise": ["章节名", ...], "reason": "简述"}'
    )
    messages = [
        {"role": "system", "content": system},
        {"role": "user", "content": user},
    ]
    content = llm_chat(
        client, tracker, "Manager", messages, json_mode=True, note="调度决策"
    )
    return _loads_lenient(content)


# ============================================================================
# 运行方式一：管理者模式（Orchestration）
# ============================================================================
def run_orchestration(chapters, out_dir, *, source_lang="英文", target_lang="中文",
                      enable_glossary=True, enable_proofreading=True, trace=None):
    """
    chapters：{chapter_name: 原文} 的有序字典
    out_dir：产物目录（术语表、各章译文、审校报告都写到这里）

    可选参数：
      source_lang / target_lang：源语言 / 目标语言（默认 英文 → 中文，与旧行为一致）。
      enable_glossary：是否启用 Glossary Agent 抽取术语表（关闭后仅保留编辑部指定术语）。
      enable_proofreading：是否启用 Proofreading Agent + Manager 修订闭环。
      trace：可选回调 trace(str)，用于打印四 Agent 协作的实时轨迹。

    返回：metrics 字典，含 tracker、manager 上下文峰值、译文映射等。
    """
    os.makedirs(out_dir, exist_ok=True)
    client = get_client()
    tracker = TokenTracker()
    emit = trace if callable(trace) else (lambda *a, **k: None)

    # ---- Manager 的上下文：只保存这些“轻量”信息，绝不含完整译文 ----
    manager_context = {
        "task": f"把一本{source_lang}技术小书翻译成流畅{target_lang}，保证术语全书一致。",
        "guide": translation_guide(target_lang),
        "plan": list(ORCHESTRATION_PLAN),
        "call_log": [],       # 各 Agent 调用记录（只记摘要，不记正文）
        "file_index": {},     # 文件索引：只存路径
        "progress": {},       # 进度状态
    }
    manager_peak = 0  # Manager 上下文（其状态序列化后的）token 峰值

    def snapshot_manager():
        nonlocal manager_peak
        size = count_tokens(json.dumps(manager_context, ensure_ascii=False))
        manager_peak = max(manager_peak, size)
        return size

    def log_call(agent, note, out_file, prompt_tokens, completion_tokens):
        # Manager 只记录“谁做了什么、产物在哪、花了多少 token”，不记录正文
        manager_context["call_log"].append(
            {
                "agent": agent,
                "note": note,
                "output": out_file,
                "prompt_tokens": prompt_tokens,
                "completion_tokens": completion_tokens,
            }
        )
        snapshot_manager()

    snapshot_manager()
    emit("Manager：制定计划并调度四个专职 Agent（各自独立上下文）")
    for step in manager_context["plan"]:
        emit(f"    计划 {step}")

    # ---- 步骤 1：Glossary Agent（独立上下文，读全书；产出后释放）----
    book_text = "\n\n".join(f"# {n}\n{t}" for n, t in chapters.items())
    if enable_glossary:
        emit(f"Manager → Glossary Agent：读全书（{len(chapters)} 章）抽取共享术语表")
        glossary = glossary_agent(client, tracker, book_text, source_lang, target_lang)
    else:
        emit("Manager：已跳过 Glossary Agent（--no-glossary），仅保留编辑部指定术语")
        glossary = []
    # 归一化：模型偶尔返回不合规条目（如 {"term": ...} 而非 {"en"/"zh": ...}，
    # 或显式 null），直接丢弃，避免后续 g["en"] / g["zh"] 索引让整轮运行崩溃。
    glossary = [
        g for g in glossary
        if isinstance(g, dict)
        and isinstance(g.get("en"), str) and g["en"].strip()
        and isinstance(g.get("zh"), str) and g["zh"].strip()
    ]
    # Manager 把“编辑部指定术语”强制写入术语表（覆盖或新增），作为全书统一契约。
    for g in glossary:
        en = g["en"].strip().lower()
        if en in EDITORIAL_MANDATE:
            g["zh"] = EDITORIAL_MANDATE[en]
    present = {g["en"].strip().lower() for g in glossary}
    for en, zh in EDITORIAL_MANDATE.items():
        if en not in present:
            glossary.append({"en": en, "zh": zh, "pos": "名词", "context": "编辑部指定术语"})
    glossary_path = os.path.join(out_dir, "glossary.json")
    with open(glossary_path, "w", encoding="utf-8") as f:
        json.dump(glossary, f, ensure_ascii=False, indent=2)
    # Manager 只在文件索引里记路径；术语表正文留在文件系统，不进 Manager 上下文
    manager_context["file_index"]["glossary"] = glossary_path
    # 仅在真正调用了 Glossary Agent 时才有 LLM usage 可记账；--no-glossary 时无调用。
    g_prompt, g_completion = (
        (tracker.calls[-1]["prompt_tokens"], tracker.calls[-1]["completion_tokens"])
        if enable_glossary and tracker.calls else (0, 0)
    )
    log_call("Glossary", f"抽取 {len(glossary)} 个术语", glossary_path,
             g_prompt, g_completion)
    if enable_glossary:
        emit(f"Glossary Agent ✓：确定 {len(glossary)} 个术语 → {os.path.basename(glossary_path)}"
             f"（Manager 只记路径，术语表正文留在文件系统）")
    else:
        emit(f"Manager：写入 {len(glossary)} 个编辑部指定术语 → {os.path.basename(glossary_path)}")

    # ---- 步骤 2：逐章 Translation Agent（每章一个独立上下文实例）----
    translations = {}
    for name, text in chapters.items():
        emit(f"Manager → Translation Agent：翻译《{name}》（独立上下文，仅见本章 + 术语表）")
        zh = translation_agent(client, tracker, text, glossary, name,
                               source_lang=source_lang, target_lang=target_lang)
        # 文件名如 chapter1_zh.md
        base = _slug(name)
        out_file = os.path.join(out_dir, f"{base}_zh.md")
        with open(out_file, "w", encoding="utf-8") as f:
            f.write(zh)
        translations[name] = zh
        manager_context["file_index"][name] = out_file
        manager_context["progress"][name] = "translated"
        last = tracker.calls[-1]
        log_call("Translation", f"翻译 {name}", out_file,
                 last["prompt_tokens"], last["completion_tokens"])
        emit(f"Translation Agent ✓：{os.path.basename(out_file)}"
             f"（上下文 {last['prompt_tokens']} tok，译文落盘不回传 Manager）")

    # ---- 步骤 3：Proofreading Agent（读所有译文 + 术语表，独立上下文）----
    if not enable_proofreading:
        emit("Manager：已跳过 Proofreading Agent 与修订闭环（--no-proofreading）")
        report = {"issues": [], "chapters_need_revision": [],
                  "summary": "（已跳过审校）"}
        snapshot_manager()
        return {
            "mode": "orchestration",
            "tracker": tracker,
            "manager_context_peak": manager_peak,
            "manager_context_final": manager_context,
            "glossary": glossary,
            "translations": translations,
            "report": report,
            "out_dir": out_dir,
        }

    emit("Manager → Proofreading Agent：读全部译文 + 术语表做一致性/流畅性审校")
    report = proofreading_agent(client, tracker, translations, glossary, target_lang)
    report_path = os.path.join(out_dir, "proofreading_report.json")
    with open(report_path, "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    manager_context["file_index"]["report"] = report_path
    last = tracker.calls[-1]
    log_call("Proofreading", "一致性审校", report_path,
             last["prompt_tokens"], last["completion_tokens"])
    emit(f"Proofreading Agent ✓：{len(_report_issues(report))} 处问题 → "
         f"{os.path.basename(report_path)}")

    # ---- 步骤 4：Manager 决策 + 至多一轮修订 ----
    # Manager 只把“文件索引 + 报告摘要”这类小上下文发给模型做决策
    report_summary = {
        "chapters_need_revision": report.get("chapters_need_revision", []) or [],
        "issues": _report_issues(report)[:5],
        "summary": report.get("summary", ""),
    }
    manager_context["progress"]["proofread"] = "done"
    snapshot_manager()

    emit("Manager：读审校报告摘要（不读正文）→ 决策哪些章节需发回修订")
    decision = manager_decision(
        client, tracker, manager_context["task"],
        manager_context["file_index"], report_summary
    )
    revise = decision.get("revise", [])
    emit(f"Manager 决策 ✓：需修订章节 {revise or '无'}")

    for name in revise:
        if name not in chapters:
            continue
        # 找到该章节的修订意见
        fb = "; ".join(
            i.get("detail", "") for i in _report_issues(report)
            if i.get("chapter") == name
        ) or "请根据术语表统一术语并提升流畅性。"
        emit(f"Manager → Translation Agent：修订《{name}》（附审校意见）")
        zh = translation_agent(client, tracker, chapters[name], glossary, name,
                               feedback=fb, source_lang=source_lang, target_lang=target_lang)
        base = _slug(name)
        out_file = os.path.join(out_dir, f"{base}_zh.md")
        with open(out_file, "w", encoding="utf-8") as f:
            f.write(zh)
        translations[name] = zh
        manager_context["progress"][name] = "revised"
        last = tracker.calls[-1]
        log_call("Translation", f"修订 {name}", out_file,
                 last["prompt_tokens"], last["completion_tokens"])

    snapshot_manager()
    emit(f"Manager：全部完成，产物目录 {out_dir}")

    return {
        "mode": "orchestration",
        "tracker": tracker,
        "manager_context_peak": manager_peak,
        "manager_context_final": manager_context,
        "glossary": glossary,
        "translations": translations,
        "report": report,
        "out_dir": out_dir,
    }


# ============================================================================
# 运行方式二：单 Agent 模式（对照组）
# ============================================================================
def run_single_agent(chapters, out_dir, *, source_lang="英文", target_lang="中文"):
    """
    朴素基线：一个 Agent 在同一条不断增长的对话里，先粗读全书，
    再逐章翻译。没有独立的术语表工具来“钉死”术语，且上下文随章节累积。

    这一模式用于暴露两个问题：
      - 上下文膨胀：单条对话的上下文峰值 = 累积到最后一章时的全部内容；
      - 术语漂移：缺少共享术语表约束，同一术语在不同章可能译法不一致。
    """
    os.makedirs(out_dir, exist_ok=True)
    client = get_client()
    tracker = TokenTracker()

    system = (
        f"你是专业技术翻译。我会逐章给你一本{source_lang}技术书，请把每一章翻译成"
        f"流畅、准确的{target_lang}。保留 Markdown 结构；代码块内的代码原样保留、不翻译。"
    )
    # 单 Agent 的“主上下文”：一条持续增长的对话
    messages = [{"role": "system", "content": system}]

    translations = {}
    for name, text in chapters.items():
        messages.append(
            {
                "role": "user",
                "content": f"请翻译下面这一章，直接输出中文译文：\n\n# {name}\n{text}",
            }
        )
        content = llm_chat(
            client, tracker, "SingleAgent", messages, note=f"翻译 {name}"
        )
        # 译文继续留在对话里 —— 这正是上下文膨胀的来源
        messages.append({"role": "assistant", "content": content})
        translations[name] = content
        base = _slug(name)
        out_file = os.path.join(out_dir, f"{base}_zh.md")
        with open(out_file, "w", encoding="utf-8") as f:
            f.write(content)

    return {
        "mode": "single_agent",
        "tracker": tracker,
        # 单 Agent 的“主上下文峰值”= 其所有调用中最大的一次 prompt_tokens
        "main_context_peak": tracker.by_agent()["SingleAgent"]["peak_context"],
        "translations": translations,
        "out_dir": out_dir,
    }

consistency.py

"""
术语一致性检查工具。

思路：对每个受关注的英文术语，预先列出它在中文里“几种常见但不同”的译法。
扫描全书各章译文，统计每个术语实际出现了几种不同译法：
  - 只出现 1 种  → 全书一致；
  - 出现 >= 2 种 → 术语漂移（不一致）。

这不是给模型评分，而是用确定性的字符串匹配，客观度量“同一术语是否全书统一”。
"""

# 每个术语：canonical 为推荐/术语表规定译法；variants 为若干“互不相同”的常见译法。
# 注意：variants 之间尽量不互为子串，避免重复计数（如“嵌入向量”归入“嵌入”一族）。
TRACKED_TERMS = [
    {"en": "token",       "canonical": "词元",  "variants": ["词元", "令牌", "标记", "token"]},
    {"en": "embedding",   "canonical": "嵌入",  "variants": ["嵌入", "词向量", "向量表示"]},
    {"en": "prompt",      "canonical": "提示词", "variants": ["提示词", "提示语", "提示"]},
    {"en": "inference",   "canonical": "推理",  "variants": ["推理", "推断"]},
    {"en": "latency",     "canonical": "时延",  "variants": ["延迟", "时延", "延时"]},
    {"en": "attention",   "canonical": "注意力", "variants": ["注意力", "关注度"]},
    {"en": "transformer", "canonical": "Transformer", "variants": ["Transformer", "变换器", "转换器"]},
    {"en": "throughput",  "canonical": "吞吐量", "variants": ["吞吐量", "吞吐率", "通量"]},
    {"en": "fine-tuning", "canonical": "微调",  "variants": ["微调", "精调"]},
]


import re


def _strip_code(text):
    """去掉围栏代码块与行内代码：代码按翻译指南原样保留英文，不应计入术语一致性统计。"""
    text = re.sub(r"```.*?```", " ", text, flags=re.DOTALL)
    text = re.sub(r"`[^`]*`", " ", text)
    return text


# 编辑部“指定术语”（house style）：为几个术语规定一个明确的、区别于模型默认译法的译名。
# 这些译法都是合法且更精确的选择，用来考察“共享术语表能否把指定译法贯彻到全书”。
#   mandated：术语表规定的译法；default：模型自由翻译时常用的默认译法。
MANDATED_TERMS = [
    {"en": "token",     "mandated": "词元",   "default": "标记"},
    {"en": "prompt",    "mandated": "提示词", "default": "提示"},
    {"en": "latency",   "mandated": "时延",   "default": "延迟"},
    {"en": "embedding", "mandated": "嵌入向量", "default": "嵌入"},
]


def check_adherence(translations):
    """
    术语表遵从率：对每个“指定术语”，统计在出现该概念的章节里，
    有多少章使用了术语表规定的译法（而非默认译法）。

    这是管理者模式的核心价值：共享术语表能把指定译法贯彻到每一章；
    单 Agent 看不到术语表，只能用自己的默认译法。
    """
    rows = []
    hit_total = 0
    concept_total = 0
    for t in MANDATED_TERMS:
        m, d = t["mandated"], t["default"]
        chapters_with_concept = 0
        chapters_adhered = 0
        for name, raw in translations.items():
            text = _strip_code(raw)
            has_m = m in text
            # default 若是 mandated 的子串（如“嵌入”是“嵌入向量”子串），需去掉 mandated 再判断
            has_d = (d in text.replace(m, "")) if d in m else (d in text)
            if has_m or has_d:
                chapters_with_concept += 1
                if has_m:
                    chapters_adhered += 1
        if chapters_with_concept:
            concept_total += chapters_with_concept
            hit_total += chapters_adhered
            rows.append({
                "en": t["en"], "mandated": m, "default": d,
                "adhered": chapters_adhered, "total": chapters_with_concept,
            })
    rate = hit_total / concept_total if concept_total else 1.0
    return {"rows": rows, "rate": rate}


def _variant_in_chapter(text, variant, other_variants):
    """
    判断某个 variant 是否在 text 中“独立”出现。
    对“提示”这种会成为“提示词/提示语”子串的情况：仅当去掉更长 variant 后仍出现才算。
    """
    longer = [v for v in other_variants if variant in v and v != variant]
    if not longer:
        return variant in text
    tmp = text
    for v in longer:
        tmp = tmp.replace(v, "")
    return variant in tmp


def analyze(translations):
    """
    translations：{chapter_name: 译文文本}
    返回：
      results：每个术语的分析（用到哪些译法、是否一致、各章用法）
      consistent_terms / total_terms / rate
    """
    results = []
    consistent = 0
    total = 0
    for term in TRACKED_TERMS:
        variants = term["variants"]
        used = {}  # variant -> [出现该译法的章节]
        for name, raw in translations.items():
            text = _strip_code(raw)
            for v in variants:
                others = [x for x in variants if x != v]
                if _variant_in_chapter(text, v, others):
                    used.setdefault(v, []).append(name)
        if not used:
            # 全书都没出现该术语，跳过统计
            continue
        total += 1
        distinct = list(used.keys())
        is_consistent = len(distinct) == 1
        if is_consistent:
            consistent += 1
        results.append(
            {
                "en": term["en"],
                "canonical": term["canonical"],
                "distinct_used": distinct,
                "consistent": is_consistent,
                "by_variant": used,
            }
        )
    rate = consistent / total if total else 1.0
    return {
        "results": results,
        "consistent_terms": consistent,
        "total_terms": total,
        "rate": rate,
    }

demo.py

"""
实验 10-3 一键演示。

  python demo.py                       # 完整跑：管理者模式 + 单 Agent 对照
  python demo.py --help                # 查看全部参数
  python demo.py --dry-run             # 离线：只画四 Agent 协作图 + token 预算，不调 API
  python demo.py --model gpt-5.6-luna        # 换用更强的模型
  python demo.py --skip-single         # 只跑管理者模式，跳过单 Agent 对照（更快）
  python demo.py --no-proofreading     # 关闭审校 Agent 与修订闭环
  python demo.py --source-lang 英文 --target-lang 日文     # 换翻译方向
  python demo.py --sample-dir path/to/book --out-dir out  # 换输入书 / 产物目录

流程：
  1) 读入 --sample-dir 下的若干英文短章节（默认 sample_book/）；
  2) 运行【管理者模式】：Glossary / Translation / Proofreading / Manager 四种 Agent 协作，
     并打印四 Agent 协作的实时轨迹；
  3) 运行【单 Agent 模式】作为对照（除非指定 --skip-single）；
  4) 打印对比表：每个 Agent 的上下文 token 消耗、Manager/主上下文峰值、术语一致性。

结论要点：
  - 管理者模式下 Manager 的上下文明显小于单 Agent 的累积上下文（控制上下文膨胀）；
  - 共享术语表让术语在各章保持一致。
"""

import argparse
import glob
import os
import sys

from dotenv import load_dotenv

load_dotenv()

HERE = os.path.dirname(os.path.abspath(__file__))
SAMPLE_DIR = os.path.join(HERE, "sample_book")
OUT_DIR = os.path.join(HERE, "output")


def parse_args():
    """命令行参数：不带任何参数运行时行为与原版完全一致。"""
    parser = argparse.ArgumentParser(
        prog="demo.py",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=(
            "实验 10-3：书籍翻译 Agent —— 管理者模式（Glossary/Translation/\n"
            "Proofreading/Manager 四种 Agent 协作）vs 单 Agent 模式，\n"
            "对比上下文膨胀与术语表遵从率。"
        ),
        epilog=(
            "示例：\n"
            "  python demo.py --dry-run                 # 离线画 Agent 图 + token 预算，不调 API\n"
            "  python demo.py --skip-single             # 只跑管理者模式\n"
            "  python demo.py --no-proofreading         # 关闭审校 Agent 与修订闭环\n"
            "  python demo.py --sample-dir book --out-dir out --model gpt-5.6-luna\n"
        ),
    )
    io = parser.add_argument_group("输入 / 输出")
    io.add_argument(
        "--sample-dir",
        default=SAMPLE_DIR,
        metavar="DIR",
        help="待翻译书籍目录（读取其中的 *.md 章节，按文件名排序）。默认 sample_book/。",
    )
    io.add_argument(
        "--out-dir",
        default=OUT_DIR,
        metavar="DIR",
        help="产物根目录（术语表 / 各章译文 / 审校报告写入其下的 orchestration|single_agent/）。"
             "默认 output/。",
    )

    lang = parser.add_argument_group("翻译方向")
    lang.add_argument(
        "--source-lang", default="英文", metavar="LANG",
        help="源语言，仅用于提示词措辞。默认 英文。",
    )
    lang.add_argument(
        "--target-lang", default="中文", metavar="LANG",
        help="目标语言，仅用于提示词措辞。默认 中文。"
             "注意：内置的术语一致性 / 遵从率统计针对 英文→中文 调校，改方向仍可翻译，"
             "但该统计表意义有限。",
    )

    agents_grp = parser.add_argument_group("启用哪些 Agent")
    agents_grp.add_argument(
        "--no-glossary", action="store_true",
        help="关闭 Glossary Agent（不做术语抽取，仅保留编辑部指定术语）。默认启用。",
    )
    agents_grp.add_argument(
        "--no-proofreading", action="store_true",
        help="关闭 Proofreading Agent 及 Manager 修订闭环。默认启用。",
    )

    run = parser.add_argument_group("运行方式")
    run.add_argument(
        "--model", default=None, metavar="MODEL",
        help="覆盖使用的模型（等价于设置 OPENAI_MODEL 环境变量）。"
             "默认沿用 OPENAI_MODEL 环境变量，缺省为 gpt-5.6-luna。",
    )
    run.add_argument(
        "--skip-single", action="store_true",
        help="只运行管理者模式，跳过单 Agent 对照组（更快，但不产出核心对比表）。默认关闭。",
    )
    run.add_argument(
        "--dry-run", action="store_true",
        help="离线预演：只打印四 Agent 协作图、Manager 计划、编辑部术语与各 Agent 的 token 预算，"
             "不调用任何 API（无需 OPENAI_API_KEY）。",
    )
    return parser.parse_args()


def load_chapters(sample_dir):
    """按文件名顺序读入 sample_dir/*.md，返回 {章节名: 原文}。"""
    files = sorted(glob.glob(os.path.join(sample_dir, "*.md")))
    chapters = {}
    for path in files:
        with open(path, "r", encoding="utf-8") as f:
            text = f.read()
        # 用文件的一级标题作为章节名，回退到文件名
        name = os.path.splitext(os.path.basename(path))[0]
        for line in text.splitlines():
            if line.startswith("# "):
                name = line[2:].strip()
                break
        chapters[name] = text
    return chapters


def hr(title=""):
    print("\n" + "=" * 72)
    if title:
        print(title)
        print("=" * 72)


def print_agent_table(tracker, title):
    hr(title)
    agg = tracker.by_agent()
    print(f"{'Agent':<14}{'调用次数':>8}{'输入tok':>12}{'输出tok':>12}{'上下文峰值':>12}")
    print("-" * 72)
    for name, a in agg.items():
        print(f"{name:<14}{a['calls']:>8}{a['in']:>12}{a['out']:>12}{a['peak_context']:>12}")
    print("-" * 72)
    print(f"{'合计':<14}{'':>8}{'':>12}{'':>12}  总 token：{tracker.total_tokens()}")


def print_consistency(analysis, label):
    print(f"\n[{label}] 术语一致性：{analysis['consistent_terms']}/{analysis['total_terms']} "
          f"个术语全书统一（{analysis['rate']*100:.0f}%）")
    for r in analysis["results"]:
        flag = "一致" if r["consistent"] else "不一致 <==="
        used = " / ".join(f"{v}({len(chs)}章)" for v, chs in r["by_variant"].items())
        print(f"  - {r['en']:<12} 实际用到：{used}  [{flag}]")


def make_tracer():
    """返回一个把子 Agent 事件缩进打印的 trace(str) 回调，展现 Manager 的实时调度轨迹。"""
    def tracer(msg):
        indent = "" if msg.startswith(("Manager", "Glossary", "Translation",
                                       "Proofreading")) else "  "
        # 已经带前导空格的“计划/子步骤”行原样输出
        print(f"  {indent}{msg}" if not msg.startswith("    ") else f"  {msg}")
    return tracer


def run_dry_run(args):
    """
    离线预演（不调用任何 API）：画出四 Agent 协作图、Manager 计划、编辑部指定术语，
    并用 tiktoken 估算各 Agent 将读到的上下文规模，直观印证“Manager 上下文与书长度基本无关”。
    """
    import agents
    import consistency

    chapters = load_chapters(args.sample_dir)
    if not chapters:
        print(f"错误：{args.sample_dir} 下没有找到任何 .md 章节。", file=sys.stderr)
        sys.exit(1)

    hr(f"实验 10-3 · 离线预演（--dry-run，不调用 API，模型={agents.MODEL}）")
    print(f"待翻译书籍：{args.sample_dir}（{len(chapters)} 章）  翻译方向："
          f"{args.source_lang} → {args.target_lang}")
    print(f"启用 Agent：Manager + " +
          ("Glossary + " if not args.no_glossary else "(Glossary 关闭) ") +
          "Translation" +
          (" + Proofreading" if not args.no_proofreading else " (Proofreading 关闭)"))

    hr("四 Agent 协作图（数据经文件系统流转，Manager 只持有路径）")
    print("""
        ┌─────────────────────── Manager Agent ───────────────────────┐
        │  只存：任务 / 计划 / 调用记录 / 文件索引（绝不存完整译文）      │
        └──┬───────────────┬────────────────────┬────────────────┬─────┘
           │ ①调度          │ ②逐章调度           │ ③调度           │ ④按报告决策
           ▼               ▼                    ▼                ▼
     Glossary Agent   Translation Agent×N   Proofreading Agent  (发回修订)
     读全书→术语表     只读本章+术语表→译文    读全部译文+术语表     命中章节重译
           │               │                    │
           ▼ glossary.json  ▼ chapterN_zh.md      ▼ proofreading_report.json
        ══════════════════ 共享文件系统（out-dir）══════════════════""")

    hr("Manager 执行计划（4 步）")
    for step in agents.ORCHESTRATION_PLAN:
        print(f"  {step}")

    hr("编辑部指定术语（house style，强制写入共享术语表，全书统一）")
    for en, zh in agents.EDITORIAL_MANDATE.items():
        print(f"  {en:<12} → {zh}")

    hr("token 预算预估（tiktoken 离线统计，非真实 API usage）")
    book_text = "\n\n".join(f"# {n}\n{t}" for n, t in chapters.items())
    book_tok = agents.count_tokens(book_text)
    print(f"  Glossary Agent   读全书           ≈ {book_tok} tok")
    per_chapter = []
    for name, text in chapters.items():
        t = agents.count_tokens(text)
        per_chapter.append(t)
        print(f"  Translation Agent 读《{name}》(独立) ≈ {t} tok")
    print(f"  Proofreading Agent 读全部译文       ≈ {sum(per_chapter)} tok（量级同全书）")

    # Manager 上下文预估：任务 + 计划 + 每章一条调用记录 + 文件索引（只有路径）
    import json as _json
    mock_manager = {
        "task": f"把一本{args.source_lang}技术小书翻译成流畅{args.target_lang}，保证术语全书一致。",
        "plan": list(agents.ORCHESTRATION_PLAN),
        "call_log": [{"agent": "Translation", "note": f"翻译 {n}",
                      "output": f"{n}_zh.md", "prompt_tokens": 0, "completion_tokens": 0}
                     for n in chapters],
        "file_index": {n: os.path.join(args.out_dir, "orchestration", f"{n}_zh.md")
                       for n in chapters},
    }
    mgr_tok = agents.count_tokens(_json.dumps(mock_manager, ensure_ascii=False))
    print(f"\n  Manager 上下文（任务/计划/调用记录/文件索引，无正文）≈ {mgr_tok} tok")
    print(f"  对照：单 Agent 累积上下文 ≥ 全书 {book_tok} tok（逐章线性增长，书越长越大）")
    print("\n  关键点：Manager 上下文只随‘章节数’加几行记录，与每章正文长度无关；")
    print("         单 Agent 把全部原文与译文都留在一条对话里，上下文随书长线性膨胀。")

    hr("术语一致性 / 遵从率将统计的术语（见 consistency.py）")
    print("  受追踪术语：" + "、".join(t["en"] for t in consistency.TRACKED_TERMS))
    print("  指定术语（遵从率）：" +
          "、".join(f'{t["en"]}→{t["mandated"]}' for t in consistency.MANDATED_TERMS))
    print("\n离线预演结束。去掉 --dry-run 并设置 OPENAI_API_KEY 即可真正运行四 Agent 协作。")


def main():
    args = parse_args()
    if args.model:
        # 必须在 import agents 之前设置：agents.py 在模块加载时读取
        # OPENAI_MODEL 环境变量来决定使用的模型。
        os.environ["OPENAI_MODEL"] = args.model

    if args.dry_run:
        # 离线路径：不需要 API Key，也不发起任何网络调用。
        run_dry_run(args)
        return

    # 延迟导入，确保上面对 OPENAI_MODEL 的覆盖能在 agents.py 读取环境变量之前生效。
    import agents
    import consistency

    if not os.environ.get("OPENAI_API_KEY") and not os.environ.get("OPENROUTER_API_KEY"):
        print("错误：未设置 OPENAI_API_KEY 或 OPENROUTER_API_KEY。请先 `export OPENAI_API_KEY=...`"
              "（或 OPENROUTER_API_KEY）或复制 env.example 为 .env 并填写（见 env.example）。\n"
              "提示：想在不联网、无 Key 的情况下查看四 Agent 协作结构，可运行 "
              "`python demo.py --dry-run`。", file=sys.stderr)
        sys.exit(1)

    chapters = load_chapters(args.sample_dir)
    if not chapters:
        print(f"错误：{args.sample_dir} 下没有找到任何 .md 章节。", file=sys.stderr)
        sys.exit(1)
    print(f"载入 {len(chapters)} 个章节：{list(chapters.keys())}  "
          f"（{args.source_lang} → {args.target_lang}）")

    # ---------------- 管理者模式 ----------------
    hr("【管理者模式】四 Agent 协作实时轨迹")
    orch = agents.run_orchestration(
        chapters, os.path.join(args.out_dir, "orchestration"),
        source_lang=args.source_lang, target_lang=args.target_lang,
        enable_glossary=not args.no_glossary,
        enable_proofreading=not args.no_proofreading,
        trace=make_tracer(),
    )
    print_agent_table(orch["tracker"], "【管理者模式】各 Agent 上下文 token 消耗")
    print(f"\nManager 上下文峰值（只存任务/计划/调用记录/文件索引）：{orch['manager_context_peak']} tokens")
    print(f"术语表（共享文件，各 Translation Agent 引用同一份）：")
    for g in orch["glossary"]:
        print(f"    {g['en']} → {g['zh']}（{g.get('pos','')}）")
    if not args.no_proofreading:
        print(f"审校报告 summary：{orch['report'].get('summary','')[:120]}")

    # ---------------- 单 Agent 模式 ----------------
    if args.skip_single:
        hr("已跳过单 Agent 对照组（--skip-single）")
        print("提示：核心对比表需要单 Agent 数据，去掉 --skip-single 可看到完整对比。")
        print(f"\n产物目录：{args.out_dir}")
        return
    single = agents.run_single_agent(
        chapters, os.path.join(args.out_dir, "single_agent"),
        source_lang=args.source_lang, target_lang=args.target_lang,
    )
    print_agent_table(single["tracker"], "【单 Agent 模式】主上下文 token 消耗")

    # ---------------- 术语一致性对比 ----------------
    hr("术语一致性对比（确定性字符串匹配，非模型打分）")
    orch_cons = consistency.analyze(orch["translations"])
    single_cons = consistency.analyze(single["translations"])
    print_consistency(orch_cons, "管理者模式")
    print_consistency(single_cons, "单 Agent 模式")

    # ---------------- 术语表遵从率对比（核心证据）----------------
    hr("术语表遵从率对比：编辑部指定术语能否贯彻全书")
    orch_adh = consistency.check_adherence(orch["translations"])
    single_adh = consistency.check_adherence(single["translations"])
    print("（管理者模式把指定术语写入共享术语表并强制下发；单 Agent 看不到术语表）\n")
    print(f"{'指定术语':<14}{'规定译法':<10}{'默认译法':<10}"
          f"{'管理者(遵从/出现)':>18}{'单Agent(遵从/出现)':>20}")
    print("-" * 78)
    o_map = {r["en"]: r for r in orch_adh["rows"]}
    s_map = {r["en"]: r for r in single_adh["rows"]}
    for r in orch_adh["rows"]:
        s = s_map.get(r["en"], {"adhered": 0, "total": 0})
        o_cell = f"{r['adhered']}/{r['total']}"
        s_cell = f"{s['adhered']}/{s['total']}"
        print(f"{r['en']:<14}{r['mandated']:<10}{r['default']:<10}"
              f"{o_cell:>18}{s_cell:>20}")
    print("-" * 78)
    print(f"术语表遵从率：管理者模式 {orch_adh['rate']*100:.0f}%  vs  "
          f"单 Agent {single_adh['rate']*100:.0f}%")

    # ---------------- 核心对比表 ----------------
    hr("核心对比表：管理者模式 vs 单 Agent 模式")
    o_tr, s_tr = orch["tracker"], single["tracker"]
    o_mgr_peak = orch["manager_context_peak"]
    # 管理者模式里，若把 Manager 当作 LLM Agent，它也有一次决策调用的上下文峰值
    o_mgr_llm_peak = o_tr.by_agent().get("Manager", {}).get("peak_context", 0)
    s_main_peak = single["main_context_peak"]

    rows = [
        ("主/Manager 上下文峰值(tokens)", o_mgr_peak, s_main_peak),
        ("Manager LLM 决策调用上下文(tokens)", o_mgr_llm_peak, "—"),
        ("全流程总 token 消耗", o_tr.total_tokens(), s_tr.total_tokens()),
        ("术语内部一致率", f"{orch_cons['rate']*100:.0f}%", f"{single_cons['rate']*100:.0f}%"),
        ("指定术语遵从率", f"{orch_adh['rate']*100:.0f}%", f"{single_adh['rate']*100:.0f}%"),
        ("参与 Agent 种类数", len(o_tr.by_agent()), 1),
    ]
    print(f"{'指标':<32}{'管理者模式':>16}{'单 Agent':>16}")
    print("-" * 72)
    for label, a, b in rows:
        print(f"{label:<32}{str(a):>16}{str(b):>16}")
    print("-" * 72)

    if isinstance(s_main_peak, int) and o_mgr_peak and s_main_peak:
        ratio = s_main_peak / o_mgr_peak
        print(f"\n结论：单 Agent 主上下文峰值是管理者模式 Manager 上下文的 "
              f"{ratio:.1f} 倍。")
        print("Manager 只保存任务/计划/调用记录/文件索引，完整译文全部落盘到文件系统，")
        print("因此无论书有多长，Manager 上下文都基本恒定 —— 这就是控制上下文膨胀的关键。")
    print(f"\n产物目录：{args.out_dir}")


if __name__ == "__main__":
    main()

test_glossary_robustness.py

"""回归测试：Glossary Agent 返回不合规 JSON 时，run_orchestration 不应崩溃。

覆盖两类模型失误（此前会让整轮管理者模式直接 KeyError/AttributeError）：
  1) glossary 条目缺 en/zh 键、或值为显式 null / 空串 -> 条目被丢弃；
  2) 顶层 JSON 是数组而非对象 -> glossary_agent 返回空表。
不依赖真实 API：llm_chat / get_client 被打桩。
"""

import json
import sys
from pathlib import Path
from types import ModuleType

sys.path.insert(0, str(Path(__file__).parent))

# 重依赖打桩：没装 openai / tiktoken 的环境里也能跑（装了则用真包）。
try:
    import openai  # noqa: F401
except ImportError:
    sys.modules["openai"] = ModuleType("openai")
    sys.modules["openai"].OpenAI = object
try:
    import tiktoken  # noqa: F401
except ImportError:
    _tk = ModuleType("tiktoken")
    _enc = type("Enc", (), {"encode": lambda self, t: list(t or "")})
    _tk.encoding_for_model = lambda model: _enc()
    _tk.get_encoding = lambda name: _enc()
    sys.modules["tiktoken"] = _tk

import agents

# 混合各种坏条目的 glossary：错键名 / null / 空串 都应被丢弃，只有合规条目保留。
GLOSSARY_JSON = json.dumps({
    "glossary": [
        {"term": "token", "translation": "词元"},              # 错键名
        {"en": None, "zh": "提示词"},                          # 显式 null
        {"en": "", "zh": "时延"},                              # 空串
        {"en": "attention", "zh": "注意力", "pos": "名词"},     # 合规
    ]
}, ensure_ascii=False)

CHAPTERS = {"Chapter 1: Intro": "# Chapter 1\nSome text about attention."}


def _install_fake_llm(glossary_payload=GLOSSARY_JSON):
    def fake_llm_chat(client, tracker, agent, messages, json_mode=False, note=""):
        tracker.record(agent, 10, 5, note)
        if agent == "Glossary":
            return glossary_payload
        return "译文"
    agents.get_client = lambda: object()
    agents.llm_chat = fake_llm_chat


def test_orchestration_skips_malformed_glossary_entries(tmp_path):
    _install_fake_llm()
    result = agents.run_orchestration(
        CHAPTERS, str(tmp_path), enable_glossary=True, enable_proofreading=False)
    glossary = result["glossary"]
    # 所有存活条目必须是非空 en/zh 字符串（下游 g["en"]/g["zh"] 索引的前提）
    for g in glossary:
        assert isinstance(g["en"], str) and g["en"].strip()
        assert isinstance(g["zh"], str) and g["zh"].strip()
    ens = {g["en"] for g in glossary}
    assert "attention" in ens                       # 合规条目保留
    assert "term" not in ens                        # 错键名条目已丢弃
    for en in agents.EDITORIAL_MANDATE:             # 编辑部指定术语仍会补齐
        assert en in ens
    assert (tmp_path / "glossary.json").exists()    # 产物正常落盘
    assert (tmp_path / "chapter1_zh.md").read_text(encoding="utf-8") == "译文"


def test_glossary_agent_tolerates_json_array():
    _install_fake_llm(glossary_payload='["not", "an", "object"]')
    assert agents.glossary_agent(None, agents.TokenTracker(), "book text") == []


def test_glossary_agent_tolerates_missing_glossary_key():
    _install_fake_llm(glossary_payload='{"terms": []}')
    assert agents.glossary_agent(None, agents.TokenTracker(), "book text") == []

test_null_issues.py

"""Null proofread issues must not TypeError when building report summaries."""
from agents import _report_issues


def test_null_issues_like_empty():
    assert _report_issues({"issues": None}) == []
    summary_issues = _report_issues({"issues": None})[:5]
    assert summary_issues == []
    details = [i.get("detail", "") for i in _report_issues({"issues": None})]
    assert details == []


def test_issues_preserved():
    issues = [{"chapter": "a", "detail": "fix me"}]
    assert _report_issues({"issues": issues}) == issues