> ## Content Index > Fetch the complete content index at: https://www.hubwiz.com/blog/llms.txt > Use this file to discover other available public pages before exploring further. # pplx-embed 多语言嵌入模型 - URL: https://www.hubwiz.com/blog/pplx-embed-multilingual/ - Published: 2026-03-09T00:57:48.000Z - Updated: 2026-03-09T00:57:48.000Z - Description: Perplexity 新的基于扩散的多语言嵌入模型如何与 Qdrant 向量数据库结合实现 Web 规模检索 —— 完整的设置、Python 代码、量化策略以及 Qdrant v1.17 的所有新功能 - Author: admin - Tags: MODEL-ZOO, PPLX-EMBED, #Import 2026-10-14 14:41 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** 上获取。有两个模型系列： | 模型 | 用途 | | --------------------- | ------------------- | | pplx-embed-v1 | 用于独立文本的标准密集文本嵌入 | | pplx-embed-context-v1 | 用于 RAG 管道中文档块的上下文嵌入 | 每个系列有两种规模可用：来源 Perplexity | 模型 | 维度 | 上下文 | 量化 | 价格（每 100 万 token） | | -------------------------- | ---- | --- | ------------- | ----------------- | | pplx-embed-v1-0.6b | 1024 | 32K | INT8 / BINARY | → $0.004 | | pplx-embed-v1-4b | 2560 | 32K | INT8 / BINARY | → $0.030 | | pplx-embed-context-v1-0.6b | 1024 | 32K | INT8 / BINARY | → $0.008 | | pplx-embed-context-v1-4b | 2560 | 32K | INT8 / BINARY | → $0.050 | ### 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 和二进制嵌入**： | 格式 | 相比 FP32 的存储节省 | | ------- | ------------- | | FP32 基线 | \- | | INT8 | 4x 减少 | | BINARY | 32x 减少 | 在 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](https://cloud.qdrant.io/?ref=hubwiz.com) → 创建免费集群 → 获取你的 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 量化深入探讨 量化是存储、速度和精度之间的权衡： | 策略 | 存储 | 速度 | 召回损失 | 适用场景 | | --------- | --- | -- | ------- | ------------ | | FP32（无量化） | 1x | 基线 | 0% | 小型数据集，需要最高精度 | | INT8 标量 | 4x | 快 | \~1-3% | 生产默认，平衡的选择 | | 二进制 | 32x | 最快 | \~5-10% | 超大规模，内存受限 | ### 策略 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 的组合代表了一个成熟的、生产级语义搜索堆栈： | 层 | 技术 | 原因 | | --------- | --------------------- | -------------------------- | | 嵌入 | pplx-embed-v1 | 双向注意力、原生 INT8、SOTA 基准 | | 上下文感知 RAG | pplx-embed-context-v1 | 块嵌入中的文档级上下文 | | 向量存储 | Qdrant v1.17 | Rust 性能、丰富的量化、生产可靠性 | | 内存效率 | INT8 标量量化 | 4x 压缩、\~1-3% 召回损失、FP32 重评分 | | 极端规模 | 二进制量化 | 十亿规模数据集的 32x 压缩 | | 延迟 | 延迟扇出 (v1.17) | 消除分布式部署中的尾部延迟 | | 相关性调优 | 相关性反馈 (v1.17) | 用户反馈随时间改进结果 | | 混合搜索 | 加权 RRF (v1.17) | 使用自定义权重结合密集 + 稀疏 | 来自 Perplexity 发布的关键洞察：**量化不是妥协 —— 它是一个设计选择**。当你的嵌入模型从零开始被训练以生成高质量的 INT8 和二进制表示时，量化成为一个功能，而不是权衡。 --- 原文链接: [pplx-embed + Qdrant: Building Production-Grade Semantic Search with Quantization](https://pub.towardsai.net/pplx-embed-qdrant-building-production-grade-semantic-search-with-quantization-cea9626b8c7c?ref=hubwiz.com) 汇智网翻译整理，转载请标明出处