当前位置：首页 > news >正文

Phi-3-mini-128k-instruct模型API接口开发教程：FastAPI快速封装

news 2026/6/1 8:58:25

Phi-3-mini-128k-instruct模型API接口开发教程：FastAPI快速封装

你是不是已经用命令行或者脚本跑通了Phi-3-mini-128k-instruct模型，感觉效果不错，但总想着怎么能把它变成一个随时可以调用的服务？比如，让其他同事、其他系统，或者你自己写的应用，都能通过一个简单的网络请求来使用这个模型的能力。

今天我们就来解决这个问题。我会带你一步步，用FastAPI这个现代、快速的Python Web框架，把Phi-3-mini-128k-instruct模型包装成一个标准的HTTP API服务。整个过程就像搭积木，我们会从最基础的接口设计开始，一直做到用Docker打包部署，最终得到一个生产环境也能用的服务。即使你之前没怎么接触过Web开发，跟着做下来也能搞定。

1. 准备工作与环境搭建

在开始写代码之前，我们得先把“舞台”搭好。这里没什么高深的理论，就是准备好工具和材料。

1.1 你需要准备的东西

首先，确保你的电脑上已经安装了Python，建议版本是3.8或以上。你可以打开终端或命令行，输入python --version来检查。

接下来，我们需要安装几个关键的Python库。你可以把它们想象成盖房子需要的不同工具：

FastAPI: 这是我们盖房子的主要框架，用来快速构建API。
Uvicorn: 一个轻量级的服务器，负责运行我们写好的FastAPI应用。
Pydantic: FastAPI的好搭档，用来定义数据的“形状”（比如，API请求里应该包含哪些字段，每个字段是什么类型），并且自动做数据验证。
Transformers: Hugging Face的库，我们用它来加载和运行Phi-3-mini模型。
Torch: PyTorch，模型运行的基础。

打开你的终端，一行命令就能搞定大部分依赖：

pip install fastapi uvicorn pydantic transformers torch

如果你的模型需要GPU来加速（运行速度会快很多），并且你的机器有NVIDIA显卡，建议安装支持CUDA的PyTorch版本。你可以去PyTorch官网根据你的系统配置生成对应的安装命令。

1.2 项目结构规划

在写代码前，先规划一下文件夹和文件怎么放，会让后续开发更清晰。在你的工作目录下，创建一个新的项目文件夹，比如叫做phi3-api-service，然后在里面创建下面这些文件：

phi3-api-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用的主入口 │ ├── models.py # 定义数据模型（请求/响应格式） │ ├── inference.py # 模型加载和推理的核心逻辑 │ └── dependencies.py # 依赖项，比如认证中间件 ├── requirements.txt # 项目依赖包列表 ├── Dockerfile # Docker镜像构建文件 └── README.md # 项目说明文档

现在，app目录还是空的，我们接下来就会往里面填充内容。先创建requirements.txt文件，把刚才安装的依赖写进去，方便以后部署：

fastapi==0.104.1 uvicorn[standard]==0.24.0 pydantic==2.5.0 transformers==4.36.0 torch==2.1.0

好了，舞台搭好了，演员（代码）可以准备上场了。

2. 核心代码实现：从模型加载到API响应

我们从最核心的部分开始：定义API怎么交互，以及模型怎么工作。

2.1 定义API的“对话规则”：数据模型

想象一下，你要给一个朋友发短信让他帮你生成一段文本。你需要告诉他：“主题是什么？”、“要写多长？”。API也是类似的，它需要一套明确的规则来理解你的请求。

我们在app/models.py文件里定义这些规则：

from pydantic import BaseModel, Field from typing import Optional class GenerationRequest(BaseModel): """ 文本生成请求的数据模型 """ prompt: str = Field(..., description="输入的提示文本，用于引导模型生成") max_length: Optional[int] = Field(100, description="生成文本的最大长度", ge=10, le=2048) temperature: Optional[float] = Field(0.7, description="控制生成随机性的温度参数", ge=0.1, le=2.0) top_p: Optional[float] = Field(0.9, description="核采样参数，影响生成多样性", ge=0.1, le=1.0) class Config: schema_extra = { "example": { "prompt": "用一段话介绍人工智能的未来发展。", "max_length": 150, "temperature": 0.8, "top_p": 0.95 } } class GenerationResponse(BaseModel): """ 文本生成响应的数据模型 """ generated_text: str = Field(..., description="模型生成的文本内容") prompt: str = Field(..., description="用户输入的原始提示") model: str = Field(..., description="使用的模型名称") processing_time: float = Field(..., description="模型推理耗时（秒）")

