pplx-embed 多语言嵌入模型

2026 年 2 月 26 日，Perplexity 发布了 pplx-embed —— 一系列最先进的多语言文本嵌入模型，从根本上重新思考了密集嵌入的生成方式。这些模型基于 Qwen3 构建，采用基于扩散的预训练和原生 INT8/二进制量化，专为处理数千万文档的现实世界、Web 规模检索任务而设计。

与此同时，Qdrant —— 领先的开源向量数据库 —— 发布了 1.17 版本，在搜索延迟、相关性反馈、写入负载性能和操作可观察性方面带来了重大改进。

本文将带你了解你需要知道的一切：

• pplx-embed 与其他嵌入模型的架构差异
• 从头开始完整设置 Qdrant（Docker、本地、Python 客户端）
• 使用 pplx-embed 进行嵌入和检索的生产就绪 Python 代码
• 深入探讨 Qdrant 的量化策略（标量、二进制、乘积）
• 完整讲解 Qdrant v1.17 的破坏性更改和新功能

到结束时，你将在你的机器上运行一个生产级的语义搜索系统。

1、了解 pplx-embed

pplx-embed 是 Perplexity 新的嵌入模型系列，在 MIT 许可证下发布，可在 Hugging Face Hub 上获取。有两个模型系列：

每个系列有两种规模可用：来源 Perplexity

1.1 核心创新：基于扩散的双向预训练

今天大多数嵌入模型都建立在仅解码器的具有因果注意力的 LLM 上 —— 这意味着每个 token 只能关注前面的 token。对于检索来说，这是一个根本性的架构限制：理解一段话通常需要知道单词之后的内容，而不仅仅是之前的内容。

pplx-embed 通过基于扩散的持续预训练解决了这个问题，这将因果 LLM（Qwen3）转换为双向编码器。这允许：

• 跨所有 token 的完整双向注意力
• 对所有 token 表示进行平均池化（不仅仅是最后一个）
• 上下文嵌入的后期分块 —— 每个块的嵌入都受到其出现的完整文档上下文的影响

这就是为什么 pplx-embed-context-v1 能够理解像"他打进了制胜一球"这样的句子在足球文章与板球文章中意味着完全不同的东西 —— 完整的文档上下文被烘焙到块嵌入中。

1.2 多阶段训练流水线

训练流水线在不同的阶段运行：

1. 仅英语预训练 —— 构建基础检索能力
2. 跨语言训练 —— 扩展到多语言检索
3. 完全多语言训练 —— 添加对 29+ 种语言的支持
4. 上下文训练（pplx-embed-context-v1） —— 结合序列内和批次内对比的双重损失，在块和文档级别
5. 三元组训练 —— 困难负例挖掘，以锐化相似但不相关文档之间的决策边界
6. SLERP 合并 —— 最终的 pplx-embed-v1 通过球面线性插值 (SLERP) 合并上下文和三元组检查点生成

1.3 基准性能

在 MTEB（多语言，v2） 上 —— 多语言检索的黄金标准：

• pplx-embed-v1-4B (INT8)：69.66% nDCG@10 —— 匹配 Qwen3-Embedding-4B (69.60%)，击败 Gemini-embedding-001 (67.71%)
• pplx-embed-v1-0.6B 在相同参数规模上优于 Qwen3-Embedding-0.6B

在 ConTEB（上下文检索基准）上：

• pplx-embed-context-v1-4B (INT8)：81.96% nDCG@10 —— 击败 voyage-context-3 (79.45%) 和 Anthropic Contextual (72.4%)

在 Perplexity 内部的 PPLXQuery2Doc（3000 万语料库）上：

• pplx-embed-v1-4B：91.7% Recall@1000 vs Qwen3-Embedding-4B 的 88.6%

1.4 原生量化：内置的存储效率

与大多数需要事后量化的模型不同，pplx-embed 原生地生成INT8 和二进制嵌入：

在 Web 规模（数十亿文档）下，这不仅仅是一个锦上添花的功能 —— 它是可行和不可行部署之间的区别。

