1. 从一张账单说起：多 Agent 协作为什么越跑越贵
如果你正在用 Cursor 的 Agent 模式跑多任务协作，或者自己搭过 Planner + Worker 的 Swarm 结构，大概率遇到过同一个问题：任务拆得越细、Agent 起得越多，API 账单涨得越快。我上个月帮一个朋友看他的项目，三个 Agent 并行处理代码重构，一天跑下来调用量比单 Agent 高出四倍多，但真正产出有效 diff 的比例不到三成。
问题不在模型本身，而在任务队列。多数 Swarm 实现把队列放在内存里，或者干脆用 JSON 文件轮询，Agent 之间靠不断重读上下文来同步状态。每同步一次，就是一次完整的 prompt 重发。任务一多，重复调用就像滚雪球。我实测下来，一个 8 子任务的 Swarm，光状态同步产生的冗余调用就占了总调用量的 60% 以上。
这篇要解决的就是这件事：用 SQLite 重写 Swarm 的任务队列，把状态同步从「反复问模型」变成「查一次本地库」。配置改完之后，同一个重构任务的 API 调用量从 1,240 次降到 160 次左右，费用直降 87%。下面把表结构、Cursor 侧配置骨架、验证动作和踩过的坑完整复盘一遍，你可以直接照着在自己的项目里复现。
2. TaoToken 前置：把模型调用收口到一个入口
在动队列之前，先解决调用入口的问题。Swarm 架构里 Planner 和 Worker 往往用不同模型，如果每个 Agent 各自配一套 key 和 endpoint，成本统计和限流都会失控。我的做法是统一走 TaoToken 的 API 入口，Planner 用强模型、Worker 用轻量模型，但都从同一个 base_url 出去，账单和调用日志集中在一处。
TaoToken 在这里的角色是模型调用的统一网关，官网地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 入口是 https://taotoken.net/api 。它本身不改变你的 Swarm 逻辑，但让「哪个 Agent 花了多少钱」变得可查——这对后面验证降费效果是刚需，否则你根本不知道省下来的钱来自队列优化还是模型切换。
具体操作上，你需要在控制台创建一个 API Key，然后把它写进 Cursor 的模型配置里。创建 Key 的入口在 https://taotoken.net/console/api-keys ，接入文档在 https://taotoken.net/doc ，里面有不同框架的 base_url 和鉴权头写法。Cursor 侧只需要在设置里把 OpenAI 兼容的 base_url 指向 TaoToken 的 API 地址，模型名按文档里支持的填。
注意：Planner 和 Worker 建议用两个不同的 Key 或者至少两个不同的模型标识，这样在调用日志里能直接区分规划开销和执行开销。我一开始图省事用同一个 Key，结果排查冗余调用时完全分不清是谁发的请求，白白多花了两小时。
如果你还没决定 Planner 和 Worker 分别用什么模型，可以先在模型对话里试一下拆解效果和代码生成质量，地址是 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。长期跑编码任务、Agent 数量比较多的，可以看下 Coding Plan 的额度方案 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite ，按调用量包月比按次计费在 Swarm 场景下更可控。
3. 可复制配置：SQLite 任务队列表结构 + Cursor 配置骨架
3.1 为什么是 SQLite 而不是内存队列
内存队列的问题在于 Agent 进程一重启，状态全丢，Planner 只能重新规划，Worker 只能重新执行。JSON 文件轮询的问题在于并发读写没有锁，多个 Worker 同时写会互相覆盖，而且每次读都要解析整个文件。SQLite 的好处是：单文件、零依赖、支持事务和行级锁、查询是增量的。对 Swarm 这种「多写多读、状态频繁变更」的场景，它刚好卡在轻量和可靠之间。
核心思路是把任务的生命周期拆成状态机：pending → claimed → running → done / failed。Planner 只负责往表里插任务，Worker 只负责认领和更新状态，Agent 之间不再通过 prompt 互相通知，而是通过查表。状态同步从「模型调用」降级成「本地 SQL 查询」，这是降费的根本来源。
3.2 表结构
-- swarm_queue.sql PRAGMA journal_mode = WAL; -- 允许读写并发 PRAGMA busy_timeout = 5000; -- 锁等待 5 秒 CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, swarm_id TEXT NOT NULL, -- 一次 Swarm 运行的标识 parent_id INTEGER, -- 父任务，用于拆解树 name TEXT NOT NULL, -- 子任务名 spec TEXT NOT NULL, -- Planner 给的规范说明 file_pattern TEXT, -- 目标文件匹配 status TEXT NOT NULL DEFAULT 'pending', worker_id TEXT, -- 认领的 Worker result TEXT, -- Worker 产出的 diff error TEXT, retry_count INTEGER NOT NULL DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_status ON tasks(status); CREATE INDEX IF NOT EXISTS idx_swarm ON tasks(swarm_id, status); -- 原子认领：避免多个 Worker 抢同一个任务 CREATE TABLE IF NOT EXISTS workers ( worker_id TEXT PRIMARY KEY, last_seen DATETIME DEFAULT CURRENT_TIMESTAMP, current_task INTEGER );status字段是整个队列的枢纽。Worker 启动后不是去问 Planner「我该干什么」，而是执行一条带条件的 UPDATE，把一条 pending 任务原子地改成 claimed 并绑定自己的 worker_id。SQLite 的事务保证同一时刻只有一个 Worker 能认领成功，不需要额外的分布式锁。
3.3 认领与回写逻辑
# queue.py import sqlite3, uuid, time DB = "swarm_queue.sql" def claim_task(worker_id: str): conn = sqlite3.connect(DB, timeout=5) cur = conn.cursor() cur.execute("BEGIN IMMEDIATE") cur.execute(""" SELECT id, name, spec, file_pattern FROM tasks WHERE status = 'pending' ORDER BY id LIMIT 1 """) row = cur.fetchone() if not row: conn.commit(); conn.close() return None task_id = row[0] cur.execute(""" UPDATE tasks SET status='claimed', worker_id=?, updated_at=CURRENT_TIMESTAMP WHERE id=? AND status='pending' """, (worker_id, task_id)) conn.commit(); conn.close() return {"id": task_id, "name": row[1], "spec": row[2], "file": row[3]} def finish_task(task_id: int, result: str, ok: bool = True): conn = sqlite3.connect(DB, timeout=5) conn.execute(""" UPDATE tasks SET status=?, result=?, updated_at=CURRENT_TIMESTAMP WHERE id=? """, ("done" if ok else "failed", result, task_id)) conn.commit(); conn.close()BEGIN IMMEDIATE是关键，它在事务开始时就拿写锁，避免两个 Worker 同时读到同一条 pending 再各自 UPDATE 的竞态。我最早用的是普通BEGIN，压测时出现过两个 Worker 认领同一任务、产出两份冲突 diff 的情况，换成 IMMEDIATE 后没再复现。
3.4 Cursor 侧配置骨架
Cursor 本身不直接读 SQLite，所以需要一个薄适配层：Worker 在 Cursor 里跑的时候，先调claim_task拿到任务，把spec拼进 prompt，执行完再调finish_task回写。Cursor 的模型配置指向 TaoToken 的 API 入口，Planner 和 Worker 用不同模型。
{ "models": [ { "name": "planner", "provider": "openai-compatible", "base_url": "https://taotoken.net/api", "model": "claude-opus-4-8", "api_key_env": "TAOTOKEN_PLANNER_KEY" }, { "name": "worker", "provider": "openai-compatible", "base_url": "https://taotoken.net/api", "model": "composer-2.5", "api_key_env": "TAOTOKEN_WORKER_KEY" } ], "swarm": { "queue_db": "./swarm_queue.sql", "max_workers": 4, "claim_timeout_sec": 300, "max_retry": 2 } }max_workers不要一上来就拉满。我试过 8 个 Worker 并行，SQLite 的写锁竞争明显，busy_timeout频繁触发，反而拖慢整体。4 个是这台机器上的甜点值，你可以从 2 开始往上加，观察busy_timeout的日志频率。
4. 验证请求：怎么确认调用量真的降了
配置改完不能只看账单，账单有延迟，而且分不清是队列优化还是模型切换的功劳。我的做法是在适配层加一个调用计数器，每次 Worker 真正发起模型请求时打一条日志，跑完一个 Swarm 后统计。
# counter.py import json, os, time LOG = "api_calls.jsonl" def log_call(role: str, task_id: int, tokens: int): with open(LOG, "a") as f: f.write(json.dumps({ "ts": time.time(), "role": role, # planner / worker "task_id": task_id, "tokens": tokens }) + "\n") def summary(): calls = {"planner": 0, "worker": 0} tokens = {"planner": 0, "worker": 0} with open(LOG) as f: for line in f: r = json.loads(line) calls[r["role"]] += 1 tokens[r["role"]] += r["tokens"] return calls, tokens跑同一个重构任务，改造前后各来一次，对比summary()的输出。我这边改造前的数据是：Planner 调用 210 次、Worker 调用 1,030 次，总 token 约 2.1M；改造后 Planner 调用 12 次、Worker 调用 148 次，总 token 约 0.27M。调用次数降了约 87%，token 降幅接近，费用自然跟着下来。
提示：验证时一定要用同一个任务、同一组模型，否则变量太多说不清。我第一次对比时顺手把 Worker 从强模型换成了轻量模型，结果省下来的钱里有多少是队列的功劳完全算不清，只能重跑一遍。
成功的结果长这样：tasks表里所有任务status='done'，retry_count大多为 0，api_calls.jsonl里 Worker 的调用次数和任务数接近 1:1 而不是 1:N。如果 Worker 调用次数远大于任务数，说明还有 Agent 在靠 prompt 轮询状态，队列没接干净。
5. 本篇常见错排查
5.1 database is locked
最常见。原因是多个 Worker 同时写，或者某个 Worker 认领后卡住没回写，事务一直挂着。先确认PRAGMA journal_mode = WAL和busy_timeout都设了。如果还报，检查是不是有 Worker 在claim_task之后抛异常没走到finish_task，导致任务永远停在 claimed。加一个超时回收：定时把updated_at超过claim_timeout_sec的 claimed 任务改回 pending。
UPDATE tasks SET status='pending', worker_id=NULL WHERE status='claimed' AND updated_at < datetime('now', '-300 seconds');5.2 Worker 认领到任务但产出为空
多半是spec字段没写清楚。Planner 拆任务时如果只给个任务名不给规范，Worker 拿到的 prompt 太模糊，模型会返回空或者一堆解释性文字。解决办法是在 Planner 的 prompt 里强制要求每个子任务输出name、file_pattern、entry_point三个字段，缺一个就重规划。这个约束比换模型有用得多。
5.3 调用量没降反升
检查两处。一是 Worker 是不是还在每次执行前重新读一遍全局上下文，如果是，那 SQLite 队列白搭，因为大头开销在上下文重发。二是 Planner 是不是被反复调用，有些实现里 Worker 遇到不确定的地方会回头问 Planner，一问就是一次强模型调用。正确做法是 Worker 只按spec执行，遇到 spec 没覆盖的情况直接标 failed，由 Planner 统一重规划，而不是让 Worker 自己发起对话。
5.4 任务重复执行
如果claim_task用的是普通BEGIN而不是BEGIN IMMEDIATE，高并发下会重复认领。另外确认UPDATE ... WHERE status='pending'这个条件带上了，只靠 SELECT 出来的 id 去 UPDATE 而不校验 status，同样会重复。
6. 收尾：把队列和调用入口一起收口
这套改造真正省下来的不是模型钱，是「Agent 之间反复确认状态」的钱。SQLite 队列把状态同步从模型层挪到本地层，Planner 和 Worker 各干各的，不再互相打断。配合 TaoToken 统一调用入口，你能清楚看到每一分钱花在规划还是执行上。
如果你准备动手，建议按这个顺序：先把任务表建起来，把现有 Swarm 的状态同步逻辑替换成claim_task/finish_task；再在 Cursor 侧把模型配置指向 TaoToken 的 API 入口，Planner 和 Worker 分开配；最后用api_calls.jsonl跑一次前后对比。API Key 在 https://taotoken.net/console/api-keys 创建，接入细节看 https://taotoken.net/doc ，模型选型可以先在 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 里试。长期跑 Agent 的，Coding Plan 那边有按量的方案可以对比一下 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。
我踩过最深的坑是急着上 8 个 Worker，结果 SQLite 写锁把整体拖慢，后来降到 4 个反而更快。队列这东西，并发不是越高越好，先让状态流转干净，再谈并行度。
