异步任务队列：Celery 处理 AI 长任务

AI 应用与普通 Web 服务有一个根本性差异：响应时间。调用 GPT-4 生成一份报告可能需要 30 秒，跑一次 RAG 检索加推理可能需要 10-20 秒，批量处理文档可能需要几分钟。而 HTTP 协议默认超时通常是 30 秒，Nginx 默认是 60 秒，大量请求会直接超时失败。

解决方案不是调大超时时间，而是改变架构：把耗时操作从 HTTP 请求链路中剥离出去，异步执行（不等任务完成，先返回"已收到"，任务在后台运行，完成后通知）。Celery 是 Python 生态中最成熟的异步任务队列（任务队列：把待处理的任务放入队列，由后台工作进程逐一取出执行），本章介绍如何用它处理 AI 长任务。

1.1 为什么 HTTP 超时是真实问题

Celery异步任务架构——API接收立即返回task_id，Worker异步执行LLM推理，结果存储后回调通知

考虑以下场景：用户上传一份 50 页的 PDF，要求 AI 生成摘要并提取关键信息。整个流程包括：解析 PDF、文本分块、批量 Embedding、存入向量数据库、调用 LLM 生成摘要。这个过程少则 30 秒，多则几分钟。

同步处理的问题不只是超时：

连接占用：每个慢请求都占用一个 worker 连接，高并发下很快耗尽
重试放大：客户端超时后重试，服务端又收到相同任务，资源双倍消耗
用户体验差：浏览器 loading 转圈两分钟，用户早已关掉页面
可见性为零：任务在哪个阶段、进度多少，完全不知道

异步队列的思路是：HTTP 请求只负责"提交任务"（毫秒级返回），实际处理在后台 Worker 中进行，客户端通过轮询或 WebSocket 获取结果。

1.2 Celery 核心概念

code

Client ──提交──► Broker ──分发──► Worker ──存结果──► Backend
                (Redis/MQ)        (真正执行)         (Redis/DB)

Broker（消息代理）：任务的中转站，Client 把任务投递进去，Worker 从这里取出来执行。常用 Redis 或 RabbitMQ（一个专业的开源消息队列系统，比 Redis 更注重消息可靠性和复杂路由）。Redis 配置简单，适合大多数场景；RabbitMQ 支持更复杂的路由规则，适合有多队列复杂调度需求的场景。

Worker（工作进程）：真正执行任务的进程，可以启动多个，支持横向扩展。每个 Worker 可以绑定特定队列，实现任务隔离。

Backend（结果存储）：Worker 执行完任务后，把结果存到 Backend，Client 通过 task_id 来查询。同样常用 Redis，也可以用数据库。

Task（任务）：Python 函数，加上 @celery.task 装饰器就变成可异步执行的任务。

1.3 安装与基础配置

bash

pip install celery redis fastapi uvicorn

项目结构：

code

ai_service/
├── main.py          # FastAPI 应用
├── celery_app.py    # Celery 配置
├── tasks.py         # 任务定义
└── config.py        # 配置项

python

# celery_app.py
from celery import Celery

# Broker 和 Backend 都指向 Redis
# 实际生产中应该是不同的 Redis 实例或不同的 DB 编号
celery_app = Celery(
    "ai_tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

celery_app.conf.update(
    # 任务序列化格式，json 比 pickle 更安全
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],

    # 结果保留 1 小时，避免 Redis 无限膨胀
    result_expires=3600,

    # Worker 每次预取 1 个任务
    # 对于 AI 长任务很重要：避免一个 Worker 囤积大量任务
    # 导致其他 Worker 空闲
    worker_prefetch_multiplier=1,

    # 每个 Worker 进程执行 10 个任务后重启
    # 防止 LLM SDK 的内存泄漏累积
    worker_max_tasks_per_child=10,
)

1.4 FastAPI + Celery 集成

核心交互流程：

python

# tasks.py
from celery_app import celery_app
from openai import OpenAI
import time

client = OpenAI()

