基于AutoDL平台的Qwen2-7B模型FastAPI部署指南

实例配置信息

配置项	规格说明
GPU	RTX 3090 (24GB) × 1
CPU	14 vCPU Intel® Xeon® Platinum 8362 CPU @ 2.80GHz
内存	45GB
系统盘	30GB
数据盘	50GB（免费）
镜像	PyTorch 2.1.0 · Python 3.10(ubuntu22.04) · Cuda 12.1

---

一、环境准备

首先，在AutoDL平台租赁一台配备RTX 3090/24G显存的显卡机器。选择适合的镜像，这里我们选择的是PyTorch-2.1.0 3.10(ubuntu20.04)-12.1，这个镜像为我们后续的操作提供了必要的基础环境。

1、镜像选择

• PyTorch版本: 2.1.0
• 操作系统: Ubuntu 22.04
• CUDA版本: 12.1

2、环境配置

在AutoDL控制台的"终端"中执行以下命令，配置Python环境和依赖：

# 创建并进入工作目录
mkdir -p /root/autodl-tmp/qwen2-api
cd /root/autodl-tmp/qwen2-api

# 创建Python虚拟环境
python -m venv venv
source venv/bin/activate

# 升级pip
pip install --upgrade pip

# 安装必要的系统依赖（如果需要）
apt-get update
apt-get install -y git curl

# 创建requirements.txt文件
cat > requirements.txt << 'EOF'
fastapi==0.104.1
uvicorn[standard]==0.24.0
transformers==4.36.0
torch==2.1.0
accelerate==0.25.0
python-multipart==0.0.6
pydantic==2.5.0
requests==2.31.0
sentencepiece==0.1.99  # Qwen2可能需要
tiktoken==0.5.2
EOF

# 安装Python依赖
pip install -r requirements.txt

二、模型下载

1、模型下载脚本示例

在AutoDL的终端中执行以下脚本下载Qwen2-7B模型。由于模型较大（约14GB），下载可能需要一些时间：

# 创建并运行模型下载脚本
cat > download_model.py << 'EOF'
import os
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
import logging

# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def download_qwen_model():
    # 模型名称（可以选择不同的版本）
    # 可选模型：Qwen/Qwen2-7B, Qwen/Qwen2-7B-Instruct, Qwen/Qwen2-7B-Chat
    model_name = "Qwen/Qwen2-7B-Instruct"
    
    # 本地保存路径
    save_path = "/root/autodl-tmp/models/qwen2-7b-instruct"
    
    print(f"开始下载模型: {model_name}")
    print(f"模型将保存到: {save_path}")
    
    try:
        # 下载tokenizer
        logger.info("正在下载tokenizer...")
        tokenizer = AutoTokenizer.from_pretrained(
            model_name,
            trust_remote_code=True
        )
        
        # 下载模型（使用float16以节省显存）
        logger.info("正在下载模型（这可能需要一些时间，请耐心等待）...")
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float16,  # 使用float16减少显存占用
            device_map="auto",
            trust_remote_code=True
        )
        
        # 保存到本地
        logger.info("保存模型到本地...")
        os.makedirs(save_path, exist_ok=True)
        model.save_pretrained(save_path)
        tokenizer.save_pretrained(save_path)
        
        print(f"✓ 模型下载并保存完成！")
        print(f"  保存路径: {save_path}")
        print(f"  模型大小: 约14GB")
        
        return model, tokenizer
        
    except Exception as e:
        logger.error(f"模型下载失败: {str(e)}")
        raise

if __name__ == "__main__":
    download_qwen_model()
EOF

# 运行下载脚本（在AutoDL终端中执行）
python download_model.py

重要说明：

• 上述脚本在AutoDL的终端中直接执行
• 下载过程可能需要30分钟到1小时，具体取决于网络速度
• 模型将下载到 /root/autodl-tmp/models/ 目录
• 如果网络较慢，可以考虑使用AutoDL的"模型下载"功能，或者提前下载到数据盘

