大模型微调（6）：开放 API 调用 - 文章 - 开发者社区

在前面的课程中，我们已经详细介绍了大模型微调相关的技术，包括 HuggingFace 和 Transformers 库的使用、开源模型的下载、部署、PEFT 微调等等。现在，我们已经拥有了微调好的模型，并且成功在本地部署，可以进行推理服务了。

那么接下来的问题就是： 如何对外暴露 API，让上游的应用服务访问这些模型呢？这就是本节课要完成的内容。话不多说，我们直接开始吧！

环境准备 =======

和之前的步骤类似，我们需要先准备下环境。具体的操作系统、GPU 相关的配置，大家可以参考前面的课程，这里就不赘述了。我们聚焦 API 服务的部分。

要开放 API 访问，一定离不开 API 框架和 Web 服务器。这里我们采用主流的 Uvicorn + FastAPI 方案。

Uvicorn ：是一个基于 uvloop 和 httptools 构建的轻量级 ASGI (Asynchronous Server Gateway Interface) 服务器，专门用于运行异步 Python Web 应用。
FastAPI ：是一个现代、快速（高性能）的 Python Web 框架，用于构建 API，基于标准 Python 类型提示。

简单理解：Uvicorn 是服务器，通过 uvloop 和 httptools 提供高性能 I/O，负责运行 FastAPI 应用；而 FastAPI 是 Web 框架，负责执行应用逻辑。

关于 Uvicorn 和 FastAPI 的具体原理，这里就不展开了，大家感兴趣可以参考相关资源自行学习。

除了 uvicorn 和 fastapi ，我们还需要 transformers 和 torch 等依赖，用于模型的加载和推理，完整的安装命令如下：

  
pip install fastapi uvicorn   
transformers torch sentencepiece

picture.image

为了加速访问，我们再对环境进行些额外的配置。(注：该步骤是可选的)

  
import subprocess  
import os  
  
# 设置模型下载路径  
os.environ["HF_HOME"] = "你的模型下载路径"  
os.environ["HF_HUB_CACHE"] = "你的模型下载路径"  
  
# 设置HuggingFace镜像站  
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"

模型的下载与测试 ===========

准备好了环境之后，下一步就是把模型下载到本地，并且部署起来。

本次课程，我们使用 DeepSeek-R1-Distill-Qwen-1.5B ，它是 DeepSeek 基于 Qwen-1.5B 蒸馏出来的小尺寸模型，非常适合在本地或者资源受限的环境下运行，在兼具了性能的同时，还可以大幅节约成本。

DeepSeek-R1-Distill-Qwen-1.5B 是开源的，可以直接在 HuggingFace 上获取：

picture.image

我们使用 Transformers API，将模型下载到本地。

  
from transformers import AutoModelForCausalLM, AutoTokenizer  
import torch  
import logging  
  
# 设置日志  
logging.basicConfig(level=logging.INFO)  
logger = logging.getLogger(__name__)  
  
# 指定模型名称  
MODEL_NAME = "deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B"  
  
# 获取设备信息  
device = "cuda" if torch.cuda.is_available() else "cpu"  
  
try:  
    # 将模型下载到本地  
    logger.info(f"开始下载模型： {MODEL_NAME}...")  
    tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)  
    model = AutoModelForCausalLM.from_pretrained(  
        MODEL_NAME,  
        torch_dtype=torch.bfloat16,  
        device_map="auto"  
    )  
    logger.info("模型下载成功！")  
except Exception as e:  
    logger.error(f"模型下载出错: {str(e)}")  
    raise

picture.image

耐心等待模型下载完成~

picture.image

经过漫长的等待后，模型终于下载完成了。我们测试下模型能否正常生成文本。

  
# 测试Prompt  
prompt = "你好，请介绍下你自己~"  
  
# 将Prompt进行编码  
inputs = tokenizer(prompt, return_tensors="pt").to(device)  
  
