ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

Cursor 多 Agent Swarm 架构实测:用 SQLite 重写任务队列,API 费直降 87% 的配置复盘

Cursor 多 Agent Swarm 架构实测:用 SQLite 重写任务队列,API 费直降 87% 的配置复盘 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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_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_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。长期跑编码任务、Agent 数量比较多的可以看下 Coding Plan 的额度方案 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 按调用量包月比按次计费在 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, timeout5) 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 statusclaimed, worker_id?, updated_atCURRENT_TIMESTAMP WHERE id? AND statuspending , (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, timeout5) conn.execute( UPDATE tasks SET status?, result?, updated_atCURRENT_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表里所有任务statusdoneretry_count大多为 0api_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 statuspending, worker_idNULL WHERE statusclaimed 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 statuspending这个条件带上了只靠 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_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 里试。长期跑 Agent 的Coding Plan 那边有按量的方案可以对比一下 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。我踩过最深的坑是急着上 8 个 Worker结果 SQLite 写锁把整体拖慢后来降到 4 个反而更快。队列这东西并发不是越高越好先让状态流转干净再谈并行度。
返回列表