Phi-3-mini-128k-instruct API服务化部署使用FastAPI构建高性能接口你是不是已经成功部署了Phi-3-mini-128k-instruct模型能在本地跑通推理但总觉得还差点意思比如想把它集成到自己的Web应用里或者让团队其他成员也能方便地调用又或者需要处理来自不同客户端的并发请求。这时候仅仅一个能跑起来的模型脚本就显得不够用了。把模型封装成一个标准的API服务就像给模型装上一个“标准插座”任何符合标准的“插头”客户端都能轻松接入。今天我就带你一步步用FastAPI把这个想法变成现实。我们不止要搭出一个能用的接口更要构建一个高性能、易维护、生产就绪的服务。从最基础的请求响应到异步处理提升并发能力再到用Nginx和Gunicorn武装起来应对真实流量这篇教程都会涵盖。跟着做下来你不仅能得到一个随时可用的模型API更能掌握一套将任何AI模型服务化的通用方法论。1. 为什么需要API服务化从脚本到服务在开始敲代码之前咱们先聊聊为什么费这个劲。你可能会想我写个Python脚本用requests库发个POST请求不也能调用吗确实可以但那只是“能用”离“好用”和“敢用”还差得远。想象一下你的模型是个大厨。直接调用脚本就像你每次想吃菜都得亲自跑到后厨跟大厨说一遍要求然后站着等。而API服务化则是给餐厅装了个前台和传菜系统。顾客客户端只需要在前台API接口点单发送请求剩下的排队、通知后厨、传菜、处理不同顾客的并发需求都由系统自动完成。具体来说API服务化能带来几个实实在在的好处标准化提供统一的RESTful接口无论是用Python、JavaScript、Java还是Curl都能以同样的方式调用极大降低了集成成本。可管理性服务可以独立部署、监控、扩缩容。你能清楚地看到请求量、响应时间、错误率而不需要去翻看杂乱的日志文件。高性能与并发通过异步框架和Web服务器可以高效处理成千上万的并发请求这是单线程脚本无法做到的。安全与稳定可以方便地添加认证、限流、负载均衡等中间件保障服务稳定和数据安全。文档与协作像FastAPI能自动生成交互式API文档前后端开发人员可以基于这份清晰的文档并行工作。所以把Phi-3-mini模型服务化是从个人玩具走向团队工具乃至产品组件的关键一步。接下来我们就选用FastAPI这个利器来开工。2. 项目初始化与环境准备工欲善其事必先利其器。我们先建立一个清晰的项目目录并安装必要的依赖。2.1 创建项目结构打开你的终端创建一个新的项目文件夹并进入其中。mkdir phi3-mini-api-service cd phi3-mini-api-service然后创建以下文件和文件夹。一个清晰的结构能让后续开发和维护轻松很多。phi3-mini-api-service/ ├── app/ # 核心应用代码 │ ├── __init__.py │ ├── main.py # FastAPI应用主入口 │ ├── api/ # 路由端点 │ │ ├── __init__.py │ │ └── endpoints.py # 模型推理等核心端点 │ ├── core/ # 核心配置与逻辑 │ │ ├── __init__.py │ │ ├── config.py # 配置文件 │ │ └── model_loader.py # 模型加载与单例管理 │ ├── models/ # Pydantic数据模型请求/响应体 │ │ ├── __init__.py │ │ └── schemas.py │ └── utils/ # 工具函数 │ ├── __init__.py │ └── logging_setup.py ├── requirements.txt # 项目依赖 ├── Dockerfile # Docker容器化配置可选 ├── docker-compose.yml # 服务编排可选 ├── nginx.conf # Nginx配置文件 ├── gunicorn_conf.py # Gunicorn配置文件 └── README.md2.2 安装依赖在项目根目录下创建requirements.txt文件并填入以下内容。这里包含了Web框架、模型推理、异步处理等核心库。fastapi0.104.1 uvicorn[standard]0.24.0 pydantic2.5.0 transformers4.35.0 torch2.1.0 accelerate0.24.0 python-multipart0.0.6 gunicorn21.2.0 httpx0.25.1 python-dotenv1.0.0 loguru0.7.2使用pip安装它们。建议使用虚拟环境。pip install -r requirements.txt如果你的Phi-3-mini模型需要特定的依赖或版本请确保也已安装。环境准备好后我们就可以开始编写核心代码了。3. 构建FastAPI应用核心我们将从内到外构建应用先处理模型加载这个最重的部分再定义清晰的数据接口最后用API把它们暴露出去。3.1 模型加载与管理core/model_loader.py模型加载耗时耗内存我们必须确保它在整个服务生命周期中只加载一次并被所有请求共享。这里采用单例模式。# app/core/model_loader.py import torch from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline from loguru import logger import asyncio from typing import Optional class ModelManager: _instance None _model None _tokenizer None _pipe None def __new__(cls): if cls._instance is None: cls._instance super(ModelManager, cls).__new__(cls) cls._instance._initialized False return cls._instance def initialize(self, model_name_or_path: str, device: Optional[str] None): 初始化模型和分词器确保只加载一次 if self._initialized: logger.info(模型已经初始化跳过重复加载。) return logger.info(f开始加载模型: {model_name_or_path}) if device is None: device cuda if torch.cuda.is_available() else cpu logger.info(f使用设备: {device}) try: # 加载分词器 self._tokenizer AutoTokenizer.from_pretrained(model_name_or_path, trust_remote_codeTrue) # 加载模型 self._model AutoModelForCausalLM.from_pretrained( model_name_or_path, torch_dtypetorch.float16 if device cuda else torch.float32, device_mapauto if device cuda else None, trust_remote_codeTrue ) # 创建文本生成pipeline简化调用 self._pipe pipeline( text-generation, modelself._model, tokenizerself._tokenizer, device0 if device cuda else -1, ) self._initialized True logger.success(模型与分词器加载成功) except Exception as e: logger.error(f模型加载失败: {e}) raise property def pipeline(self): if not self._initialized: raise RuntimeError(模型未初始化请先调用 initialize() 方法。) return self._pipe property def tokenizer(self): if not self._initialized: raise RuntimeError(模型未初始化请先调用 initialize() 方法。) return self._tokenizer # 创建全局单例 model_manager ModelManager()这个ModelManager类确保了无论你在代码中哪里引用model_manager得到的都是同一个实例背后的模型也只加载一次。3.2 定义请求与响应模型models/schemas.py使用Pydantic定义清晰的数据结构这不仅能自动验证请求数据还能为自动生成的API文档提供完美的说明。# app/models/schemas.py from pydantic import BaseModel, Field from typing import Optional, List class GenerationRequest(BaseModel): 文本生成请求体 prompt: str Field(..., description输入的提示文本, example请用Python写一个快速排序函数。) max_new_tokens: int Field(default512, ge1, le4096, description生成的最大新令牌数) temperature: float Field(default0.7, ge0.1, le2.0, description采样温度控制随机性) top_p: float Field(default0.9, ge0.1, le1.0, description核采样参数) do_sample: bool Field(defaultTrue, description是否使用采样) repetition_penalty: float Field(default1.1, ge1.0, le2.0, description重复惩罚系数) class Config: schema_extra { example: { prompt: 请用Python写一个快速排序函数。, max_new_tokens: 256, temperature: 0.8, } } class GeneratedText(BaseModel): 单条生成结果 generated_text: str Field(..., description模型生成的文本) class GenerationResponse(BaseModel): 文本生成响应体 status: str Field(defaultsuccess, description请求状态) request_id: Optional[str] Field(None, description本次请求的唯一ID) data: List[GeneratedText] Field(..., description生成的结果列表) inference_time: Optional[float] Field(None, description推理耗时秒) prompt_tokens: Optional[int] Field(None, description提示词令牌数) total_tokens: Optional[int] Field(None, description总令牌数) class HealthResponse(BaseModel): 健康检查响应 status: str Field(..., description服务状态) model_loaded: bool Field(..., description模型是否已加载) device: Optional[str] Field(None, description模型运行设备)3.3 实现API路由端点api/endpoints.py现在我们来创建真正的API接口。我们将创建两个端点一个用于健康检查一个用于核心的文本生成。# app/api/endpoints.py from fastapi import APIRouter, HTTPException, BackgroundTasks from app.models.schemas import GenerationRequest, GenerationResponse, HealthResponse from app.core.model_loader import model_manager from app.core.config import settings import time import uuid from loguru import logger router APIRouter() router.get(/health, response_modelHealthResponse, summary健康检查, tags[系统]) async def health_check(): 检查API服务及模型状态 is_loaded model_manager._initialized return HealthResponse( statushealthy if is_loaded else model_not_loaded, model_loadedis_loaded, devicecuda if settings.use_cuda else cpu ) router.post(/generate, response_modelGenerationResponse, summary文本生成, tags[推理]) async def generate_text(request: GenerationRequest, background_tasks: BackgroundTasks): 根据提示词生成文本。 - **prompt**: 必需的提示文本 - **max_new_tokens**: 控制生成长度 - **temperature**: 值越高输出越随机值越低输出越确定 - **top_p**: 仅从累积概率超过此值的令牌中采样 - **...其他参数见请求体模型描述** if not model_manager._initialized: raise HTTPException(status_code503, detail模型未加载服务不可用) request_id str(uuid.uuid4())[:8] logger.info(f收到生成请求 [{request_id}]: prompt前缀{request.prompt[:50]}...) start_time time.time() try: # 调用模型pipeline进行生成 results model_manager.pipeline( request.prompt, max_new_tokensrequest.max_new_tokens, temperaturerequest.temperature, top_prequest.top_p, do_samplerequest.do_sample, repetition_penaltyrequest.repetition_penalty, num_return_sequences1, # 每次请求返回一个序列 eos_token_idmodel_manager.tokenizer.eos_token_id, ) inference_time time.time() - start_time # 计算令牌数粗略估算 prompt_tokens len(model_manager.tokenizer.encode(request.prompt)) total_tokens prompt_tokens request.max_new_tokens # 近似值 logger.info(f请求 [{request_id}] 完成耗时 {inference_time:.2f}s) response_data [{generated_text: res[generated_text]} for res in results] return GenerationResponse( statussuccess, request_idrequest_id, dataresponse_data, inference_timeinference_time, prompt_tokensprompt_tokens, total_tokenstotal_tokens ) except Exception as e: logger.error(f请求 [{request_id}] 处理失败: {e}) raise HTTPException(status_code500, detailf生成过程中发生错误: {str(e)})注意我们使用了async def来定义端点函数。虽然模型推理本身是CPU/GPU密集型操作通常是阻塞的但将其放在FastAPI的异步环境中配合后文会提到的Gunicorn工作进程可以更好地处理高并发下的I/O等待如网络请求排队提升整体吞吐量。真正的模型计算会在独立的线程池中执行。3.4 应用配置与主入口core/config.py main.py我们需要一个地方来管理配置比如模型路径、服务端口等。# app/core/config.py from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): 应用配置 api_title: str Phi-3-mini-128k-instruct API api_description: str 基于FastAPI构建的高性能Phi-3-mini模型API服务 api_version: str v1.0.0 model_path: str ./models/phi-3-mini-128k-instruct # 你的模型本地路径 use_cuda: bool True # 是否使用GPU host: str 0.0.0.0 port: int 8000 reload: bool False # 生产环境设为False class Config: env_file .env # 可以从.env文件读取配置 settings Settings()最后创建FastAPI应用实例并集成我们写的所有模块。# app/main.py from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api.endpoints import router as api_router from app.core.config import settings from app.core.model_loader import model_manager from app.utils.logging_setup import setup_logging import uvicorn # 设置日志 setup_logging() # 创建FastAPI应用 app FastAPI( titlesettings.api_title, descriptionsettings.api_description, versionsettings.api_version, docs_url/docs, # Swagger UI 文档地址 redoc_url/redoc, # ReDoc 文档地址 ) # 添加CORS中间件允许前端跨域请求生产环境应严格限制 app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境请替换为具体的前端地址 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 启动时加载模型 app.on_event(startup) async def startup_event(): logger.info(正在启动API服务...) try: device cuda if settings.use_cuda else cpu model_manager.initialize(settings.model_path, devicedevice) logger.success(服务启动完成模型已就绪。) except Exception as e: logger.critical(f服务启动失败模型加载异常: {e}) # 根据实际情况可以选择让服务启动失败 raise # 注册API路由 app.include_router(api_router, prefix/api/v1) # 根路径重定向到文档 app.get(/) async def root(): return {message: f欢迎访问 {settings.api_title}请访问 /docs 查看API文档。} # 本地开发直接运行 if __name__ __main__: uvicorn.run( app.main:app, hostsettings.host, portsettings.port, reloadsettings.reload, log_levelinfo )4. 本地运行与测试代码都写好了让我们先在本地点火测试一下。首先确保你的Phi-3-mini模型已经下载并且settings.model_path指向了正确的路径。然后在项目根目录下运行python -m app.main如果一切顺利你会看到服务启动的日志最后显示类似Uvicorn running on http://0.0.0.0:8000的信息。打开浏览器访问http://127.0.0.1:8000/docs。你会看到一个自动生成的、交互式的Swagger API文档页面。这就是FastAPI的魅力之一代码即文档。在文档页面里你可以点击/api/v1/health的Try it out再点Execute看看健康检查是否返回model_loaded: true。点击/api/v1/generate的Try it out在请求体里填入你的提示词例如{prompt: 请用Python写一个简单的HTTP服务器。}然后执行。稍等片刻你就能在“Responses”部分看到模型生成的代码了你也可以用curl命令测试curl -X POST http://127.0.0.1:8000/api/v1/generate \ -H Content-Type: application/json \ -d {prompt: 解释一下机器学习中的过拟合现象。, max_new_tokens: 200}本地测试通过证明我们的核心逻辑没问题。但这样的服务还比较“脆弱”不适合直接对外。接下来我们把它武装起来。5. 生产环境部署Gunicorn Nginx直接用python命令运行是单进程的性能有限也不稳定。我们需要一个ASGI服务器来管理多个工作进程再用一个Web服务器反向代理在前面处理静态文件、负载均衡和SSL。5.1 使用Gunicorn作为ASGI服务器Gunicorn是一个成熟的WSGI/ASGI服务器可以管理多个工作进程提高并发能力。为FastAPI这样的异步应用我们需要使用Uvicorn的工作器Worker。首先创建一个Gunicorn的配置文件gunicorn_conf.py# gunicorn_conf.py import multiprocessing # 服务器套接字 bind 0.0.0.0:8000 # 工作进程数通常设置为 CPU核心数 * 2 1 workers multiprocessing.cpu_count() * 2 1 # 每个工作进程的线程数 (对于异步框架通常为1使用异步worker) threads 1 # 工作器类型使用Uvicorn的异步工作器 worker_class uvicorn.workers.UvicornWorker # 工作进程最大请求数防止内存泄漏 max_requests 1000 max_requests_jitter 50 # 超时设置 timeout 120 keepalive 2 # 日志配置 accesslog - # 访问日志输出到标准输出 errorlog - # 错误日志输出到标准错误 loglevel info # 进程名 proc_name phi3_mini_api然后使用Gunicorn启动服务gunicorn -c gunicorn_conf.py app.main:app现在你的服务运行在多个工作进程上能更好地利用多核CPU并发处理能力大大提升。5.2 使用Nginx作为反向代理Nginx可以处理静态文件、SSL/TLS加密、负载均衡并将请求高效地转发给后端的Gunicorn。它还能提供一层安全防护。创建一个简单的Nginx配置文件nginx.conf# nginx.conf events { worker_connections 1024; } http { upstream api_server { # 指向Gunicorn服务地址这里假设Gunicorn运行在本机8000端口 server 127.0.0.1:8000; # 可以添加多个server行实现负载均衡 # server 127.0.0.1:8001; } server { listen 80; server_name your_domain.com; # 替换为你的域名或IP # 静态文件服务如果有的话 # location /static { # alias /path/to/your/static/files; # } location / { # 将请求代理到Gunicorn proxy_pass http://api_server; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; # 一些超时设置 proxy_connect_timeout 60s; proxy_send_timeout 60s; proxy_read_timeout 60s; } # 限制客户端请求体大小防止过大提示词攻击 client_max_body_size 10M; } }启动Nginx假设你已安装Nginx# 测试配置文件语法 sudo nginx -t -c /path/to/your/phi3-mini-api-service/nginx.conf # 使用指定配置文件启动或重新加载 sudo nginx -c /path/to/your/phi3-mini-api-service/nginx.conf # 或者如果已有一个默认nginx在运行可以修改默认配置后重载 # sudo systemctl reload nginx现在你的API服务可以通过Nginx监听的80端口或配置的域名访问了。外部流量先到Nginx再由Nginx转发给Gunicorn集群。5.3 使用系统服务管理Systemd为了确保服务在服务器重启后能自动运行我们将其配置为系统服务。创建一个服务文件/etc/systemd/system/phi3-api.service[Unit] DescriptionPhi-3-mini API Service Afternetwork.target [Service] Useryour_username # 替换为你的用户名 Groupyour_groupname # 替换为你的用户组 WorkingDirectory/path/to/your/phi3-mini-api-service EnvironmentPATH/path/to/your/venv/bin # 虚拟环境路径 ExecStart/path/to/your/venv/bin/gunicorn -c gunicorn_conf.py app.main:app Restartalways RestartSec3 [Install] WantedBymulti-user.target然后启用并启动服务sudo systemctl daemon-reload sudo systemctl enable phi3-api.service sudo systemctl start phi3-api.service # 查看状态和日志 sudo systemctl status phi3-api.service sudo journalctl -u phi3-api.service -f至此一个具备生产级基础能力的Phi-3-mini模型API服务就部署完成了。它具备了并发处理、反向代理、服务自启等关键特性。6. 总结走完这一趟我们从本地的一个模型脚本构建出了一个结构清晰、性能可观、便于集成的标准API服务。回顾一下核心步骤首先是项目结构和环境搭建这是所有好项目的基础然后是用单例模式稳妥地加载和管理模型这个“重资产”接着用Pydantic定义清晰的接口契约并用FastAPI快速实现端点本地测试通过后再用Gunicorn和Nginx这套经典组合为服务穿上“铠甲”提升其并发能力和安全性最后通过Systemd让服务能够稳定运行。现在你的Phi-3-mini模型已经不再是一个孤立的程序而是一个可以通过网络被各种应用调用的智能服务。你可以在此基础上继续添加API密钥认证、请求限流、更完善的监控和日志甚至结合Docker容器化部署让它更加健壮和易用。希望这篇教程能为你打开AI模型工程化落地的一扇门。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
Phi-3-mini-128k-instruct API服务化部署：使用FastAPI构建高性能接口
Phi-3-mini-128k-instruct API服务化部署使用FastAPI构建高性能接口你是不是已经成功部署了Phi-3-mini-128k-instruct模型能在本地跑通推理但总觉得还差点意思比如想把它集成到自己的Web应用里或者让团队其他成员也能方便地调用又或者需要处理来自不同客户端的并发请求。这时候仅仅一个能跑起来的模型脚本就显得不够用了。把模型封装成一个标准的API服务就像给模型装上一个“标准插座”任何符合标准的“插头”客户端都能轻松接入。今天我就带你一步步用FastAPI把这个想法变成现实。我们不止要搭出一个能用的接口更要构建一个高性能、易维护、生产就绪的服务。从最基础的请求响应到异步处理提升并发能力再到用Nginx和Gunicorn武装起来应对真实流量这篇教程都会涵盖。跟着做下来你不仅能得到一个随时可用的模型API更能掌握一套将任何AI模型服务化的通用方法论。1. 为什么需要API服务化从脚本到服务在开始敲代码之前咱们先聊聊为什么费这个劲。你可能会想我写个Python脚本用requests库发个POST请求不也能调用吗确实可以但那只是“能用”离“好用”和“敢用”还差得远。想象一下你的模型是个大厨。直接调用脚本就像你每次想吃菜都得亲自跑到后厨跟大厨说一遍要求然后站着等。而API服务化则是给餐厅装了个前台和传菜系统。顾客客户端只需要在前台API接口点单发送请求剩下的排队、通知后厨、传菜、处理不同顾客的并发需求都由系统自动完成。具体来说API服务化能带来几个实实在在的好处标准化提供统一的RESTful接口无论是用Python、JavaScript、Java还是Curl都能以同样的方式调用极大降低了集成成本。可管理性服务可以独立部署、监控、扩缩容。你能清楚地看到请求量、响应时间、错误率而不需要去翻看杂乱的日志文件。高性能与并发通过异步框架和Web服务器可以高效处理成千上万的并发请求这是单线程脚本无法做到的。安全与稳定可以方便地添加认证、限流、负载均衡等中间件保障服务稳定和数据安全。文档与协作像FastAPI能自动生成交互式API文档前后端开发人员可以基于这份清晰的文档并行工作。所以把Phi-3-mini模型服务化是从个人玩具走向团队工具乃至产品组件的关键一步。接下来我们就选用FastAPI这个利器来开工。2. 项目初始化与环境准备工欲善其事必先利其器。我们先建立一个清晰的项目目录并安装必要的依赖。2.1 创建项目结构打开你的终端创建一个新的项目文件夹并进入其中。mkdir phi3-mini-api-service cd phi3-mini-api-service然后创建以下文件和文件夹。一个清晰的结构能让后续开发和维护轻松很多。phi3-mini-api-service/ ├── app/ # 核心应用代码 │ ├── __init__.py │ ├── main.py # FastAPI应用主入口 │ ├── api/ # 路由端点 │ │ ├── __init__.py │ │ └── endpoints.py # 模型推理等核心端点 │ ├── core/ # 核心配置与逻辑 │ │ ├── __init__.py │ │ ├── config.py # 配置文件 │ │ └── model_loader.py # 模型加载与单例管理 │ ├── models/ # Pydantic数据模型请求/响应体 │ │ ├── __init__.py │ │ └── schemas.py │ └── utils/ # 工具函数 │ ├── __init__.py │ └── logging_setup.py ├── requirements.txt # 项目依赖 ├── Dockerfile # Docker容器化配置可选 ├── docker-compose.yml # 服务编排可选 ├── nginx.conf # Nginx配置文件 ├── gunicorn_conf.py # Gunicorn配置文件 └── README.md2.2 安装依赖在项目根目录下创建requirements.txt文件并填入以下内容。这里包含了Web框架、模型推理、异步处理等核心库。fastapi0.104.1 uvicorn[standard]0.24.0 pydantic2.5.0 transformers4.35.0 torch2.1.0 accelerate0.24.0 python-multipart0.0.6 gunicorn21.2.0 httpx0.25.1 python-dotenv1.0.0 loguru0.7.2使用pip安装它们。建议使用虚拟环境。pip install -r requirements.txt如果你的Phi-3-mini模型需要特定的依赖或版本请确保也已安装。环境准备好后我们就可以开始编写核心代码了。3. 构建FastAPI应用核心我们将从内到外构建应用先处理模型加载这个最重的部分再定义清晰的数据接口最后用API把它们暴露出去。3.1 模型加载与管理core/model_loader.py模型加载耗时耗内存我们必须确保它在整个服务生命周期中只加载一次并被所有请求共享。这里采用单例模式。# app/core/model_loader.py import torch from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline from loguru import logger import asyncio from typing import Optional class ModelManager: _instance None _model None _tokenizer None _pipe None def __new__(cls): if cls._instance is None: cls._instance super(ModelManager, cls).__new__(cls) cls._instance._initialized False return cls._instance def initialize(self, model_name_or_path: str, device: Optional[str] None): 初始化模型和分词器确保只加载一次 if self._initialized: logger.info(模型已经初始化跳过重复加载。) return logger.info(f开始加载模型: {model_name_or_path}) if device is None: device cuda if torch.cuda.is_available() else cpu logger.info(f使用设备: {device}) try: # 加载分词器 self._tokenizer AutoTokenizer.from_pretrained(model_name_or_path, trust_remote_codeTrue) # 加载模型 self._model AutoModelForCausalLM.from_pretrained( model_name_or_path, torch_dtypetorch.float16 if device cuda else torch.float32, device_mapauto if device cuda else None, trust_remote_codeTrue ) # 创建文本生成pipeline简化调用 self._pipe pipeline( text-generation, modelself._model, tokenizerself._tokenizer, device0 if device cuda else -1, ) self._initialized True logger.success(模型与分词器加载成功) except Exception as e: logger.error(f模型加载失败: {e}) raise property def pipeline(self): if not self._initialized: raise RuntimeError(模型未初始化请先调用 initialize() 方法。) return self._pipe property def tokenizer(self): if not self._initialized: raise RuntimeError(模型未初始化请先调用 initialize() 方法。) return self._tokenizer # 创建全局单例 model_manager ModelManager()这个ModelManager类确保了无论你在代码中哪里引用model_manager得到的都是同一个实例背后的模型也只加载一次。3.2 定义请求与响应模型models/schemas.py使用Pydantic定义清晰的数据结构这不仅能自动验证请求数据还能为自动生成的API文档提供完美的说明。# app/models/schemas.py from pydantic import BaseModel, Field from typing import Optional, List class GenerationRequest(BaseModel): 文本生成请求体 prompt: str Field(..., description输入的提示文本, example请用Python写一个快速排序函数。) max_new_tokens: int Field(default512, ge1, le4096, description生成的最大新令牌数) temperature: float Field(default0.7, ge0.1, le2.0, description采样温度控制随机性) top_p: float Field(default0.9, ge0.1, le1.0, description核采样参数) do_sample: bool Field(defaultTrue, description是否使用采样) repetition_penalty: float Field(default1.1, ge1.0, le2.0, description重复惩罚系数) class Config: schema_extra { example: { prompt: 请用Python写一个快速排序函数。, max_new_tokens: 256, temperature: 0.8, } } class GeneratedText(BaseModel): 单条生成结果 generated_text: str Field(..., description模型生成的文本) class GenerationResponse(BaseModel): 文本生成响应体 status: str Field(defaultsuccess, description请求状态) request_id: Optional[str] Field(None, description本次请求的唯一ID) data: List[GeneratedText] Field(..., description生成的结果列表) inference_time: Optional[float] Field(None, description推理耗时秒) prompt_tokens: Optional[int] Field(None, description提示词令牌数) total_tokens: Optional[int] Field(None, description总令牌数) class HealthResponse(BaseModel): 健康检查响应 status: str Field(..., description服务状态) model_loaded: bool Field(..., description模型是否已加载) device: Optional[str] Field(None, description模型运行设备)3.3 实现API路由端点api/endpoints.py现在我们来创建真正的API接口。我们将创建两个端点一个用于健康检查一个用于核心的文本生成。# app/api/endpoints.py from fastapi import APIRouter, HTTPException, BackgroundTasks from app.models.schemas import GenerationRequest, GenerationResponse, HealthResponse from app.core.model_loader import model_manager from app.core.config import settings import time import uuid from loguru import logger router APIRouter() router.get(/health, response_modelHealthResponse, summary健康检查, tags[系统]) async def health_check(): 检查API服务及模型状态 is_loaded model_manager._initialized return HealthResponse( statushealthy if is_loaded else model_not_loaded, model_loadedis_loaded, devicecuda if settings.use_cuda else cpu ) router.post(/generate, response_modelGenerationResponse, summary文本生成, tags[推理]) async def generate_text(request: GenerationRequest, background_tasks: BackgroundTasks): 根据提示词生成文本。 - **prompt**: 必需的提示文本 - **max_new_tokens**: 控制生成长度 - **temperature**: 值越高输出越随机值越低输出越确定 - **top_p**: 仅从累积概率超过此值的令牌中采样 - **...其他参数见请求体模型描述** if not model_manager._initialized: raise HTTPException(status_code503, detail模型未加载服务不可用) request_id str(uuid.uuid4())[:8] logger.info(f收到生成请求 [{request_id}]: prompt前缀{request.prompt[:50]}...) start_time time.time() try: # 调用模型pipeline进行生成 results model_manager.pipeline( request.prompt, max_new_tokensrequest.max_new_tokens, temperaturerequest.temperature, top_prequest.top_p, do_samplerequest.do_sample, repetition_penaltyrequest.repetition_penalty, num_return_sequences1, # 每次请求返回一个序列 eos_token_idmodel_manager.tokenizer.eos_token_id, ) inference_time time.time() - start_time # 计算令牌数粗略估算 prompt_tokens len(model_manager.tokenizer.encode(request.prompt)) total_tokens prompt_tokens request.max_new_tokens # 近似值 logger.info(f请求 [{request_id}] 完成耗时 {inference_time:.2f}s) response_data [{generated_text: res[generated_text]} for res in results] return GenerationResponse( statussuccess, request_idrequest_id, dataresponse_data, inference_timeinference_time, prompt_tokensprompt_tokens, total_tokenstotal_tokens ) except Exception as e: logger.error(f请求 [{request_id}] 处理失败: {e}) raise HTTPException(status_code500, detailf生成过程中发生错误: {str(e)})注意我们使用了async def来定义端点函数。虽然模型推理本身是CPU/GPU密集型操作通常是阻塞的但将其放在FastAPI的异步环境中配合后文会提到的Gunicorn工作进程可以更好地处理高并发下的I/O等待如网络请求排队提升整体吞吐量。真正的模型计算会在独立的线程池中执行。3.4 应用配置与主入口core/config.py main.py我们需要一个地方来管理配置比如模型路径、服务端口等。# app/core/config.py from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): 应用配置 api_title: str Phi-3-mini-128k-instruct API api_description: str 基于FastAPI构建的高性能Phi-3-mini模型API服务 api_version: str v1.0.0 model_path: str ./models/phi-3-mini-128k-instruct # 你的模型本地路径 use_cuda: bool True # 是否使用GPU host: str 0.0.0.0 port: int 8000 reload: bool False # 生产环境设为False class Config: env_file .env # 可以从.env文件读取配置 settings Settings()最后创建FastAPI应用实例并集成我们写的所有模块。# app/main.py from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api.endpoints import router as api_router from app.core.config import settings from app.core.model_loader import model_manager from app.utils.logging_setup import setup_logging import uvicorn # 设置日志 setup_logging() # 创建FastAPI应用 app FastAPI( titlesettings.api_title, descriptionsettings.api_description, versionsettings.api_version, docs_url/docs, # Swagger UI 文档地址 redoc_url/redoc, # ReDoc 文档地址 ) # 添加CORS中间件允许前端跨域请求生产环境应严格限制 app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境请替换为具体的前端地址 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 启动时加载模型 app.on_event(startup) async def startup_event(): logger.info(正在启动API服务...) try: device cuda if settings.use_cuda else cpu model_manager.initialize(settings.model_path, devicedevice) logger.success(服务启动完成模型已就绪。) except Exception as e: logger.critical(f服务启动失败模型加载异常: {e}) # 根据实际情况可以选择让服务启动失败 raise # 注册API路由 app.include_router(api_router, prefix/api/v1) # 根路径重定向到文档 app.get(/) async def root(): return {message: f欢迎访问 {settings.api_title}请访问 /docs 查看API文档。} # 本地开发直接运行 if __name__ __main__: uvicorn.run( app.main:app, hostsettings.host, portsettings.port, reloadsettings.reload, log_levelinfo )4. 本地运行与测试代码都写好了让我们先在本地点火测试一下。首先确保你的Phi-3-mini模型已经下载并且settings.model_path指向了正确的路径。然后在项目根目录下运行python -m app.main如果一切顺利你会看到服务启动的日志最后显示类似Uvicorn running on http://0.0.0.0:8000的信息。打开浏览器访问http://127.0.0.1:8000/docs。你会看到一个自动生成的、交互式的Swagger API文档页面。这就是FastAPI的魅力之一代码即文档。在文档页面里你可以点击/api/v1/health的Try it out再点Execute看看健康检查是否返回model_loaded: true。点击/api/v1/generate的Try it out在请求体里填入你的提示词例如{prompt: 请用Python写一个简单的HTTP服务器。}然后执行。稍等片刻你就能在“Responses”部分看到模型生成的代码了你也可以用curl命令测试curl -X POST http://127.0.0.1:8000/api/v1/generate \ -H Content-Type: application/json \ -d {prompt: 解释一下机器学习中的过拟合现象。, max_new_tokens: 200}本地测试通过证明我们的核心逻辑没问题。但这样的服务还比较“脆弱”不适合直接对外。接下来我们把它武装起来。5. 生产环境部署Gunicorn Nginx直接用python命令运行是单进程的性能有限也不稳定。我们需要一个ASGI服务器来管理多个工作进程再用一个Web服务器反向代理在前面处理静态文件、负载均衡和SSL。5.1 使用Gunicorn作为ASGI服务器Gunicorn是一个成熟的WSGI/ASGI服务器可以管理多个工作进程提高并发能力。为FastAPI这样的异步应用我们需要使用Uvicorn的工作器Worker。首先创建一个Gunicorn的配置文件gunicorn_conf.py# gunicorn_conf.py import multiprocessing # 服务器套接字 bind 0.0.0.0:8000 # 工作进程数通常设置为 CPU核心数 * 2 1 workers multiprocessing.cpu_count() * 2 1 # 每个工作进程的线程数 (对于异步框架通常为1使用异步worker) threads 1 # 工作器类型使用Uvicorn的异步工作器 worker_class uvicorn.workers.UvicornWorker # 工作进程最大请求数防止内存泄漏 max_requests 1000 max_requests_jitter 50 # 超时设置 timeout 120 keepalive 2 # 日志配置 accesslog - # 访问日志输出到标准输出 errorlog - # 错误日志输出到标准错误 loglevel info # 进程名 proc_name phi3_mini_api然后使用Gunicorn启动服务gunicorn -c gunicorn_conf.py app.main:app现在你的服务运行在多个工作进程上能更好地利用多核CPU并发处理能力大大提升。5.2 使用Nginx作为反向代理Nginx可以处理静态文件、SSL/TLS加密、负载均衡并将请求高效地转发给后端的Gunicorn。它还能提供一层安全防护。创建一个简单的Nginx配置文件nginx.conf# nginx.conf events { worker_connections 1024; } http { upstream api_server { # 指向Gunicorn服务地址这里假设Gunicorn运行在本机8000端口 server 127.0.0.1:8000; # 可以添加多个server行实现负载均衡 # server 127.0.0.1:8001; } server { listen 80; server_name your_domain.com; # 替换为你的域名或IP # 静态文件服务如果有的话 # location /static { # alias /path/to/your/static/files; # } location / { # 将请求代理到Gunicorn proxy_pass http://api_server; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; # 一些超时设置 proxy_connect_timeout 60s; proxy_send_timeout 60s; proxy_read_timeout 60s; } # 限制客户端请求体大小防止过大提示词攻击 client_max_body_size 10M; } }启动Nginx假设你已安装Nginx# 测试配置文件语法 sudo nginx -t -c /path/to/your/phi3-mini-api-service/nginx.conf # 使用指定配置文件启动或重新加载 sudo nginx -c /path/to/your/phi3-mini-api-service/nginx.conf # 或者如果已有一个默认nginx在运行可以修改默认配置后重载 # sudo systemctl reload nginx现在你的API服务可以通过Nginx监听的80端口或配置的域名访问了。外部流量先到Nginx再由Nginx转发给Gunicorn集群。5.3 使用系统服务管理Systemd为了确保服务在服务器重启后能自动运行我们将其配置为系统服务。创建一个服务文件/etc/systemd/system/phi3-api.service[Unit] DescriptionPhi-3-mini API Service Afternetwork.target [Service] Useryour_username # 替换为你的用户名 Groupyour_groupname # 替换为你的用户组 WorkingDirectory/path/to/your/phi3-mini-api-service EnvironmentPATH/path/to/your/venv/bin # 虚拟环境路径 ExecStart/path/to/your/venv/bin/gunicorn -c gunicorn_conf.py app.main:app Restartalways RestartSec3 [Install] WantedBymulti-user.target然后启用并启动服务sudo systemctl daemon-reload sudo systemctl enable phi3-api.service sudo systemctl start phi3-api.service # 查看状态和日志 sudo systemctl status phi3-api.service sudo journalctl -u phi3-api.service -f至此一个具备生产级基础能力的Phi-3-mini模型API服务就部署完成了。它具备了并发处理、反向代理、服务自启等关键特性。6. 总结走完这一趟我们从本地的一个模型脚本构建出了一个结构清晰、性能可观、便于集成的标准API服务。回顾一下核心步骤首先是项目结构和环境搭建这是所有好项目的基础然后是用单例模式稳妥地加载和管理模型这个“重资产”接着用Pydantic定义清晰的接口契约并用FastAPI快速实现端点本地测试通过后再用Gunicorn和Nginx这套经典组合为服务穿上“铠甲”提升其并发能力和安全性最后通过Systemd让服务能够稳定运行。现在你的Phi-3-mini模型已经不再是一个孤立的程序而是一个可以通过网络被各种应用调用的智能服务。你可以在此基础上继续添加API密钥认证、请求限流、更完善的监控和日志甚至结合Docker容器化部署让它更加健壮和易用。希望这篇教程能为你打开AI模型工程化落地的一扇门。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