三、代码准备

1、FastAPI 应用代码示例

在AutoDL实例中创建以下文件结构：

/root/autodl-tmp/qwen2-api/
├── main.py              # FastAPI主应用
├── requirements.txt     # Python依赖
├── download_model.py    # 模型下载脚本
└── models/              # 模型存放目录

main.py - FastAPI应用主文件：

"""
Qwen2-7B FastAPI部署脚本
在AutoDL实例中运行：uvicorn main:app --host 0.0.0.0 --port 6006
"""
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, List
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
import logging
import time
from datetime import datetime
import os

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="Qwen2-7B API服务 - AutoDL部署",
    description="基于Qwen2-7B-Instruct大语言模型的API接口",
    version="2.0.0",
    docs_url="/docs",  # 启用Swagger UI
    redoc_url="/redoc"  # 启用ReDoc
)

# 配置CORS（允许跨域）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境应限制域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 定义请求/响应模型
class ChatMessage(BaseModel):
    role: str  # system, user, assistant
    content: str

class ChatRequest(BaseModel):
    messages: List[ChatMessage]
    max_tokens: Optional[int] = 512
    temperature: Optional[float] = 0.7
    top_p: Optional[float] = 0.9
    stream: Optional[bool] = False

class ChatResponse(BaseModel):
    response: str
    tokens_used: int
    processing_time: float
    timestamp: str
    model: str = "Qwen2-7B-Instruct"

class HealthResponse(BaseModel):
    status: str
    model_loaded: bool
    device: str
    gpu_memory: Optional[str]
    timestamp: str

# 全局变量存储模型和tokenizer
model = None
tokenizer = None
device = None

def get_gpu_memory_info():
    """获取GPU内存信息"""
    if torch.cuda.is_available():
        total = torch.cuda.get_device_properties(0).total_memory / 1024**3
        allocated = torch.cuda.memory_allocated(0) / 1024**3
        free = total - allocated
        return f"GPU内存: {allocated:.2f}/{total:.2f} GB (空闲: {free:.2f} GB)"
    return "无GPU信息"

@app.on_event("startup")
async def startup_event():
    """启动时加载模型"""
    global model, tokenizer, device
    
    try:
        logger.info("正在加载Qwen2-7B模型...")
        start_time = time.time()
        
        # 设置设备
        device = "cuda" if torch.cuda.is_available() else "cpu"
        logger.info(f"使用设备: {device}")
        
        # 模型路径（根据实际情况调整）
        model_path = "/root/autodl-tmp/models/qwen2-7b-instruct"
        
        if not os.path.exists(model_path):
            logger.error(f"模型路径不存在: {model_path}")
            logger.info("请先运行 download_model.py 下载模型")
            return
        
        # 加载tokenizer
        logger.info("加载tokenizer...")
        tokenizer = AutoTokenizer.from_pretrained(
            model_path,
            trust_remote_code=True,
            padding_side="left"
        )
        
        if tokenizer.pad_token is None:
            tokenizer.pad_token = tokenizer.eos_token
        
        # 加载模型
        logger.info("加载模型（这可能需要几分钟）...")
        model = AutoModelForCausalLM.from_pretrained(
            model_path,
            torch_dtype=torch.float16,  # 使用float16节省显存
            device_map="auto" if device == "cuda" else None,
            trust_remote_code=True
        )
        
        if device == "cpu":
            model = model.to(device)
        
        model.eval()
        
        load_time = time.time() - start_time
        logger.info(f"模型加载完成！耗时: {load_time:.2f}秒")
        logger.info(get_gpu_memory_info())
        
    except Exception as e:
        logger.error(f"模型加载失败: {str(e)}")
        raise

@app.get("/", response_model=HealthResponse)
async def root():
    """根端点 - 服务状态检查"""
    return HealthResponse(
        status="运行中",
        model_loaded=model is not None,
        device=device,
        gpu_memory=get_gpu_memory_info() if device == "cuda" else None,
        timestamp=datetime.now().isoformat()
    )

