【AI】在本地 Docker 环境中搭建使用 Hugging Face 托管的 Llama 模型
@TOC
推荐超级课程：
Hugging Face 已经成为机器学习（ML）领域的强大力量。他们大量的预训练模型和用户友好的接口彻底改变了我们对人工智能/机器学习部署和空间的看法。如果你对深入了解 Docker 和 Hugging Face 模型的集成感兴趣，可以在文章“使用 Hugging Face 的 Docker Spaces 构建机器学习应用 ”中找到一份全面的指南。
大语言模型（LLM）——语言生成的奇迹——是一个令人惊叹的发明。在本文中，我们将探讨如何在 Docker 环境中使用 Hugging Face 托管的 Llama 模型，为自然语言处理（NLP）爱好者和研究人员开辟新的机会。
Hugging Face 和 LLMs 简介
Hugging Face（HF）为训练、微调和部署ML 模型 提供了一个综合平台。而 LLMS 则提供了一种能够执行文本生成、补全和分类等任务的尖端模型。
利用 Docker 进行 ML
强大的Docker 容器化技术使程序的打包、分发和运行变得更加简单。将 ML 模型封装在 Docker 容器中确保了在不同环境下模型的一致运行。这样就能保证可重现性，解决了“在我的机器上可以运行”的问题。
格式的类型
Hugging Face 上的大多数模型有两种选择：
AI 模型量化中使用的示例量化技术包括 GGML 和 GPTQ 模型。这可以意味着在训练期间或训练之后的量化。通过将模型权重减少到较低的精度，GGML 和 GPTQ 模型——两个众所周知的量化模型——减少了模型大小和计算需求。
HF 模型加载在 GPU 上，相比 CPU 执行推断要快得多。一般来说，这种模型非常庞大，你也需要很多 GPU 内存。在本文中，我们将使用 GGML 模型，该模型在 CPU 上运行良好，如果你没有一个好的 GPU，它可能会更快。
在本次演示中，我们还将使用 transformers 和 ctransformers，因此让我们先了解一下它们：
- transformers ：现代预训练模型可以通过 transformers 的 API 和工具轻松下载并训练。通过使用预先训练的模型，你可以减少从头开始训练模型所需的时间和资源，以及计算开销和碳足迹。
- ctransformers ：使用 GGML 库开发的 C/C++ 中的 transformer 模型的 Python 绑定。
请求 Llama 模型访问
我们将利用 Meta Llama 模型 ，注册并请求访问。
创建 Hugging Face 令牌
要创建将来使用的访问令牌，请转到你的 Hugging Face 个人资料设置，并从左侧边栏选择 访问令牌（图 1）。保存创建的访问令牌的值。
图 1. 生成访问令牌。
设置 Docker 环境
在探索 LLM 领域之前，我们必须首先配置 Docker 环境。首先安装 Docker，根据你的操作系统在官方Docker 网站 上按照说明进行。安装完成后，执行以下命令以确认你的设置：
docker --version
快速演示
以下命令将使用 Hugging Face 的 harsh-manvar-llama-2-7b-chat-test:latest 映像运行一个容器，并将容器的端口 7860 暴露到主机机器。它还将设置环境变量 HUGGING_FACE_HUB_TOKEN 为你提供的值。
docker run -it -p 7860:7860 --platform=linux/amd64 \
-e HUGGING_FACE_HUB_TOKEN="YOUR_VALUE_HERE" \
registry.hf.space/harsh-manvar-llama-2-7b-chat-test:latest python app.py
-it标志告诉 Docker 以交互模式运行容器并将终端附加到它。这将允许你与容器及其进程进行交互。-p标志告诉 Docker 将容器的端口 7860 暴露到主机机器。这意味着你将能够在主机机器的端口 7860 上访问容器的 Web 服务器。--platform=linux/amd64标志告诉 Docker 在具有 AMD64 架构的 Linux 机器上运行容器。-e HUGGING_FACE_HUB_TOKEN="YOUR_VALUE_HERE"标志告诉 Docker 将环境变量HUGGING_FACE_HUB_TOKEN设置为你提供的值。这对于从容器中访问 Hugging Face 模型是必需的。
app.py 脚本是你要在容器中运行的 Python 脚本。这将启动容器并打开一个终端。然后你可以在终端中与容器及其进程进行交互。要退出容器，请按 Ctrl+C。
访问页面
要访问容器的 Web 服务器，请打开 Web 浏览器并导航至 http://localhost:7860。你应该会看到 Hugging Face 模型的初始页面（图 2）。
打开你的浏览器并转至http://localhost:7860 。
图 2. 访问本地 Docker LLM。
入门
克隆项目
要开始，你可以克隆或下载 Hugging Face 现有的space/repository 。
git clone https://huggingface.co/spaces/harsh-manvar/llama-2-7b-chat-test
文件: requirements.txt
requirements.txt 文件是一个列出项目需要运行的 Python 包和模块的文本文件。它用于管理项目的依赖项并确保所有开发人员使用的都是所需包的相同版本。
为了运行 Hugging Face 的 llama-2-13b-chat 模型，需要以下 Python 包。请注意，这个模型很大，可能需要一些时间来下载和安装。你可能还需要增加分配给 Python 进程的内存以运行该模型。
gradio==3.37.0
protobuf==3.20.3
scipy==1.11.1
torch==2.0.1
sentencepiece==0.1.99
transformers==4.31.0
ctransformers==0.2.27
文件: Dockerfile
FROM python:3.9
RUN useradd -m -u 1000 user
WORKDIR /code
COPY ./requirements.txt /code/requirements.txt
RUN pip install --upgrade pip
RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
USER user
COPY --link --chown=1000 ./ /code
接下来的部分将对 Dockerfile 进行详细说明。第一行告诉 Docker 使用官方的 Python 3.9 镜像作为我们镜像的基础镜像：
FROM python:3.9
接下来的一行使用用户 ID 1000 创建一个名为 user 的新用户。-m 标志告诉 Docker 为用户创建一个 home 目录。
RUN useradd -m -u 1000 user
接下来，这一行将容器的工作目录设置为 /code。
WORKDIR /code
现在将 requirements 文件从当前目录复制到容器中的 /code 目录中。此外，该行升级容器中的 pip 软件包管理器。
RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
这一行将容器的默认用户设置为 user。
USER user
下一行将当前目录的内容复制到容器中的 /code 目录中。--link 标志告诉 Docker 创建硬链接而不是复制文件，这可以提高性能并减小镜像的大小。--chown=1000 标志告诉 Docker 将拷贝文件的所有权更改为用户 user。
COPY --link --chown=1000 ./ /code
构建了 Docker 镜像之后，你可以使用 docker run 命令运行它。这将以非 root 用户 user 运行 Python 3.9 镜像的新容器。然后你可以使用终端与容器进行交互。
文件: app.py
这段 Python 代码展示了如何使用 Gradio 创建一个使用 transformers 训练的文本生成模型的演示。该代码允许用户输入文本提示并生成文本的延续。
Gradio 是一个 Python 库，使您可以轻松创建和分享交互式机器学习演示。它提供了一个简单直观的界面来创建和部署演示，并支持多种机器学习框架和库，包括 transformers。
这个 Python 脚本是一个文本聊天机器人的 Gradio 演示。它使用一个预训练的文本生成模型来生成用户输入的响应。我们将对文件进行拆分，并查看每个部分。
以下行从 typing 模块导入 Iterator 类型。这种类型用于表示可以遍历的值序列。下一行也导入了 gradio 库。
from typing import Iterator
import gradio as gr
接下来的一行从 transformers 库导入 logging 模块，该库是一个流行的自然语言处理机器学习库。
from transformers.utils import logging
from model import get_input_token_length, run
接下来的一行从 model 模块中导入 get_input_token_length() 和 run() 函数。这些函数用于计算文本的输入令牌长度和使用预训练文本生成模型生成文本，分别。接下来的两行配置了 logging 模块，以打印信息级别消息并使用 transformers 记录器。
from model import get_input_token_length, run
logging.set_verbosity_info()
logger = logging.get_logger("transformers")
以下行定义了在代码中使用的一些常量。还有定义在 Gradio 演示中显示的文本。
DEFAULT_SYSTEM_PROMPT = """"""
MAX_MAX_NEW_TOKENS = 2048
DEFAULT_MAX_NEW_TOKENS = 1024
MAX_INPUT_TOKEN_LENGTH = 4000
DESCRIPTION = """"""
LICENSE = """"""
这行记录一个信息级别消息，指示代码正在启动。该函数清除文本框并将输入消息保存到 saved_input 状态变量中。
logger.info("Starting")
def clear_and_save_textbox(message: str) -> tuple[str, str]:
return '', message
以下函数显示输入消息的聊天机器人，并将消息添加到聊天历史记录中。
def display_input(message: str,
history: list[tuple[str, str]]) -> list[tuple[str, str]]:
history.append((message, ''))
logger.info("display_input=%s",message)
return history
此函数从聊天历史记录中删除之前的响应，并返回已更新的聊天历史记录和之前的响应。
def delete_prev_fn(
history: list[tuple[str, str]]) -> tuple[list[tuple[str, str]], str]:
try:
message, _ = history.pop()
except IndexError:
message = ''
return history, message or ''
以下函数使用预训练文本生成模型和给定的参数生成文本。它返回一个迭代器，该迭代器生成一个元组列表，每个元组包含输入消息和生成的响应。
def generate(
message: str,
history_with_input: list[tuple[str, str]],
system_prompt: str,
max_new_tokens: int,
temperature: float,
top_p: float,
top_k: int,
) -> Iterator[list[tuple[str, str]]]:
#logger.info("message=%s",message)
if max_new_tokens > MAX_MAX_NEW_TOKENS:
raise ValueError
history = history_with_input[:-1]
generator = run(message, history, system_prompt, max_new_tokens, temperature, top_p, top_k)
try:
first_response = next(generator)
yield history + [(message, first_response)]
except StopIteration:
yield history + [(message, '')]
for response in generator:
yield history + [(message, response)]
以下函数对给定消息生成响应，并返回空字符串和生成的响应。
def process_example(message: str) -> tuple[str, list[tuple[str, str]]]:
generator = generate(message, [], DEFAULT_SYSTEM_PROMPT, 1024, 1, 0.95, 50)
for x in generator:
pass
return '', x
你可以查看完整的 Python 代码 。
from typing import Iterator
import gradio as gr
from transformers.utils import logging
from model import get_input_token_length, run
logging.set_verbosity_info()
logger = logging.get_logger("transformers")
DEFAULT_SYSTEM_PROMPT = """"""
MAX_MAX_NEW_TOKENS = 2048
DEFAULT_MAX_NEW_TOKENS = 1024
MAX_INPUT_TOKEN_LENGTH = 4000
DESCRIPTION = """"""
LICENSE = """"""
logger.info("Starting")
def clear_and_save_textbox(message: str) -> tuple[str, str]:
return '', message
def display_input(message: str,
history: list[tuple[str, str]]) -> list[tuple[str, str]]:
history.append((message, ''))
logger.info("display_input=%s",message)
return history
def delete_prev_fn(
history: list[tuple[str, str]]) -> tuple[list[tuple[str, str]], str]:
try:
message, _ = history.pop()
except IndexError:
message = ''
return history, message or ''
def generate(
message: str,
history_with_input: list[tuple[str, str]],
system_prompt: str,
max_new_tokens: int,
temperature: float,
top_p: float,
top_k: int,
) -> Iterator[list[tuple[str, str]]]:
#logger.info("message=%s",message)
if max_new_tokens > MAX_MAX_NEW_TOKENS:
raise ValueError
history = history_with_input[:-1]
generator = run(message, history, system_prompt, max_new_tokens, temperature, top_p, top_k)
try:
first_response = next(generator)
yield history + [(message, first_response)]
except StopIteration:
yield history + [(message, '')]
for response in generator:
yield history + [(message, response)]
def process_example(message: str) -> tuple[str, list[tuple[str, str]]]:
generator = generate(message, [], DEFAULT_SYSTEM_PROMPT, 1024, 1, 0.95, 50)
for x in generator:
pass
return '', x
def check_input_token_length(message: str, chat_history: list[tuple[str, str]], system_prompt: str) -> None:
#logger.info("check_input_token_length=%s",message)
input_token_length = get_input_token_length(message, chat_history, system_prompt)
#logger.info("input_token_length",input_token_length)
#logger.info("MAX_INPUT_TOKEN_LENGTH",MAX_INPUT_TOKEN_LENGTH)
if input_token_length > MAX_INPUT_TOKEN_LENGTH:
logger.info("Inside IF condition")
raise gr.Error(f'The accumulated input is too long ({input_token_length} > {MAX_INPUT_TOKEN_LENGTH}). Clear your chat history and try again.')
#logger.info("End of check_input_token_length function")
with gr.Blocks(css='style.css') as demo:
gr.Markdown(DESCRIPTION)
gr.DuplicateButton(value='Duplicate Space for private use',
elem_id='duplicate-button')
with gr.Group():
chatbot = gr.Chatbot(label='Chatbot')
with gr.Row():
textbox = gr.Textbox(
container=False,
show_label=False,
placeholder='Type a message...',
scale=10,
)
submit_button = gr.Button('Submit',
variant='primary',
scale=1,
min_width=0)
with gr.Row():
retry_button = gr.Button('Retry', variant='secondary')
undo_button = gr.Button('Undo', variant='secondary')
clear_button = gr.Button('Clear', variant='secondary')
saved_input = gr.State()
with gr.Accordion(label='Advanced options', open=False):
system_prompt = gr.Textbox(label='System prompt',
value=DEFAULT_SYSTEM_PROMPT,
lines=6)
max_new_tokens = gr.Slider(
label='Max new tokens',
minimum=1,
maximum=MAX_MAX_NEW_TOKENS,
step=1,
value=DEFAULT_MAX_NEW_TOKENS,
)
temperature = gr.Slider(
label='Temperature',
minimum=0.1,
maximum=4.0,
step=0.1,
value=1.0,
)
top_p = gr.Slider(
label='Top-p (nucleus sampling)',
minimum=0.05,
maximum=1.0,
step=0.05,
value=0.95,
)
top_k = gr.Slider(
label='Top-k',
minimum=1,
maximum=1000,
step=1,
value=50,
)
gr.Markdown(LICENSE)
textbox.submit(
fn=clear_and_save_textbox,
inputs=textbox,
outputs=[textbox, saved_input],
api_name=False,
queue=False,
).then(
fn=display_input,
inputs=[saved_input, chatbot],
outputs=chatbot,
api_name=False,
queue=False,
).then(
fn=check_input_token_length,
inputs=[saved_input, chatbot, system_prompt],
api_name=False,
queue=False,
).success(
fn=generate,
inputs=[
saved_input,
chatbot,
system_prompt,
max_new_tokens,
temperature,
top_p,
top_k,
],
outputs=chatbot,
api_name=False,
)
button_event_preprocess = submit_button.click(
fn=clear_and_save_textbox,
inputs=textbox,
outputs=[textbox, saved_input],
api_name=False,
queue=False,
).then(
fn=display_input,
inputs=[saved_input, chatbot],
outputs=chatbot,
api_name=False,
queue=False,
).then(
fn=check_input_token_length,
inputs=[saved_input, chatbot, system_prompt],
api_name=False,
queue=False,
).success(
fn=generate,
inputs=[
saved_input,
chatbot,
system_prompt,
max_new_tokens,
temperature,
top_p,
top_k,
],
outputs=chatbot,
api_name=False,
)
retry_button.click(
fn=delete_prev_fn,
inputs=chatbot,
outputs=[chatbot, saved_input],
api_name=False,
queue=False,
).then(
fn=display_input,
inputs=[saved_input, chatbot],
outputs=chatbot,
api_name=False,
queue=False,
).then(
fn=generate,
inputs=[
saved_input,
chatbot,
system_prompt,
max_new_tokens,
temperature,
top_p,
top_k,
],
outputs=chatbot,
api_name=False,
)
undo_button.click(
fn=delete_prev_fn,
inputs=chatbot,
outputs=[chatbot, saved_input],
api_name=False,
queue=False,
).then(
fn=lambda x: x,
inputs=[saved_input],
outputs=textbox,
api_name=False,
queue=False,
)
clear_button.click(
fn=lambda: ([], ''),
outputs=[chatbot, saved_input],
queue=False,
api_name=False,
)
demo.queue(max_size=20).launch(share=False, server_name="0.0.0.0")
check_input_token_length和generate函数组成了代码的主要部分。generate函数负责根据消息、先前消息的历史记录以及各种生成参数生成响应，包括：
- max_new_tokens：这是一个整数，表示生成响应的模型被允许生成的最多标记数量。
- temperature：这是一个浮点值，调节生成的输出有多随机。值越高（如1.0），结果就越随机，而值越低（如0.2），结果就越可预测。
- top_p：这是一个浮点值，通常介于0到1之间，用于确定核心采样的累积概率的截止点。
- top_k：这个整数表示需要考虑的下一个标记的数量。数字越大，输出越集中。
UI组件和运行API服务器由app.py处理。基本上，app.py是您初始化应用程序和其他配置的地方。
Model.py 这个Python脚本是一个聊天机器人，它使用LLM来生成对用户输入的回复。脚本使用以下步骤生成回复：
- 通过结合用户输入、聊天历史和系统提示为LLM创建提示。
- 计算提示的输入标记长度。
- 使用LLM和以下参数生成回复：
- max_new_tokens：生成的最大新标记数量。
- temperature：生成回复时使用的温度。温度越高，产生的回复越有创意和多样化，但也可能导致一些不连贯的回复。
- top_p：此参数控制生成回复所使用的核心采样算法。更高的top_p值将产生更加专注和信息丰富的回复，而较低的值则将产生更加有创意和多样化的回复。
- top_k：此参数控制生成回复时要考虑的最高概率标记数量。更高的top_k值将产生更加可预测和一致的回复，而较低的值则将产生更加有创意和多样化的回复。
TextIteratorStreamer类的主要功能是将准备好的文本存储在队列中。然后下游应用程序可以使用此队列作为迭代器以非阻塞的方式访问生成的文本。
from threading import Thread
from typing import Iterator
#import torch
from transformers.utils import logging
from ctransformers import AutoModelForCausalLM
from transformers import TextIteratorStreamer, AutoTokenizer
logging.set_verbosity_info()
logger = logging.get_logger("transformers")
config = {"max_new_tokens": 256, "repetition_penalty": 1.1,
"temperature": 0.1, "stream": True}
model_id = "TheBloke/Llama-2-7B-Chat-GGML"
device = "cpu"
model = AutoModelForCausalLM.from_pretrained(model_id, model_type="llama", lib="avx2", hf=True)
tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-2-7b-chat-hf")
def get_prompt(message: str, chat_history: list[tuple[str, str]],
system_prompt: str) -> str:
#logger.info("get_prompt chat_history=%s",chat_history)
#logger.info("get_prompt system_prompt=%s",system_prompt)
texts = [f'<s>[INST] <<SYS>>\n{system_prompt}\n<</SYS>>\n\n']
#logger.info("texts=%s",texts)
do_strip = False
for user_input, response in chat_history:
user_input = user_input.strip() if do_strip else user_input
do_strip = True
texts.append(f'{user_input} [/INST] {response.strip()} </s><s>[INST] ')
message = message.strip() if do_strip else message
#logger.info("get_prompt message=%s",message)
texts.append(f'{message} [/INST]')
#logger.info("get_prompt final texts=%s",texts)
return ''.join(texts) def get_input_token_length(message: str, chat_history: list[tuple[str, str]], system_prompt: str) -> int:
#logger.info("get_input_token_length=%s",message)
prompt = get_prompt(message, chat_history, system_prompt)
#logger.info("prompt=%s",prompt)
input_ids = tokenizer([prompt], return_tensors='np', add_special_tokens=False)['input_ids']
#logger.info("input_ids=%s",input_ids)
return input_ids.shape[-1]
def run(message: str,
chat_history: list[tuple[str, str]],
system_prompt: str,
max_new_tokens: int = 1024,
temperature: float = 0.8,
top_p: float = 0.95,
top_k: int = 50) -> Iterator[str]:
prompt = get_prompt(message, chat_history, system_prompt)
inputs = tokenizer([prompt], return_tensors='pt', add_special_tokens=False).to(device)
streamer = TextIteratorStreamer(tokenizer,
timeout=15.,
skip_prompt=True,
skip_special_tokens=True)
generate_kwargs = dict(
inputs,
streamer=streamer,
max_new_tokens=max_new_tokens,
do_sample=True,
top_p=top_p,
top_k=top_k,
temperature=temperature,
num_beams=1,
)
t = Thread(target=model.generate, kwargs=generate_kwargs)
t.start()
outputs = []
for text in streamer:
outputs.append(text)
yield "".join(outputs)
要导入用于与transformers进行文本生成所需的模块和库，可以使用以下代码：
from transformers import AutoTokenizer, AutoModelForCausalLM
这将导入用于使用transformers进行标记化和生成文本的必要模块。
要定义要导入的模型，可以使用：
model_id = "TheBloke/Llama-2-7B-Chat-GGML"
这一步将模型ID定义为TheBloke/Llama-2-7B-Chat-GGML，这是Meta 7B chat LLama模型的缩减版本。
一旦导入了必要的模块和库并定义了要导入的模型，可以使用以下代码加载标记器和模型：
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForCausalLM.from_pretrained(model_id)
这将从Hugging Face Hub加载标记器和模型。标记器的作用是准备模型的输入。每个模型的标记器都可以在库中找到。再次定义要导入的模型，我们使用的是TheBloke/Llama-2-7B-Chat-GGML。
您需要在配置中设置max_new_tokens，temperature，repetition_penalty和stream的变量和值：
max_new_tokens：最大可能的标记数，忽略提示中指定的标记数量。temperature：用于修改下一个标记的概率的数量。repetition_penalty：重复惩罚参数。1.0表示没有惩罚。stream：是否以流方式生成响应文本还是一次性生成。
您还可以创建空间并将文件提交到其中，以在Hugging Face上托管应用程序并进行直接测试。
构建镜像
以下命令将在linux/amd64平台上为llama-2-13b-chat模型构建一个Docker镜像。该镜像将被命名为local-llm:v1。
docker buildx build --platform=linux/amd64 -t local-llm:v1 .
运行容器
以下命令将启动一个新容器，运行local-llm:v1 Docker镜像，并在主机机器上公开端口7860。-e HUGGING_FACE_HUB_TOKEN="YOUR_VALUE_HERE"环境变量设置Hugging Face Hub令牌，该令牌用于从Hugging Face Hub下载llama-2-13b-chat模型。
docker run -it -p 7860:7860 --platform=linux/amd64 -e HUGGING_FACE_HUB_TOKEN="YOUR_VALUE_HERE" local-llm:v1 python app.py
接下来，打开浏览器并访问http://localhost:7860 ，以查看本地LLM Docker容器输出（图3）。
图3. 本地LLM Docker容器输出。
您还可以通过Docker Desktop查看容器（图4）。
图4. 使用Docker Desktop监视容器。
结论
在本地使用Docker部署LLM GGML模型是一种方便有效的使用自然语言处理的方式。将模型docker化使其易于在不同环境之间移动，并确保其能够一致运行。在浏览器中测试模型提供了用户友好的界面，让您可以快速评估其性能。
这样的设置使您对基础架构和数据有更多控制，并更容易部署先进的语言模型，适用于各种应用。这是在部署大型语言模型方面迈出的一大步。