@celery_app.task(
    bind=True,          # bind=True 让任务能访问 self，用于更新进度
    name="tasks.rag_answer",
    max_retries=3,      # AI API 调用失败时自动重试
    default_retry_delay=5,  # 重试间隔 5 秒
)
def rag_answer_task(self, question: str, doc_ids: list[str]) -> dict:
    """
    异步 RAG 问答任务。
    bind=True 的原因：需要通过 self.update_state() 上报中间进度，
    让前端知道任务在哪个阶段，而不是只能等最终结果。
    """
    try:
        # 阶段 1：检索相关文档
        # 用自定义状态 PROGRESS 而不是 STARTED，携带进度信息
        self.update_state(
            state="PROGRESS",
            meta={"stage": "retrieving", "progress": 20}
        )
        relevant_docs = retrieve_docs(question, doc_ids)

        # 阶段 2：构建 prompt
        self.update_state(
            state="PROGRESS",
            meta={"stage": "building_prompt", "progress": 40}
        )
        context = "\n\n".join([doc["content"] for doc in relevant_docs])
        prompt = f"基于以下内容回答问题：\n{context}\n\n问题：{question}"

        # 阶段 3：调用 LLM
        self.update_state(
            state="PROGRESS",
            meta={"stage": "generating", "progress": 60}
        )
        try:
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": prompt}],
                timeout=120,  # LLM 调用单独设置超时
            )
        except Exception as exc:
            # API 调用失败，触发 Celery 重试机制
            raise self.retry(exc=exc)

        answer = response.choices[0].message.content

        # 返回最终结果
        return {
            "answer": answer,
            "sources": [doc["id"] for doc in relevant_docs],
            "tokens_used": response.usage.total_tokens,
        }

    except Exception as exc:
        # 非可重试错误，直接标记失败
        self.update_state(
            state="FAILURE",
            meta={"error": str(exc), "stage": "failed"}
        )
        raise


def retrieve_docs(question: str, doc_ids: list[str]) -> list[dict]:
    """模拟向量检索，实际替换为你的向量数据库查询。"""
    time.sleep(2)  # 模拟检索耗时
    return [{"id": doc_id, "content": f"文档 {doc_id} 的相关内容..."} for doc_id in doc_ids[:3]]

python

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from celery.result import AsyncResult
from tasks import rag_answer_task

app = FastAPI()


class AnalyzeRequest(BaseModel):
    question: str
    doc_ids: list[str]


@app.post("/api/analyze", status_code=202)
async def submit_task(request: AnalyzeRequest):
    """
    提交任务接口。
    返回 202 Accepted 而非 200 OK，
    语义上表示"已收到请求，但尚未完成处理"。
    """
    task = rag_answer_task.delay(
        question=request.question,
        doc_ids=request.doc_ids,
    )
    return {"task_id": task.id, "status": "submitted"}


@app.get("/api/task/{task_id}")
async def get_task_result(task_id: str):
    """
    查询任务状态和结果。
    前端每隔 2-3 秒轮询一次，直到 status 变为 SUCCESS 或 FAILURE。
    """
    result = AsyncResult(task_id)

    if result.state == "PENDING":
        # 任务还在队列中等待，或者 task_id 不存在
        return {"status": "pending", "progress": 0}

    elif result.state == "PROGRESS":
        # 任务执行中，有进度信息
        return {
            "status": "running",
            "stage": result.info.get("stage"),
            "progress": result.info.get("progress", 0),
        }

    elif result.state == "SUCCESS":
        return {
            "status": "success",
            "result": result.result,
        }

    elif result.state == "FAILURE":
        # result.info 在失败时包含异常信息
        return {
            "status": "failed",
            "error": str(result.info),
        }

    return {"status": result.state}

1.5 任务优先级与队列隔离

不同类型的 AI 任务对响应速度的要求不同。实时对话需要快速响应，批量报告生成可以慢慢排队。把它们混在一个队列里，批量任务会拖慢实时任务。

python

# celery_app.py 中增加队列配置
from kombu import Queue

celery_app.conf.task_queues = (
    # 优先级高：实时问答，Worker 数量多
    Queue("high_priority", routing_key="high"),
    # 优先级低：批量处理，Worker 数量少，不占用核心资源
    Queue("low_priority", routing_key="low"),
    # 默认队列
    Queue("default", routing_key="default"),
)

celery_app.conf.task_default_queue = "default"

python

# tasks.py 中指定任务路由到对应队列
@celery_app.task(bind=True, name="tasks.realtime_chat", queue="high_priority")
def realtime_chat_task(self, message: str) -> str:
    """实时对话，走高优先级队列。"""
    pass