@app.get("/health", response_model=HealthResponse)
async def health_check():
    """健康检查端点"""
    if model is None or tokenizer is None:
        raise HTTPException(
            status_code=503,
            detail="模型未加载完成，请稍后再试"
        )
    
    return HealthResponse(
        status="健康",
        model_loaded=True,
        device=device,
        gpu_memory=get_gpu_memory_info() if device == "cuda" else None,
        timestamp=datetime.now().isoformat()
    )

@app.post("/v1/chat/completions", response_model=ChatResponse)
async def chat_completion(request: ChatRequest):
    """
    对话生成接口（兼容OpenAI API格式）
    
    示例请求体:
    {
        "messages": [
            {"role": "system", "content": "你是一个AI助手"},
            {"role": "user", "content": "你好"}
        ],
        "max_tokens": 100,
        "temperature": 0.7
    }
    """
    if model is None or tokenizer is None:
        raise HTTPException(
            status_code=503,
            detail="模型未就绪，请检查服务状态"
        )
    
    try:
        start_time = time.time()
        
        # 构建对话文本
        conversation_text = ""
        for msg in request.messages:
            if msg.role == "system":
                conversation_text += f"<|im_start|>system\n{msg.content}<|im_end|>\n"
            elif msg.role == "user":
                conversation_text += f"<|im_start|>user\n{msg.content}<|im_end|>\n"
            elif msg.role == "assistant":
                conversation_text += f"<|im_start|>assistant\n{msg.content}<|im_end|>\n"
        
        # 添加assistant提示
        conversation_text += "<|im_start|>assistant\n"
        
        # Tokenize输入
        inputs = tokenizer(
            conversation_text,
            return_tensors="pt",
            padding=True,
            truncation=True,
            max_length=2048  # 限制输入长度
        ).to(device)
        
        # 生成配置
        gen_kwargs = {
            "max_new_tokens": min(request.max_tokens, 2048),  # 限制生成长度
            "temperature": request.temperature,
            "top_p": request.top_p,
            "do_sample": True if request.temperature > 0 else False,
            "pad_token_id": tokenizer.pad_token_id,
            "eos_token_id": tokenizer.eos_token_id,
        }
        
        # 生成响应
        with torch.no_grad():
            outputs = model.generate(
                **inputs,
                **gen_kwargs
            )
        
        # 解码响应（跳过输入部分）
        input_length = inputs["input_ids"].shape[1]
        response_ids = outputs[0][input_length:]
        response_text = tokenizer.decode(response_ids, skip_special_tokens=True)
        
        # 清理响应中的特殊标记
        response_text = response_text.replace("<|im_end|>", "").strip()
        
        # 计算使用token数和处理时间
        tokens_used = len(response_ids)
        processing_time = time.time() - start_time
        
        logger.info(f"请求处理完成 - Token数: {tokens_used}, 耗时: {processing_time:.2f}秒")
        
        return ChatResponse(
            response=response_text,
            tokens_used=tokens_used,
            processing_time=processing_time,
            timestamp=datetime.now().isoformat()
        )
        
    except torch.cuda.OutOfMemoryError:
        logger.error("GPU内存不足，请减少max_tokens或使用更小的模型")
        raise HTTPException(
            status_code=500,
            detail="GPU内存不足，请减少生成长度"
        )
    except Exception as e:
        logger.error(f"生成失败: {str(e)}")
        raise HTTPException(
            status_code=500,
            detail=f"生成失败: {str(e)}"
        )