1.5 支持的推理框架

• HuggingFace Transformers
• SentenceTransformers
• Text Embeddings Inference (TEI)
• Transformers.js（浏览器/Node.js）
• ONNX Runtime
• Perplexity API（托管）

2、设置 Qdrant 向量数据库

Qdrant 是一个用 Rust 编写的开源、高性能向量数据库。它支持 HNSW 索引、多种距离度量、负载过滤、稀疏向量、量化和分布式部署。

选项 1：Docker（推荐用于生产）

docker run -d \
  --name qdrant \
  -p 6333:6333 \
  -p 6334:6334 \
  -v $(pwd)/qdrant_storage:/qdrant/storage \
  qdrant/qdrant

这会暴露：

• 端口 6333：REST API
• 端口 6334：gRPC API
• qdrant_storage/：持久化数据卷

选项 2：Docker Compose（推荐用于配置）

创建一个 docker-compose.yml：

version: "3.8"
services:
  qdrant:
    image: qdrant/qdrant:v1.17.0
    container_name: qdrant
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - ./qdrant_storage:/qdrant/storage
      - ./qdrant_config.yaml:/qdrant/config/production.yaml
    environment:
      - QDRANT__SERVICE__GRPC_PORT=6334
    restart: unless-stopped

可选的 qdrant_config.yaml：

storage:
  on_disk_payload: true  # 在磁盘上存储负载以节省 RAM
service:
  max_request_size_mb: 32

log_level: INFO

启动它：

docker compose up -d

选项 3：本地二进制（macOS/Linux）

# 下载最新版本
curl -LO https://github.com/qdrant/qdrant/releases/latest/download/qdrant-x86_64-unknown-linux-gnu.tar.gz
tar -xzf qdrant-*.tar.gz
./qdrant

选项 4：Qdrant Cloud（完全托管）

访问 cloud.qdrant.io → 创建免费集群 → 获取你的 API 密钥和集群 URL。

安装 Python 客户端

pip install qdrant-client openai sentence-transformers transformers torch

对于 Perplexity 的托管 API：

pip install qdrant-client openai  # Perplexity 使用 OpenAI 兼容的 API

验证连接

from qdrant_client import QdrantClient

# 本地 Docker 实例
client = QdrantClient(host="localhost", port=6333)

# 或 Qdrant Cloud
# client = QdrantClient(
#     url="https://your-cluster-url.qdrant.io",
#     api_key="your-api-key"
# )

# 健康检查
info = client.get_collections()
print(f"✅ 已连接到 Qdrant | 集合: {len(info.collections)}")

3、使用 pplx-embed 与 Qdrant

设置：API 密钥和导入

import os
from openai import OpenAI
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance,
    VectorParams,
    PointStruct,
    ScalarQuantizationConfig,
    ScalarType,
    QuantizationSearchParams,
    SearchParams,
    Filter,
    FieldCondition,
    MatchValue,
)
import uuid
import time
from typing import List, Dict, Any, Optional

# Perplexity API 客户端（OpenAI 兼容）
pplx_client = OpenAI(
    api_key=os.environ.get("PERPLEXITY_API_KEY"),
    base_url="https://api.perplexity.ai"
)

# Qdrant 客户端
client = QdrantClient(host="localhost", port=6333)

# 选择你的模型
EMBEDDING_MODEL = "pplx-embed-v1-0.6b"      # 快速、轻量级
# EMBEDDING_MODEL = "pplx-embed-context-v1-4b"  # 最适合 RAG 文档块

步骤 1：使用 pplx-embed 生成嵌入

def embed_texts(texts: List[str]) -> List[List[float]]:
    """
    使用 pplx-embed API 生成嵌入。
    支持批处理以获得更高吞吐量。
    """
    response = pplx_client.embeddings.create(
        model=EMBEDDING_MODEL,
        input=texts
    )
    return [item.embedding for item in response.data]

