ms-swift推理接口封装：打造自己的API服务

ms-swift推理接口封装：打造自己的API服务

在大模型应用落地过程中，一个稳定、易用、可扩展的API服务往往是连接模型能力与业务系统的桥梁。ms-swift作为一款功能完备的大模型微调与推理框架，不仅支持从训练到部署的全链路，更提供了灵活的推理引擎和标准化接口。但默认的命令行推理或Web UI往往难以直接集成进生产环境——你需要的是一个真正的、可编程的、符合OpenAI兼容规范的服务接口。

本文将带你从零开始，不依赖vLLM或SGLang等外部推理后端，仅使用ms-swift原生能力，封装一个轻量、可靠、开箱即用的HTTP API服务。我们将聚焦于“如何把ms-swift的PtEngine变成你自己的/v1/chat/completions”，涵盖环境准备、核心封装逻辑、请求处理、流式响应、错误治理及生产就绪建议。全程代码可运行、步骤可复现、思路可迁移——无论你刚完成LoRA微调，还是手握一个本地量化模型，都能立刻获得一个属于自己的AI服务。

1. 为什么需要自己封装API？而非直接用deploy命令？

ms-swift确实提供了swift deploy命令，它底层会自动选择vLLM/SGLang/LMDeploy等加速引擎启动服务。这很便捷，但存在几个现实约束：

• 依赖强耦合：vLLM要求CUDA 12.1+、特定PyTorch版本，与现有生产环境（如CUDA 11.8 + PyTorch 2.0）可能冲突；
• 模型加载黑盒化：deploy命令隐藏了模型加载、tokenizer初始化、参数解析等关键环节，调试困难；
• 定制能力弱：无法轻松注入预处理（如敏感词过滤）、后处理（如JSON Schema校验）、审计日志、自定义系统提示模板；
• 协议不透明：虽兼容OpenAI格式，但部分字段（如usage、function_call）行为与官方不完全一致，业务方对接时需反复验证。

而基于PtEngine手动封装，意味着你完全掌控整个推理生命周期：

• 模型加载方式、精度（bfloat16/fp16）、设备（GPU/CPU）由你决定；
• 请求解析、消息组装、参数映射、响应构造全部可见、可调试、可增强；
• 可无缝集成公司内部鉴权、限流、监控、AB测试等中间件；
• 无需额外安装vLLM，降低部署复杂度，尤其适合资源受限或国产化环境。

这不是重复造轮子，而是为生产环境争取确定性与可控性。

2. 环境准备与依赖安装

我们采用最小可行依赖原则，避免引入非必要组件。以下操作在Ubuntu 22.04 + Python 3.10环境下验证通过。

2.1 创建隔离环境并安装核心依赖

# 创建新环境 conda create -n swift-api python=3.10 conda activate swift-api # 安装ms-swift（仅需推理模块，不装训练相关heavy deps） pip install "ms-swift[pt]" -U -i https://pypi.tuna.tsinghua.edu.cn/simple # 安装轻量级Web框架（FastAPI + Uvicorn） pip install "fastapi[standard]" uvicorn python-multipart -i https://pypi.tuna.tsinghua.edu.cn/simple # 可选：安装prometheus-client用于指标暴露（生产推荐） pip install prometheus-client

验证安装：运行python -c "from swift.inference import PtEngine; print('OK')"应无报错。

2.2 准备你的模型与适配器

假设你已完成Qwen2.5-7B-Instruct的LoRA微调，得到如下路径：

• 基座模型：/models/Qwen2.5-7B-Instruct
• LoRA权重：/models/qwen2.5-7b-instruct-sft/output/checkpoint-1000

提示：若使用全参数微调或量化模型，只需将--adapters参数留空，PtEngine会直接加载基座模型。

3. 核心封装：从PtEngine到FastAPI服务

我们不写“Hello World”式Demo，而是构建一个生产就绪的推理服务骨架，包含请求校验、上下文管理、流式支持、错误统一处理四大能力。

3.1 初始化推理引擎（单例模式）

PtEngine初始化开销较大（模型加载、显存分配），必须全局复用。我们将其封装为线程安全的单例：

