让 FastAPI Agent 思考不阻塞：手把手教你实现异步任务与后台处理方案
- 解决长耗时 Agent 任务导致的接口超时问题- 搞清楚什么时候用自带的任务什么时候上专业队列- 实现任务状态的实时查询和进度推送- 避开那些看似聪明实则坑人的并发陷阱️ 第一部分方案选择——杀鸡要用牛刀吗在 FastAPI 里处理后台任务主要有两种路子大白话比喻1. FastAPI BackgroundTasks轻便的小推车这是 FastAPI 内置的功能。它的原理是在 HTTP 响应返回后在同一个进程里顺便把任务做了。适用场景发个确认邮件、打个简单的日志。吐槽一句别拿它去跑重型 Agent因为它和你的 Web 服务抢 CPU任务一多你整个 API 都会变得卡顿。2. Celery / ARQ专业的运输车队这是把任务彻底发到另一个地方Worker去执行通常配合 Redis 或者 RabbitMQ。适用场景这种才是 Agent 的正解。不管 Agent 跑 1 分钟还是 1 小时你的 API 都是瞬间回复一个task_id。 第二部分实战演练——让 Agent “隐身”思考原理解释的差不多了咱们直接上干货。假设我们要实现一个“批量生成行业分析报告”的 Agent。1. 提交任务接口这里重点来了接口的任务不是“做报告”而是“接订单”。app.post(/analyze) async def start_analysis(topic: str): # 这里我们生成一个唯一ID task_id str(uuid.uuid4()) # 异步把任务塞进队列千万别在这里 await agent.run() analyze_task.delay(task_id, topic) return {status: accepted, task_id: task_id}2. 状态查询接口前端拿到task_id后可以每隔几秒来问问“好了吗好了吗”app.get(/task/{task_id}) async def get_status(task_id: str): result analyze_task.AsyncResult(task_id) return { task_id: task_id, status: result.status, # PENDING, SUCCESS, etc. result: result.result if result.ready() else None }经验之谈很多同学在这儿容易翻车记得在 Agent 运行过程中不断更新 Redis 里的“中间状态”。比如“正在搜索资料”、“正在起草文案”这样前端显示进度条时用户才不会觉得程序死了。⚠️ 第三部分避坑指南——那些我踩过的血泪坑接下来重点来了这都是反复Debug换来的教训 危险的 asyncio.create_task有些朋友喜欢在 FastAPI 里直接用这个函数来启动 Agent。看上去很美好但它是“内存寄生型”。如果你的服务器突然重启或者并发稍微高一点内存会瞬间爆炸而且任务丢了你都不知道在哪儿哭 数据库连接池耗尽Agent 在跑异步任务时如果每个步骤都去读写数据库记得一定要在任务结束时手动释放连接。不然等你 Agent 跑完你的 Web 服务也就挂了。 实时进度建议用 WebSocket如果你的 Agent 步骤非常多轮询接口Polling会给服务器增加很多没必要的压力。这时候可以考虑用 FastAPI 的 WebSocket 建立长连接让 Worker 主动把进度“推”给前端。这里多说一句你的 Agent 通常跑在 Celery 的 Worker 进程里而 WebSocket 连在 FastAPI 的进程里。Worker 里的 Agent 进度怎么传给 Web 进程千万别尝试在 Worker 里直接调用 WebSocket 的 send 方法那不在一个次元最稳妥的办法是用Redis 的 Pub/Sub发布/订阅机制。这里给个参考# 搞个“连接管理器”万一用户断网了、刷新页面了你得能及时把废弃的连接清理掉否则内存溢出分分钟的事 class ConnectionManager: def __init__(self): # 存放活跃连接key是task_id或者user_id self.active_connections: dict[str, WebSocket] {} async def connect(self, task_id: str, websocket: WebSocket): await websocket.accept() self.active_connections[task_id] websocket def disconnect(self, task_id: str): if task_id in self.active_connections: del self.active_connections[task_id] async def send_progress(self, task_id: str, message: dict): if task_id in self.active_connections: await self.active_connections[task_id].send_json(message) manager ConnectionManager() # 在 Celery Task 内部发布者 redis_client.publish(ftask_progress_{task_id}, json.dumps({step: 正在搜索资料, percent: 30})) # FastAPI 端订阅者 app.websocket(/ws/{task_id}) async def websocket_endpoint(websocket: WebSocket, task_id: str): await manager.connect(task_id, websocket) pubsub redis_client.pubsub() await pubsub.subscribe(ftask_progress_{task_id}) try: while True: # 持续监听 Redis 的消息 message await pubsub.get_message(ignore_subscribe_initTrue) if message: data json.loads(message[data]) await manager.send_progress(task_id, data) if data.get(percent) 100: # 任务完成关掉对讲机 break await asyncio.sleep(0.1) # 稍微喘口气别空转太猛 except WebSocketDisconnect: manager.disconnect(task_id) 进阶思考Agent 实例的并发管理你可能会问如果我有 100 个用户同时点“生成报告”我的 Worker 塞满了怎么办这时候你需要给任务加优先级队列。VIP 客户的 Agent 优先跑普通用户的慢慢排队。而且针对 LLM 的 API 限流Rate Limit你也需要在 Celery 这一层做控制别让你的 Agent 跑太快把 OpenAI 的