# 示例用法
sample_texts = [
    "语义搜索通过理解文本的含义而非关键字来工作",
    "Qdrant 是一个用 Rust 编写的高性能向量数据库",
    "量化减少了内存使用，同时保持检索质量"
]

embeddings = embed_texts(sample_texts)
print(f"✅ 生成了 {len(embeddings)} 个嵌入，每个 {len(embeddings[0])} 维")

步骤 2：使用标量量化创建 Qdrant 集合

def create_collection(
    collection_name: str,
    vector_size: int,
    quantile: float = 0.99
):
    """
    创建优化的 Qdrant 集合。

    配置：
    - 余弦距离（最适合嵌入）
    - INT8 标量量化（4x 压缩，~1-3% 召回损失）
    - HNSW 索引（可调优）
    """
    client.create_collection(
        collection_name=collection_name,
        vectors_config=VectorParams(
            size=vector_size,
            distance=Distance.COSINE,
        ),
        quantization_config=ScalarQuantizationConfig(
            type=ScalarType.INT8,
            quantile=quantile,
            always_ram=True,  # 将量化向量保留在 RAM 中以实现快速搜索
        ),
        hnsw_config={
            "m": 16,           # 连接数（权衡精度 vs 速度）
            "ef_construct": 100,  # 索引构建速度
            "on_disk": False    # 将 HNSW 图保持在 RAM 中
        }
    )
    print(f"✅ 已创建集合: {collection_name}")

# 创建我们的集合
create_collection(
    collection_name="semantic_search",
    vector_size=1024  # pplx-embed-v1-0.6b 的维度
)

步骤 3：索引文档

def index_documents(
    documents: List[Dict[str, Any]],
    collection_name: str = "semantic_search",
    batch_size: int = 100
):
    """
    批量索引文档到 Qdrant。

    每个文档：
    - id: 唯一标识符
    - vector: pplx-embed 嵌入
    - payload: 元数据（文本、类别等）
    """
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]
        
        # 为批次生成嵌入
        texts = [d["text"] for d in batch]
        embeddings = embed_texts(texts)
        
        # 创建 Qdrant 点
        points = [
            PointStruct(
                id=d.get("id", str(uuid.uuid4())),
                vector=emb,
                payload={
                    "text": d["text"],
                    "category": d.get("category", "general"),
                    "timestamp": d.get("timestamp", int(time.time()))
                }
            )
            for d, emb in zip(batch, embeddings)
        ]
        
        # 上传到 Qdrant
        client.upsert(
            collection_name=collection_name,
            points=points,
            wait=True
        )
        
        print(f"✅ 已索引 {i+len(batch)}/{len(documents)} 个文档")

# 示例：索引示例数据集
sample_documents = [
    {
        "id": "1",
        "text": "Qdrant 支持稀疏向量，这对于混合检索很有用",
        "category": "technical"
    },
    {
        "id": "2",
        "text": "语义搜索优于关键词搜索，因为它理解意图",
        "category": "search"
    },
    {
        "id": "3",
        "text": "pplx-embed 使用双向注意力以提高检索质量",
        "category": "ai"
    }
]

index_documents(sample_documents)

步骤 4：使用量化感知参数进行语义搜索

def search(
    query: str,
    collection_name: str = "semantic_search",
    top_k: int = 5,
    filters: Optional[Dict] = None,
    rescore: bool = True
) -> List[Dict]:
    """
    使用量化感知参数进行搜索。
    使用 INT8 实现速度 + FP32 重评分保证质量。
    """
    # 嵌入查询
    query_vector = embed_texts([query])[0]
    
    # 构建过滤器
    query_filter = None
    if filters:
        query_filter = Filter(
            must=[
                FieldCondition(key=k, match=MatchValue(value=v))
                for k, v in filters.items()
            ]
        )
    
    # 使用优化参数搜索
    results = client.search(
        collection_name=collection_name,
        query_vector=query_vector,
        query_filter=query_filter,
        limit=top_k,
        search_params=SearchParams(
            quantization=QuantizationSearchParams(
                ignore=False,    # 不忽略量化
                rescore=rescore,    # 使用 FP32 重评分
                oversampling=2.0     # 过采样以提高召回
            ),
            hnsw_ef=128    # HNSW 搜索深度
        ),
        with_payload=True
    )
    
    # 格式化结果
    return [
        {
            "rank": i + 1,
            "score": round(r.score, 4),
            "text": r.payload.get("text", ""),
            "category": r.payload.get("category", ""),
            "metadata": {k: v for k, v in r.payload.items() if k != "text"}
        }
        for i, r in enumerate(results)
    ]

