Qwen2.5异步推理部署：Celery任务队列整合案例
1. 引言
1.1 业务场景描述
在当前大模型应用快速落地的背景下，通义千问系列模型（Qwen）凭借其强大的语言理解与生成能力，广泛应用于智能客服、内容创作、代码辅助等高并发场景。然而，随着用户请求量的增长，传统的同步推理服务面临响应延迟高、资源利用率不均等问题。
以Qwen2.5-7B-Instruct模型为例，该模型参数规模达76亿，在NVIDIA RTX 4090 D（24GB显存）上单次推理平均耗时约8-12秒。若采用同步调用方式，多个并发请求将导致线程阻塞，严重影响用户体验。
因此，构建一个支持异步处理、可扩展性强的推理服务架构成为工程实践中的关键需求。
1.2 痛点分析
现有基于Gradio或直接Flask/FastAPI的同步部署方案存在以下问题：
- 阻塞性强：每个请求需等待前一个完成，无法应对突发流量。
- 资源浪费：GPU长时间处于空闲状态，而CPU和内存持续被占用。
- 缺乏任务管理机制：无法实现任务排队、重试、超时控制等功能。
- 难以监控与调试：日志分散，任务执行状态不可追踪。
为解决上述问题，本文提出一种基于Celery 分布式任务队列的异步推理部署方案，结合 Redis 作为消息中间件，实现对 Qwen2.5-7B-Instruct 模型的安全、高效、可扩展调用。
1.3 方案预告
本文将详细介绍如何将 Qwen2.5-7B-Instruct 模型从同步服务改造为异步任务系统，涵盖以下核心内容：
- Celery + Redis 架构设计
- 模型加载与推理封装
- 异步任务定义与调用流程
- 前端接口集成与结果轮询机制
- 性能优化与错误处理策略
最终实现一个稳定、低延迟、高可用的大模型异步推理平台。
2. 技术方案选型
2.1 为什么选择 Celery？
| 对比项 | 同步服务 | 多线程/协程 | Celery |
|---|---|---|---|
| 并发能力 | 差 | 中等 | 高 |
| 可靠性 | 无持久化 | 进程崩溃即丢失 | 支持任务持久化 |
| 扩展性 | 单节点 | 有限 | 支持多worker横向扩展 |
| 错误恢复 | 不可恢复 | 局部恢复 | 支持重试机制 |
| 监控能力 | 弱 | 弱 | 支持 Flower 等可视化工具 |
Celery 具备以下优势：
- 解耦前后端逻辑：Web 接口仅负责接收请求并返回任务ID，推理由独立 worker 执行。
- 支持多种 Broker：Redis、RabbitMQ 等均可作为消息队列。
- 灵活的任务调度：支持定时、延迟、周期性任务。
- 容错能力强：任务失败可自动重试，支持异常捕获与日志记录。
2.2 整体架构设计
+------------------+ +-------------------+ | Web Server | | Celery Worker | | (FastAPI/Flask) |<--->| (Model Inference) | +------------------+ +-------------------+ | | v v +------------------+ +-------------------+ | Redis Broker | | GPU Resource | | (Task Queue) | | (RTX 4090 D) | +------------------+ +-------------------+工作流程如下：
- 用户通过 HTTP 请求提交 prompt；
- Web 服务将其封装为 Celery 任务，放入 Redis 队列；
- Worker 从队列中取出任务，加载模型或复用已加载实例进行推理；
- 推理完成后将结果写回 Redis 或数据库；
- 前端通过任务 ID 轮询获取结果。
3. 实现步骤详解
3.1 环境准备
确保已安装以下依赖：
pip install celery redis fastapi uvicorn[standard] transformers torch gradio启动 Redis 服务（默认端口 6379）：
redis-server --daemonize yes3.2 模型加载与推理封装
创建inference.py封装模型初始化与推理逻辑：
# inference.py from transformers import AutoModelForCausalLM, AutoTokenizer import torch _model = None _tokenizer = None def load_model(): global _model, _tokenizer if _model is None: model_path = "/Qwen2.5-7B-Instruct" _tokenizer = AutoTokenizer.from_pretrained(model_path) _model = AutoModelForCausalLM.from_pretrained( model_path, device_map="auto", torch_dtype=torch.float16 ) return _model, _tokenizer def generate_response(prompt: str, max_new_tokens: int = 512) -> str: model, tokenizer = load_model() inputs = tokenizer(prompt, return_tensors="pt").to(model.device) with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=max_new_tokens, do_sample=True, temperature=0.7, top_p=0.9 ) response = tokenizer.decode( outputs[0][len(inputs.input_ids[0]):], skip_special_tokens=True ) return response注意：使用全局变量缓存模型，避免每次任务重复加载。
3.3 Celery 任务定义
创建celery_app.py初始化 Celery 实例：
# celery_app.py from celery import Celery from .inference import generate_response app = Celery( 'qwen_tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/1' ) @app.task(bind=True, max_retries=3, default_retry_delay=30) def async_qwen_inference(self, messages, max_new_tokens=512): try: # 使用 apply_chat_template 构造输入 from transformers import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained("/Qwen2.5-7B-Instruct") prompt = tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) result = generate_response(prompt, max_new_tokens) return {"status": "success", "response": result} except Exception as exc: raise self.retry(exc=exc)3.4 Web 接口服务（FastAPI）
创建app.py提供 RESTful API：
# app.py from fastapi import FastAPI from celery.result import AsyncResult from .celery_app import async_qwen_inference app = FastAPI() @app.post("/v1/chat/completions") async def create_completion(data: dict): messages = data.get("messages", []) max_tokens = data.get("max_tokens", 512) task = async_qwen_inference.delay(messages, max_tokens) return {"task_id": task.id, "status": "submitted"} @app.get("/v1/tasks/{task_id}") async def get_task_status(task_id: str): task_result = AsyncResult(task_id, app=async_qwen_inference.app) if task_result.ready(): return {"status": "completed", "result": task_result.result} else: return {"status": "processing"}启动服务：
uvicorn app:app --host 0.0.0.0 --port 7860启动 Celery Worker：
celery -A celery_app worker -l info -c 1
-c 1表示只运行一个 worker，防止多进程竞争 GPU 资源。
3.5 前端调用示例
import requests import time # 提交任务 data = { "messages": [{"role": "user", "content": "请解释什么是深度学习？"}], "max_tokens": 512 } response = requests.post("http://localhost:7860/v1/chat/completions", json=data) task_id = response.json()["task_id"] # 轮询结果 while True: result = requests.get(f"http://localhost:7860/v1/tasks/{task_id}") result_data = result.json() if result_data["status"] == "completed": print("Response:", result_data["result"]["response"]) break else: print("Waiting for response...") time.sleep(2)4. 实践问题与优化
4.1 实际遇到的问题及解决方案
| 问题 | 原因 | 解决方案 |
|---|---|---|
| CUDA Out of Memory | 多个 worker 同时加载模型 | 限制 worker 数量为 1，共享模型实例 |
| 任务超时中断 | 默认超时时间过短 | 设置task_time_limit=600 |
| Redis 内存溢出 | 结果未及时清理 | 设置 backend TTL，定期清理已完成任务 |
| 模型加载慢 | 每次重启需重新加载 | 使用 preload 加载模型到 worker |
修改 Celery 启动命令以预加载模型：
celery -A celery_app worker -l info -c 1 --preload-module inference并在inference.py中添加：
if __name__ != "__main__": load_model() # 预加载4.2 性能优化建议
- 启用半精度推理：使用
torch.float16减少显存占用，提升推理速度。 - 批处理优化（Batching）：对于非实时场景，可收集多个请求合并推理（需自定义调度器）。
- 结果缓存机制：对常见问答对建立缓存，减少重复计算。
- 动态缩放 worker：根据负载自动启停 worker（需配合 Kubernetes 或 Docker Swarm）。
- 使用更高效的序列化格式：如
pickle protocol 5提升数据传输效率。
5. 总结
5.1 实践经验总结
通过本次 Qwen2.5-7B-Instruct 模型的异步部署实践，我们验证了 Celery 在大模型推理场景下的可行性与稳定性。主要收获包括：
- 成功将同步服务改造为异步任务系统，显著提升系统吞吐量；
- 利用 Redis 实现任务队列与结果存储，保障任务可靠性；
- 通过单 worker 控制 GPU 资源竞争，避免 OOM；
- 实现完整的任务生命周期管理：提交 → 执行 → 查询 → 完成。
同时，也发现了 Celery 在长耗时任务场景下的局限性，例如心跳检测可能导致误判超时，需合理配置broker_transport_options和task_acks_late。
5.2 最佳实践建议
- 生产环境务必使用 RabbitMQ 替代 Redis 作为 Broker：Redis 在高并发下可能出现消息丢失。
- 为任务设置合理的超时与重试策略：避免僵尸任务堆积。
- 结合 Prometheus + Grafana 监控任务队列长度与执行时间。
- 使用 Flower 可视化工具实时查看任务状态：
pip install flower celery -A celery_app flower
本方案不仅适用于 Qwen 系列模型，也可推广至 Llama、ChatGLM、Baichuan 等主流开源大模型的异步部署，具备良好的通用性和工程价值。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
