1. 项目缘起为什么我们需要一个自己的Embedding服务最近在折腾一些RAG检索增强生成或者语义搜索相关的项目时我遇到了一个非常典型且恼人的问题no embedding model is loaded. set rag_embedding_model to a valid sentence transformer model。这个错误提示就像一盆冷水浇灭了我快速验证想法的热情。它背后反映的是一个更普遍的需求——我们总是需要一个稳定、可控、且能按需定制的文本转向量服务。市面上的云服务当然方便OpenAI的text-embedding-ada-002、百度的文心、阿里的通义调用一个API就能拿到高质量的向量。但问题也随之而来成本、网络延迟、数据隐私、模型固定无法微调。尤其是在做一些内部工具、离线应用或者对响应速度要求极高的场景时依赖外部API就成了瓶颈。更别提当你兴致勃勃地拉下一个开源项目准备跑起来看看效果却卡在模型下载或环境配置上时的那种挫败感。所以我决定动手从零搭建一个属于自己的Embedding服务。这个“从零”并不意味着从零开始训练一个BERT那不现实。我们的“从零”指的是从选择一个成熟的开源Embedding模型开始完成本地部署、服务化封装、性能优化最终提供一个类似云服务那样可以通过HTTP接口调用的文本向量化服务。目标很明确摆脱对外部服务的强依赖掌握核心组件的自主权并且能根据业务需求灵活调整模型、优化性能。这次实践我选择了目前中文社区口碑很好的BGEBAAI General Embedding系列模型特别是BGE-M3和轻量级的BGE embedding 4b作为主要实验对象。整个过程涉及模型选型、环境搭建、服务框架选择、接口设计、性能压测以及一些实战中的“坑”和技巧。下面我就把这套完整的实现路径和思考过程分享出来。2. 核心组件选型模型、框架与部署环境搭建服务的第一步是选择“砖瓦”。我们需要确定三样东西用哪个模型把文本变成向量用什么框架来包装和提供这个能力以及最终把这个服务放在哪里运行2.1 Embedding模型选型为什么是BGE开源Embedding模型的选择很多像Sentence-BERT、Instructor、E5系列都很优秀。我最终聚焦在BGE上主要是基于以下几点考虑中文优化与社区热度BGE由智源研究院推出针对中文场景做了大量优化在MTEB中文榜单上长期名列前茅。这意味着用它来处理中文文本开箱即用的效果就很有保障。从网络热词bge embedding、embedding 4b bge的搜索热度也能看出它已经是中文开发者的事实标准之一。模型规格丰富BGE提供了从大到小各种规格的模型。例如BGE-M3最新的多功能模型支持稠密向量、稀疏向量和多重向量检索能力全面但体积较大约2.2GB。BGE-large-zh-v1.5经典的高质量稠密向量模型适用于大多数检索和语义相似度任务。BGE embedding 4b这里需要澄清一个常见的误解。4b并非指4亿参数而是指模型文件名为bge-small-zh-v1.5的量化版本可能指4-bit量化模型本身很小约50MB速度极快非常适合对精度要求不高、但对延迟和资源极其敏感的场景。这正好解决了我们轻量级、快速部署的需求。易用性BGE模型完美兼容sentence-transformers库而后者是Python生态中处理句子嵌入的标杆库API设计优雅社区支持好。注意模型选择没有银弹。BGE-M3功能强但耗资源bge-small速度快但精度有损。我的建议是生产环境可以先从BGE-large-zh开始它在效果和资源消耗上取得了很好的平衡对于需要快速原型验证或资源受限的环境bge-small即常说的embedding 4b是绝佳的起点。2.2 服务化框架选型FastAPI的压倒性优势把模型封装成HTTP服务我们有几个选择Flask、FastAPI、或者直接用gradio快速构建一个Web界面。对于纯API服务FastAPI几乎是当前的最优解原因如下性能卓越基于Starlette和Pydantic异步支持原生且强大天生适合IO密集型的推理服务网络请求、模型加载都是IO。开发效率极高自动生成交互式API文档Swagger UI和ReDoc类型提示Type Hints带来极佳的开发体验和代码可靠性。生态契合与机器学习部署库如ray serve,text-generation-inference的理念很契合社区活跃。因此我们的技术栈就确定为sentence-transformersFastAPIUvicornASGI服务器。这是一个轻量、高效、且易于维护的组合。2.3 部署环境准备避开第一个坑环境是这一切的基础。这里最大的坑就是网络问题。sentence-transformers在第一次使用某个模型时会自动从Hugging Face Hub下载模型。如果你身处国内网络环境这个过程可能极其缓慢甚至失败直接导致服务启动报错no embedding model is loaded。解决方案不是修改代码而是预先准备好模型。有两种推荐做法手动下载与离线加载通过镜像站如hf-mirror.com或能稳定访问的环境提前下载好模型文件。将模型文件夹包含pytorch_model.bin、config.json、tokenizer.json等放置到服务器本地目录例如./models/bge-large-zh-v1.5。在代码中初始化模型时指定本地路径from sentence_transformers import SentenceTransformer model SentenceTransformer(/path/to/your/local/models/bge-large-zh-v1.5)这是最稳定、最推荐的方式尤其适合生产环境。环境变量配置备选如果还是希望通过库自动下载可以设置HF镜像的环境变量export HF_ENDPOINThttps://hf-mirror.com但这并非百分百可靠取决于镜像站的同步情况和网络状况。我的实战经验是对于核心的生产依赖永远优先选择离线部署。这能避免在关键时刻如服务重启、扩容被网络问题“背刺”。准备好模型文件后我们就可以开始编写服务核心代码了。3. 服务核心实现从模型加载到API暴露有了清晰的选型和准备好的模型接下来就是编码实现。我们的服务核心要做两件事1. 高效加载并管理Embedding模型2. 通过HTTP接口暴露向量化能力。3.1 模型加载与单例模式在Web服务中我们肯定不希望每个请求都去加载一次模型耗时耗内存。正确的做法是在服务启动时加载一次之后所有请求共享这个模型实例。这可以通过FastAPI的lifespan事件或直接在全局作用域初始化来实现。我更喜欢使用lifespan因为它管理起来更清晰。from contextlib import asynccontextmanager from fastapi import FastAPI from sentence_transformers import SentenceTransformer import numpy as np # 全局变量存放模型 _model None asynccontextmanager async def lifespan(app: FastAPI): # 启动时加载模型 global _model print(Loading Embedding Model...) # 这里替换为你的实际模型路径 _model SentenceTransformer(./models/bge-large-zh-v1.5) # 可以在这里进行一次预热推理避免第一次请求过慢 _model.encode(预热文本) print(Model loaded successfully.) yield # 关闭时清理如果需要 print(Shutting down...) app FastAPI(lifespanlifespan)为什么用lifespan它提供了明确的启动和关闭钩子比在模块顶层直接写加载代码更优雅也便于未来扩展例如连接数据库池。预热推理那一步很重要因为PyTorch/TensorFlow在第一次推理时会有图构建等开销预热能确保第一个真实请求的延迟不会异常高。3.2 API接口设计兼顾简单与灵活Embedding服务的核心API通常很简单输入文本输出向量。但设计时需要考虑扩展性。基础编码接口from pydantic import BaseModel from typing import List class EncodeRequest(BaseModel): texts: List[str] # 可扩展参数是否归一化归一化后便于计算余弦相似度 normalize_embeddings: bool True # 可扩展参数批处理大小针对大量文本 batch_size: int 32 app.post(/encode) async def encode_text(request: EncodeRequest): 将文本列表编码为向量列表 try: # 调用模型进行编码 embeddings _model.encode( request.texts, normalize_embeddingsrequest.normalize_embeddings, batch_sizerequest.batch_size, show_progress_barFalse # 服务端不需要进度条 ) # 将numpy数组转换为列表 embeddings_list embeddings.tolist() return {embeddings: embeddings_list, model: _model.get_sentence_embedding_dimension()} except Exception as e: return {error: str(e)}关键点使用List[str]支持批量处理能极大提升吞吐量。提供normalize_embeddings参数。对于大多数语义相似度或检索任务将向量归一化为单位向量是标准操作这样余弦相似度就等于点积计算更高效。batch_size参数允许调用方根据自身文本长度和数量微调以平衡内存和速度。健康检查与元信息接口app.get(/health) async def health_check(): return {status: healthy, model: _model.__class__.__name__} app.get(/model_info) async def model_info(): return { model_name: _model.model_name_or_path, embedding_dimension: _model.get_sentence_embedding_dimension(), max_seq_length: _model.max_seq_length }这两个接口对于服务运维至关重要。健康检查用于负载均衡器或K8s的存活探针模型信息接口让客户端能动态获取向量维度避免硬编码。3.3 处理长文本截断与分块策略所有Embedding模型都有一个max_seq_length例如BGE-large是512。当文本超过这个长度时必须处理。sentence-transformers的默认行为是静默截断但这可能丢失重要信息。更优的做法是在服务层提供明确的策略并在API响应中告知客户端。我们可以修改/encode接口增加一个truncation_strategy参数或者更简单一点在服务内部实现一个智能分块函数对于远长于512字的文档。def smart_truncate(text: str, max_length: int 500) - str: 简单智能截断尽量在句末截断 if len(text) max_length: return text # 找到max_length之前的最后一个句号、问号或感叹号 truncate_at text.rfind(。, 0, max_length) if truncate_at -1: truncate_at text.rfind(, 0, max_length) if truncate_at -1: truncate_at text.rfind(, 0, max_length) if truncate_at -1: truncate_at text.rfind(., 0, max_length) # 英文句号 # 如果还是没找到就在max_length处硬截断 truncate_at truncate_at if truncate_at ! -1 else max_length return text[:truncate_at 1] # 包含截断的标点 # 在encode函数中调用 processed_texts [smart_truncate(t, _model.max_seq_length) for t in request.texts] embeddings _model.encode(processed_texts, ...)这是一个基础示例。对于真正的长文档检索更好的做法是“分块-分别嵌入-再聚合”例如使用langchain的文本分割器但这通常在上游应用层完成而非在基础的Embedding服务内。我们的服务应保持职责单一主要提供基础的向量化能力。4. 性能优化与生产级考量一个能用的服务和一个好用的服务之间隔着性能优化和稳定性建设。当你的服务开始接收真实流量时以下几个方面的考量至关重要。4.1 并发处理与异步优化FastAPI是异步框架但sentence_transformers的encode方法是CPU/GPU密集型的同步操作。如果在异步路径中直接调用会阻塞整个事件循环导致服务并发能力急剧下降。解决方案使用run_in_executor将同步的模型推理任务丢到线程池中执行避免阻塞异步事件循环。import asyncio from concurrent.futures import ThreadPoolExecutor # 创建线程池 _executor ThreadPoolExecutor(max_workers4) # worker数量根据CPU核心数调整 app.post(/encode) async def encode_text(request: EncodeRequest): loop asyncio.get_event_loop() try: # 将同步的model.encode函数放到线程池中运行 embeddings await loop.run_in_executor( _executor, lambda: _model.encode( request.texts, normalize_embeddingsrequest.normalize_embeddings, batch_sizerequest.batch_size, show_progress_barFalse ) ) embeddings_list embeddings.tolist() return {embeddings: embeddings_list} except Exception as e: return {error: str(e)}参数max_workers设置多少合适这没有固定答案。如果模型推理是CPU瓶颈例如在CPU机器上运行设置成CPU核心数或稍多一点。如果是GPU推理GPU本身是瓶颈线程数可以略多于GPU流处理器数量但主要目的是不让CPU成为调度瓶颈。建议通过压测来确定观察GPU利用率和请求延迟。4.2 批处理吞吐量的关键Embedding模型在GPU上运行时批处理能极大提升吞吐量因为GPU擅长并行计算。我们的API设计已经支持批量文本输入。但这里有一个重要的权衡点批大小batch_size。批大小太小GPU算力无法被充分利用大量时间浪费在kernel启动和数据传输上。批大小太大可能导致GPU内存溢出OOM特别是文本长度不一、动态padding后显存占用激增。如何找到最佳批大小需要实测。写一个脚本用不同长度的文本和不同的批大小进行测试监控GPU内存使用量和每秒处理的token数或句子数。一个常见的经验是对于固定长度的输入可以逐步增加批大小直到接近OOM然后留出20%的安全余量。对于变长输入可以设定一个“最大总token数”作为批处理的上限而不是简单的句子数。4.3 模型量化与轻量化部署如果你对延迟和资源消耗极其敏感或者需要在边缘设备部署模型量化是必须考虑的步骤。前面提到的BGE embedding 4b很可能就是一个量化版本。量化分为多种精度FP16半精度、INT8、甚至INT4。sentence-transformers本身对量化支持有限但我们可以借助其他工具使用ONNX Runtime将模型导出为ONNX格式并使用ONNX Runtime进行推理它支持多种硬件加速和量化。# 示例使用 optimum 库导出 pip install optimum[onnxruntime] optimum-cli export onnx --model BAAI/bge-large-zh-v1.5 ./bge-large-zh-onnx/使用Pytorch原生量化对于高级用户可以对模型进行动态量化或静态量化但过程相对复杂。直接使用社区提供的量化模型在Hugging Face上搜索模型时可以关注是否有-int8、-4bit等后缀的版本。量化带来的影响精度会有轻微损失对于Embedding任务通常损失在可接受范围内但模型体积和推理速度会有显著改善。建议在决定量化前务必在你的业务数据集上评估量化前后向量相似度的保真度。4.4 监控、日志与容错一个健壮的服务离不开可观测性。日志使用标准的logging模块记录每个请求的摘要如文本数量、总字符数、处理耗时、错误信息。这有助于问题排查和用量分析。监控指标可以考虑集成Prometheus客户端暴露一些指标如requests_total总请求数request_duration_seconds请求耗时直方图batch_size_distribution批大小分布text_length_distribution输入文本长度分布容错与限流输入验证使用Pydantic严格校验输入防止恶意或异常数据导致服务崩溃。超时控制在FastAPI层面或反向代理如Nginx设置请求超时防止慢请求拖垮服务。限流对于公开服务必须实施限流如使用slowapi或fastapi-limiter防止被滥用。优雅降级如果模型加载失败或GPU不可用是否有后备方案如降级到CPU模式或返回特定错误码5. 实战部署与测试让服务跑起来代码写好了我们需要把它部署到一个稳定的环境中并验证其功能和性能。5.1 使用Docker容器化部署Docker是保证环境一致性的最佳实践。编写一个Dockerfile# 使用带有CUDA的Python基础镜像如果使用GPU FROM pytorch/pytorch:2.1.0-cuda11.8-cudnn8-runtime # 或使用CPU镜像 # FROM python:3.10-slim WORKDIR /app # 复制依赖文件并安装 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple # 复制模型文件假设模型已下载到本地./models目录 COPY ./models /app/models # 复制应用代码 COPY . . # 暴露端口 EXPOSE 8000 # 启动命令 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000, --workers, 4]requirements.txt内容fastapi0.104.1 uvicorn[standard]0.24.0 sentence-transformers2.2.2 numpy1.24.3 pydantic2.5.0构建与运行# 构建镜像 docker build -t embedding-service . # 运行容器CPU版本 docker run -p 8000:8000 --name embedding-api embedding-service # GPU版本需要加参数 docker run --gpus all -p 8000:8000 --name embedding-api embedding-service5.2 功能与性能测试服务启动后访问http://localhost:8000/docs可以看到自动生成的API文档我们需要进行测试。基础功能测试使用curl或Python的requests库调用接口。curl -X POST http://localhost:8000/encode \ -H Content-Type: application/json \ -d {texts: [今天天气真好, 人工智能是未来科技的核心], normalize_embeddings: true}检查返回的向量维度是否正确例如BGE-large是1024维以及两个语义不太相关的句子的向量余弦相似度是否较低。压力测试使用wrk、locust或apache benchmark进行压测。# 使用wrk示例 wrk -t4 -c100 -d30s --scriptpost.lua --latency http://localhost:8000/encode在post.lua文件中定义POST请求体和Header。通过压测你可以找到服务的QPS每秒查询数上限以及在不同并发下的延迟分布P50, P95, P99。这是调整workers数量、线程池大小和批处理参数的核心依据。长文本与边界测试输入超长文本、空文本列表、特殊字符等观察服务是否稳定响应是否符合预期如截断或返回错误信息。5.3 集成到现有项目最后如何在你自己的RAG或搜索项目中使用这个服务非常简单只需将原来直接调用云API或本地模型库的代码替换为HTTP调用。# 以前直接使用sentence-transformers # from sentence_transformers import SentenceTransformer # model SentenceTransformer(model_name) # embeddings model.encode(texts) # 现在调用自建服务 import requests import numpy as np def encode_with_service(texts, service_urlhttp://localhost:8000): resp requests.post( f{service_url}/encode, json{texts: texts, normalize_embeddings: True} ) resp.raise_for_status() result resp.json() return np.array(result[embeddings]) # 使用 vectors encode_with_service([查询文本1, 查询文本2])这样你的应用就和Embedding模型的具体实现解耦了。未来如果需要切换模型比如从BGE-large换成BGE-M3或者进行模型版本升级只需要重启或替换后端的Embedding服务而无需修改所有上游应用的代码。从选择一个合适的开源模型到用FastAPI将其封装成服务再到考虑性能、并发和生产部署的方方面面这个过程让我对“Embedding即服务”有了更深的体会。它不再是一个神秘的黑盒而是一个可以根据自己需求随意拆解、组装和优化的组件。最重要的是当再次看到no embedding model is loaded的报错时你心里有底知道问题出在哪并且有能力去解决它。这种掌控感或许就是自建基础设施最大的乐趣和价值所在。