让 FastAPI Agent 思考不阻塞：手把手教你实现异步任务与后台处理方案

发布时间：2026/7/8 3:57:05

- 解决长耗时 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(f"task_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(f"task_progress_{task_id}") try: while True: # 持续监听 Redis 的消息 message = await pubsub.get_message(ignore_subscribe_init=True) 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)