Qwen2.5异步推理部署：Celery任务队列整合案例-开发者社区

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 yes

3.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 等主流开源大模型的异步部署，具备良好的通用性和工程价值。