Qwen2.5-72B-GPTQ-Int4代码实例：vLLM异步API+Chainlit流式响应实现

1. 引言：当大模型遇上高效部署与交互

想象一下，你手头有一个拥有720亿参数的“巨无霸”语言模型——Qwen2.5-72B。它知识渊博，能写代码、解数学题、处理长文档，还支持多国语言。但问题来了：这么大的模型，怎么才能让它快速响应你的问题，并且让你能像聊天一样和它自然交互呢？

这就是我们今天要解决的问题。单纯把模型跑起来只是第一步，关键在于如何让它“跑得快”且“聊得爽”。本文将带你一步步实现一个高性能的解决方案：使用 vLLM 来部署经过 GPTQ-Int4 量化压缩的 Qwen2.5-72B-Instruct 模型，并通过 Chainlit 构建一个支持流式响应的前端聊天界面。

你将学到什么？

• 理解为什么选择 vLLM 和 GPTQ-Int4 这套组合拳。
• 掌握如何编写异步API服务，让大模型推理不阻塞。
• 实现一个美观、实时的前端聊天应用，看着答案一个字一个字“流”出来。
• 获得一套完整的、可运行的代码实例，直接部署使用。

无论你是想快速搭建一个内部知识问答工具，还是探索大模型的高效服务化方案，这篇教程都能给你清晰的路径和可落地的代码。

2. 技术选型：为什么是它们？

在开始动手之前，我们先花几分钟了解一下核心组件，明白它们各自解决了什么问题，这样后面的操作会更有方向感。

2.1 模型核心：Qwen2.5-72B-Instruct-GPTQ-Int4

Qwen2.5-72B-Instruct 是通义千问系列的最新大模型，能力非常全面：

• 规模巨大：720亿参数，在编程、数学、长文本理解、多语言支持上表现突出。
• 指令精调：专门针对理解和遵循人类指令进行了优化，对话体验更好。
• 超长上下文：支持长达128K tokens的输入，能处理整本书或长篇报告。

但72B的原始模型对显存要求极高（约140GB+），普通机器根本无法加载。这时就需要 GPTQ-Int4量化。

• GPTQ 是一种前沿的模型压缩技术，能在几乎不损失精度的情况下，将模型权重从高精度（如FP16）压缩到低精度（如INT4）。
• Int4 意味着每个权重参数只用4比特存储，相比原始的16比特（FP16），显存占用直接减少到约1/4。这使得72B模型在单张或多张高性能消费级显卡（如RTX 4090*2）上运行成为可能。

简单说，Qwen2.5-72B-Instruct-GPTQ-Int4 就是那个“能力又强、身材又苗条”的选手。

2.2 推理引擎：vLLM

vLLM 是一个专为大规模语言模型设计的高吞吐量、低延迟推理和服务引擎。它的核心优势在于：

• PagedAttention：这是vLLM的“杀手锏”。它像操作系统管理内存一样管理Attention的KV Cache，极大减少了内存碎片，使得同时处理多个用户请求（批处理）时效率飙升。
• 高性能：相比原始的 Hugging Face Transformers 推理，vLLM 通常能带来数倍甚至数十倍的吞吐量提升。
• 易于部署：它提供了简单易用的Python API和OpenAI兼容的API服务器，让我们可以轻松地将模型封装成服务。

选择vLLM，就是为了让我们的量化大模型“跑得更快、服务得更稳”。

2.3 交互前端：Chainlit

Chainlit 是一个专门为构建大语言模型应用而设计的开源框架，可以理解为“LLM应用的Streamlit”。

• 快速构建：用很少的代码就能创建一个功能丰富的聊天界面。
• 流式响应：原生支持流式输出，答案可以逐字逐句实时显示，用户体验极佳。
• 会话管理：自动处理聊天历史、文件上传等常见功能。
• 可定制性强：UI元素、回调函数等都支持深度定制。

选择Chainlit，就是为了打造一个“开箱即用、体验流畅”的聊天前端。

技术栈总结：我们用 vLLM 驱动 量化后的Qwen2.5大模型，并通过 Chainlit 提供友好的交互界面。接下来，我们进入实战环节。