# 示例搜索
results = search(
    "语义搜索的好处是什么？",
    top_k=3,
    filters={"category": "search"}
)

for r in results:
    print(f"#{r['rank']} [{r['score']}] {r['text'][:80]}...")

步骤 5：使用 pplx-embed-context-v1 的 RAG 管道

def contextual_rag_search(
    query: str,
    collection_name: str = "semantic_search",
    top_k: int = 3
) -> Dict:
    """
    使用 pplx-embed-context-v1 的上下文感知 RAG 搜索。
    
    上下文嵌入包含完整文档的语义信息，
    使得 RAG 检索更准确。
    """
    # 使用上下文模型嵌入
    context_model = "pplx-embed-context-v1-0.6b"
    
    # 嵌入查询和文档（使用上下文）
    query_embedding = pplx_client.embeddings.create(
        model=context_model,
        input=[query]
    ).data[0].embedding
    
    # 搜索
    results = client.search(
        collection_name=collection_name,
        query_vector=query_embedding,
        limit=top_k,
        with_payload=True
    )
    
    # 返回上下文感知的结果
    return {
        "query": query,
        "results": [
            {
                "text": r.payload.get("text", ""),
                "score": r.score,
                "context_note": "此嵌入包含完整文档的上下文信息"
            }
            for r in results
        ]
    }

# 使用示例
rag_result = contextual_rag_search(
    "量化如何影响向量搜索？"
)

print(f"查询: {rag_result['query']}")
for i, r in enumerate(rag_result['results'], 1):
    print(f"{i}. [{r['score']}] {r['text'][:60]}")

4、Qdrant 量化深入探讨

量化是存储、速度和精度之间的权衡：

策略 1：标量量化（INT8）—— 生产默认

INT8 标量量化是最常见的量化方法。它将每个向量维度从 FP32 缩减到 8 位整数。

from qdrant_client.models import (
    ScalarQuantizationConfig,
    ScalarType,
)

# 创建具有 INT8 量化的集合
client.create_collection(
    collection_name="int8_collection",
    vectors_config=VectorParams(
        size=1024,
        distance=Distance.COSINE,
    ),
    quantization_config=ScalarQuantizationConfig(
        type=ScalarType.INT8,
        quantile=0.99,    # 使用 99% 分位数用于范围
        always_ram=True,      # 保持量化向量在 RAM 中
    ),
)

工作原理：

1. 分析范围： 扫描所有向量以找到每个维度的范围
2. 缩放： 将 FP32 值线性缩放到 [-128, 127]
3. 舍入： 舍入到最接近的整数
4. 重评分： 搜索时使用 INT8 获得候选，FP32 重评分前 k 个结果

使用 FP32 重评分进行最大质量搜索：

# 使用重评分进行搜索
results = client.search(
    collection_name="int8_collection",
    query_vector=query_embedding,
    limit=10,
    search_params=SearchParams(
        quantization=QuantizationSearchParams(
            ignore=False,      # 使用量化向量
            rescore=True,       # 用原始 FP32 重评分
            oversampling=2.0    # 过采样以补偿召回损失
        )
    )
)

策略 2：二进制量化 —— 最大压缩

二进制量化将每个维度存储为单个位（0 或 1），实现 32x 的压缩。

from qdrant_client.models import BinaryQuantizationConfig

