如何在本地部署大模型并实现接口访问（ Llama3、Qwen、DeepSeek等）

如何在本地部署 大模型并实现接口访问（ Llama3、Qwen、DeepSeek等）

如何在本地部署大模型并实现接口访问（ Llama3、Qwen、DeepSeek等）
- 模型地址
- 模型下载
- 模型部署
- - 指定显卡运行
  - app.py
- 运行环境
- - requirements
- 调用接口
- - 代码调用
- 结语

如何在本地部署 大模型并实现接口访问（ Llama3、Qwen、DeepSeek等）

本文详细介绍了如何在本地服务器上部署大模型（如DeepSeek、Llama3、Qwen等），并通过接口实现外部调用。首先，从HuggingFace或魔搭网站下载模型，使用git lfs和screen确保大文件完整下载；接着，使用FastAPI封装模型推理过程，支持多GPU运行并通过CUDA_VISIBLE_DEVICES指定显卡，提供完整的app.py代码实现模型加载和接口响应；然后，通过conda创建Python 3.10环境并安装依赖，使用nohup后台运行服务；最后，展示如何通过Postman或代码调用接口，发送请求并获取模型生成的文本。本文提供了从模型下载、部署到接口调用的完整流程，适合在本地服务器上运行大模型并实现高效的推理服务。

模型地址

在HuggingFace中搜索对应模型，并选择git clone
比如

git clone https://huggingface.co/deepseek-ai/DeepSeek-V2-Chat

huggingface网站可能会访问超时，因此可以选择国内的魔搭网站，同样登录网站后，搜索对应模型并下载

在这里插入图片描述

git clone https://www.modelscope.cn/deepseek-ai/DeepSeek-V2-Lite-Chat.git

模型下载

由于模型比较大，所以需要用git lfs install

如果是下载70B或者更大的模型，可能git clone还是会遗漏一些文件，所以推荐使用Screen在后台下载

创建conda环境后，使用conda install -c conda-forge screen安装screen

为 screen 会话指定一个名称，可以使用 -S 参数：

screen -S mysession

暂时离开当前 screen 会话（但保持会话在后台运行）

Ctrl + A，然后按 D

重新连接到之前创建的 screen 会话，可以使用以下命令：

screen -r mysession

screen相当于启动了一个新的终端，与其他的终端互不影响，其中的工作都在后台运行

在screen中运行 git clone命令即可

如果出现文件遗漏，可以使用sh脚本指定文件下载，比如指定三个文件进行下载的download_files.sh：

#!/bin/bash# 目标 URL 和文件名的前缀
BASE_URL="https://www.modelscope.cn/models/LLM-Research/Llama-3.3-70B-Instruct/resolve/master"
FILES=("model-00028-of-00030.safetensors""model-00029-of-00030.safetensors""model-00030-of-00030.safetensors"
)# 循环下载文件
for FILE in "${FILES[@]}"; doecho "开始下载 $FILE ..."curl -O "$BASE_URL/$FILE"  # 使用 curl 下载文件if [ $? -eq 0 ]; thenecho "$FILE 下载成功"elseecho "$FILE 下载失败"fi
doneecho "所有文件下载完毕！"

在终端中运行脚本：

./download_files.sh

./ 表示当前目录，确保系统知道从当前目录查找脚本。

模型部署

模型下载后，本地就有了完整的模型仓库

在这里插入图片描述
接下来就要进行模型的部署，采用fastapi进行模型部署,使用下列代码作为app.py

指定显卡运行

如果你有多张显卡，并且想指定使用第4和第5张GPU运行，可以在环境变量中设置仅4，5 GPU可见，此时第4变为0 第5变为1

if "DeepSeek" in MODEL_PATH_OR_NAME:os.environ['CUDA_VISIBLE_DEVICES'] = '4'
else:os.environ['CUDA_VISIBLE_DEVICES'] = '4,5'

然后加载模型时设置device_map="auto"

model = AutoModelForCausalLM.from_pretrained(MODEL_PATH_OR_NAME, torch_dtype=torch.float16, device_map="auto")

app.py

