bge-large-zh-v1.5教程：使用FastAPI封装embedding服务接口
1. 引言
随着大模型应用的不断深入，文本嵌入（Embedding）技术在语义检索、相似度计算、问答系统等场景中发挥着越来越关键的作用。bge-large-zh-v1.5作为一款高性能中文嵌入模型，凭借其强大的语义表达能力，已成为众多NLP任务中的首选模型之一。
然而，直接调用模型进行推理存在部署复杂、接口不统一等问题。为了提升服务的可用性与可集成性，本文将介绍如何基于已通过sglang部署的bge-large-zh-v1.5模型，使用FastAPI构建一个高效、易用的RESTful风格embedding服务接口。该方案不仅便于前后端系统集成，也支持快速扩展和监控。
本教程适用于已有sglang部署环境的开发者，目标是实现从本地模型服务到标准化Web API的封装升级。
2. 技术背景与核心架构
2.1 bge-large-zh-v1.5简介
bge-large-zh-v1.5是一款基于深度学习的中文嵌入模型，通过大规模语料库训练，能够捕捉中文文本的深层语义信息。其特点包括：
- 高维向量表示：输出向量维度高，语义区分度强。
- 支持长文本处理：能够处理长达512个token的文本输入。
- 领域适应性：在通用领域和特定垂直领域均表现优异。
这些特性使得bge-large-zh-v1.5在需要高精度语义匹配的场景中成为理想选择，但同时也对计算资源提出了较高要求。
目前，该模型已通过sglang框架部署为本地推理服务，默认监听http://localhost:30000/v1，提供类OpenAI格式的API接口，极大简化了后续封装流程。
2.2 整体架构设计
本方案采用分层架构设计，确保模块解耦与可维护性：
[客户端] ↓ (HTTP POST /embeddings) [FastAPI 封装层] → 转发请求 ↓ (HTTP POST http://localhost:30000/v1/embeddings) [sglang 模型服务层] ↓ 返回embedding结果 [FastAPI 层] ← 接收并格式化响应 ↓ [返回JSON结果给客户端]FastAPI作为中间层，承担以下职责： - 提供标准RESTful接口 - 验证输入参数 - 统一错误处理 - 支持跨域（CORS） - 可扩展日志、鉴权等功能
3. 环境准备与依赖安装
3.1 前置条件
在开始之前，请确保满足以下条件：
- 已成功部署
bge-large-zh-v1.5模型并通过 sglang 启动 - 模型服务运行在
http://localhost:30000/v1 - Python 3.9 或以上版本
- pip 包管理工具可用
3.2 创建项目目录结构
mkdir bge-fastapi-service cd bge-fastapi-service python -m venv venv source venv/bin/activate # Linux/Mac # 或 venv\Scripts\activate # Windows3.3 安装必要依赖
创建requirements.txt文件，内容如下：
fastapi>=0.110.0 uvicorn>=0.29.0 httpx>=0.27.0 pydantic>=2.6.0执行安装命令：
pip install -r requirements.txt说明：
-fastapi：用于构建Web API
-uvicorn：ASGI服务器，运行FastAPI应用
-httpx：异步HTTP客户端，用于转发请求至sglang服务
4. 实现FastAPI封装服务
4.1 编写主服务文件
创建main.py文件，内容如下：
from fastapi import FastAPI, HTTPException from pydantic import BaseModel import httpx import logging # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI( title="BGE Large zh v1.5 Embedding API", description="基于sglang部署的bge-large-zh-v1.5模型，使用FastAPI封装的embedding服务", version="1.0.0" ) # 定义请求数据模型 class EmbeddingRequest(BaseModel): input: str | list[str] model: str = "bge-large-zh-v1.5" # 定义响应数据模型（可根据实际返回结构调整） class EmbeddingData(BaseModel): object: str embedding: list[float] index: int class UsageInfo(BaseModel): prompt_tokens: int total_tokens: int class EmbeddingResponse(BaseModel): data: list[EmbeddingData] model: str usage: UsageInfo object: str # 全局HTTP客户端（支持连接池） client = httpx.AsyncClient(base_url="http://localhost:30000/v1", timeout=30.0) @app.post("/embeddings", response_model=EmbeddingResponse) async def create_embeddings(request: EmbeddingRequest): """ 创建文本嵌入向量 """ try: logger.info(f"收到embedding请求，输入类型: {type(request.input)}") # 转发请求到sglang服务 payload = request.model_dump() response = await client.post("/embeddings", json=payload) if response.status_code != 200: raise HTTPException(status_code=response.status_code, detail=response.text) return response.json() except httpx.RequestError as e: logger.error(f"请求sglang服务失败: {str(e)}") raise HTTPException(status_code=500, detail=f"模型服务不可达: {str(e)}") except Exception as e: logger.error(f"内部错误: {str(e)}") raise HTTPException(status_code=500, detail="内部服务器错误") @app.get("/") async def root(): return {"message": "BGE Embedding Service is running", "model": "bge-large-zh-v1.5"} @app.on_event("shutdown") async def shutdown_event(): await client.aclose() logger.info("HTTP客户端已关闭")4.2 代码解析
| 模块 | 功能说明 |
|---|---|
EmbeddingRequest | 接收客户端传入的文本和模型名，兼容单条或批量输入 |
httpx.AsyncClient | 使用异步客户端提高并发性能，复用连接 |
/embeddings接口 | 标准OpenAI兼容接口，便于迁移现有系统 |
| 日志记录 | 记录请求与异常，便于调试与监控 |
| 错误处理 | 分层捕获网络异常与内部错误，返回清晰提示 |
5. 启动与验证服务
5.1 启动FastAPI服务
在终端执行以下命令启动服务：
uvicorn main:app --host 0.0.0.0 --port 8000 --reload参数说明： ---host 0.0.0.0：允许外部访问 ---port 8000：服务监听端口 ---reload：开发模式下自动重载（生产环境应移除）
启动成功后，访问http://localhost:8000/docs可查看自动生成的Swagger文档界面。
5.2 验证sglang模型服务状态
在继续前，请确认sglang模型服务已正常运行。
进入工作目录
cd /root/workspace查看启动日志
cat sglang.log注意：若日志中显示类似
"Model bge-large-zh-v1.5 loaded successfully"或服务监听在:30000，则说明模型已成功加载。
5.3 使用Jupyter Notebook调用验证
启动Jupyter并执行以下Python代码进行测试：
import httpx # FastAPI封装后的服务地址 FASTAPI_URL = "http://localhost:8000/embeddings" data = { "input": "今天天气怎么样？", "model": "bge-large-zh-v1.5" } response = httpx.post(FASTAPI_URL, json=data) result = response.json() print("Status Code:", response.status_code) print("Embedding Vector Length:", len(result["data"][0]["embedding"])) print("Model:", result["model"])预期输出示例：
{ "data": [ { "object": "embedding", "embedding": [0.12, -0.45, ..., 0.67], "index": 0 } ], "model": "bge-large-zh-v1.5", "usage": { "prompt_tokens": 9, "total_tokens": 9 }, "object": "list" }若能成功获取长度为1024（或其他固定维度）的浮点数列表，则表明整个链路调用成功。
6. 进阶优化建议
6.1 添加请求校验与限流
可在FastAPI中集成pydantic更严格的字段校验，例如限制输入长度：
from typing import Annotated from pydantic import AfterValidator from functools import wraps def check_input_length(v): if isinstance(v, str): assert len(v) <= 512, "单段文本不得超过512字符" elif isinstance(v, list): assert len(v) <= 10, "最多支持10条文本批量处理" assert all(len(item) <= 512 for item in v), "每条文本不得超过512字符" return v class EmbeddingRequest(BaseModel): input: Annotated[str | list[str], AfterValidator(check_input_length)] model: str = "bge-large-zh-v1.5"6.2 支持API密钥认证（可选）
添加简单Token验证：
from fastapi import Header, Depends def verify_api_key(x_api_key: str = Header(...)): if x_api_key != "your-secret-token": raise HTTPException(status_code=401, detail="Invalid API Key") return x_api_key @app.post("/embeddings", dependencies=[Depends(verify_api_key)]) async def create_embeddings(...): ...调用时需添加头信息：
headers = {"X-API-Key": "your-secret-token"} httpx.post(FASTAPI_URL, json=data, headers=headers)6.3 性能监控与日志增强
建议集成Prometheus + Grafana进行指标采集，或使用loguru替代原生日志模块，记录请求耗时、成功率等关键指标。
7. 总结
7.1 核心价值回顾
本文详细介绍了如何将一个已通过sglang部署的bge-large-zh-v1.5中文嵌入模型，封装为标准化的RESTful API服务。通过引入FastAPI框架，我们实现了：
- ✅ 统一的HTTP接口规范
- ✅ 高性能异步处理能力
- ✅ 易于集成的JSON通信格式
- ✅ 自带API文档（Swagger UI）
- ✅ 可扩展的安全与监控机制
这一封装方式显著降低了模型服务的接入门槛，使前端、移动端或其他后端服务可以轻松调用embedding能力。
7.2 最佳实践建议
- 生产环境去
--reload：避免因热重载导致性能下降或状态异常。 - 使用反向代理：建议配合Nginx或Traefik做负载均衡与SSL终止。
- 容器化部署：可将服务打包为Docker镜像，便于CI/CD与集群管理。
- 健康检查接口：增加
/healthz接口供Kubernetes等平台探活。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