# 创建具有二进制量化的集合
client.create_collection(
    collection_name="binary_collection",
    vectors_config=VectorParams(
        size=1024,
        distance=Distance.COSINE,  # 注意：余弦需要汉明距离等价物
    ),
    quantization_config=BinaryQuantizationConfig(
        always_ram=True,
    ),
)

权衡：

• 优势： 最大存储节省（32x），超快搜索（汉明距离）
• 劣势： 较高的召回损失（~5-10%），需要更大的向量大小

何时使用：

• 数十亿文档的数据集
• 内存是主要约束
• 可以容忍较低的召回以换取巨大的存储节省

策略 3：乘积量化 —— 可配置压缩

乘积量化（PQ）将向量分解为子空间，每个子空间独立量化。

from qdrant_client.models import ProductQuantizationConfig

# 创建具有 PQ 量化的集合
client.create_collection(
    collection_name="pq_collection",
    vectors_config=VectorParams(
        size=2560,  # pplx-embed-v1-4b 的维度
        distance=Distance.COSINE,
    ),
    quantization_config=ProductQuantizationConfig(
        compression_ratio=0.25,  # 4x 压缩（可调优）
        always_ram=True,
    ),
)

工作原理：

1. 分解： 将 2560 维向量分成子空间
2. 聚类： 每个子空间聚类为代码本
3. 编码： 存储每个子向量的最近代码本索引
4. 解码： 搜索时解码以近似向量

压缩比：

# 可配置压缩比
compression_ratios = {
    0.5: "2x 压缩",
    0.25: "4x 压缩",
    0.125: "8x 压缩",
}

更新现有集合的量化

# 将大型集合从标量升级为二进制
client.update_collection(
    collection_name="large_collection",
    quantization_config=BinaryQuantizationConfig()
)

print("✅ 量化配置已更新")

选择正确的量化策略

def choose_quantization(
    num_documents: int,
    memory_constraint: str,
    accuracy_requirement: str
) -> str:
    """
    根据约束选择量化策略。
    """
    # 超大规模
    if num_documents > 1_000_000_000:
        if memory_constraint == "severe":
            return "binary"
        return "product"
    
    # 生产默认
    if accuracy_requirement == "high":
        return "int8"
    
    # 平衡
    return "int8"

5、Qdrant v1.17 的新功能

5.1 相关性反馈查询（新功能）

相关性反馈允许根据用户反馈改进搜索结果。

# 相关性反馈：根据已知相关/不相关的示例改进结果
user_feedback = {
    "relevant_ids": ["doc_1", "doc_5", "doc_10"],
    "irrelevant_ids": ["doc_3", "doc_7"]
}

# 向用户展示结果并收集反馈后：
# 使用推荐 API 进行相关性反馈
recommend_result = client.recommend(
    collection_name="semantic_search",
    positive=user_feedback["relevant_ids"],
    negative=user_feedback["irrelevant_ids"],
    limit=10,
    using="vector",  # 使用向量（而非原始文本）
)

print(f"✅ 基于反馈推荐了 {len(recommend_result)} 个结果")

5.2 搜索延迟改进

2a. 避免大型未优化分段

# qdrant_config.yaml — v1.17 新配置
storage:
  # v1.17: 优化分段大小以避免大型未优化分段
  optimizers:
    default_segment_number: 4        # 默认段数
    indexing_threshold: 20000      # 索引阈值
    flush_interval_sec: 5           # 刷新间隔

2b. 尾部延迟的延迟扇出

# v1.17: 配置每个搜索请求的延迟扇出
client.search(
    collection_name="semantic_search",
    query_vector=query_embedding,
    limit=10,
    search_params=SearchParams(
        # v1.17 — 延迟扇出配置
        timeout=2.0,  # 2 秒超时
        on_disk=False,  # 从 RAM 搜索以减少延迟
    )
)

# v1.17 — 延迟扇出配置
# 此功能在后台异步扇出请求到副本，
# 减少分布式部署中的尾部延迟

5.3 加权倒数排名融合 (RRF)