@app.post("/generate")
async def generate_text(prompt: str, max_length: int = 100):
    """简单的文本生成接口"""
    if model is None or tokenizer is None:
        raise HTTPException(status_code=503, detail="模型未就绪")
    
    try:
        inputs = tokenizer(prompt, return_tensors="pt").to(device)
        
        with torch.no_grad():
            outputs = model.generate(
                **inputs,
                max_new_tokens=min(max_length, 1024),
                do_sample=True,
                temperature=0.7,
                pad_token_id=tokenizer.pad_token_id
            )
        
        generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
        
        return {
            "prompt": prompt,
            "generated_text": generated_text,
            "model": "Qwen2-7B-Instruct",
            "timestamp": datetime.now().isoformat()
        }
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    # 在AutoDL中，通常使用6006端口
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=6006,
        log_level="info"
    )

四、API 部署

在AutoDL平台上部署API的步骤：

1. 启动API服务

在AutoDL终端中执行：

# 进入项目目录
cd /root/autodl-tmp/qwen2-api

# 激活虚拟环境
source venv/bin/activate

# 启动FastAPI服务（使用6006端口，这是AutoDL常用的端口）
nohup uvicorn main:app --host 0.0.0.0 --port 6006 --workers 1 > api.log 2>&1 &

# 查看日志
tail -f api.log

2. 设置AutoDL自定义服务

为了在AutoDL上通过Web界面访问API，需要设置自定义服务：

1. 在AutoDL控制台，点击"自定义服务"
2. 点击"添加"
3. 配置如下：
   • 名称: Qwen2-7B API
   • 端口: 6006
   • 启动命令:

     cd /root/autodl-tmp/qwen2-api && source venv/bin/activate && uvicorn main:app --host 0.0.0.0 --port 6006
     

   • 环境变量: 无特殊要求可不填

3. 访问API文档

服务启动后，可以通过以下方式访问：

1. AutoDL Web界面: 点击"自定义服务"中的链接
2. 直接访问:
   • API文档: http://你的实例IP:6006/docs
   • 健康检查: http://你的实例IP:6006/health
   • 根端点: http://你的实例IP:6006/

五、API 测试

1、使用curl调用API

在本地终端或AutoDL终端中测试：

# 健康检查
curl http://localhost:6006/health

# 简单文本生成
curl -X POST "http://localhost:6006/generate" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "请介绍人工智能", "max_length": 50}'

# 对话接口（兼容OpenAI格式）
curl -X POST "http://localhost:6006/v1/chat/completions" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"role": "system", "content": "你是一个AI助手"},
      {"role": "user", "content": "你好，请介绍一下你自己"}
    ],
    "max_tokens": 100,
    "temperature": 0.7
  }'

2、使用Python的requests库调用API

创建测试脚本 test_api.py：

import requests
import json
import time

class Qwen2APIClient:
    def __init__(self, base_url="http://localhost:6006"):
        self.base_url = base_url
    
    def check_health(self):
        """检查服务健康状态"""
        try:
            response = requests.get(f"{self.base_url}/health", timeout=10)
            return response.json()
        except Exception as e:
            return {"error": str(e), "status": "unhealthy"}
    
    def generate_text(self, prompt, max_length=100):
        """调用文本生成接口"""
        payload = {
            "prompt": prompt,
            "max_length": max_length
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/generate",
                json=payload,
                timeout=30
            )
            return response.json()
        except Exception as e:
            return {"error": str(e)}
    
    def chat_completion(self, messages, max_tokens=200, temperature=0.7):
        """调用对话接口"""
        payload = {
            "messages": messages,
            "max_tokens": max_tokens,
            "temperature": temperature
        }
        
        try:
            start_time = time.time()
            response = requests.post(
                f"{self.base_url}/v1/chat/completions",
                json=payload,
                timeout=60  # 对话可能需要更长时间
            )
            processing_time = time.time() - start_time
            
            if response.status_code == 200:
                result = response.json()
                result["processing_time"] = processing_time
                return result
            else:
                return {
                    "error": f"请求失败: {response.status_code}",
                    "detail": response.text
                }
        except Exception as e:
            return {"error": str(e)}
    
    def test_all(self):
        """运行所有测试"""
        print("=== Qwen2-7B API 测试 ===")
        
        # 1. 健康检查
        print("\n1. 健康检查:")
        health = self.check_health()
        print(json.dumps(health, indent=2, ensure_ascii=False))
        
        if health.get("status") != "健康":
            print("服务不健康，停止测试")
            return
        
        # 2. 文本生成测试
        print("\n2. 文本生成测试:")
        gen_result = self.generate_text("什么是机器学习？", max_length=80)
        if "error" not in gen_result:
            print(f"Prompt: {gen_result.get('prompt')}")
            print(f"Generated: {gen_result.get('generated_text')}")
        else:
            print(f"错误: {gen_result.get('error')}")
        
        # 3. 对话测试
        print("\n3. 对话测试:")
        messages = [
            {"role": "system", "content": "你是一个AI助手，请用中文回答"},
            {"role": "user", "content": "请写一首关于春天的短诗"}
        ]
        chat_result = self.chat_completion(messages, max_tokens=100)
        
        if "error" not in chat_result:
            print(f"AI回复: {chat_result.get('response')}")
            print(f"使用token数: {chat_result.get('tokens_used')}")
            print(f"处理时间: {chat_result.get('processing_time', 0):.2f}秒")
        else:
            print(f"错误: {chat_result.get('error')}")
        
        print("\n=== 测试完成 ===")