@celery_app.task(bind=True, name="tasks.batch_report", queue="low_priority")
def batch_report_task(self, doc_ids: list[str]) -> dict:
    """批量报告生成，走低优先级队列。"""
    pass

启动不同数量的 Worker 来处理不同队列：

bash

# 高优先级队列：4 个并发 Worker
celery -A celery_app worker --queues=high_priority --concurrency=4 --loglevel=info

# 低优先级队列：1 个并发 Worker，不占用核心资源
celery -A celery_app worker --queues=low_priority --concurrency=1 --loglevel=info

1.6 流式结果的特殊处理

LLM 支持流式输出（streaming），逐 token 返回。但 Celery 任务只有一个最终返回值，无法直接传递流式数据。常见解法是把每个 token 写入 Redis List，客户端通过 SSE 或 WebSocket 实时读取。

python

import redis
import json

redis_client = redis.Redis(host="localhost", port=6379, db=2)


@celery_app.task(bind=True, name="tasks.streaming_answer")
def streaming_answer_task(self, task_id: str, question: str) -> dict:
    """
    流式输出任务：每个 token 写入 Redis，
    FastAPI SSE 接口实时从 Redis 读取并推送给前端。
    """
    stream_key = f"stream:{task_id}"

    full_answer = ""
    for chunk in client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": question}],
        stream=True,
    ):
        if chunk.choices[0].delta.content:
            token = chunk.choices[0].delta.content
            full_answer += token
            # 把每个 token 推入 Redis List，设置 TTL 防止泄漏
            redis_client.rpush(stream_key, json.dumps({"token": token}))
            redis_client.expire(stream_key, 300)  # 5 分钟后自动清理

    # 写入结束标志
    redis_client.rpush(stream_key, json.dumps({"done": True}))
    return {"answer": full_answer}

python

# main.py 中的 SSE 接口，从 Redis 读取流式 token 推送给客户端
from fastapi.responses import StreamingResponse
import asyncio


@app.get("/api/stream/{task_id}")
async def stream_result(task_id: str):
    """SSE 接口，实时推送 AI 生成的 token。"""
    stream_key = f"stream:{task_id}"

    async def event_generator():
        index = 0
        while True:
            # 非阻塞读取 Redis List 中新增的 token
            items = redis_client.lrange(stream_key, index, -1)
            if items:
                for item in items:
                    data = json.loads(item)
                    if data.get("done"):
                        yield "data: [DONE]\n\n"
                        return
                    yield f"data: {item.decode()}\n\n"
                index += len(items)
            else:
                await asyncio.sleep(0.05)  # 50ms 轮询间隔

    return StreamingResponse(event_generator(), media_type="text/event-stream")

1.7 Flower：任务监控面板

Flower 是 Celery 的 Web 监控界面，提供任务状态、Worker 状态、队列积压等实时视图。

bash

pip install flower

# 启动 Flower，监听 5555 端口
celery -A celery_app flower --port=5555

访问 http://localhost:5555，可以看到：

活跃 Worker 列表及负载
每个任务的状态、耗时、参数
队列中等待执行的任务数量
失败任务的错误信息和可重试操作

生产环境建议给 Flower 加上认证：

bash

celery -A celery_app flower --port=5555 --basic_auth=admin:your_password

1.8 各方案对比

方案	适合场景	复杂度	可扩展性
同步 HTTP（直接调用）	任务 < 5s，并发低	低	差
BackgroundTasks（FastAPI 内置）	任务 < 30s，不需要结果	低	差（无法跨进程）
Celery + Redis	长任务，需要结果和进度	中	好
Celery + RabbitMQ	复杂路由，消息可靠性要求高	高	很好
云托管队列（SQS/Pub/Sub）	无运维能力，云原生部署	中	很好

1.9 小结

异步任务队列是 AI 应用生产化的必经之路。核心思路是：HTTP 接口只负责"接单"，Worker 在后台"干活"，客户端通过 task_id 查进度取结果。

Celery 的关键配置要点：worker_prefetch_multiplier=1 防止长任务被少数 Worker 囤积；worker_max_tasks_per_child 防止内存泄漏；队列隔离保证高优先级任务不被低优先级任务阻塞。

下一步可以关注 Celery Beat（定时任务，适合定期重建索引、清理过期数据）以及如何用 Kubernetes 对 Worker 进行弹性伸缩，根据队列积压量自动增减 Worker 数量。