- 解决长耗时 Agent 任务导致的接口超时问题

- 搞清楚什么时候用自带的任务，什么时候上专业队列

- 实现任务状态的实时查询和进度推送

- 避开那些看似聪明实则坑人的并发陷阱

🎯 危险的 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)