梳理 BGE-M3 这一兼具多功能、多语言与多粒度特性的文本检索模型，并给出私有化部署方案：用 FastAPI 与 sentence-transformers 封装编码接口，配合 uvicorn 提供向量检索服务。

bge-m3模型

BGE-M3是第一个具有多功能、多语言和多粒度特性的文本检索模型。

多功能:可以同时执行三种检索功能：单向量检索、多向量检索和稀疏检索。
多语言:支持100多种工作语言。
多粒度:它能够处理不同粒度的输入，从短句子到长达8192个词汇的长文档。

私有化部署（bge-m3）

import json
import os
import sys

import torch
import uvicorn
from fastapi import FastAPI
from loguru import logger

project_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
sys.path.insert(0, project_path)

from config.message import BgeM3EmbeddingRequest
from config.paths import bge_m3_model_path
from utils.bge_m3_model_util import BGEM3FlagModel

os.environ["KMP_DUPLICATE_LIB_OK"] = "TRUE"
os.environ["TOKENIZERS_PARALLELISM"] = "TRUE"


def load_bge_m3(model_name_or_path: str, device="cuda"):
    if device is None:
        device = "cuda" if torch.cuda.is_available() else "cpu"

    model = BGEM3FlagModel(model_name_or_path=model_name_or_path, device=device)

    return model


embedding_server = FastAPI()
bge_m3_model = load_bge_m3(bge_m3_model_path, device="cuda")


@embedding_server.post("/bge_m3/encode/")
def encode(request: BgeM3EmbeddingRequest):
    """
    搜索应用构建接口
    :param request:{
                    "texts": [
                        "你好",
                        "你是谁"
                    ]
                }
    :return:{
                embeddings: {
                    "dense":[],
                    "sparse":[]}
            }
    """

    embeddings = {}
    result = {
        "state": True,
        "embeddings": embeddings
    }
    try:
        request = json.loads(request.json())
        logger.info(request)
        texts = request["texts"]
        return_dense = request.get("return_dense", True)
        return_sparse = request.get("return_sparse", False)
        return_colbert_vecs = request.get("return_colbert_vecs", False)
        batch_size = request.get("batch_size", 12)
        output = bge_m3_model.encode(texts, return_dense=return_dense, return_sparse=return_sparse,
                                     return_colbert_vecs=return_colbert_vecs, batch_size=batch_size)
        if return_dense:
            embeddings["dense"] = [dense.tolist() for dense in output["dense_vecs"]]

        if return_sparse:
            sparse_dim = len(bge_m3_model.tokenizer)
            sparse_vecs = []
            for sparse_vec in output["lexical_weights"]:
                sparse_vecs.append({key: float(value) for key, value in sparse_vec.items()})
            embeddings["sparse"] = [sparse_vecs, sparse_dim]

        if return_colbert_vecs:
            embeddings["multi"] = [multi.tolist() for multi in output["colbert_vecs"]]

    except KeyError as e:
        result['state'] = False
        result['error'] = repr(e)
        logger.error(result)
    return result


if __name__ == "__main__":
    """
    nohup python bge_m3.py > bge_m3.log 2>&1 &
    """
    uvicorn.run(app=embedding_server,
                host='0.0.0.0',
                port=8098)

AI 大模型研发与应用

BGE-M3 检索模型与私有化部署：FastAPI 向量服务搭建

bge-m3模型