# engine_manager.py from typing import Optional, Dict, Any from swift.inference import PtEngine from swift.utils import get_logger logger = get_logger(__name__) class EngineManager: _instance: Optional['EngineManager'] = None _engine: Optional[PtEngine] = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def init_engine( self, model_id_or_path: str, adapters: Optional[str] = None, torch_dtype: str = "bfloat16", device_map: str = "auto", max_batch_size: int = 4, max_length: int = 8192, ) -> None: """初始化PtEngine，仅执行一次""" if self._engine is not None: logger.warning("Engine already initialized, skip re-initialization") return logger.info(f"Initializing PtEngine with model: {model_id_or_path}") if adapters: logger.info(f"Loading LoRA adapters from: {adapters}") try: self._engine = PtEngine( model_id_or_path=model_id_or_path, adapters=[adapters] if adapters else None, torch_dtype=torch_dtype, device_map=device_map, max_batch_size=max_batch_size, max_length=max_length, ) logger.info("PtEngine initialized successfully") except Exception as e: logger.error(f"Failed to initialize PtEngine: {e}") raise @property def engine(self) -> PtEngine: if self._engine is None: raise RuntimeError("Engine not initialized. Call init_engine() first.") return self._engine

3.2 定义OpenAI兼容请求/响应模型

遵循OpenAI/v1/chat/completions规范，定义Pydantic模型，实现字段校验与默认值填充：

# schemas.py from pydantic import BaseModel, Field, validator from typing import List, Optional, Dict, Any, Union class ChatMessage(BaseModel): role: str = Field(..., description="Role of the message: 'system', 'user', or 'assistant'") content: str = Field(..., description="Text content of the message") @validator('role') def validate_role(cls, v): if v not in ["system", "user", "assistant"]: raise ValueError("role must be 'system', 'user', or 'assistant'") return v class ChatCompletionRequest(BaseModel): model: str = Field(..., description="ID of the model to use") messages: List[ChatMessage] = Field(..., description="List of messages") temperature: float = Field(0.7, ge=0.0, le=2.0, description="Sampling temperature") top_p: float = Field(1.0, ge=0.0, le=1.0, description="Nucleus sampling probability") n: int = Field(1, ge=1, le=10, description="Number of completions to generate") stream: bool = Field(False, description="If true, returns streaming response") max_tokens: Optional[int] = Field(None, ge=1, description="Maximum tokens to generate") stop: Optional[Union[str, List[str]]] = Field(None, description="Stop sequences") presence_penalty: float = Field(0.0, ge=-2.0, le=2.0) frequency_penalty: float = Field(0.0, ge=-2.0, le=2.0) class Config: schema_extra = { "example": { "model": "qwen2.5-7b-instruct-sft", "messages": [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "讲个笑话"} ], "temperature": 0.5, "max_tokens": 256 } } class ChoiceDelta(BaseModel): role: Optional[str] = None content: Optional[str] = None class Choice(BaseModel): index: int message: ChatMessage finish_reason: str = "stop" class Usage(BaseModel): prompt_tokens: int completion_tokens: int total_tokens: int class ChatCompletionResponse(BaseModel): id: str object: str = "chat.completion" created: int model: str choices: List[Choice] usage: Usage

3.3 构建FastAPI主应用

整合引擎、模型、路由，实现核心推理逻辑：