3. 环境准备与模型部署

假设你已经通过CSDN星图镜像或其他方式，获得了预装好基础环境（Python, CUDA等）和Qwen2.5-72B-GPTQ-Int4模型的服务器环境。我们直接从验证和编写代码开始。

3.1 验证模型服务

首先，我们需要确认vLLM服务是否已经正常启动。通常，部署日志会记录在特定文件。

# 查看模型服务日志，确认加载成功
cat /root/workspace/llm.log

如果看到类似下面的输出，表明模型已成功加载至GPU，vLLM引擎准备就绪：

INFO 07-28 10:30:15 llm_engine.py:73] Initializing an LLM engine (vLLM version 0.3.3)...
INFO 07-28 10:30:15 llm_engine.py:74] Engine args: model='/path/to/Qwen2.5-72B-Instruct-GPTQ-Int4', ...
INFO 07-28 10:31:45 model_runner.py:181] CUDA device: NVIDIA GeForce RTX 4090 (sm_89)
INFO 07-28 10:32:20 llm_engine.py:199] # GPU blocks: 1245, # CPU blocks: 512
INFO 07-28 10:32:21 llm_engine.py:212] KV cache usage: 0.0%
INFO 07-28 10:32:21 llm_engine.py:213] Available: 1245 GPU blocks (100.0%)
INFO 07-28 10:32:21 async_llm_engine.py:69] Initialized LLM engine.

关键信息是看到 Initialized LLM engine 且没有报错。此时，模型已经在后台运行，等待我们的API调用。

3.2 安装必要的Python库

我们需要安装vLLM和Chainlit。建议创建一个干净的Python虚拟环境。

# 创建并激活虚拟环境（可选，但推荐）
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate  # Windows

# 安装核心库
pip install vllm chainlit

如果你的环境需要特定版本的vLLM以兼容GPTQ模型，请参考官方文档安装。

4. 核心代码实现：异步API与流式响应

一切就绪，现在我们来编写核心的Python代码。我们将创建两个主要文件：

1. api_server.py：基于vLLM的异步API服务。
2. app.py：Chainlit前端应用，调用后端API。

4.1 第一步：构建异步API服务器 (api_server.py)

这个文件负责加载模型并提供一个HTTP接口，接收问题，返回流式的模型响应。

# api_server.py
from vllm import AsyncLLMEngine, AsyncEngineArgs, SamplingParams
from vllm.utils import random_uuid
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import uvicorn
import json

# 1. 定义模型路径和参数
MODEL_PATH = "/root/workspace/Qwen2.5-72B-Instruct-GPTQ-Int4" # 请替换为你的实际模型路径
MAX_MODEL_LEN = 8192  # 模型最大生成长度
GPU_MEMORY_UTILIZATION = 0.9  # GPU显存使用率

# 2. 初始化异步LLM引擎
engine_args = AsyncEngineArgs(
    model=MODEL_PATH,
    max_model_len=MAX_MODEL_LEN,
    gpu_memory_utilization=GPU_MEMORY_UTILIZATION,
    tensor_parallel_size=2,  # 如果使用多张GPU，例如2张，请设置为2
    quantization="gptq",  # 指定使用GPTQ量化
    trust_remote_code=True,  # 信任远程代码（对于Qwen模型是必须的）
    enforce_eager=True,  # 对于某些量化模型，可能需要启用eager模式
)

# 创建异步引擎
llm_engine = AsyncLLMEngine.from_engine_args(engine_args)

# 3. 创建FastAPI应用
app = FastAPI(title="Qwen2.5-72B GPTQ API Server")

@app.get("/")
async def root():
    return {"message": "Qwen2.5-72B GPTQ API Server is running!"}