if __name__ == "__main__":
    # 如果API在远程服务器上，修改这里的地址
    # client = Qwen2APIClient("http://你的实例IP:6006")
    client = Qwen2APIClient()  # 本地测试
    
    client.test_all()

在AutoDL终端中运行测试：

# 进入项目目录
cd /root/autodl-tmp/qwen2-api

# 运行测试
python test_api.py

六、性能优化建议

1. 显存优化

由于RTX 3090只有24GB显存，加载Qwen2-7B模型（约14GB）后，还需要考虑以下优化：

# 在加载模型时使用以下配置节省显存
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    torch_dtype=torch.float16,  # 使用半精度
    device_map="auto",
    low_cpu_mem_usage=True,     # 减少CPU内存使用
    trust_remote_code=True
)

# 或者使用量化版本（如果可用）
# model = AutoModelForCausalLM.from_pretrained(
#     model_path,
#     torch_dtype=torch.float16,
#     load_in_8bit=True,  # 8位量化
#     device_map="auto",
#     trust_remote_code=True
# )

2. 批处理优化

# 在生成时设置合适的参数
gen_kwargs = {
    "max_new_tokens": 512,      # 限制生成长度
    "temperature": 0.7,
    "top_p": 0.9,
    "repetition_penalty": 1.1,  # 防止重复
    "do_sample": True,
}

3. AutoDL特殊配置

• 设置Jupyter密码后，可以通过Web终端操作
• 使用nohup让服务在后台运行
• 监控GPU使用情况：watch -n 1 nvidia-smi
• 如果显存不足，考虑使用Qwen2-1.5B或Qwen2-4B等更小模型

结语

本文详细介绍了在AutoDL平台上部署Qwen2-7B模型并创建FastAPI接口的完整流程。通过本指南，您可以：

1. 快速搭建：在AutoDL平台30分钟内完成环境搭建和模型部署
2. 灵活调用：提供兼容OpenAI API的接口，方便集成到各种应用
3. 监控管理：包含健康检查、性能监控等生产级功能

注意事项：

• AutoDL实例关机后数据会保留，但需要注意保存重要代码
• 长期运行请关注GPU使用时长和费用
• 生产环境建议添加身份验证和安全限制
• 对于高并发场景，可能需要使用多个实例和负载均衡

后续扩展方向：

1. 添加流式输出支持
2. 实现多轮对话记忆
3. 集成RAG（检索增强生成）功能
4. 添加Rate Limiting和API密钥管理
5. 实现模型热加载和版本管理

通过本方案，您可以快速将Qwen2-7B等大语言模型部署为可调用的API服务，为AI应用开发提供强大的基础能力。