复杂任务交给单个智能体，常见的结果是：前面想得很细、后面越写越糙，或者干脆漏掉一半要求。Agent Swarm 的思路是把一件大事切成几块互不依赖的小事，让多个智能体同时开工，最后把各自的产出合并、对冲突、补缺口。这篇教程按"拆分 → 并行 → 合并 → 校验"的顺序，带你搭一套可以反复使用的 swarm 骨架。
文中涉及模型接口的具体字段、SDK 写法、平台按钮位置，以官方文档当前版本为准；骨架代码本身不依赖某个具体版本，换模型只需要改一个文件。
适用场景
适合手上有一类"信息量大、可以分区推进"的任务：多厂商产品对比、一份长文档拆章摘要、一次活动的多渠道文案、一批资料的分类归集。也适合单个智能体已经跑不动、输出质量随长度明显下滑的场景。
不适合强顺序依赖的任务（第二步必须等第一步结论），那类任务用单链条的 Agent 更稳。
环境与前置条件
- 操作系统：macOS、Linux、Windows（WSL2 更省心），无特殊要求。
- 运行时：Python 3.10 及以上（代码用到
asyncio.to_thread），pip 可用。 - 内存：8 GB 起步即可。整套流程只做调度和文本拼接，模型推理在云端完成，本地不需要 GPU。若你打算改成调用本地模型，显存需求以该模型官方说明为准。
- 磁盘：预留 5 GB 以上，主要给日志和中间结果；一次几百个子任务的运行日志通常在几十 MB 量级。
- 账号与密钥：已在 Carbon 上开通对应模型能力，拿到 API Key，并确认账号的并发额度（QPS / 每分钟 token 数）。这个数字决定下面
SWARM_MAX_PARALLEL该设多少。 - 依赖：
httpx用于 HTTP 调用；pydantic可选，用来做结构化校验。
```bash
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install httpx
```
分步骤部署
第 1 步：建立目录结构
把调度逻辑和模型调用分开，后面换模型、换 prompt 都不至于牵一发动全身。
```bash
mkdir -p swarm/logs
cd swarm
touch __init__.py config.py llm.py planner.py worker.py merger.py verifier.py run.py
```
目录长这样：llm.py 是全项目唯一和模型打交道的地方，其余文件只调用它。
第 2 步：配置参数
环境变量优于硬编码，换环境不用改代码。
```python
config.py
import os
API_BASE = os.getenv("SWARM_API_BASE", "")
API_KEY = os.getenv("SWARM_API_KEY", "")
MODEL = os.getenv("SWARM_MODEL", "")
MAX_PARALLEL = int(os.getenv("SWARM_MAX_PARALLEL", "4"))
TIMEOUT = float(os.getenv("SWARM_TASK_TIMEOUT", "120"))
MAX_RETRY = int(os.getenv("SWARM_MAX_RETRY", "3"))
```
```bash
export SWARM_API_BASE="你的接口地址"
export SWARM_API_KEY="你的密钥"
export SWARM_MODEL="你的模型名"
export SWARM_MAX_PARALLEL=4
```
成功的标志：python -c "import config; print(config.MODEL)" 能打印出模型名而不是空字符串。
第 3 步：写一个模型适配层
这是整个骨架里唯一需要你根据官方文档改动的地方。它对外只暴露一个 call_model(prompt) -> str，内部怎么请求随你。
```python
llm.py
import os, time, random, httpx
from config import API_BASE, API_KEY, MODEL, TIMEOUT, MAX_RETRY
class ModelError(RuntimeError):
pass
def call_model(prompt: str, temperature: float = 0.2) -> str:
"""最小适配层。把下面的请求体和响应解析换成官方 SDK / REST 的写法，
字段名以官方文档当前版本为准。函数签名保持不变，其余文件无需改动。"""
if not (API_BASE and API_KEY and MODEL):
raise ModelError("SWARM_API_BASE / SWARM_API_KEY / SWARM_MODEL 未配置")
payload = {"model": MODEL, "input": prompt, "temperature": temperature}
headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
last = None
for i in range(MAX_RETRY):
try:
with httpx.Client(timeout=TIMEOUT) as client:
resp = client.post(f"{API_BASE.rstrip('/')}/generate",
json=payload, headers=headers)
if resp.status_code in (429, 500, 502, 503, 504):
raise ModelError(f"HTTP {resp.status_code}")
resp.raise_for_status()
return _extract_text(resp.json())
except Exception as e:
last = e
time.sleep(min(2 ** i, 20) + random.random())
raise ModelError(f"重试 {MAX_RETRY} 次仍失败：{last}")
def _extract_text(data: dict) -> str:
"""按实际响应结构调整。常见是 data["output"]["text"] 或 data["candidates"][0]..."""
raise NotImplementedError("请按官方文档填写响应解析")
```
指数退避加抖动的意义在于：多个 worker 同时撞上限流时，不会在同一秒一起重试。
第 4 步：写任务拆分器
拆分质量决定后面一切。提示词里必须卡死三件事：子任务互不依赖、每个有验收标准、只输出 JSON。
```python
planner.py
import json
from llm import call_model
PLANNER_PROMPT = """你是任务拆分器。把下面的复杂任务拆成 2-6 个可以并行执行的子任务。
硬性要求：
1. 子任务之间不得依赖彼此的输出；
2. 每个子任务给出明确交付物和可检查的验收标准；
3. 只输出 JSON，不要任何解释、不要 Markdown 代码块。
输出格式：
{"goal": "...", "subtasks": [
{"id": "t1", "title": "...", "instruction": "...", "acceptance": "..."}
]}
任务：{task}
"""
def strip_fence(s: str) -> str:
s = s.strip()
if s.startswith("```"):
s = s.split("\n", 1)[1]
s = s.rsplit("```", 1)[0]
return s.strip()
def plan(task: str) -> dict:
raw = call_model(PLANNER_PROMPT.format(task=task), temperature=0.1)
data = json.loads(strip_fence(raw))
if not data.get("subtasks"):
raise ValueError("拆分结果为空")
return data
```
成功的标志：单独跑一次 plan("比较三个云厂商的对象存储计费差异")，能拿到合法 JSON，且子任务数量在 2 到 6 之间。
第 5 步：写并行执行器
并发数用信号量控制，单个子任务失败不影响其他子任务。
```python
worker.py
import asyncio, json
from llm import call_model
from config import MAX_PARALLEL
from planner import strip_fence
WORKER_PROMPT = """你在独立执行一个子任务，不要假设其他子任务会提供信息。
子任务：{title}
具体要求：{instruction}
验收标准：{acceptance}
只输出 JSON：
{{"id": "{id}", "status": "ok|partial|failed", "result": "...",
"evidence": ["支撑结论的具体依据"], "confidence": 0.0}}
"""
async def run_one(sub: dict, sem: asyncio.Semaphore) -> dict:
async with sem:
for attempt in range(3):
try:
prompt = WORKER_PROMPT.format(**sub)
raw = await asyncio.to_thread(call_model, prompt, 0.3)
data = json.loads(strip_fence(raw))
data.setdefault("id", sub["id"])
return data
except Exception as e:
if attempt == 2:
return {"id": sub["id"], "status": "failed",
"result": f"执行失败：{e}", "evidence": [],
"confidence": 0.0}
await asyncio.sleep(2 ** attempt)
async def run_swarm(subtasks: list[dict]) -> list[dict]:
sem = asyncio.Semaphore(MAX_PARALLEL)
return await asyncio.gather(*[run_one(s, sem) for s in subtasks])
```
call_model 是同步阻塞的，所以套 asyncio.to_thread 丢进线程池，这样信号量才真正起作用。gather 保证任何一个子任务抛错都不会中断整批——错误在 run_one 内部就被收成了一条 failed 记录。
第 6 步：写合并与校验
合并器不是简单拼接，要做三件事：按主题重组、标出冲突、指出缺口。校验器再独立看一眼，决定是否需要返工。
```python
merger.py
import json
from llm import call_model
from planner import strip_fence
MERGER_PROMPT = """你是结果合并器。下面是同一任务下多个子任务的结果，
可能存在重复、冲突或缺失。
请完成：
1. 按主题重新组织结论，不要按子任务原样罗列；
2. 指出互相矛盾之处，并说明哪一方证据更充分；
3. 列出仍未覆盖的信息缺口。
只输出 JSON：
{"summary": "...", "sections": [{"title": "...", "points": ["..."]}],
"conflicts": [{"topic": "...", "detail": "..."}], "gaps": ["..."]}
输入：
{payload}
"""
def merge(task: str, results: list[dict]) -> dict:
payload = json.dumps({"task": task, "results": results},
ensure_ascii=False, indent=2)
raw = call_model(MERGER_PROMPT.format(payload=payload), temperature=0.1)
return json.loads(strip_fence(raw))
```
```python
verifier.py
import json
from llm import call_model
from planner import strip_fence
VERIFY_PROMPT = """你是独立校验员。对照原始任务与子任务验收标准，
检查下面的合并结论是否达标。
只输出 JSON：
{"pass": true, "missing": ["..."], "weak_points": ["..."],
"redo": [{"id": "子任务id", "reason": "..."}]}
原始任务：{task}
子任务验收标准：{acceptance}
合并结论：{merged}
"""
def verify(task: str, subtasks: list[dict], merged: dict) -> dict:
acc = json.dumps([{"id": s["id"], "acceptance": s.get("acceptance", "")}
for s in subtasks], ensure_ascii=False)
raw = call_model(VERIFY_PROMPT.format(
task=task, acceptance=acc,
merged=json.dumps(merged, ensure_ascii=False)), temperature=0.0)
return json.loads(strip_fence(raw))
```
第 7 步：串起来跑完整流程
```python
run.py
import asyncio, json, time, pathlib
from planner import plan
from worker import run_swarm
from merger import merge
from verifier import verify
TASK = "对比三款主流笔记类应用在协作、离线、导出三方面的差异，给出选型建议"
async def main():
t0 = time.time()
p = plan(TASK)
print(f"[1/4] 拆出 {len(p['subtasks'])} 个子任务")
results = await run_swarm(p["subtasks"])
ok = sum(1 for r in results if r.get("status") == "ok")
print(f"[2/4] 完成 {ok}/{len(results)}")
merged = merge(TASK, results)
print(f"[3/4] 合并出 {len(merged.get('sections', []))} 个主题、"
f"{len(merged.get('conflicts', []))} 处冲突")
report = verify(TASK, p["subtasks"], merged)
print(f"[4/4] 校验 pass={report.get('pass')}，需返工 {len(report.get('redo', []))} 项")
out = pathlib.Path("logs") / f"run-{int(t0)}.json"
out.write_text(json.dumps(
{"task": TASK, "plan": p, "results": results,
"merged": merged, "verify": report}, ensure_ascii=False, indent=2))
print(f"耗时 {time.time()-t0:.1f}s，结果已写入 {out}")
asyncio.run(main())
```
这一版故意只跑一轮，不做自动返工。返工逻辑建议在稳定跑过十几次之后再加，否则容易陷入反复重试的循环。
验证部署是否成功
按下面顺序逐层验证，出问题好定位：
```bash
1. 配置是否到位
python -c "import config; print(config.MODEL, config.MAX_PARALLEL)"
2. 模型连通性（单独一个最小请求）
python -c "from llm import call_model; print(call_model('回复两个字：正常')[:50])"
3. 拆分器是否稳定输出结构化 JSON
python -c "from planner import plan; import json; print(json.dumps(plan('列出三种常见的数据备份策略及适用场景'), ensure_ascii=False, indent=2))"
4. 并行链路（把并发降到 1，先排除限流因素）
SWARM_MAX_PARALLEL=1 python run.py
5. 恢复并发跑全量
python run.py
```
预期结果：
- 第 2 步返回非空字符串，没有异常。
- 第 3 步输出是合法 JSON，含
goal和subtasks，子任务 2 到 6 个。 - 第 4 步串行跑通，四个阶段都有打印。
- 第 5 步并发跑完，
logs/下生成一份完整 JSON，[4/4]行显示pass=True或给出明确缺口。
判断并发是否真的生效：把 SWARM_MAX_PARALLEL 从 1 调到 4，总耗时应该明显下降（前提是子任务之间耗时接近、且未触达账号限流）。如果耗时几乎不变，检查信号量是不是写在了 asyncio.to_thread 外面。
常见报错与解决
1. json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
→ 模型在 JSON 前后加了说明文字或 Markdown 代码块，json.loads 直接失败。
```bash
解决：确认 strip_fence 被调用；提示词已强调"只输出 JSON"后仍复现，
就在解析失败时追加一次更硬的追问重试
python -c "
from llm import call_model
from planner import strip_fence
raw = call_model('只输出一个 JSON 对象：{\"ok\": true}，不要任何其他字符')
print(repr(strip_fence(raw))[:120])
"
```
同时在 run_one 的重试分支里，对 JSON 解析失败的情形额外拼一句"上一次输出不是合法 JSON，请只返回 JSON 本体"。
2. ModelError: HTTP 429
→ 并发超过了账号的速率额度，多个 worker 同时被拒。
```bash
export SWARM_MAX_PARALLEL=2
export SWARM_MAX_RETRY=5
python run.py
```
先降并发跑通，再以 1 为步长往上加，找到稳定不出 429 的值，写进环境变量长期使用。
3. ModelError: 重试 3 次仍失败：HTTP 504
→ 单个子任务太大，请求耗时超过网关上限。拆得更细比调大超时更有效。
```bash
export SWARM_TASK_TIMEOUT=180
同时在 planner 提示词里把"2-6 个子任务"改成"4-8 个"，逼它切得更碎
python -c "from planner import plan; print(len(plan('整理一份万字技术文档的摘要')['subtasks']))"
```
4. KeyError: 'subtasks' 或 ValueError: 拆分结果为空
→ 模型输出结构漂移，字段名换了。加一层兜底：解析失败时直接把原任务当成单个子任务跑，至少不让整次运行挂掉。
```python
planner.py 末尾追加兜底
except Exception:
return {"goal": task, "subtasks": [
{"id": "t1", "title": task, "instruction": task,
"acceptance": "覆盖任务的全部要求"}]}
```
5. 合并结果前后自相矛盾
→ 不同 worker 拿到了互相冲突的信息，而合并器只是被动拼接。解决办法是让 worker 必填 evidence 字段，并在合并提示词里要求对每处冲突标注"哪一方证据更充分"；校验器输出里出现 conflicts 时，人工抽查一眼对应的子任务原始输出。
后续维护
备份：logs/ 下按时间戳命名的 JSON 是完整现场记录（拆分方案、每个子任务产出、合并结果、校验意见），把整个目录纳入定时备份即可。想复现某次结果，从这份 JSON 出发比从零重跑划算。
升级：llm.py 是唯一耦合模型接口的文件，升级时只动它。换模型后先跑第 4 步验证清单，重点看拆分质量有没有变化——模型换了，同一段 planner 提示词拆出来的粒度可能明显不同。
日志：现在只有运行结果落盘，建议补两列：每个子任务的耗时、以及 token 用量。有了这两列，才能算出"并行把总时长压下去多少""一次运行大概花多少"，也才能在限流时判断是并发太高还是单个子任务太重。
监控指标：优先盯四个——子任务成功率、单任务 P95 耗时、429 出现次数、合并结果里的冲突数。前三个掉下去说明调度或额度有问题，冲突数持续偏高通常意味着拆分边界划得不好，该回头改 planner 提示词了。
成本控制：在 call_model 里按输入输出长度记一笔账，超过阈值时打印告警。子任务数量是最容易失控的变量，planner 提示词里的上限（2-6）别随手放宽到两位数。
提示词版本化：planner、worker、merger、verifier 这四段提示词各存在哪个文件里、改过几次，值得在文件头用注释记下日期和改动原因。改完之后回跑同一批测试任务，对比拆分数量和合并质量，比凭感觉判断靠谱。