@app.post("/generate_stream")
async def generate_stream(request: Request):
    """流式生成文本的API端点"""
    try:
        # 解析请求数据
        data = await request.json()
        prompt = data.get("prompt", "")
        stream = data.get("stream", True)
        max_tokens = data.get("max_tokens", 512)
        temperature = data.get("temperature", 0.7)
        top_p = data.get("top_p", 0.9)

        if not prompt:
            return {"error": "Prompt is required"}

        # 设置生成参数
        sampling_params = SamplingParams(
            temperature=temperature,
            top_p=top_p,
            max_tokens=max_tokens,
            stream=stream,  # 启用流式输出
        )

        # 生成唯一的请求ID
        request_id = random_uuid()

        # 准备异步生成器
        async def stream_generator():
            # 发起异步生成请求
            results_generator = llm_engine.generate(
                prompt, sampling_params, request_id
            )

            # 流式输出结果
            async for output in results_generator:
                if output.finished:
                    # 生成结束，发送结束标记
                    yield f"data: [DONE]\n\n"
                    break
                # 获取最新生成的文本片段
                text = output.outputs[0].text
                # 以SSE (Server-Sent Events) 格式发送
                yield f"data: {json.dumps({'text': text})}\n\n"

        # 返回流式响应
        return StreamingResponse(
            stream_generator(),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no",  # 禁用Nginx缓冲
            }
        )

    except Exception as e:
        return {"error": f"Generation failed: {str(e)}"}

if __name__ == "__main__":
    # 启动服务器，监听所有网络接口的8000端口
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

代码关键点解释：

1. AsyncLLMEngine：使用vLLM的异步引擎，这是实现高并发流式响应的核心。
2. SamplingParams：控制模型生成行为的参数，如temperature（创造性）、top_p（核采样）等。
3. StreamingResponse：FastAPI的流式响应类，我们用它来逐段发送生成的文本。
4. SSE (Server-Sent Events)：我们采用SSE协议向前端推送数据，这是一种简单的HTTP流式传输标准。
5. [DONE]标记：我们约定当收到[DONE]时，表示流式传输结束。

4.2 第二步：启动API服务器

在终端运行以下命令启动API服务：

python api_server.py

如果一切正常，你会看到服务器启动日志，并监听在 http://0.0.0.0:8000。你可以用curl测试一下：

curl -X POST http://localhost:8000/generate_stream \
  -H "Content-Type: application/json" \
  -d '{"prompt": "你好，请介绍一下你自己。", "stream": true}' \
  --no-buffer

你应该能看到文本流式地返回。API服务准备就绪，接下来我们构建前端。

4.3 第三步：创建Chainlit前端应用 (app.py)

这个文件创建聊天界面，并调用我们刚启动的API。

# app.py
import chainlit as cl
import aiohttp
import json
import asyncio
from typing import Optional

# 配置后端API地址
API_BASE_URL = "http://localhost:8000"  # 如果API服务在别的机器，请修改此地址

async def query_llm_stream(prompt: str, session: aiohttp.ClientSession) -> str:
    """异步调用后端流式API，并收集完整响应"""
    full_response = ""
    url = f"{API_BASE_URL}/generate_stream"
    
    payload = {
        "prompt": prompt,
        "stream": True,
        "max_tokens": 1024,
        "temperature": 0.7,
        "top_p": 0.9,
    }
    
    try:
        async with session.post(url, json=payload, timeout=60) as response:
            if response.status != 200:
                error_text = await response.text()
                return f"API请求错误: {response.status}, {error_text}"
            
            # 处理流式响应
            buffer = ""
            async for chunk in response.content:
                if chunk:
                    chunk_str = chunk.decode('utf-8')
                    buffer += chunk_str
                    
                    # 处理可能的多条消息
                    while "\n\n" in buffer:
                        line, buffer = buffer.split("\n\n", 1)
                        if line.startswith("data: "):
                            data_str = line[6:]  # 去掉 "data: " 前缀
                            if data_str == "[DONE]":
                                return full_response
                            try:
                                data = json.loads(data_str)
                                token = data.get("text", "")
                                if token:
                                    full_response += token
                                    # 实时流式发送到前端
                                    await cl.Message(content=token).send()
                            except json.JSONDecodeError:
                                continue
    except asyncio.TimeoutError:
        return full_response + "\n\n【请求超时，响应可能不完整】"
    except Exception as e:
        return f"请求过程中发生错误: {str(e)}"
    
    return full_response