import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import AutoTokenizer, AutoModelForCausalLM, GenerationConfig
import torch
import torch.nn as nn
# Initialize FastAPI app
app = FastAPI()MODEL_PATH_OR_NAME = "/yourpath/DeepSeek-V2-Lite-Chat"# Set CUDA devices for visibility
if "DeepSeek" in MODEL_PATH_OR_NAME:os.environ['CUDA_VISIBLE_DEVICES'] = '4'
else:os.environ['CUDA_VISIBLE_DEVICES'] = '4,5'
device = torch.device('cuda:0')  # Using GPU 2# Declare global variables for model and tokenizer
model = None
tokenizer = None# Use startup and shutdown events for model loading and unloading
@app.on_event("startup")
async def startup():global model, tokenizerprint("Loading model and tokenizer...")if "DeepSeek" in MODEL_PATH_OR_NAME:tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH_OR_NAME, trust_remote_code=True)model = AutoModelForCausalLM.from_pretrained(MODEL_PATH_OR_NAME, trust_remote_code=True, torch_dtype=torch.bfloat16, device_map="auto").cuda()model.generation_config = GenerationConfig.from_pretrained(MODEL_PATH_OR_NAME)model.generation_config.pad_token_id = model.generation_config.eos_token_idelse:tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH_OR_NAME, use_fast=False)tokenizer.pad_token_id = tokenizer.eos_token_id  # 假设eos_token_id就是合适的填充词元标识model = AutoModelForCausalLM.from_pretrained(MODEL_PATH_OR_NAME, torch_dtype=torch.float16, device_map="auto")print("Model loaded successfully!")@app.on_event("shutdown")
async def shutdown():global model, tokenizerprint("Shutting down the model...")del modeldel tokenizer# Define request model using Pydantic
class ChatCompletionRequest(BaseModel):model: strmessages: listtemperature: float = 0max_tokens: int = 256top_p: float = 1stop: list = None  # Ensure stop is a list or None# Define response model
class ChatCompletionResponse(BaseModel):choices: list# Define the /generate route
@app.post("/generate", response_model=ChatCompletionResponse)
async def generate_response(request: ChatCompletionRequest):# Get user prompt (last message)prompt = request.messages[-1]["content"]print('INPUT ：'+prompt+'\n')if "DeepSeek" in MODEL_PATH_OR_NAME:try:messages = [{"role": "user", "content": prompt}]input_tensor = tokenizer.apply_chat_template(messages, add_generation_prompt=True, return_tensors="pt")outputs = model.generate(input_tensor.to(device), max_new_tokens=100)result = tokenizer.decode(outputs[0][input_tensor.shape[1]:], skip_special_tokens=True)print('OUTPUT ：'+result)return ChatCompletionResponse(choices=[{"message": {"role": "assistant", "content": result},"finish_reason": "stop"}])except Exception as e:raise HTTPException(status_code=500, detail=str(e))else:try:# Tokenize the input text (with padding and truncation to ensure consistency)input_ids = tokenizer.encode(prompt, return_tensors="pt").to(device)# Optionally, get eos_token_id for stopping criteriaeos_token_id = tokenizer.eos_token_id if tokenizer.eos_token_id else Noneattention_mask = torch.ones(input_ids.shape,dtype=torch.long,device=device)# Generate the response using the modeloutput = model.generate(input_ids,num_beams=1,  # Use greedy decodingnum_return_sequences=1,early_stopping=True,do_sample=False,max_length=(getattr(request, 'max_tokens', 50) + len(input_ids[0])),  # 默认 max_tokens=50top_p=getattr(request, 'top_p', 0.9),  # 默认 top_p=0.9temperature=getattr(request, 'temperature', 0.9),  # 默认 temperature=1.0repetition_penalty=getattr(request, 'repetition_penalty', 1.2),  # 默认 repetition_penalty=1.2eos_token_id=eos_token_id,  # Use eos_token_id for stoppingattention_mask=attention_mask,pad_token_id=eos_token_id)# Decode the generated text (skip special tokens)generated_text = tokenizer.decode(output[0], skip_special_tokens=True)# Strip the prompt (remove the input part from the output)output_tokens = output[0][len(input_ids[0]):]  # Remove input tokens from output tokensgenerated_text = tokenizer.decode(output_tokens, skip_special_tokens=True)print('OUTPUT:'+generated_text)# Return the generated text in the expected formatreturn ChatCompletionResponse(choices=[{"message": {"role": "assistant", "content": generated_text.strip()},"finish_reason": "stop"}])except Exception as e:raise HTTPException(status_code=500, detail=str(e))# Health check route
@app.get("/")
async def health_check():return {"status": "ok", "message": "Text Generation API is running!"}if __name__ == "__main__":import uvicornuvicorn.run("app:app", host="0.0.0.0", port=8890, reload=True)  # reload=True 用于开发时的热重载