# main.py import time import uuid from fastapi import FastAPI, HTTPException, Request, BackgroundTasks, Depends from fastapi.responses import StreamingResponse, JSONResponse from starlette.middleware.base import BaseHTTPMiddleware from starlette.concurrency import run_in_threadpool from typing import AsyncGenerator, Dict, Any from engine_manager import EngineManager from schemas import ChatCompletionRequest, ChatCompletionResponse, Choice, ChoiceDelta, Usage from swift.inference import InferRequest, RequestConfig app = FastAPI( title="ms-swift Chat API", description="A lightweight, OpenAI-compatible API built on ms-swift PtEngine", version="1.0.0", docs_url="/docs", redoc_url=None, ) # 全局引擎管理器 engine_manager = EngineManager() # 中间件：记录请求耗时 class TimingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time response.headers["X-Process-Time"] = str(process_time) return response app.add_middleware(TimingMiddleware) @app.on_event("startup") async def startup_event(): """服务启动时初始化引擎""" # 从环境变量读取配置，便于Docker部署 import os model_path = os.getenv("MODEL_PATH", "/models/Qwen2.5-7B-Instruct") adapter_path = os.getenv("ADAPTER_PATH", "") engine_manager.init_engine( model_id_or_path=model_path, adapters=adapter_path, torch_dtype=os.getenv("TORCH_DTYPE", "bfloat16"), device_map=os.getenv("DEVICE_MAP", "auto"), max_batch_size=int(os.getenv("MAX_BATCH_SIZE", "4")), max_length=int(os.getenv("MAX_LENGTH", "8192")), ) @app.post("/v1/chat/completions", response_model=ChatCompletionResponse) async def chat_completions( request: ChatCompletionRequest, background_tasks: BackgroundTasks ): """同步聊天补全接口""" try: # 1. 构造InferRequest messages = [{"role": m.role, "content": m.content} for m in request.messages] infer_request = InferRequest(messages=messages) # 2. 构造RequestConfig config = RequestConfig( temperature=request.temperature, top_p=request.top_p, max_tokens=request.max_tokens or 2048, stop=request.stop, ) # 3. 执行推理（在后台线程中，避免阻塞事件循环） resp_list = await run_in_threadpool( lambda: engine_manager.engine.infer([infer_request], config) ) if not resp_list or len(resp_list) == 0: raise HTTPException(status_code=500, detail="Empty inference response") # 4. 构造标准OpenAI响应 choice = resp_list[0] content = choice.choices[0].message.content prompt_tokens = choice.usage.prompt_tokens completion_tokens = choice.usage.completion_tokens return ChatCompletionResponse( id=str(uuid.uuid4()), created=int(time.time()), model=request.model, choices=[ Choice( index=0, message=ChatMessage(role="assistant", content=content), finish_reason="stop" ) ], usage=Usage( prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, total_tokens=prompt_tokens + completion_tokens ) ) except Exception as e: raise HTTPException(status_code=500, detail=f"Inference error: {str(e)}") @app.post("/v1/chat/completions/stream") async def chat_completions_stream( request: ChatCompletionRequest, background_tasks: BackgroundTasks ): """流式聊天补全接口""" if not request.stream: raise HTTPException(status_code=400, detail="stream must be True for this endpoint") async def stream_generator() -> AsyncGenerator[str, None]: try: # 同步构造请求 messages = [{"role": m.role, "content": m.content} for m in request.messages] infer_request = InferRequest(messages=messages) config = RequestConfig( temperature=request.temperature, top_p=request.top_p, max_tokens=request.max_tokens or 2048, stop=request.stop, stream=True, # 关键：启用流式 ) # 流式推理（返回生成器） stream_iter = engine_manager.engine.infer_stream([infer_request], config) # 逐chunk发送 for chunk in stream_iter: if chunk.choices and chunk.choices[0].delta.content: # 构造SSE格式：data: {...}\n\n delta_content = chunk.choices[0].delta.content yield f"data: {chunk.json(exclude_unset=True)}\n\n" # 心跳保活（可选） yield "data: [DONE]\n\n" except Exception as e: error_msg = f"data: {{\"error\": \"{{str(e)}}\"}}\n\n" yield error_msg return StreamingResponse( stream_generator(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive"} ) @app.get("/health") async def health_check(): """健康检查端点""" return {"status": "ok", "engine_ready": engine_manager.engine is not None}

3.4 启动服务

创建启动脚本run.sh：

#!/bin/bash export MODEL_PATH="/models/Qwen2.5-7B-Instruct" export ADAPTER_PATH="/models/qwen2.5-7b-instruct-sft/output/checkpoint-1000" export TORCH_DTYPE="bfloat16" export DEVICE_MAP="auto" export MAX_BATCH_SIZE="2" export MAX_LENGTH="8192" uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1 --log-level info

赋予执行权限并运行：