@cl.on_chat_start
async def on_chat_start():
    """聊天开始时的初始化"""
    # 初始化一个aiohttp会话，保持连接复用
    cl.user_session.set("http_session", aiohttp.ClientSession())
    # 发送欢迎消息
    await cl.Message(
        content="你好！我是基于 Qwen2.5-72B-GPTQ 模型驱动的助手。我可以帮你解答问题、编写代码、进行创意写作等。请问有什么可以帮你的？"
    ).send()

@cl.on_message
async def on_message(message: cl.Message):
    """处理用户消息"""
    user_input = message.content
    
    # 获取或创建HTTP会话
    session = cl.user_session.get("http_session")
    if session is None:
        session = aiohttp.ClientSession()
        cl.user_session.set("http_session", session)
    
    # 发送一个初始消息来占位，后续流式内容会追加到其中
    msg = cl.Message(content="")
    await msg.send()
    
    # 调用后端API获取流式响应
    full_response = await query_llm_stream(user_input, session)
    
    # 如果流式发送过程中出现问题，确保最终消息完整
    if msg.content != full_response:
        await cl.Message(content=full_response).send()

@cl.on_chat_end
async def on_chat_end():
    """聊天结束时清理资源"""
    session = cl.user_session.get("http_session")
    if session and not session.closed:
        await session.close()

# Chainlit应用配置
if __name__ == "__main__":
    # 可以在这里添加更多Chainlit配置
    cl.run()

代码关键点解释：

1. query_llm_stream 函数：这是核心函数，它使用aiohttp异步地调用我们的后端API，并处理SSE流式数据。每收到一个文本片段，就通过cl.Message(content=token).send()实时发送到前端界面。
2. @cl.on_message 装饰器：这是Chainlit的核心，它定义了当用户发送消息时的处理逻辑。
3. 会话管理：我们使用cl.user_session来管理HTTP会话，避免为每个请求都创建新连接，提升性能。
4. 错误处理：包含了超时、网络错误等基本处理，保证应用健壮性。

4.4 第四步：配置Chainlit (chainlit.md)

创建一个名为 chainlit.md 的文件，作为聊天界面的说明和配置。

# 欢迎使用 Qwen2.5-72B 智能助手

这是一个基于 **Qwen2.5-72B-Instruct-GPTQ-Int4** 大模型构建的对话应用。

## 功能特点
- 🚀 **高性能**：采用vLLM推理引擎，响应迅速。
- 💬 **流式交互**：答案逐字显示，体验流畅。
- 🧠 **强大能力**：支持复杂推理、代码编写、长文本分析等。
- ⚙️ **量化模型**：使用GPTQ-Int4量化，高效利用显存。

## 使用提示
- 你可以问我任何问题，我会尽力提供准确、详细的回答。
- 支持多轮对话，上下文长度可达128K。
- 如需生成代码，请明确说明编程语言和需求。

## 模型信息
- **基础模型**: Qwen2.5-72B-Instruct
- **量化方式**: GPTQ-Int4
- **服务引擎**: vLLM
- **前端框架**: Chainlit

开始聊天吧！

5. 运行与测试：体验流式对话

现在，让我们把整个应用跑起来。

5.1 启动应用

打开两个终端窗口。

终端1：确保API服务器在运行

cd /你的项目路径
python api_server.py

终端2：启动Chainlit前端

cd /你的项目路径
chainlit run app.py

Chainlit会自动在默认浏览器打开一个页面（通常是 http://localhost:8000）。如果没自动打开，你可以手动访问终端中显示的地址。

5.2 进行对话测试

在打开的Chainlit界面中，你会看到我们设置的欢迎语。现在，尝试问一些问题：

1. 简单问候：“你好，请介绍一下你自己。”
2. 代码生成：“用Python写一个快速排序函数，并加上注释。”
3. 逻辑推理：“如果昨天是明天的话就好了，这样今天就是周五了。请问实际的今天是星期几？”
4. 长文本总结（可以输入一段长文本让它总结）。

