同时从多个网站搜集信息的 Agent（★★）

源文件：chapter10/parallel-web-research/README.md

实验 10-6 · 同时从多个网站搜集信息的 Agent（★★）

《深入理解 AI Agent》配套实验。演示多个同构 Agent 的并行搜索 + 中心协调：
主协调器同时启动 N 个子 Agent，每个子 Agent 访问一个"网站/来源"找答案；
一旦某个子 Agent 命中目标，其余立即优雅停止。

书中的原型是"10 个并行的 Computer Use Agent 同时访问不同网站找信息"。为便于
自动验证，本实验不启动真实浏览器，而是用一批可控的模拟信息源代替；把重点
完整放在协调机制上——消息总线、并行派发、实时监控、级联终止、竞态处理均为真实实现。

目录结构

架构与机制

                         ┌────────────────────────────┐
                         │   Coordinator（主协调器）    │
                         │  · 并行派发 task_assigned    │
                         │  · 维护任务状态表(状态机)     │
                         │  · 首个命中→加锁结算(幂等)    │
                         │  · 广播一轮 terminate        │
                         └───────────┬────────────────┘
                                     │
                    ┌────────────────┴─────── MessageBus（异步消息总线）───────┐
                    │  Envelope{ sender_id, target, type, payload, seq, ts }    │
                    │  type: task_assigned/status_update/result/terminate/ack   │
                    └──┬─────────┬─────────┬─────────┬──────────────┬──────────┘
                       │         │         │         │              │
                    ┌──▼──┐  ┌──▼──┐  ┌──▼──┐  ┌──▼──┐   ...   ┌──▼──┐
                    │W-00 │  │W-01 │  │W-02 │  │W-03 │         │W-09 │  子 Agent
                    │网站A│  │网站B│  │网站C│  │网站D│         │网站J│  （同构·并行）
                    └─────┘  └─────┘  └─────┘  └─────┘         └─────┘
​

对应书中强调的五个机制：

1. 消息总线（Message Bus）：所有通信都是发布到总线的带信封消息，按 type
   订阅。日志中每条 BUS ... 行就是一次发布/投递（Redis Pub/Sub 语义的进程内实现）。
2. 并行派发：Coordinator 同时给 10 个子 Agent 发 task_assigned 并 asyncio.create_task 并发执行。
3. 实时监控（push 范式）：子 Agent 执行中主动 status_update 上报进度，
   主 Agent 维护任务状态表并在状态机跳变时实时刷新打印。
   状态机：已提交 → 执行中 →（需要输入）→ 已完成 / 失败 / 已终止。
4. 级联终止：某子 Agent 命中后，主 Agent 广播 terminate；其余子 Agent 在
   循环的安全检查点发现信号后回 ack 并优雅退出（状态置为"已终止"）。
5. 竞态处理：多个子 Agent 可能几乎同时命中，主 Agent 用 asyncio.Lock +
   幂等标志 _settled 保证只结算一次、只广播一轮终止；迟到的命中被记录并忽略。

为让"竞态""级联终止"可复现，各来源被赋予不同的模拟延迟，其中 geo-journal 与
forum-qa 两个正确源被设成相同延迟，从而稳定地在同一时刻命中、触发竞态。

运行

cd chapter10/parallel-web-research
pip install -r requirements.txt   # 仅离线演示的话可跳过，纯标准库即可运行
python demo.py
​

默认走离线关键词判断（无需联网、结果可复现）。若要让子 Agent 用真实 LLM 判断：

cp env.example .env
# 在 .env 填入 OPENAI_API_KEY（也支持 Moonshot / 火山方舟 ARK 的 OpenAI 兼容网关）
python demo.py
# 或不改 .env，直接用命令行开关（仍需配置 key 才会真正生效）：
python demo.py --use-llm --model gpt-5.6-luna
​

可用 key：OPENAI_API_KEY（默认模型 gpt-5.6-luna）/ MOONSHOT_API_KEY / ARK_API_KEY
（填到 OPENAI_API_KEY 并按需设置 OPENAI_BASE_URL、OPENAI_MODEL）。

通用回退：若未设置 OPENAI_API_KEY 但设了 OPENROUTER_API_KEY，则真实 LLM 判断
自动改走 OpenRouter，并把模型名映射到其命名空间（gpt-5.6-luna → openai/gpt-5.6-luna）。
不影响协调机制，仅改变"是否命中"的判断。

命令行参数

python demo.py --help 可查看完整帮助。所有参数都不改变默认行为——不传任何参数即为
原有的「10 个 Agent + 内置问题 + 离线可复现 + 详细 BUS 日志」演示。

python demo.py --agents 6 --compare       # 6 个并行 Agent，并对比串行墙钟耗时
python demo.py --output result.json        # 结论落盘为 JSON
​

