Atlas 300i A2 实战：BGE-M3 嵌入模型与 Qdrant 向量数据库的高效集成方案

1次阅读

共计 2484 个字符，预计需要花费 7 分钟才能阅读完成。

在边缘计算设备如 Atlas 300i A2 上部署大模型嵌入服务面临多重挑战：

内存限制 ：边缘设备通常内存有限，而 BGE-M3 这类嵌入模型参数量大，直接加载可能导致 OOM
计算资源竞争 ：CPU/GPU 资源需同时处理模型推理和数据库操作，容易产生瓶颈
延迟敏感 ：边缘场景对实时性要求高，传统的云服务架构难以满足低延迟需求
网络约束 ：带宽受限环境下，频繁的云服务交互不切实际

对比主流向量数据库在边缘场景的表现：

数据库	内存占用	查询延迟	社区支持	边缘适配
Qdrant	中	低	良好	优秀
Milvus	高	中	优秀	一般
Weaviate	高	高	一般	差
FAISS	低	最低	优秀	无服务端

选择 Qdrant 的核心原因：

原生支持 gRPC 协议，通信效率远超 HTTP
提供嵌入式模式，可直接运行在应用进程内
动态分片机制适合资源受限环境
完善的量化索引（Scalar Quantization）支持

采用 8-bit 动态量化策略：

from transformers import AutoModel, AutoTokenizer
import torch

model = AutoModel.from_pretrained('BAAI/bge-m3', torch_dtype=torch.float16)
model = torch.quantization.quantize_dynamic(
    model,
    {torch.nn.Linear},  # 仅量化线性层
    dtype=torch.qint8
)
tokenizer = AutoTokenizer.from_pretrained('BAAI/bge-m3')

量化后模型大小减少 65%，推理速度提升 2.3 倍，精度损失 <2%

关键参数（qdrant_config.yml）：

storage:
  # 使用内存映射文件减少内存占用
  mmap_threshold_kb: 10000 
  # 限制单个分片内存使用  
  memmap_limit: 1073741824  # 1GB

optimizers:
  # 更频繁的段合并降低内存碎片
  default_segment_number: 2  
  max_segment_size: 100000

quantization:
  # 启用标量量化  
  scalar:
    type: int8
    always_ram: false  # 允许磁盘缓存

flowchart TD
    A[客户端请求] --> B{本地缓存?}
    B -->| 命中 | C[返回缓存结果]
    B -->| 未命中 | D[BGE-M3 嵌入计算]
    D --> E[Qdrant 向量检索]
    E --> F[结果后处理]
    F --> G[写入缓存]
    G --> H[返回结果]

class EmbeddingService:
    def __init__(self):
        self.model = load_quantized_model()
        self.tokenizer = load_tokenizer()
        self.pool = ThreadPoolExecutor(max_workers=4)  # 限制并发数

    async def embed(self, texts: List[str]) -> List[np.ndarray]:
        # 批处理优化
        inputs = self.tokenizer(
            texts, 
            padding=True, 
            truncation=True, 
            max_length=512,
            return_tensors="pt"
        )
        with torch.no_grad():
            outputs = self.model(**inputs)
        return outputs.last_hidden_state.mean(dim=1).numpy()

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams

client = QdrantClient(
    "localhost", 
    port=6333,
    grpc_port=6334,
    prefer_grpc=True  # 强制使用 gRPC
)

# 集合配置
client.recreate_collection(
    collection_name="docs",
    vectors_config=VectorParams(
        size=1024,  # BGE-M3 输出维度
        distance=Distance.COSINE
    )
)

# 带重试的搜索实现
def search_with_retry(query_vector, k=5, retries=3):
    for attempt in range(retries):
        try:
            return client.search(
                collection_name="docs",
                query_vector=query_vector,
                limit=k
            )
        except Exception as e:
            if attempt == retries - 1:
                raise
            time.sleep(2 ** attempt)  # 指数退避

测试环境：Atlas 300i A2 (4 核 /16GB)