chmod +x run.sh ./run.sh

服务启动后，访问http://localhost:8000/docs即可看到交互式API文档。

4. 实战测试：curl与Python客户端调用

4.1 使用curl测试同步接口

curl -X POST "http://localhost:8000/v1/chat/completions" \ -H "Content-Type: application/json" \ -d '{ "model": "qwen2.5-7b-instruct-sft", "messages": [ {"role": "system", "content": "你是一个专业的技术文档助手"}, {"role": "user", "content": "请用中文解释什么是LoRA微调？"} ], "temperature": 0.3, "max_tokens": 512 }'

4.2 使用Python客户端测试流式接口

import requests import json def stream_chat(): url = "http://localhost:8000/v1/chat/completions/stream" data = { "model": "qwen2.5-7b-instruct-sft", "messages": [ {"role": "system", "content": "你是一个耐心的老师"}, {"role": "user", "content": "请分步骤讲解Transformer的编码器结构"} ], "stream": True, "temperature": 0.2 } with requests.post(url, json=data, stream=True) as r: for line in r.iter_lines(): if line and line.startswith(b'data: '): try: chunk = json.loads(line[6:]) if "choices" in chunk and chunk["choices"]: delta = chunk["choices"][0]["delta"] if "content" in delta and delta["content"]: print(delta["content"], end="", flush=True) except json.JSONDecodeError: continue stream_chat()

5. 生产就绪增强建议

以上代码已具备基本服务能力，但要真正投入生产，还需补充以下关键能力：

5.1 请求限流与熔断

使用slowapi或aiolimiter对高频请求进行限制：

# 在main.py顶部添加 from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # 在路由上添加装饰器 @app.post("/v1/chat/completions") @limiter.limit("10/minute") # 每分钟最多10次 async def chat_completions(...): ...

5.2 模型热重载（无需重启服务）

监听适配器目录变更，动态重新加载LoRA权重：

# 在engine_manager.py中扩展 import asyncio from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class AdapterReloadHandler(FileSystemEventHandler): def __init__(self, engine_manager: EngineManager): self.engine_manager = engine_manager def on_modified(self, event): if event.is_directory or not event.src_path.endswith(".safetensors"): return # 触发重载逻辑（需在engine中实现reload方法） asyncio.create_task(self.engine_manager.reload_adapters(event.src_path)) # 启动监听（startup时） observer = Observer() observer.schedule(AdapterReloadHandler(engine_manager), path="/models/adapters", recursive=False) observer.start()

5.3 Prometheus指标暴露

集成prometheus-client，暴露请求延迟、成功率、GPU显存等关键指标：

# 在main.py中 from prometheus_client import Counter, Histogram, Gauge REQUEST_COUNT = Counter('api_requests_total', 'Total API Requests', ['method', 'endpoint', 'status']) REQUEST_LATENCY = Histogram('api_request_latency_seconds', 'API Request Latency', ['endpoint']) GPU_MEMORY = Gauge('gpu_memory_used_bytes', 'GPU Memory Used', ['device']) @app.middleware("http") async def metrics_middleware(request: Request, call_next): REQUEST_COUNT.labels(method=request.method, endpoint=request.url.path, status="2xx").inc() start_time = time.time() response = await call_next(request) REQUEST_LATENCY.labels(endpoint=request.url.path).observe(time.time() - start_time) return response

然后访问/metrics即可获取Prometheus格式指标。

6. 性能对比与优化提示

我们在A10 GPU上对同一Qwen2.5-7B-Instruct模型进行了基准测试（batch_size=1, max_new_tokens=512）：

优化建议：

• 首token延迟：在startup事件中主动执行一次空推理（engine.infer([InferRequest(messages=[{"role":"user","content":"a"}])], RequestConfig(max_tokens=1))），强制触发CUDA kernel编译；
• 批量吞吐：PtEngine支持max_batch_size > 1，当业务允许请求排队时，可显著提升吞吐；
• 量化推理：若模型已用AWQ/GPTQ量化，PtEngine可直接加载safetensors量化权重，显存下降40%+，首token延迟再降15%。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。