并行 vs 串行的墙钟收益（--compare）

对应书中实验要求「记录并对比并行/串行时间差异」。--compare 会在并行演示之后，用完全相同的
来源集合再跑一遍串行基线（逐个 await source.fetch() + 判断，命中即止），耗时是实测而非
估算。示例输出（默认 10 源，离线判断）：

5) 并行执行墙钟耗时：1.57s（含收敛静默期）
------------------------------------------------------------------------------
并行 vs 串行 墙钟对比（--compare，串行基线为实测）
------------------------------------------------------------------------------
   串行：命中前逐个抓取了 3/10 个源，墙钟耗时 2.60s，winner=geo-journal
   并行：墙钟耗时 1.57s，winner=worker-02
   加速比 ≈ 1.66×，节省约 1.03s（并行让最快的源立即结束全局搜索）。
​

串行必须依次抓完 baike-wiki→news-portal 才轮到最快命中的 geo-journal（累计 2.6s）；并行则让
所有源同时开跑，最快的源一命中就触发级联终止、立刻结束全局搜索。并行墙钟包含级联终止的收敛
开销，因此不是理想的「首个源延迟」，但仍显著快于串行——这正是并行 + 级联终止的价值所在。

演示说明什么（真实运行输出关键片段）

(a) 消息总线的发布/订阅在工作——每条带信封的消息都打印出来：

BUS [t=  0.00s #3  ] coordinator -> worker-02   | task_assigned  | {"question": "...", "source": "geo-journal"}
BUS [t=  0.00s #13 ]   worker-02 -> coordinator | status_update  | {"state": "执行中", ...}
​

(b) N 个子 Agent 并行执行 + 主 Agent 实时刷新状态表：

── 任务状态表（worker-02 -> 执行中） ──
   worker-00  源=baike-wiki   状态=执行中   | 开始抓取来源
   worker-02  源=geo-journal  状态=执行中   | 开始抓取来源
   ...
​

(c) 级联终止——命中后广播 terminate，其余子 Agent ack 并优雅退出：

BUS [t=0.60s #41 ] coordinator -> ALL         | terminate | {"reason":"answer_found","winner":"worker-02"}
BUS [t=0.67s #43 ]   worker-09 -> coordinator | ack       | {"acked":"terminate","source":"map-service"}
[ack] worker-09 已确认终止（1 个已 ack）
...最终 8 个未命中的 Worker 全部状态=已终止
​

(d) 竞态：即使几乎同时命中，也只结算一次、只广播一轮终止：

BUS [t=0.60s #37 ]   worker-02 -> coordinator | result | {"found":true, "answer":"...珠穆朗玛峰...8848 米..."}
BUS [t=0.60s #38 ]   worker-04 -> coordinator | result | {"found":true, "answer":"...珠穆朗玛峰...8848.86 米..."}
[结算] 首个命中来自 worker-02 —— 加锁结算，广播一轮 terminate。
[竞态] worker-04 也命中，但已由 worker-02 结算 —— 忽略此次命中，不重复广播终止。
​

demo.py 末尾有断言式自检：terminate 广播轮数 == 1、只结算一次 == True、
winner 非空。跑通即证明机制正确：

4) terminate 广播轮数：1（应为 1，证明只广播一轮）
   迟到/并发的重复命中被忽略：['worker-04']
[自检通过] 单次结算 + 单轮终止广播 + 级联 ack 均符合预期。
​

局限与注意事项

本实验把重点放在协调机制（消息总线/并行派发/级联终止/竞态处理）上，这些均为真实
实现；但为了可离线运行与自动验证，以下三处做了简化，是已知局限：

• 局限·模拟源非真实浏览器：不启动真实浏览器，来源是可控的模拟数据 + 延迟。若要接
  真实 Computer Use，只需把 WorkerAgent.run() 里的"抓取一步 + 判断"换成真实浏览器
  操作，协调层无需改动。
• 局限·竞态靠相同延迟稳定复现：geo-journal 与 forum-qa 两个正确源被人为设成
  相同延迟，才能稳定触发"同时命中"；真实环境里竞态是偶发的，但加锁 + 幂等的结算逻辑
  对偶发竞态同样成立，不依赖这个人为设定。
• 局限·进程内总线非真实 Redis：MessageBus 用进程内 async 队列模拟 Redis
  Pub/Sub，无需真装 Redis；语义（信封、按类型订阅、点对点/广播投递）一致，但不具备
  跨进程/跨机器能力，便于单机复现与自动验证。
• 子 Agent 在循环里定期检查终止信号（每步抓取前后），因此终止是"安全点响应"而非
  强杀，能保证资源被正常收尾。