这里我们定义了两个“规则”：

GenerationRequest: 规定了客户端发来的请求必须包含一个prompt（提示词），还可以选择性地包含控制生成长度、随机性的参数。Field里的description会自动变成API文档的说明，ge/le用来限制数值范围。
GenerationResponse: 规定了服务器返回的响应里应该包含哪些信息，比如生成的文本、原始提示、模型名字和处理时间。

2.2 让模型“干活”：推理逻辑封装

接下来，我们在app/inference.py里创建一个“工人”，它的职责就是加载模型并执行生成任务。为了服务能同时处理多个请求，我们使用异步（async）的方式。

import torch from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline from typing import Dict, Any import time import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class Phi3InferenceEngine: """ Phi-3-mini模型推理引擎 """ def __init__(self, model_name: str = "microsoft/Phi-3-mini-128k-instruct"): self.model_name = model_name self.tokenizer = None self.model = None self.generator = None self.device = "cuda" if torch.cuda.is_available() else "cpu" logger.info(f"将使用设备: {self.device}") async def load_model(self): """ 异步加载模型和分词器 """ try: logger.info(f"开始加载模型: {self.model_name}") self.tokenizer = AutoTokenizer.from_pretrained(self.model_name) self.model = AutoModelForCausalLM.from_pretrained( self.model_name, torch_dtype=torch.float16 if self.device == "cuda" else torch.float32, device_map="auto" if self.device == "cuda" else None, trust_remote_code=True ) # 如果使用CPU，需要手动将模型移到设备上 if self.device == "cpu": self.model = self.model.to(self.device) self.generator = pipeline( "text-generation", model=self.model, tokenizer=self.tokenizer, device=0 if self.device == "cuda" else -1 ) logger.info("模型加载完成！") except Exception as e: logger.error(f"模型加载失败: {e}") raise async def generate_text(self, request_data: Dict[str, Any]) -> Dict[str, Any]: """ 执行文本生成 """ if not self.generator: raise RuntimeError("模型未加载，请先调用 load_model()") start_time = time.time() try: # 从请求数据中提取参数 prompt = request_data["prompt"] max_length = request_data.get("max_length", 100) temperature = request_data.get("temperature", 0.7) top_p = request_data.get("top_p", 0.9) # 构建生成参数 generation_args = { "max_length": max_length, "temperature": temperature, "top_p": top_p, "do_sample": temperature > 0, # 当temperature>0时启用采样 "num_return_sequences": 1 } # 执行生成 result = self.generator(prompt, **generation_args) generated_text = result[0]["generated_text"] # 移除重复的prompt（如果模型在生成时包含了输入） if generated_text.startswith(prompt): generated_text = generated_text[len(prompt):].strip() processing_time = time.time() - start_time return { "generated_text": generated_text, "prompt": prompt, "model": self.model_name, "processing_time": round(processing_time, 3) } except Exception as e: logger.error(f"文本生成过程中出错: {e}") raise # 创建全局推理引擎实例 inference_engine = Phi3InferenceEngine()

这段代码做了几件事：

定义了一个Phi3InferenceEngine类，它负责管理模型的生命周期。
在__init__中自动检测是否有GPU可用。
load_model方法用transformers库加载指定的模型和分词器，并根据设备选择合适的数据类型。
generate_text方法是核心，它接收请求参数，调用模型管道（pipeline）进行生成，计算耗时，并整理返回结果。

2.3 创建API的“接待处”：路由与主应用

现在，我们需要把“对话规则”和“工人”连接起来，创建一个Web服务入口。这就是app/main.py要做的事情。

