1. 从一次接口超时说起FastAPI 并发任务为什么容易卡线上有个 FastAPI 服务接口本身逻辑不复杂接收请求、调用一次大模型、把结果写回数据库。单机 QPS 一上来P99 直接从 300ms 飙到 8s日志里全是Task exception was never retrieved和连接池耗尽。排查下来问题不在 FastAPI 本身而是三件事叠在一起同步阻塞调用堵死了事件循环、每个请求都新建 HTTP 客户端、多模型调用各自维护一套 Key 和重试逻辑。FastAPI 的并发能力来自async/await加事件循环它适合 IO 密集型场景。但只要你在一段async def里写了同步的requests.post整个事件循环就被这一个请求占住其他协程只能排队。这就是为什么很多人的 FastAPI 压测数据远低于预期——框架没问题是调用方式把并发吃掉了。这篇要解决的是在 FastAPI 高并发场景下怎么用异步任务队列把耗时的大模型调用从请求主链路里剥离同时用 TaoToken 的统一 Key 和 API 通道收敛多模型接入让配置可复制、指标可观测。适合已经在写 FastAPI、准备做多模型接入、或者正被并发瓶颈卡住的开发者。下面给到的settings.json、config.toml骨架和压测动作都可以直接拿去改。2. 前置准备TaoToken 统一 Key 与通道接入多模型接入最烦的不是调用本身是每个厂商一套鉴权、一套限流、一套错误码。TaoToken 在这里的作用是把这些收敛成一个入口一个 Key、一个 Base URL模型名通过参数切换。对 FastAPI 这种要横向扩展的服务来说配置项越少环境变量和密钥管理越干净。接入前先拿到 Key。打开控制台创建 API Key建议按环境分 Key比如fastapi-dev、fastapi-prod各一个方便单独吊销和统计用量。创建入口在控制台的 API Keys 页面https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentfastapi_concurrencyutm_campaignrewrite拿到 Key 之后接口地址统一用https://taotoken.net/api不要在每个请求里硬编码。模型名、超时、重试这些放进配置文件代码只读配置。这样做的直接好处是压测时改并发参数不用动代码切模型也不用重新部署。如果你还在选模型阶段可以先用模型对话页面手动跑几条请求确认目标模型的响应特征和延迟量级再决定放进队列的并发上限https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentfastapi_concurrencyutm_campaignrewrite需要说明的是TaoToken 是合规的 API 聚合通道不是让你绕过任何限制的工具。它的价值在于统一鉴权和调用格式减少多厂商适配的重复代码。密钥只放服务端环境变量绝对不要下发到前端或写进仓库。3. 可复制配置settings.json 与 config.toml 骨架配置分两层settings.json管运行时参数并发、超时、队列config.toml管模型与通道定义。分开的原因是前者经常调后者相对稳定。先看settings.json重点是队列并发和连接池参数{ app: { host: 0.0.0.0, port: 8000, workers: 4 }, queue: { backend: redis, redis_url: redis://127.0.0.1:6379/0, max_concurrency: 32, task_timeout_seconds: 60, max_retries: 3, retry_backoff_seconds: 1.5, result_ttl_seconds: 300 }, http_client: { max_connections: 100, max_keepalive_connections: 40, keepalive_expiry_seconds: 30, connect_timeout_seconds: 5, read_timeout_seconds: 55 }, observability: { log_level: INFO, metrics_enabled: true, slow_task_threshold_ms: 3000 } }max_concurrency是队列同时处理的任务数不是 FastAPI 的 worker 数两者要分开理解。http_client里的连接池参数决定了对下游通道的复用能力max_connections要大于等于max_concurrency否则任务会卡在等连接上。再看config.toml把通道和模型定义集中管理[provider.taotoken] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY default_timeout 55 max_retries 3 [models.fast] name gpt-4o-mini provider taotoken temperature 0.3 max_tokens 1024 [models.reasoning] name claude-3-5-sonnet provider taotoken temperature 0.2 max_tokens 2048 [queue.routing] summarize fast analyze reasoningapi_key_env指向环境变量名而不是明文 Key启动时读取。queue.routing把任务类型映射到模型业务代码只传任务名不关心具体模型。这样切模型只改一行配置。加载配置的代码大致这样import json import os import tomllib from functools import lru_cache lru_cache def load_settings(path: str settings.json) - dict: with open(path, r, encodingutf-8) as f: return json.load(f) lru_cache def load_config(path: str config.toml) - dict: with open(path, rb) as f: return tomllib.load(f) def get_api_key(cfg: dict) - str: env_name cfg[provider][taotoken][api_key_env] key os.environ.get(env_name) if not key: raise RuntimeError(fmissing env: {env_name}) return keytomllib是 Python 3.11 起内置的低版本用tomli替代。配置加载用lru_cache缓存避免每个请求重复读盘。4. 异步任务队列与并发参数落地队列选型上轻量场景用arq或taskiq就够它们原生异步、和 FastAPI 同事件循环模型契合。Celery 更重但生态成熟如果已有 Redis/RabbitMQ 基础设施可以沿用。这里以异步客户端加 Redis 队列的思路展开核心是把大模型调用封装成可并发执行的任务。先建一个复用的 HTTP 客户端这是并发优化的关键。每次请求新建客户端会反复握手连接池形同虚设import httpx from settings import load_settings, load_config, get_api_key settings load_settings() cfg load_config() _client: httpx.AsyncClient | None None def get_client() - httpx.AsyncClient: global _client if _client is None: hc settings[http_client] _client httpx.AsyncClient( base_urlcfg[provider][taotoken][base_url], headers{Authorization: fBearer {get_api_key(cfg)}}, limitshttpx.Limits( max_connectionshc[max_connections], max_keepalive_connectionshc[max_keepalive_connections], keepalive_expiryhc[keepalive_expiry_seconds], ), timeouthttpx.Timeout( connecthc[connect_timeout_seconds], readhc[read_timeout_seconds], ), ) return _client客户端全局单例在 FastAPI 的lifespan里关闭from contextlib import asynccontextmanager from fastapi import FastAPI asynccontextmanager async def lifespan(app: FastAPI): yield client get_client() await client.aclose() app FastAPI(lifespanlifespan)任务函数本身要控制并发用信号量限制同时在飞的大模型请求数避免把下游打爆import asyncio from settings import load_settings settings load_settings() _sem asyncio.Semaphore(settings[queue][max_concurrency]) async def call_model(task_type: str, prompt: str) - dict: model_key cfg[queue][routing].get(task_type, fast) model cfg[models][model_key] async with _sem: client get_client() resp await client.post( /v1/chat/completions, json{ model: model[name], messages: [{role: user, content: prompt}], temperature: model[temperature], max_tokens: model[max_tokens], }, ) resp.raise_for_status() return resp.json()信号量放在任务层而不是客户端层是因为连接池管的是 TCP 复用信号量管的是业务并发上限两者职责不同。max_concurrency设 32 意味着最多 32 个任务同时调用通道超出的排队等待不会无限堆积。接口层用BackgroundTasks或队列投递把任务异步化主链路只返回任务 IDfrom fastapi import BackgroundTasks from pydantic import BaseModel class TaskIn(BaseModel): task_type: str prompt: str app.post(/tasks) async def create_task(payload: TaskIn, bg: BackgroundTasks): task_id ftask-{uuid4().hex[:12]} bg.add_task(run_and_store, task_id, payload.task_type, payload.prompt) return {task_id: task_id, status: queued}run_and_store里做重试和结果落库重试次数读max_retries退避用retry_backoff_seconds做指数增长。这样请求响应时间稳定在毫秒级耗时的大模型调用在后台跑。5. 验证请求与预期指标配置写完必须压测验证否则并发参数就是拍脑袋。先起服务用uvicorn多 workerexport TAOTOKEN_API_KEY你的Key uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4单请求功能验证curl -X POST http://127.0.0.1:8000/tasks \ -H Content-Type: application/json \ -d {task_type:summarize,prompt:用三句话说明异步任务队列的价值}预期返回task_id和queued状态响应时间应在 50ms 以内因为主链路没等模型返回。压测用wrk或locust重点看三个指标接口 P99、任务完成率、通道错误率。下面用wrk打接口层wrk -t8 -c200 -d60s -s post.lua http://127.0.0.1:8000/taskspost.lua里构造 JSON body。实测下来接口层 P99 能稳定在 100ms 内因为请求只做入队。真正要盯的是后台任务的完成情况用日志统计grep task_completed app.log | wc -l grep task_failed app.log | wc -l预期指标参考max_concurrency32时单实例每分钟能完成约 800 到 1200 个短任务取决于模型延迟通道错误率低于 1%任务平均耗时和模型本身延迟接近说明队列没有额外拖慢。如果完成率明显偏低先看是不是max_connections小于max_concurrency导致等连接。慢任务要单独标记slow_task_threshold_ms设 3000超过就记一条 warn 日志方便定位是哪个模型或哪类 prompt 拖慢了整体。6. 本篇常见错排查事件循环被同步调用堵死。症状是并发上不去、CPU 不高但延迟高。检查代码里有没有requests、同步数据库驱动、time.sleep。全部换成httpx、asyncpg、asyncio.sleep。这是最常见的坑。连接池参数不匹配。max_connections小于max_concurrency时任务会卡在获取连接上表现为队列积压但通道没压力。把max_connections设成max_concurrency的 1.5 到 2 倍留余量。Key 读取失败。api_key_env指向的环境变量没导出或者多 worker 下某个进程没读到。启动前统一export容器里用 secret 注入别写进镜像。重试放大流量。max_retries3加上高并发失败时会瞬间放大 3 倍请求。退避时间要够retry_backoff_seconds建议 1.5 起并对 4xx 类错误不重试只重试超时和 5xx。任务结果丢失。result_ttl_seconds太短前端还没取结果就过期了。按业务最长处理时间设一般 300 秒起步。多 worker 下信号量失效。asyncio.Semaphore是进程内的4 个 worker 就是 4 份独立信号量实际并发是max_concurrency × workers。要全局限制就得用 Redis 做分布式信号量或者把max_concurrency按 worker 数除一下。排障时优先看接入文档里的错误码说明对照日志定位是鉴权、限流还是超时https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentfastapi_concurrencyutm_campaignrewrite7. 长期编码与 Agent 场景的接入选择如果你的 FastAPI 服务后面要接 Coding Agent、批量代码生成这类长期跑的任务单次调用延迟不是重点稳定性和额度管理才是。这种场景建议用 Coding Plan它更适合持续性的编码类调用配合前面的队列方案能把长任务和短任务分开调度https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentfastapi_concurrencyutm_campaignrewrite回到工程本身并发优化的核心不是把参数调大而是让每一层职责清晰FastAPI 管请求接入队列管任务调度信号量管业务并发连接池管 TCP 复用TaoToken 管多模型鉴权收敛。任何一层越界都会变成瓶颈。我踩过的坑基本都集中在把同步调用混进异步链路以及连接池和并发数没对齐这两点上。把settings.json和config.toml拆开、把 Key 收进环境变量、把模型路由做成配置后面无论加模型还是调并发改动面都很小。压测数据别只看接口层后台任务的完成率和错误率才是真实水位。