观察回答的显示方式：答案应该是一个词一个词或一句话一句话地“流”出来，而不是等待全部生成完再一次性显示。这就是流式响应的魅力，它能极大提升用户等待时的体验。

5.3 验证部署成功

如果一切顺利，你的聊天界面应该类似下图，可以流畅地进行问答： （此处可描述：界面左侧为对话历史，右侧主区域显示用户问题和模型流式生成的回答，回答逐字出现。）

6. 进阶配置与优化建议

基础功能已经实现，但要让这个服务更健壮、更好用，还可以考虑以下几点：

6.1 API服务器增强

当前的API服务器比较简单，生产环境需要考虑更多：

# 可以在api_server.py的FastAPI app中添加以下功能

from fastapi.middleware.cors import CORSMiddleware

# 添加CORS支持，允许前端跨域访问
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境应指定具体前端地址
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 添加健康检查端点
@app.get("/health")
async def health_check():
    return {"status": "healthy", "model": "Qwen2.5-72B-GPTQ"}

# 添加批量推理端点（非流式）
@app.post("/generate")
async def generate_batch(request: Request):
    data = await request.json()
    # ... 类似逻辑，但使用stream=False
    # 返回完整结果

6.2 前端体验优化

在 app.py 中，可以增强用户体验：

# 1. 添加消息历史管理
@cl.on_message
async def on_message(message: cl.Message):
    # 获取历史消息
    history = cl.user_session.get("message_history", [])
    history.append({"role": "user", "content": message.content})
    
    # 构建包含历史的prompt（简单示例，实际需按模型格式）
    context = "\n".join([f"{h['role']}: {h['content']}" for h in history[-6:]])  # 保留最近6轮
    full_prompt = f"{context}\nassistant:"
    
    # 使用full_prompt调用API...
    
    # 保存助手回复到历史
    history.append({"role": "assistant", "content": full_response})
    cl.user_session.set("message_history", history)

# 2. 添加文件上传处理
@cl.on_file_upload
async def on_file_upload(files: List[cl.File]):
    # 处理用户上传的文件（如txt, pdf, docx）
    for file in files:
        # 读取文件内容，提取文本
        # 将文本内容作为上下文的一部分发送给模型
        pass

6.3 性能与稳定性

1. 调整vLLM参数：根据你的GPU显存大小，调整 gpu_memory_utilization 和 tensor_parallel_size。
2. 设置超时与重试：在前端代码中为API调用设置合理的超时时间，并添加重试逻辑。
3. 使用反向代理：生产环境建议使用Nginx等反向代理服务器，处理SSL、负载均衡和静态文件。
4. 监控与日志：添加详细的日志记录，监控API的响应时间、错误率等指标。

7. 总结

通过本教程，我们完成了一个从模型部署到交互前端的完整链路：

1. 模型准备：利用了 Qwen2.5-72B-Instruct-GPTQ-Int4 这个强大的量化模型，在保持高性能的同时大幅降低了部署门槛。
2. 高效推理：使用 vLLM 作为推理引擎，其PagedAttention技术确保了高并发下的吞吐量和低延迟。
3. 流式服务：通过 FastAPI 构建了支持SSE流式传输的异步API，让模型可以“边想边说”。
4. 友好交互：借助 Chainlit 快速搭建了一个美观、支持流式响应的聊天界面，提供了丝滑的用户体验。

这套组合拳的核心优势在于平衡了性能、效率和体验。GPTQ量化解决了大模型“装不下”的问题，vLLM解决了“跑得慢”的问题，而异步流式API和Chainlit则解决了“等得急”和“不好用”的问题。

你可以在此基础上继续探索：

• 集成RAG（检索增强生成），让模型能够基于你的私有知识库回答问题。
• 添加Function Calling功能，让模型可以调用外部工具（如搜索、计算器）。
• 部署到云服务器，通过公网提供稳定的服务。
• 尝试其他前端框架，如Gradio、Streamlit，或自己开发Web界面。

希望这个完整的代码实例能成为你探索大模型应用的一个坚实起点。从“跑通”到“用好”，还有更多有趣的可能性等待你去实现。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。