from fastapi import FastAPI, Depends, HTTPException, status from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager import logging from app.models import GenerationRequest, GenerationResponse from app.inference import inference_engine from app.dependencies import verify_api_key # 我们稍后会创建这个依赖项 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 使用生命周期管理器管理模型加载和卸载 @asynccontextmanager async def lifespan(app: FastAPI): # 启动时加载模型 logger.info("正在启动服务，加载模型中...") await inference_engine.load_model() logger.info("服务启动完成，模型就绪。") yield # 关闭时清理资源（如果有需要的话） logger.info("服务正在关闭...") # 创建FastAPI应用实例，并传入生命周期管理器 app = FastAPI( title="Phi-3-mini-128k-instruct API服务", description="基于FastAPI封装的Phi-3-mini-128k-instruct模型文本生成API", version="1.0.0", lifespan=lifespan ) # 添加CORS中间件，允许前端跨域访问（根据实际情况调整origins） app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境应替换为具体的域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/") async def root(): """ 根路径，返回服务基本信息 """ return { "message": "Phi-3-mini-128k-instruct API服务正在运行", "docs": "/docs", "health": "/health" } @app.get("/health") async def health_check(): """ 健康检查端点 """ return {"status": "healthy", "model_loaded": inference_engine.model is not None} @app.post("/generate", response_model=GenerationResponse) async def generate_text( request: GenerationRequest # 如果需要API密钥认证，可以在这里添加: api_key: str = Depends(verify_api_key) ): """ 文本生成接口 - **prompt**: 输入的提示文本 - **max_length**: 生成文本的最大长度 (默认: 100) - **temperature**: 温度参数，控制随机性 (默认: 0.7) - **top_p**: 核采样参数 (默认: 0.9) """ try: logger.info(f"收到生成请求，prompt长度: {len(request.prompt)}") # 将Pydantic模型转换为字典，传给推理引擎 result = await inference_engine.generate_text(request.dict()) return GenerationResponse(**result) except Exception as e: logger.error(f"API处理请求时出错: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"文本生成失败: {str(e)}" )

这个文件是服务的核心：

lifespan函数确保了服务启动时自动加载模型，关闭时（如果有需要）清理资源。
创建了FastAPI应用实例，并配置了标题、描述等信息。
添加了CORS中间件，方便前后端分离的项目调用。
定义了三个API端点：
- GET /: 根路径，返回服务状态。
- GET /health: 健康检查，用于监控服务是否正常。
- POST /generate:核心接口，接收GenerationRequest格式的数据，调用推理引擎，并返回GenerationResponse格式的结果。这里我们还预留了API密钥认证的接口。

3. 进阶功能：让API更安全、更健壮

一个基础的API服务已经完成了。但要想在生产环境使用，我们还需要给它加上一些“安全锁”和“防护栏”。

3.1 添加API密钥认证

我们不想让任何人都能随意调用我们的API，尤其是如果它部署在公网上。一个简单有效的方法是使用API密钥。在app/dependencies.py中实现：

from fastapi import HTTPException, status, Header import os # 在实际环境中，应该从环境变量或安全的配置管理中读取 # 这里为了演示，使用一个固定的密钥。生产环境请务必使用复杂密钥并从环境变量读取。 API_KEYS = { os.getenv("API_KEY", "your-secret-api-key-here-12345") } async def verify_api_key(api_key: str = Header(..., alias="X-API-Key")): """ 验证API密钥的依赖项。 客户端需要在请求头中携带 `X-API-Key: your-api-key`。 """ if api_key not in API_KEYS: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="无效或缺失的API密钥" ) return api_key

然后，回到app/main.py，修改/generate接口，加入这个依赖：

