原文：
towardsdatascience.com/minimum-viable-mle-306877dd6030?source=collection_archive---------9-----------------------#2024-10-31
构建一个最小化的生产就绪情感分析模型
https://medium.com/@lenixc210?source=post_page---byline--306877dd6030--------------------------------https://towardsdatascience.com/?source=post_page---byline--306877dd6030-------------------------------- Lenix Carter
·发表于Towards Data Science ·阅读时长 7 分钟·2024 年 10 月 31 日
–
https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/27789732b7a4449b4ff94b4e82876275.png
图片由Stephen Dawson提供，来自Unsplash
什么是生产就绪模型？
我们经常听到“生产化的机器学习”，但要让模型在实际应用中茁壮成长究竟意味着什么呢？在生产中，许多因素都影响并有助于机器学习模型的效能。为了本文的目的，我们将重点关注其中的五个因素。
可复现性
监控
测试
自动化
版本控制
提供推断
构建生产就绪机器学习模型最重要的部分是能够访问它。
为此，我们构建了一个 FastAPI 客户端，用于提供情感分析的响应。我们利用 Pydantic 来确保输入和输出的结构化。我们使用的模型是来自 Huggingface 的 Transformers 库中的基本情感分析管道，这使我们能够使用预训练模型开始进行测试。
# Filename: main.pyfromfastapiimportFastAPIfrompydanticimportBaseModelfromtransformersimportpipeline app=FastAPI()classifier=pipeline("sentiment-analysis")classTextInput(BaseModel):text:strclassSentimentOutput(BaseModel):text:strsentiment:strscore:float@app.post("/predict",response_model=SentimentOutput)asyncdefpredict_sentiment(input_data:TextInput):result=classifier(input_data.text)[0]returnSentimentOutput(text=input_data.text,sentiment=result["label"],score=result["score"])为了确保我们的工作是可复现的，我们可以使用 requirements.txt 文件和 pip。
# Filename: requirements.txt# Note: This has all required packages for the final result.fastapi==0.68.1uvicorn==0.15.0transformers==4.30.0torch==2.0.0pydantic==1.10.0numpy==1.24.3sentencepiece==0.1.99protobuf==3.20.3prometheus-client==0.17.1要安装此项，请初始化您的文件中的 venv并运行：pip install -r requirements.txt。
要托管此 API，只需运行：uvicorn main:app --reload。
现在您拥有一个可以通过以下方式查询的 API：
curl-X POST"http://localhost:8000/predict"\-H"Content-Type: application/json"\-d'{"text": "I love using FastAPI!"}'或者您想要的任何 API 工具（即Postman）。您应该能收到一个返回结果，其中包括文本查询、预测的情感以及预测的置信度。
我们稍后将使用 GitHub 进行 CI/CD，因此我建议在此目录中初始化并使用 git。
我们现在有一个本地托管的机器学习推理 API。
进一步提高可复现性
为了让我们的代码能够更一致地执行，我们将使用 Docker。Docker 模拟了一个轻量级的环境，允许应用程序在隔离的容器中运行，类似于虚拟机。这种隔离确保应用程序能够在任何安装了 Docker 的计算机上稳定执行，而不受底层系统的影响。
首先，为你的操作系统设置 Docker。
# Filename: Dockerfile# Use the official Python 3.9 slim image as the baseFROM python:3.9-slim# Set the working directory inside the container to /appWORKDIR/app# Copy the requirements.txt file to the working directoryCOPY requirements.txt.# Install the Python dependencies listed in requirements.txtRUN pip install-r requirements.txt# Copy the main application file (main.py) to the working directoryCOPY main.py.# Define the command to run the FastAPI application with UvicornCMD["uvicorn","main:app","--host","0.0.0.0","--port","8000"]此时，你应该拥有如下的目录结构。
your-project/├── Dockerfile ├── requirements.txt └── main.py现在，你可以构建镜像并使用以下命令运行这个 API：
# Build the Docker imagedocker build-t sentiment-api.# Run the containerdocker run-p8000:8000sentiment-api你现在应该能够像之前一样进行查询。
curl-X POST"http://localhost:8000/predict"\-H"Content-Type: application/json"\-d'{"text": "I love using FastAPI!"}'我们现在拥有一个容器化的、本地托管的机器学习推理 API。
添加基础监控
在机器学习应用中，监控对于理解模型性能以及确保其达到预期的准确性和效率至关重要。像Prometheus这样的工具帮助跟踪诸如预测延迟、请求计数和模型输出分布等指标，使你能够识别诸如模型漂移或资源瓶颈等问题。这种主动的方法确保你的机器学习模型随着时间的推移保持有效，并能够适应不断变化的数据或使用模式。在我们的案例中，我们专注于预测时间、请求和收集查询信息。
fromfastapiimportFastAPIfrompydanticimportBaseModelfromtransformersimportpipelinefromprometheus_clientimportCounter,Histogram,start_http_serverimporttime# Start prometheus metrics server on port 8001start_http_server(8001)app=FastAPI()# MetricsPREDICTION_TIME=Histogram('prediction_duration_seconds','Time spent processing prediction')REQUESTS=Counter('prediction_requests_total','Total requests')SENTIMENT_SCORE=Histogram('sentiment_score','Histogram of sentiment scores',buckets=[0.0,0.25,0.5,0.75,1.0])classTextInput(BaseModel):text:strclassSentimentOutput(BaseModel):text:strsentiment:strscore:float@app.post("/predict",response_model=SentimentOutput)asyncdefpredict_sentiment(input_data:TextInput):REQUESTS.inc()start_time=time.time()result=classifier(input_data.text)[0]score=result["score"]SENTIMENT_SCORE.observe(score)# Record the sentiment scorePREDICTION_TIME.observe(time.time()-start_time)returnSentimentOutput(text=input_data.text,sentiment=result["label"],score=score)使用自定义模型
尽管构建和微调模型的过程并不是本项目的目的，但理解如何将模型添加到此过程中非常重要。
# Filename: train.pyimporttorchfromtransformersimportAutoTokenizer,AutoModelForSequenceClassificationfromdatasetsimportload_datasetfromtorch.utils.dataimportDataLoaderdeftrain_model():# Load datasetfull_dataset=load_dataset("stanfordnlp/imdb",split="train")dataset=full_dataset.shuffle(seed=42).select(range(10000))model_name="distilbert-base-uncased"tokenizer=AutoTokenizer.from_pretrained(model_name)model=AutoModelForSequenceClassification.from_pretrained(model_name,num_labels=2)optimizer=torch.optim.AdamW(model.parameters(),lr=2e-5)# Use GPU if availabledevice=torch.device("cuda"iftorch.cuda.is_available()else"cpu")model.to(device)model.train()# Create a DataLoader for batchingdataloader=DataLoader(dataset,batch_size=8,shuffle=True)# Training loopnum_epochs=3# Set the number of epochsforepochinrange(num_epochs):total_loss=0forbatchindataloader:inputs=tokenizer(batch["text"],truncation=True,padding=True,return_tensors="pt",max_length=512).to(device)labels=torch.tensor(batch["label"]).to(device)optimizer.zero_grad()outputs=model(**inputs,labels=labels)loss=outputs.loss loss.backward()optimizer.step()total_loss+=loss.item()avg_loss=total_loss/len(dataloader)print(f"Epoch{epoch+1}/{num_epochs}, Loss:{avg_loss:.4f}")# Save the modelmodel.save_pretrained("./model/")tokenizer.save_pretrained("./model/")# Test the model with sample sentencestest_sentences=["This movie was fantastic!","I absolutely hated this film.","It was just okay, not great.","An absolute masterpiece!","Waste of time!","A beautiful story and well acted.","Not my type of movie.","It could have been better.","A thrilling adventure from start to finish!","Very disappointing."]# Switch model to evaluation modemodel.eval()# Prepare tokenizer for test inputsinputs=tokenizer(test_sentences,truncation=True,padding=True,return_tensors="pt",max_length=512).to(device)withtorch.no_grad():outputs=model(**inputs)predictions=torch.argmax(outputs.logits,dim=1)# Print predictionsforsentence,predictioninzip(test_sentences,predictions):sentiment="positive"ifprediction.item()==1else"negative"print(f"Input: \"{sentence}\" -> Predicted sentiment:{sentiment}")# Call the function to train the model and test ittrain_model()为了确保我们能够查询我们训练的新模型，我们需要更新一些现有的文件。例如，在main.py中，我们现在使用来自./model的模型，并将其加载为预训练模型。此外，为了对比，我们现在有两个可用的端点，/predict/naive和predict/trained。
# Filename: main.pyfromfastapiimportFastAPIfrompydanticimportBaseModelfromtransformersimportAutoModelForSequenceClassification,AutoTokenizerfromtransformersimportpipelinefromprometheus_clientimportCounter,Histogram,start_http_serverimporttime# Start prometheus metrics server on port 8001start_http_server(8001)app=FastAPI()# Load the trained model and tokenizer from the local directorymodel_path="./model"# Path to your saved modeltokenizer=AutoTokenizer.from_pretrained(model_path)trained_model=AutoModelForSequenceClassification.from_pretrained(model_path)# Create pipelinesnaive_classifier=pipeline("sentiment-analysis",device=-1)trained_classifier=pipeline("sentiment-analysis",model=trained_model,tokenizer=tokenizer,device=-1)# MetricsPREDICTION_TIME=Histogram('prediction_duration_seconds','Time spent processing prediction')REQUESTS=Counter('prediction_requests_total','Total requests')SENTIMENT_SCORE=Histogram('sentiment_score','Histogram of sentiment scores',buckets=[0.0,0.25,0.5,0.75,1.0])classTextInput(BaseModel):text:strclassSentimentOutput(BaseModel):text:strsentiment:strscore:float@app.post("/predict/naive",response_model=SentimentOutput)asyncdefpredict_naive_sentiment(input_data:TextInput):REQUESTS.inc()start_time=time.time()result=naive_classifier(input_data.text)[0]score=result["score"]SENTIMENT_SCORE.observe(score)# Record the sentiment scorePREDICTION_TIME.observe(time.time()-start_time)returnSentimentOutput(text=input_data.text,sentiment=result["label"],score=score)@app.post("/predict/trained",response_model=SentimentOutput)asyncdefpredict_trained_sentiment(input_data:TextInput):REQUESTS.inc()start_time=time.time()result=trained_classifier(input_data.text)[0]score=result["score"]SENTIMENT_SCORE.observe(score)# Record the sentiment score我们还必须更新我们的 Dockerfile，以包含我们的模型文件。
# Filename: DockerfileFROM python:3.9-slim WORKDIR/app COPY requirements.txt.RUN pip install-r requirements.txt COPY main.py.COPY./model./model CMD["uvicorn","main:app","--host","0.0.0.0","--port","8000"]重要的是，如果你使用 git，确保将pytorch_model.bin文件添加到git lfs，这样你就可以推送到 GitHub。git lfs 允许你对非常大的文件使用版本控制。
添加测试和 CI/CD
CI/CD 和测试通过确保代码更改自动集成、测试和部署，从而简化了机器学习模型的部署，这减少了错误的风险并提高了模型的可靠性。这个过程促进了持续改进和更快的迭代周期，使团队能够更高效地交付高质量、准备投入生产的模型。首先，我们创建了两个非常基础的测试，确保我们的模型表现得足够好。
# Filename: test_model.pyimportpytestfromfastapi.testclientimportTestClientfrommainimportapp client=TestClient(app)deftest_positive_sentiment():response=client.post("/predict/trained",json={"text":"This is amazing!"})assertresponse.status_code==200data=response.json()assertdata["sentiment"]=="LABEL_1"assertdata["score"]>0.5deftest_negative_sentiment():response=client.post("/predict/trained",json={"text":"This is terrible!"})assertresponse.status_code==200data=response.json()assertdata["sentiment"]=="LABEL_0"assertdata["score"]<0.5要测试你的代码，你可以在端点运行时简单地执行pytest或python -m pytest。
然而，当推送到 GitHub 时，我们将添加自动化测试 CI/CD（持续集成与持续交付）。
# Filename: .github/workflows/ci_cd.ymlname:CI/CD on:[push]jobs:test:runs-on:ubuntu-latest steps:-name:Checkout code uses:actions/checkout@v2with:lfs:true-name:Set up Python uses:actions/setup-python@v2with:python-version:'3.9'-name:Install dependencies run:|pip install-r requirements.txt pip install pytest httpx-name:Run tests run:pytest我们最终的项目结构应如下所示。
sentiment-analysis-project/├──.github/│ └── workflows/│ └── ci_cd.yml ├── test_model.py ├── main.py ├── Dockerfile ├── requirements.txt └── train.py现在，每当我们推送到 GitHub 时，它将运行一个自动化过程，检出代码，设置 Python 3.9 环境，安装依赖，并使用 pytest 运行我们的测试。
结论
在这个项目中，我们开发了一个生产就绪的情感分析 API，突出了部署机器学习模型的关键方面。虽然它并不涵盖该领域的每个方面，但它提供了涉及这一过程的基本任务的代表性样本。通过检查这些组件，我希望能澄清你可能遇到过的概念，但不确定它们如何在实际环境中配合工作。
