SuperAGI 做 AI 智能体调度,任务队列接入 TaoToken

SuperAGI 做 AI 智能体调度,任务队列接入 TaoToken 1. SuperAGI 队列卡在 queued先改模型入口再谈 Agent 策略在 SuperAGI 里把多个 Agent 挂到同一个任务队列后我遇到的第一个运维问题不是 Agent 提示词而是 worker 从队列取到任务后一直停在 queued日志里 provider 鉴权失败任务重试三次仍没有最终产出。这种情况如果只盯 Agent 策略很容易在工具权限、任务依赖上绕圈真正要先把模型入口固定下来。做法是先去 TaoToken 官网拿 Keyhttps://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentsuperagi_queue_intro 再把 SuperAGI 的模型 Base URL 填为 https://taotoken.net/apiKey 用 YOUR_API_KEY 占位。近期围绕 AI 智能体“逃逸环境”的讨论升温这提醒 Agent 运维不能只看单次调用是否成功还要看模型入口、任务队列和审计记录是否在可控边界内。本文从 Agent 运维视角给出一套可复现的 SuperAGI 模型配置和任务队列调用记录方案先用 TaoToken 统一模型入口再让 SuperAGI 的调度器、worker 和本地记录表对齐同一组 request_id。SuperAGI 的任务调度通常包含几个阶段任务进入队列、worker 取任务、Agent 规划、调用模型、执行工具、回写结果。模型调用失败会直接让任务堆积而队列堆积又会掩盖真正的错误码。所以排障顺序应该是先确认模型通道可用再确认 worker 能取到任务最后才去调 Agent 策略和工具权限。下面从拿到 Key、填 Base URL、记录调用链、复用本地客户端配置几个部分展开。2. TaoToken Key 与 Base URLSuperAGI 模型配置的最小闭环先完成最小闭环Key 从 TaoToken 官网获取Base URL 填https://taotoken.net/api。打开 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentsuperagi_key_baseurl 登录后在控制台创建 API Key。把创建出来的 Key 放在环境变量里不要写进 Agent 提示词也不要提交到代码仓库。SuperAGI 不同版本的模型配置入口可能不同常见有三处Web UI 里的 Models / LLM Provider 配置页项目根目录的.envdocker-compose.yml或部署平台的环境变量。无论用哪一种核心都是让 SuperAGI 的模型客户端读到同一组值# SuperAGI 模型通道配置示例 # 如果使用 OpenAI 兼容变量按你的版本保留实际生效的那一组 OPENAI_API_KEYYOUR_API_KEY OPENAI_API_BASEhttps://taotoken.net/api OPENAI_BASE_URLhttps://taotoken.net/api # 建议单独保留一套 TaoToken 变量方便自定义 provider 读取 TAOTOKEN_API_KEYYOUR_API_KEY TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELYOUR_MODEL如果 SuperAGI 通过 Web UI 配置模型Provider 选 OpenAI 兼容类型API Key 填YOUR_API_KEYBase URL 填https://taotoken.net/api。模型名不要凭感觉填先从 TaoToken 控制台或模型列表里确认可用模型再写入TAOTOKEN_MODEL。改完配置后重启相关服务。具体服务名按你的 compose 文件调整常见做法是重启 SuperAGI 主服务和 workerdocker compose restart superagi worker docker compose logs -f --tail200 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_keyos.environ[TAOTOKEN_API_KEY], base_urlhttps://taotoken.net/api, timeout60, max_retries2, ) resp client.chat.completions.create( modelos.environ.get(TAOTOKEN_MODEL, YOUR_MODEL), messages[ {role: system, content: 你只需要回复 OK。}, {role: user, content: 连通性测试}, ], timeout30, ) 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_idSuperAGI 做 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_tokens0, completion_tokens0, error_codeNone, retry_count0, ): 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_keyos.environ[TAOTOKEN_API_KEY], base_urlhttps://taotoken.net/api, timeout60, max_retries2, ) start time.time() status success request_id None prompt_tokens 0 completion_tokens 0 error_code None try: resp client.chat.completions.create( modelos.environ.get(TAOTOKEN_MODEL, YOUR_MODEL), messages[ {role: system, content: 你是 SuperAGI 调度中的执行 Agent。}, {role: user, content: 返回当前任务的一句话摘要。}, ], timeout30, ) 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_idTASK_FROM_SUPERAGI, queue_nameagent_queue, agent_idAGENT_ID, modelos.environ.get(TAOTOKEN_MODEL, YOUR_MODEL), request_idrequest_id, statusstatus, latency_mslatency_ms, prompt_tokensprompt_tokens, completion_tokenscompletion_tokens, error_codeerror_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.tomlCC 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_KEYYOUR_API_KEYCodex 这里读取的是TAOTOKEN_API_KEY不是ANTHROPIC_AUTH_TOKEN。如果你在同一个 shell 里同时用 Claude Code 和 Codex建议分别开终端或者用不同 profile 隔离环境变量避免互相覆盖。CC Switch 三件套可以理解为供应商名称TaoTokenBase URLhttps://taotoken.net/apiAPI KeyYOUR_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_attempts4): 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(fattempt{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_sourcetaotoken_aicg_blog_endutm_contentsuperagi_repro 创建YOUR_API_KEY。在 SuperAGI 的.env或模型配置页里填入 Base URLhttps://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_sourcetaotoken_aicg_blog_endutm_contentsuperagi_chat如果你需要长期做本地 Agent 开发和代码辅助可以看 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentsuperagi_coding当你准备把 SuperAGI、Claude Code、Codex 都接到同一套入口时先创建并管理 API Key https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentsuperagi_keysClaude Code 的具体配置和变量说明可以直接对照文档 https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentsuperagi_claudecode