# v1.17: 加权 RRF — 使用自定义权重结合密集和稀疏搜索
results = client.search(
    collection_name="hybrid_search",
    query_vector=dense_embedding,
    limit=10,
    # v1.17: 添加稀疏向量
    query_filter=Filter(
        must=[FieldCondition(key="category", match=MatchValue(value="tech"))]
    ),
    # 使用 RRF 配置
    search_params=SearchParams(
        # v1.17: 自定义 RRF 权重
        # dense_vector_weight + sparse_vector_weight
        # 默认: 1.0 dense, 0.5 sparse
    )
)

5.4 使用更新模式的 Upsert（仅插入 / 仅更新）

# 仅插入：如果点 ID 已存在则失败（严格去重）
client.upsert(
    collection_name="strict_collection",
    points=[PointStruct(id="doc_1", vector=vec, payload={...})],
    wait=True
)

# 仅更新：如果 ID 不存在则失败（避免意外插入）
client.upsert(
    collection_name="update_only_collection",
    points=[PointStruct(id="doc_1", vector=vec, payload={...})],
    wait=True
)

5.5 集群范围遥测 API

# v1.17 新集群遥测端点
telemetry = client.get_telemetry_data()

# 监控：
# - 内存使用
# - CPU 使用
# - 搜索延迟
# - 索引大小

print(f"内存使用: {telemetry['memory_usage']}")
print(f"搜索延迟: {telemetry['search_latency']}")

5.6 分段优化监控

# 监控分段优化状态
info = client.get_collection_info("semantic_search")

# v1.17: 检查分段状态
if info.status == "green":
    print("✅ 分段已优化")
elif info.status == "yellow":
    print("⚠️  分段需要优化")

# 详细集合遥测
details = client.get_collection("semantic_search")

print(f"向量计数: {details.result.vectors_count}")
print(f"段数: {details.result.segments_count}")

5.7 审计日志

# qdrant_config.yaml
# v1.17: 审计日志用于合规性
service:
  enable_audit_log: true
  audit_log_path: "/var/log/qdrant/audit.log"
  audit_log_filters:
    - "write_operations"      # 记录写入操作
    - "search_queries"        # 记录搜索查询
    - "admin_actions"          # 记录管理操作

5.8 改进的副本恢复 WAL

# v1.17: 副本恢复的 WAL 自动扩展
storage:
  # v1.17: 配置 WAL 容量和扩展
  wal_capacity_mb: 32  # 基本 WAL 容量
  
  # v1.17: 当远程副本不可用时的乘数
  # WAL 扩展以保留离线副本的操作
  # （自动配置 —— 无需手动设置）

5.9 通过 Web UI 进行重分片

Qdrant Cloud Web UI 现在支持重分片集合 —— 在不停机的情况下调整实时集合的分片数量。以前这需要手动 API 调用和仔细协调。

5.10 请求头中的外部提供者 API 密钥

# v1.17: 在请求头中传递外部推理 API 密钥（不仅仅是服务器配置）
# 对于每个用户都有自己的 API 密钥的多租户部署很有用
results = client.search(
    collection_name=COLLECTION_NAME,
    query_vector=query_embedding,
    limit=10,
    # 新：external_inference_key 在头中传递以进行每个请求的 API 密钥路由
)

6、生产架构 —— 端到端

这是一个结合一切的完整生产就绪模式：