运行环境

运行app.py前需要安装环境依赖

首先
conda 创建环境：

conda create -n llmapi python=3.10 -y

conda activate llmapi

安装以下依赖，保存为requirements,txt

requirements

accelerate==1.1.1
annotated-types==0.7.0
anyio==4.6.2.post1
certifi==2024.8.30
charset-normalizer==3.4.0
click==8.1.7
exceptiongroup==1.2.2
fastapi==0.115.5
filelock==3.16.1
fsspec==2024.10.0
gputil==1.4.0
h11==0.14.0
huggingface-hub==0.26.3
idna==3.10
jinja2==3.1.4
markupsafe==3.0.2
mpmath==1.3.0
# networkx==3.2.1
numpy==2.0.2
nvidia-cublas-cu12==12.4.5.8
nvidia-cuda-cupti-cu12==12.4.127
nvidia-cuda-nvrtc-cu12==12.4.127
nvidia-cuda-runtime-cu12==12.4.127
nvidia-cudnn-cu12==9.1.0.70
nvidia-cufft-cu12==11.2.1.3
nvidia-curand-cu12==10.3.5.147
nvidia-cusolver-cu12==11.6.1.9
nvidia-cusparse-cu12==12.3.1.170
nvidia-nccl-cu12==2.21.5
nvidia-nvjitlink-cu12==12.4.127
nvidia-nvtx-cu12==12.4.127
packaging==24.2
pillow==11.0.0
protobuf==5.29.0
psutil==6.1.0
pydantic==2.10.2
pydantic-core==2.27.1
pyyaml==6.0.2
regex==2024.11.6
requests==2.32.3
safetensors==0.4.5
sentencepiece==0.2.0
sniffio==1.3.1
starlette==0.41.3
sympy==1.13.1
tokenizers==0.20.3
torch==2.5.1
torchaudio==2.5.1
torchvision==0.20.1
tqdm==4.67.1
transformers==4.46.3
triton==3.1.0
typing-extensions==4.12.2
urllib3==2.2.3
uvicorn==0.32.1

运行下列命令安装：

pip install -r requirements,txt

然后运行下列命令启动app.py ,这个命令会在后台启动app.py，并且输出日志在app.log 文件内

nohup python app.py > app.log 2>&1 &

调用接口

查看日志模型是否成功启动：

在这里插入图片描述

然后使用Post请求对应接口

{"model": "deep seek", "messages": [{"role": "user", "content": "AI是什么意思?"}],"temperature": 0.9,"max_tokens": 100,"repetition_penalty": 1.2, "top_p": 0.9,"stop": ["\n"] }

使用PostMan测试结果：

在这里插入图片描述

代码调用

 import requestsimport json# Define the FastAPI URLurl = "http://YourIPAddress:YourPort/generate"prompt = '你是什么模型？'# Define the request payload (data)data = {"model": "ModelName",  # Example model name (change according to your setup)"messages": [{'role': 'user', 'content': prompt}],"temperature": 0.7,"max_tokens": 100,"top_p": 1.0,"frequency_penalty": 0.0,"presence_penalty": 0.0,"stop": []  # Ensure 'stop' is an empty list}response = ''# Send a POST request to the FastAPI endpointresponse_data = requests.post(url, json=data)# Check the responseif response_data.status_code == 200:# print("Response received successfully:")# print(json.dumps(response.json(), indent=4))result = response_data.json()# 从响应中获取生成的文本response = str(result['choices'][0]['message']['content']).strip()else:print(f"Request failed with status code {response_data.status_code}")print(response_data.text)return response