# 调用模型，生成文本  
with torch.no_grad():  
    outputs = model.generate(  
        **inputs,  
        max_length=128,  
        temperature=0.7,  
        num_return_sequences=1,  
        pad_token_id=tokenizer.eos_token_id  
    )  
  
# 解码输出  
generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)  
print(f"模型回复内容：{generated_text}")

执行这段代码，成功生成了回复，说明我们的模型可以正常运行了！

picture.image

封装 API =========

本地的模型已经部署好了，最后一步就是启动 Web 服务、并开放 API 访问了。

先定义好请求和响应参数的结构：

  
from pydantic import BaseModel  
from typing import Optional  
  
class GenerationRequest(BaseModel):  
    """请求参数"""  
  
    prompt: str  
    max_length: Optional[int] = 512  
    temperature: Optional[float] = 0.7  
    top_p: Optional[float] = 0.9  
    top_k: Optional[int] = 50  
    repetition_penalty: Optional[float] = 1.1  
    num_return_sequences: Optional[int] = 1  
  
  
class GenerationResponse(BaseModel):  
    """响应参数"""  
      
    generated_text: str  
    status: str

接下来，创建 FastAPI 的 Web 客户端：

  
from fastapi import FastAPI, HTTPException  
  
app = FastAPI(  
    title="DeepSeek-R1开放API",  
    description="Open API for DeepSeek-R1",  
    version="1.0"  
)

然后，通过 @app 装饰器，定义路由与相应的处理函数：

  
@app.post("/generate", response_model=GenerationResponse)  
async def generate_text(request: GenerationRequest):  
    try:  
        # 编码输入  
        inputs = tokenizer(request.prompt, return_tensors="pt").to(device)  
  
        # 生成文本  
        with torch.no_grad():  
            outputs = model.generate(  
                **inputs,  
                max_length=request.max_length,  
                temperature=request.temperature,  
                top_p=request.top_p,  
                top_k=request.top_k,  
                repetition_penalty=request.repetition_penalty,  
                num_return_sequences=request.num_return_sequences,  
                pad_token_id=tokenizer.eos_token_id  
            )  
  
        # 解码输出  
        generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)  
  
        return {  
            "generated_text": generated_text,  
            "status": "success"  
        }  
    except Exception as e:  
        logger.error(f"Generation error: {str(e)}")  
        raise HTTPException(status_code=500, detail=str(e))  
  
@app.get("/health")  
async def health_check():  
    return {"status": "healthy"}

最后，启动 uvicorn ，运行 Web 服务：

  
if __name__ == "__main__":  
    import uvicorn  
    uvicorn.run(app, host="0.0.0.0", port=8000)

picture.image

这样一来，Web 服务就成功启动在 8000 端口上了。

我们本地调用 API 测试一下：

  
curl -X POST "http://localhost:8000/generate" \  
-H "Content-Type: application/json" \  
-d '{"prompt": "你好，请介绍一下你自己~", "max_length": 200}'

picture.image

大功告成~

总结与扩展 ========

本节课程，我们基于本地部署好的模型，结合 Uvicorn + FastAPI 框架，成功封装了 Open API，并开放给外部应用访问。这是一套通用的开发范式，希望可以对大家实际的开发有所帮助。

这套方案如果要部署在生产环境，还需要进行一系列的优化，下面是一些具体的优化方向，供大家参考：

性能调优：如调整Uvicorn 的 worker 数量（通常 CPU 核心数 × 2 + 1），并结合 --limit-concurrency 等参数，防止服务过载。
网关与负载均衡：可以使用 Nginx 或 Kong 等网关服务器，进行反向代理与负载均衡，并结合实际业务场景，定制路由、限流、黑白名单、鉴权等策略。
可观测性：结合 Prometheus + Grafana，对核心性能指标进行监控和告警。
OpenAI 格式兼容：可以通过 Text Generation Inference (TGI)，生成兼容 OpenAI 格式的接口，这样上游的应用可以以统一的形式调用模型服务。