class PPLXEmbedQdrantRetriever:
    """
    使用 pplx-embed + Qdrant v1.17 的生产语义搜索系统。

    功能：
    - pplx-embed-v1 或 pplx-embed-context-v1 嵌入
    - INT8 标量量化与 FP32 重评分
    - 用于基于元数据缩小的负载过滤
    - 异步就绪批处理嵌入
    """

    def __init__(
        self,
        collection_name: str = "production_kb",
        embedding_model: str = "pplx-embed-v1-4b",
        vector_dim: int = 2560,
        qdrant_host: str = "localhost",
        qdrant_port: int = 6333,
        pplx_api_key: str = None
    ):
        self.collection_name = collection_name
        self.embedding_model = embedding_model
        self.vector_dim = vector_dim

        self.qdrant = QdrantClient(host=qdrant_host, port=qdrant_port)
        self.pplx = OpenAI(
            api_key=pplx_api_key or os.environ["PERPLEXITY_API_KEY"],
            base_url="https://api.perplexity.ai"
        )

        self._ensure_collection()

    def _ensure_collection(self):
        """如果集合不存在，则创建具有最佳设置的集合。"""
        existing = [c.name for c in self.qdrant.get_collections().collections]
        if self.collection_name not in existing:
            self.qdrant.create_collection(
                collection_name=self.collection_name,
                vectors_config=VectorParams(
                    size=self.vector_dim,
                    distance=Distance.COSINE,
                ),
                quantization_config=ScalarQuantizationConfig(
                    type=ScalarType.INT8,
                    quantile=0.99,
                    always_ram=True,
                ),
                hnsw_config={"m": 16, "ef_construct": 100, "on_disk": False}
            )

    def embed(self, texts: List[str]) -> List[List[float]]:
        """通过 pplx-embed API 生成嵌入。"""
        response = self.pplx.embeddings.create(
            model=self.embedding_model,
            input=texts
        )
        return [item.embedding for item in response.data]

    def add_documents(self, documents: List[Dict], batch_size: int = 100):
        """使用自动批处理索引文档。"""
        for i in range(0, len(documents), batch_size):
            batch = documents[i:i + batch_size]
            embeddings = self.embed([d["text"] for d in batch])

            self.qdrant.upsert(
                collection_name=self.collection_name,
                points=[
                    PointStruct(
                        id=d.get("id", str(uuid.uuid4())),
                        vector=emb,
                        payload={
                            "text": d["text"],
                            **d.get("metadata", {})
                        }
                    )
                    for d, emb in zip(batch, embeddings)
                ],
                wait=True
            )

    def search(
        self,
        query: str,
        top_k: int = 5,
        filters: Optional[Dict] = None,
        rescore: bool = True
    ) -> List[Dict]:
        """
        使用量化感知参数进行搜索。
        使用 INT8 实现速度 + FP32 重评分保证质量。
        """
        query_vector = self.embed([query])[0]

        query_filter = None
        if filters:
            query_filter = Filter(
                must=[
                    FieldCondition(key=k, match=MatchValue(value=v))
                    for k, v in filters.items()
                ]
            )

        results = self.qdrant.search(
            collection_name=self.collection_name,
            query_vector=query_vector,
            query_filter=query_filter,
            limit=top_k,
            search_params=SearchParams(
                quantization=QuantizationSearchParams(
                    ignore=False,
                    rescore=rescore,
                    oversampling=2.0
                ),
                hnsw_ef=128
            ),
            with_payload=True
        )

        return [
            {
                "rank": i + 1,
                "score": round(r.score, 4),
                "text": r.payload.get("text", ""),
                "metadata": {k: v for k, v in r.payload.items() if k != "text"}
            }
            for i, r in enumerate(results)
        ]

# 使用示例
retriever = PPLXEmbedQdrantRetriever(
    collection_name="production_kb",
    embedding_model="pplx-embed-v1-4b",
    vector_dim=2560
)

retriever.add_documents(sample_documents)

results = retriever.search(
    "语义搜索的好处是什么？",
    top_k=3,
    filters={"category": "search"}
)

for r in results:
    print(f"#{r['rank']} [{r['score']}] {r['text'][:80]}...")

7、结束语

pplx-embed + Qdrant v1.17 的组合代表了一个成熟的、生产级语义搜索堆栈：

来自 Perplexity 发布的关键洞察：量化不是妥协 —— 它是一个设计选择。当你的嵌入模型从零开始被训练以生成高质量的 INT8 和二进制表示时，量化成为一个功能，而不是权衡。

原文链接: pplx-embed + Qdrant: Building Production-Grade Semantic Search with Quantization

汇智网翻译整理，转载请标明出处