liqiang
b119135836
Build latest book artifacts / build (push) Canceled after 0s
dependency resolution / resolve (3.11) (push) Canceled after 0s
dependency resolution / resolve (3.13) (push) Canceled after 0s
deploy-pages / build (push) Canceled after 0s
deploy-pages / deploy (push) Canceled after 0s
i18n consistency check / check (push) Canceled after 0s
provider adoption tests / test (chapter2/context-compression) (push) Canceled after 0s
provider adoption tests / test (chapter2/prompt-injection) (push) Canceled after 0s
provider adoption tests / test (chapter2/system-hint) (push) Canceled after 0s
provider adoption tests / test (chapter3/log-sanitization) (push) Canceled after 0s
web-search-agent tests / test (push) Canceled after 0s
web-search-agent tests / agentbook (push) Canceled after 0s
362 lines
17 KiB
Python
362 lines
17 KiB
Python
"""
|
||
实验 10-2 一键演示。
|
||
|
||
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-2：书籍翻译 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-2 · 离线预演（--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()
|