1. SuperAGI 队列卡在 queued:先改模型入口,再谈 Agent 策略
在 SuperAGI 里把多个 Agent 挂到同一个任务队列后,我遇到的第一个运维问题不是 Agent 提示词,而是 worker 从队列取到任务后一直停在 queued:日志里 provider 鉴权失败,任务重试三次仍没有最终产出。这种情况如果只盯 Agent 策略,很容易在工具权限、任务依赖上绕圈;真正要先把模型入口固定下来。做法是先去 TaoToken 官网拿 Key:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=superagi_queue_intro ,再把 SuperAGI 的模型 Base URL 填为 https://taotoken.net/api,Key 用 YOUR_API_KEY 占位。
近期围绕 AI 智能体“逃逸环境”的讨论升温,这提醒 Agent 运维不能只看单次调用是否成功,还要看模型入口、任务队列和审计记录是否在可控边界内。本文从 Agent 运维视角,给出一套可复现的 SuperAGI 模型配置和任务队列调用记录方案:先用 TaoToken 统一模型入口,再让 SuperAGI 的调度器、worker 和本地记录表对齐同一组 request_id。
SuperAGI 的任务调度通常包含几个阶段:任务进入队列、worker 取任务、Agent 规划、调用模型、执行工具、回写结果。模型调用失败会直接让任务堆积,而队列堆积又会掩盖真正的错误码。所以排障顺序应该是:先确认模型通道可用,再确认 worker 能取到任务,最后才去调 Agent 策略和工具权限。下面从拿到 Key、填 Base URL、记录调用链、复用本地客户端配置几个部分展开。
2. TaoToken Key 与 Base URL:SuperAGI 模型配置的最小闭环
先完成最小闭环:Key 从 TaoToken 官网获取,Base URL 填https://taotoken.net/api。打开 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=superagi_key_baseurl ,登录后在控制台创建 API Key。把创建出来的 Key 放在环境变量里,不要写进 Agent 提示词,也不要提交到代码仓库。
SuperAGI 不同版本的模型配置入口可能不同,常见有三处:
- Web UI 里的 Models / LLM Provider 配置页;
- 项目根目录的
.env; docker-compose.yml或部署平台的环境变量。
无论用哪一种,核心都是让 SuperAGI 的模型客户端读到同一组值:
# SuperAGI 模型通道配置示例 # 如果使用 OpenAI 兼容变量,按你的版本保留实际生效的那一组 OPENAI_API_KEY=YOUR_API_KEY OPENAI_API_BASE=https://taotoken.net/api OPENAI_BASE_URL=https://taotoken.net/api # 建议单独保留一套 TaoToken 变量,方便自定义 provider 读取 TAOTOKEN_API_KEY=YOUR_API_KEY TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL=YOUR_MODEL如果 SuperAGI 通过 Web UI 配置模型,Provider 选 OpenAI 兼容类型,API Key 填YOUR_API_KEY,Base URL 填https://taotoken.net/api。模型名不要凭感觉填,先从 TaoToken 控制台或模型列表里确认可用模型,再写入TAOTOKEN_MODEL。
改完配置后重启相关服务。具体服务名按你的 compose 文件调整,常见做法是重启 SuperAGI 主服务和 worker:
docker compose restart superagi worker docker compose logs -f --tail=200 superagi worker日志里重点看三类信息:worker 是否成功取到 task、模型客户端是否走到你的 Base URL、失败时返回的是 401 还是 404 还是超时。只要 worker 日志里出现明确的模型调用结果,就说明 SuperAGI 的模型入口已经接管成功。下一步才是把每次调用和任务队列关联起来。
如果你暂时不想在 SuperAGI 里反复改配置,也可以先用 Python 做一次本地探针。它只验证 Key 和 Base URL 是否可用,不涉及生产库,也不让 Agent 直接操作任何数据库:
import os from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url="https://taotoken.net/api", timeout=60, max_retries=2, ) resp = client.chat.completions.create( model=os.environ.get("TAOTOKEN_MODEL", "YOUR_MODEL"), messages=[ {"role": "system", "content": "你只需要回复 OK。"}, {"role": "user", "content": "连通性测试"}, ], timeout=30, ) print("request_id:", resp.id) print("content:", resp.choices[0].message.content) print("usage:", resp.usage)这段脚本能跑通后,再回到 SuperAGI 配置页或.env,把相同的 Key、Base URL、模型名写进去。注意:Claude Code 使用ANTHROPIC_*变量,Codex 使用config.toml,两者不要混用。这个边界在后面的客户端复用一节会单独说明。
3. 任务队列调用记录:用 SQLite 给每个 Agent 调用留 request_id
SuperAGI 做 AI 智能体调度时,最怕的是“任务失败但不知道失败在哪一层”。队列里只看到任务状态,模型侧只有一次调用,工具侧又有自己的日志。Agent 运维要把它们串起来,最简单的方式是本地 SQLite 记录表:每次 worker 调用模型前后写一条记录,字段至少包括 task_id、queue_name、agent_id、model、base_url、request_id、status、latency_ms、error_code。
先在你的本地环境建表。SQL 由你本地执行,不要让 Agent 直接连生产库:
CREATE TABLE IF NOT EXISTS agent_llm_calls ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, queue_name TEXT, agent_id TEXT, provider TEXT, base_url TEXT, model TEXT, request_id TEXT, status TEXT, latency_ms INTEGER, prompt_tokens INTEGER, completion_tokens INTEGER, error_code TEXT, retry_count INTEGER DEFAULT 0, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_agent_llm_calls_task_id ON agent_llm_calls(task_id); CREATE INDEX IF NOT EXISTS idx_agent_llm_calls_request_id ON agent_llm_calls(request_id);然后在 SuperAGI 的 worker 侧或者自定义 LLM provider 适配层里,把任务上下文和模型调用绑定。下面是一个最小记录函数,重点是字段结构,不是绑定到某个固定版本的 SuperAGI 文件:
import os import sqlite3 import time from datetime import datetime, timezone DB_PATH = os.environ.get("AGENT_LLM_DB", "agent_llm_calls.db") def init_db(): con = sqlite3.connect(DB_PATH) con.execute(""" CREATE TABLE IF NOT EXISTS agent_llm_calls ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, queue_name TEXT, agent_id TEXT, provider TEXT, base_url TEXT, model TEXT, request_id TEXT, status TEXT, latency_ms INTEGER, prompt_tokens INTEGER, completion_tokens INTEGER, error_code TEXT, retry_count INTEGER DEFAULT 0, created_at TEXT NOT NULL ) """) con.commit() con.close() def record_call( task_id, queue_name, agent_id, model, request_id, status, latency_ms, prompt_tokens=0, completion_tokens=0, error_code=None, retry_count=0, ): con = sqlite3.connect(DB_PATH) con.execute( """ INSERT INTO agent_llm_calls ( task_id, queue_name, agent_id, provider, base_url, model, request_id, status, latency_ms, prompt_tokens, completion_tokens, error_code, retry_count, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( task_id, queue_name, agent_id, "taotoken", "https://taotoken.net/api", model, request_id, status, latency_ms, prompt_tokens, completion_tokens, error_code, retry_count, datetime.now(timezone.utc).isoformat(), ), ) con.commit() con.close() init_db()实际调用时可以这样包一层:
import time from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url="https://taotoken.net/api", timeout=60, max_retries=2, ) start = time.time() status = "success" request_id = None prompt_tokens = 0 completion_tokens = 0 error_code = None try: resp = client.chat.completions.create( model=os.environ.get("TAOTOKEN_MODEL", "YOUR_MODEL"), messages=[ {"role": "system", "content": "你是 SuperAGI 调度中的执行 Agent。"}, {"role": "user", "content": "返回当前任务的一句话摘要。"}, ], timeout=30, ) request_id = resp.id prompt_tokens = resp.usage.prompt_tokens if resp.usage else 0 completion_tokens = resp.usage.completion_tokens if resp.usage else 0 except Exception as exc: status = "failed" error_code = type(exc).__name__ raise finally: latency_ms = int((time.time() - start) * 1000) record_call( task_id="TASK_FROM_SUPERAGI", queue_name="agent_queue", agent_id="AGENT_ID", model=os.environ.get("TAOTOKEN_MODEL", "YOUR_MODEL"), request_id=request_id, status=status, latency_ms=latency_ms, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, error_code=error_code, )这样你就能在本地查询某个 task_id 下所有模型调用:
SELECT task_id, queue_name, agent_id, model, request_id, status, latency_ms, error_code, created_at FROM agent_llm_calls WHERE task_id = 'TASK_FROM_SUPERAGI' ORDER BY created_at DESC;这张表就是“任务队列调用记录”的基线。它不依赖生产数据库,也不会让 Agent 直接持有高权限连接。对 Agent 运维来说,能按 task_id 回看每一次模型调用,已经能解决大部分“任务卡住但不知道卡在哪”的问题。
4. Claude Code、Codex、CC Switch:同一把 Key 的本地运维配置模板
SuperAGI 负责调度,但本地排障时经常还要用 Claude Code、Codex、CC Switch 做辅助验证。这时最容易犯的错是把不同工具的变量混在一起。记住边界:Claude Code 用settings.json和ANTHROPIC_*;Codex 用config.toml;CC Switch 关注供应商、Base URL、API Key 三件套。不要把ANTHROPIC_*套到 Codex。
Claude Code 的settings.json示例:
{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_AUTH_TOKEN": "YOUR_API_KEY", "ANTHROPIC_MODEL": "YOUR_MODEL" } }这里的ANTHROPIC_AUTH_TOKEN用YOUR_API_KEY占位。改完后重新打开 Claude Code,让它重新读取配置。若你使用项目级配置,注意不要把它提交到公共仓库。
Codex 的config.toml示例:
model = "YOUR_MODEL" model_provider = "taotoken" [model_providers.taotoken] name = "TaoToken" base_url = "https://taotoken.net/api" env_key = "TAOTOKEN_API_KEY" wire_api = "chat"对应环境变量:
export TAOTOKEN_API_KEY=YOUR_API_KEYCodex 这里读取的是TAOTOKEN_API_KEY,不是ANTHROPIC_AUTH_TOKEN。如果你在同一个 shell 里同时用 Claude Code 和 Codex,建议分别开终端,或者用不同 profile 隔离环境变量,避免互相覆盖。
CC Switch 三件套可以理解为:
- 供应商名称:TaoToken;
- Base URL:
https://taotoken.net/api; - API Key:
YOUR_API_KEY。
不同版本的 CC Switch 字段名可能不同,但本质就是这三项。你可以先用一个本地 JSON 做对照:
{ "provider": "taotoken", "baseUrl": "https://taotoken.net/api", "apiKey": "YOUR_API_KEY" }如果你的 CC Switch 界面要求选择协议类型,优先选 OpenAI 兼容或 Anthropic 兼容中与你当前工具匹配的那一项,不要为了“看起来能连”而混填协议。Claude Code 走 Anthropic 风格配置,Codex 走 Codex 自己的 provider 配置,SuperAGI 走 OpenAI 兼容 provider。三条链路都指向同一个 Base URL,但变量名和配置文件必须分开。
5. SuperAGI worker 排障:401、404、429、超时的处理顺序
当 SuperAGI 任务队列堆积时,不要先改 Agent 策略,按下面顺序看模型调用错误。
401 通常出现在鉴权阶段。检查YOUR_API_KEY是否被正确替换,检查 worker 容器里是否真的读到了环境变量,检查 Key 前面是否被误加了Bearer。如果你在 UI 里填 Key,不要把Bearer一起填进去。
404 常见于 Base URL 或模型名不对。先确认 Base URL 是https://taotoken.net/api,再确认模型名来自 TaoToken 控制台或模型列表。不要自己拼接不存在的路径,也不要在 Base URL 后面随手加/v1或重复加/api。
429 说明调用频率或并发触发了限制。Agent 运维上要做两件事:一是给 worker 设置合理的并发数,二是给模型调用加指数退避。不要用无限重试,否则队列会更堵。
超时则要区分是模型响应慢,还是 worker 侧超时太短。可以先在本地探针脚本里把timeout设成 60 秒,确认模型侧能返回;再回到 SuperAGI 调整 worker 的超时和重试次数。对于长任务,建议在 Agent 执行详情里记录开始时间、模型调用时间、工具执行时间,避免把工具耗时误判为模型耗时。
一个简单的退避思路如下:
import time import random def call_with_backoff(fn, max_attempts=4): for attempt in range(1, max_attempts + 1): try: return fn() except Exception as exc: if attempt == max_attempts: raise sleep_s = min(2 ** attempt + random.random(), 10) print(f"attempt={attempt} error={type(exc).__name__} sleep={sleep_s:.2f}s") time.sleep(sleep_s)把它包在模型调用外层即可。记录表里的retry_count要同步写入,否则你看到的失败次数和实际重试次数会对不上。
6. 从智能体风险讨论回到工程边界:队列隔离、审计与人工闸门
外部关于 AI 智能体风险的讨论,落到 Agent 运维上不是一句“要小心”,而是几个具体工程动作。
第一,模型入口统一。所有 SuperAGI worker 都通过 TaoToken 的 Base URL 调用模型,Key 只存在环境变量或密钥管理系统里,不进入 Agent 提示词,不写进任务描述。
第二,队列隔离。不要把所有 Agent 塞进同一个队列。按业务域、风险等级或工具权限拆队列,例如agent_low_risk_queue、agent_review_queue。某个队列模型调用失败时,不会拖垮全部任务。
第三,审计记录。每次模型调用至少保留 task_id、agent_id、model、request_id、latency_ms、status、error_code。出现异常时,先按 request_id 查模型侧,再按 task_id 查队列侧,最后查工具执行日志。
第四,人工闸门。涉及数据写入、外部通知、批量操作的步骤,不要让 Agent 直接执行到底。把 SuperAGI 的任务拆成“生成计划”和“执行计划”两段,执行前进入人工确认队列。这样即使模型输出异常,也不会直接作用到生产环境。
第五,本地执行 SQL 和命令。Agent 可以生成待执行的 SQL 或命令文本,但最终由人在本地或受控终端执行。不要让 Agent 直连生产库,也不要把高权限连接串配置进调度器。
这些动作不需要一次做完。对大多数团队来说,先把模型入口统一、调用记录留全、队列按风险拆分,已经能显著降低 Agent 调度失控的概率。
7. 可复现产出:把 SuperAGI 模型配置和调用记录跑成一套基线
如果你要按本文复现,建议按这个顺序产出:
- 打开 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=superagi_repro ,创建
YOUR_API_KEY。 - 在 SuperAGI 的
.env或模型配置页里填入 Base URL:https://taotoken.net/api。 - 用本地 Python 探针确认 Key、Base URL、模型名三者可用。
- 在 worker 或自定义 provider 里接入
record_call,把 task_id 和 request_id 写入 SQLite。 - 在 SuperAGI 里触发一个简单 Agent 任务,确认任务从 queued 进入 running,再进入 success。
- 用
SELECT * FROM agent_llm_calls WHERE task_id = '...'检查调用记录。 - 如果失败,按 401、404、429、超时的顺序排查,不要先改 Agent 策略。
完成这套基线后,你得到的不是一次“能跑”的配置,而是一条可追踪的调度链路:SuperAGI 负责任务入队和 Agent 调度,TaoToken 负责模型入口,SQLite 负责调用记录。后续要换模型、调并发、拆队列,都有数据可依。
8. 下一步入口:模型对话、Coding Plan、API Key 与 Claude Code 文档
如果你还没有确定模型或想先验证对话效果,可以从模型对话开始: https://taotoken.net/models/detail/chat?utm_source=taotoken_aicg_blog_end&utm_content=superagi_chat
如果你需要长期做本地 Agent 开发和代码辅助,可以看 Coding Plan: https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=superagi_coding
当你准备把 SuperAGI、Claude Code、Codex 都接到同一套入口时,先创建并管理 API Key: https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=superagi_keys
Claude Code 的具体配置和变量说明,可以直接对照文档: https://taotoken.net/doc/ClaudeCodeAnthropic?utm_source=taotoken_aicg_blog_end&utm_content=superagi_claudecode