@app.post("/generate", response_model=GenerationResponse) async def generate_text( request: GenerationRequest, api_key: str = Depends(verify_api_key) # 添加这行 ): # ... 函数体其他部分不变

这样，客户端在调用/generate接口时，必须在HTTP请求头中带上正确的X-API-Key，否则会被拒绝访问。

3.2 用Docker进行容器化部署

手动在服务器上配置Python环境、安装依赖是一件繁琐且容易出错的事情。Docker可以帮助我们把应用及其所有依赖打包成一个独立的“集装箱”，在任何支持Docker的环境中都能以相同的方式运行。

在项目根目录创建Dockerfile：

# 使用官方Python运行时作为父镜像 FROM python:3.10-slim # 设置工作目录 WORKDIR /app # 设置环境变量，防止Python输出被缓冲 ENV PYTHONUNBUFFERED=1 # 安装系统依赖（如果需要） RUN apt-get update && apt-get install -y \ gcc \ g++ \ && rm -rf /var/lib/apt/lists/* # 将依赖文件复制到容器中 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 将当前目录内容复制到容器的/app中 COPY . . # 暴露端口（FastAPI默认运行在8000端口） EXPOSE 8000 # 定义环境变量（例如API密钥，在生产环境中应通过docker run或编排工具传入） # ENV API_KEY=your-production-key-here # 启动应用 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

同时，创建一个.dockerignore文件，告诉Docker哪些文件不需要打包进镜像，可以减小镜像体积：

__pycache__ *.pyc *.pyo *.pyd .Python env venv .venv .env .git .DS_Store README.md test*

现在，你可以使用以下命令来构建Docker镜像并运行：

# 在项目根目录（phi3-api-service）下执行 # 1. 构建Docker镜像 docker build -t phi3-api-service . # 2. 运行Docker容器 # -p 8000:8000: 将本地的8000端口映射到容器的8000端口 # -e API_KEY=my-secret-key: 设置环境变量（覆盖Dockerfile中的默认值） docker run -d --name phi3-api -p 8000:8000 -e API_KEY=my-secret-key phi3-api-service

运行成功后，打开浏览器访问http://localhost:8000/docs，你就能看到自动生成的、交互式的API文档（Swagger UI），并可以直接在那里测试/generate接口！

4. 测试与运行你的API服务

让我们最后检查一下所有环节，并启动服务。

4.1 本地运行测试

如果你不想用Docker，也可以在本地直接运行。确保你在项目根目录，然后执行：

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

--reload: 代码修改后自动重启服务，适合开发。
--host 0.0.0.0: 让服务监听所有网络接口，方便其他设备访问。
--port 8000: 指定端口。

访问http://localhost:8000/docs，你应该能看到API文档页面。尝试点击/generate接口的 “Try it out” 按钮，输入一个prompt，比如“讲一个关于太空探险的短故事”，然后点击 “Execute”。稍等片刻，你就能在 “Responses” 部分看到模型生成的文本和详细的响应信息了。

4.2 使用cURL或Python客户端测试

除了在网页上测试，你也可以用命令行工具cURL或者写一个简单的Python脚本来调用API。

使用cURL:

curl -X POST "http://localhost:8000/generate" \ -H "Content-Type: application/json" \ -H "X-API-Key: my-secret-key" \ -d '{ "prompt": "用Python写一个简单的Hello World程序", "max_length": 50 }'

使用Python脚本 (test_client.py):

import requests import json url = "http://localhost:8000/generate" headers = { "Content-Type": "application/json", "X-API-Key": "my-secret-key" # 如果启用了认证 } data = { "prompt": "解释一下机器学习的基本概念", "max_length": 150, "temperature": 0.8 } response = requests.post(url, headers=headers, data=json.dumps(data)) if response.status_code == 200: result = response.json() print("生成结果:") print(result["generated_text"]) print(f"\n处理耗时: {result['processing_time']}秒") else: print(f"请求失败，状态码: {response.status_code}") print(response.text)

运行这个脚本，你就能在终端看到API返回的结果了。

走到这里，你已经成功地将一个本地运行的Phi-3-mini模型，封装成了一个具备标准HTTP接口、基础认证、并且可以容器化部署的Web服务。整个过程其实就像搭积木，把模型加载、请求处理、网络响应、安全认证、部署打包这几个模块组合起来。

这个服务现在虽然基础，但已经具备了核心功能。在实际项目中，你可能还需要考虑更多，比如加入更完善的日志记录、性能监控、请求限流、以及使用Nginx等反向代理做负载均衡。但无论如何，你现在拥有的这个API服务，已经可以很方便地集成到你的其他应用、自动化脚本，或者分享给团队成员使用了。下次当你需要把某个AI模型的能力开放出去时，不妨试试这个FastAPI的“配方”，相信它能帮你节省不少时间。