企业官网建设流程全解析

Haystack × Weaviate 集成实战：WeaviateDocumentStore 与 BM25 / Embedding / Hybrid 三种检索器完全指南

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack

本篇技术指南围绕 Haystack（version-2.23）与 Weaviate 的官方集成展开，系统讲解WeaviateDocumentStore的初始化、认证、文档写入与元数据管理能力，以及WeaviateBM25Retriever、WeaviateEmbeddingRetriever、WeaviateHybridRetriever三种检索器的构造参数、run/run_async调用方式与filter_policy的底层合并逻辑。读完本文，你将能够基于 Weaviate 独立搭建可运行的关键词检索、向量检索与混合检索链路，并将它们嵌入 Haystack Pipeline 构成 RAG 应用。

集成概览与接入方式

Weaviate 是一个可同时存储向量嵌入与数据对象的多用途向量数据库，适合多模态场景。WeaviateDocumentStore可连接任意 Weaviate 实例，无论是 Weaviate Cloud Services、Kubernetes 还是本地 Docker 容器。

安装方式只有一个命令：

pip install weaviate-haystack

该集成包含四个核心模块，均可在haystack_integrations命名空间下找到：

• haystack_integrations.document_stores.weaviate.document_store——WeaviateDocumentStore
• haystack_integrations.document_stores.weaviate.auth—— 四类认证凭据类
• haystack_integrations.components.retrievers.weaviate.bm25_retriever——WeaviateBM25Retriever
• haystack_integrations.components.retrievers.weaviate.embedding_retriever/hybrid_retriever—— 向量与混合检索器

更详细的 API 参考见 Weaviate 集成 API 文档，对应的用户指南见 weaviatedocumentstore.mdx 与 weaviatebm25retriever.mdx 等页面。

方式一：Weaviate Embedded（临时实例）

若只想快速试验，无需单独部署，可直接在客户端内创建嵌入式 Weaviate 集群：

from haystack_integrations.document_stores.weaviate.document_store import ( WeaviateDocumentStore, ) from weaviate.embedded import EmbeddedOptions document_store = WeaviateDocumentStore(embedded_options=EmbeddedOptions())

embedded_options对应 Weaviate 官方客户端中的weaviate.embedded.EmbeddedOptions，可用于配置嵌入式实例的完整选项列表。该方式适合开发与测试，不适合生产。

方式二：本地 Docker 容器

一份最小可用的docker-compose.yml如下（暴露 REST 8080 与 gRPC 50051 端口）：

services: weaviate: command: - --host - 0.0.0.0 - --port - '8080' - --scheme - http image: semitechnologies/weaviate:1.36.2 ports: - 8080:8080 - 50051:50051 volumes: - weaviate_data:/var/lib/weaviate restart: 'no' environment: QUERY_DEFAULTS_LIMIT: 25 AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'true' PERSISTENCE_DATA_PATH: '/var/lib/weaviate' DEFAULT_VECTORIZER_MODULE: 'none' ENABLE_MODULES: '' CLUSTER_HOSTNAME: 'node1' volumes: weaviate_data:

启动容器并初始化 Document Store：

docker compose up -d

from haystack_integrations.document_stores.weaviate.document_store import ( WeaviateDocumentStore, ) from haystack import Document document_store = WeaviateDocumentStore(url="http://localhost:8080") document_store.write_documents( [Document(content="This is first"), Document(content="This is second")], ) print(document_store.count_documents())

注意：上述 compose 配置显式开启了无认证访问，仅适合本地开发，生产环境强烈不建议。认证方案见下文"认证体系"章节。

方式三：Weaviate Cloud Service（WCS）

先在 Weaviate 云控制台创建集群，取得 URL 与 API Key，然后：

import os from haystack_integrations.document_stores.weaviate import ( WeaviateDocumentStore, AuthApiKey, ) os.environ["WEAVIATE_API_KEY"] = "YOUR-API-KEY" document_store = WeaviateDocumentStore( url="YOUR-WEAVIATE-URL", auth_client_secret=AuthApiKey(), )

认证体系：四类 AuthCredentials 与环境变量

haystack_integrations.document_stores.weaviate.auth模块基于AuthCredentials（ABC 抽象基类）定义了四种认证方式。每个类都将敏感信息存储为 Haystack 的Secret，在需要时从环境变量加载，并实现resolve_value()将密钥解析为 Weaviate 客户端对应的认证对象（WeaviateAuthApiKey、WeaviateAuthBearerToken、WeaviateAuthClientCredentials、WeaviateAuthClientPassword）。SupportedAuthTypes枚举通过from_class()将认证类映射为枚举值，供序列化使用。

其中WEAVIATE_SCOPE可选，若设置可为单个字符串（如"scope1"）或空格分隔的字符串列表（如"scope1 scope2"）。

每个认证类都支持to_dict()/from_dict()序列化，并可通过from_dict反序列化任意受支持的认证凭据。若想更换环境变量名，可显式传入Secret：

from haystack_integrations.document_stores.weaviate.auth import AuthApiKey from haystack.utils.auth import Secret AuthApiKey(api_key=Secret.from_env_var("MY_ENV_VAR"))

WeaviateDocumentStore：核心 Document Store 详解

构造参数

__init__( *, url: str | None = None, collection_settings: dict[str, Any] | None = None, auth_client_secret: AuthCredentials | None = None, additional_headers: dict | None = None, embedded_options: EmbeddedOptions | None = None, additional_config: AdditionalConfig | None = None, grpc_port: int = 50051, grpc_secure: bool = False ) -> None

各参数作用：

• url：Weaviate 实例地址。自托管时为http://localhost:8080，WCS 时为云集群 URL。

• collection_settings：集合（Collection）配置。为None时使用名为default的集合，属性如下：

  • _original_id：text（保留原始文档 ID）
  • content：text（文档正文）
  • blob_data：blob（二进制数据）
  • blob_mime_type：text（MIME 类型）
  • score：number（分数）

  注意：默认集合配置刻意省略了 Document 的meta字段，因为无法对 meta 结构做假设。官方强烈建议为你的业务场景创建包含正确 meta 属性的自定义集合；也可依赖自动 schema 生成，但官方不推荐用于生产。

• auth_client_secret：认证凭据，取AuthBearerToken、AuthClientPassword、AuthClientCredentials、AuthApiKey之一。

• additional_headers：附加请求头，常用于携带模型服务商密钥，例如{"X-OpenAI-Api-Key": "<THE-KEY>"}或{"X-HuggingFace-Api-Key": "<THE-KEY>"}（当 Weaviate 侧配置了向量化模块时使用）。

• embedded_options：设置后在客户端内部创建嵌入式 Weaviate 集群。

• additional_config：传给 Weaviate 的额外高级配置。

• grpc_port：gRPC 连接端口，默认 50051。

• grpc_secure：是否对底层 gRPC API 使用安全通道，默认False。

惰性客户端与集合访问

WeaviateDocumentStore提供四个惰性属性，首次访问时才创建并连接：

• client：同步 Weaviate 客户端
• async_client：异步客户端
• collection：同步集合对象（Collection[dict[str, Any], None]）
• async_collection：异步集合对象

配套的close()/close_async()分别释放同步与异步资源；三个检索器与 Document Store 也各自实现了close/close_async，用于释放底层 Document Store 资源。

文档写入：write_documents 与 DuplicatePolicy

write_documents( documents: list[Document], policy: DuplicatePolicy = DuplicatePolicy.NONE ) -> int

DuplicatePolicy定义在 haystack/document_stores/types/policy.py，取值包括NONE（默认，不检查重复）、SKIP（跳过重复）、OVERWRITE（覆盖）、FAIL（报错）。

一个值得注意的实现细节：官方推荐使用OVERWRITE策略，因为它是唯一可以使用 Weaviate batch API 的策略，写入速度更快；其余策略无法使用 batch API——batch 接口不返回"文档是否已存在"的信息，这会导致FAIL策略无法返回错误、SKIP策略无法跳过重复文档。

异常情况：输入无效时抛ValueError；使用FAIL策略发现重复文档时抛DuplicateDocumentError；批量写入失败时抛DocumentStoreError。异步版本为write_documents_async。

文档检索：filter_documents 与元数据过滤

filter_documents(filters: dict[str, Any] | None = None) -> list[Document]

按 Haystack 的DocumentStore.filter_documents()协议执行过滤。一个易踩的坑：contains过滤操作符是大小写敏感的（子串匹配）。如需大小写不敏感匹配，请在构造 filter 前先对值做归一化处理。

文档管理：计数、删除、更新

• count_documents()/count_documents_async()：返回存储中的文档总数。
• count_documents_by_filter(filters)/ 异步版：按过滤器统计文档数量。
• delete_documents(document_ids)/ 异步版：按文档 ID 列表删除。
• delete_by_filter(filters)/ 异步版：删除所有匹配过滤器的文档，返回删除数量。
• update_by_filter(filters, meta)/ 异步版：更新所有匹配文档的元数据（meta会与现有元数据合并），返回更新数量。
• delete_all_documents(*, recreate_index=False, batch_size=1000)/ 异步版：
  • recreate_index=False：保留集合，按批次迭代删除（batch_size默认 1000，需小于等于 Weaviate 部署的QUERY_MAXIMUM_RESULTS配置，默认 10000）；
  • recreate_index=True：直接 drop 集合再重建，官方推荐用于性能优化。

元数据统计能力

WeaviateDocumentStore实现了丰富的元数据统计方法，为检索结果可视化、faceting 类功能提供数据支撑：

• get_metadata_fields_info()/ 异步版：返回元数据字段名与类型映射，排除特殊字段（content、blob_data、blob_mime_type、_original_id、score）。示例返回：{'number': {'type': 'int'}, 'date': {'type': 'date'}, 'category': {'type': 'text'}}。
• get_metadata_field_min_max(metadata_field)/ 异步版：返回数值或日期字段的最小/最大值，字段名可加meta.前缀（如meta.year或year）；字段不存在或不支持 min/max 时抛ValueError。
• count_unique_metadata_by_filter(filters, metadata_fields)/ 异步版：对指定字段统计去重值数量；字段不存在时抛ValueError。
• get_metadata_field_unique_values(metadata_field, search_term=None, from_=0, size=10, filters=None)/ 异步版：分页返回字段去重值，search_term做大小写不敏感的子串过滤（无词干还原），返回(去重值列表, 去重总数)。

一个由 weaviate-client 协议决定的类型陷阱值得注意：标量int元数据值取回时会变成float。原因是 weaviate-client 没有标量 int 的线上协议字段——非列表属性被打包进google.protobuf.Struct，其Value类型只有number_value（double），int/float 的区分在到达 Weaviate 之前就丢失了；GroupByAggregate对数值分组键的解码方式相同，因此即使 schema 显式声明DataType.INT也如此。而列表型 int 字段（如meta={"tags": [1, 2]}）不受影响，它们走专用的IntArrayProperties线上类型。

序列化与资源管理

to_dict()/from_dict()对所有组件通用：to_dict返回可序列化字典，from_dict从字典反序列化。配合 Haystack 的Pipeline.dumps()/Pipeline.loads()，整个 Weaviate 检索链路（含认证凭据、过滤器配置）都可以 YAML/JSON 形式持久化并在反序列化时完整还原。

WeaviateBM25Retriever：关键词检索

__init__( *, document_store: WeaviateDocumentStore, filters: dict[str, Any] | None = None, top_k: int = 10, filter_policy: str | FilterPolicy = FilterPolicy.REPLACE ) -> None

基于 BM25 算法的关键词检索组件。基本用法：

from haystack_integrations.document_stores.weaviate.document_store import ( WeaviateDocumentStore, ) from haystack_integrations.components.retrievers.weaviate.bm25_retriever import ( WeaviateBM25Retriever, ) document_store = WeaviateDocumentStore(url="http://localhost:8080") retriever = WeaviateBM25Retriever(document_store=document_store) retriever.run(query="How to make a pizza", top_k=3)

• document_store：必需的WeaviateDocumentStore实例。
• filters：初始化时设定的自定义过滤器。
• top_k：最多返回的文档数，默认 10。
• filter_policy：过滤器应用策略，默认FilterPolicy.REPLACE。

run(query, filters=None, top_k=None)返回{"documents": [...]}；run_async为异步版本。运行时传入的filters如何生效取决于初始化时选定的filter_policy。

WeaviateEmbeddingRetriever：向量检索

__init__( *, document_store: WeaviateDocumentStore, filters: dict[str, Any] | None = None, top_k: int = 10, distance: float | None = None, certainty: float | None = None, filter_policy: str | FilterPolicy = FilterPolicy.REPLACE ) -> None

基于查询向量与文档向量相似度检索的组件。

• distance：文档嵌入与查询向量之间允许的最大距离阈值。
• certainty：结果项与搜索向量之间的归一化距离（即"确定性"分数）。
• 重要约束：distance与certainty不能同时提供，否则抛ValueError。二者都作用于 Weaviate 的向量相似度阈值，语义互斥，请按部署与业务需求二选一。

run(query_embedding: list[float], filters=None, top_k=None, distance=None, certainty=None)返回{"documents": [...]}，同样提供run_async。注意它的输入是query_embedding（查询向量）而非查询文本，因此在使用前需要先用 Text Embedder 之类的组件将查询文本转换为向量。

WeaviateHybridRetriever：混合检索与 alpha 调参

__init__( *, document_store: WeaviateDocumentStore, filters: dict[str, Any] | None = None, top_k: int = 10, alpha: float = 0.7, max_vector_distance: float | None = None, filter_policy: str | FilterPolicy = FilterPolicy.REPLACE ) -> None

WeaviateHybridRetriever将 BM25 关键词检索与向量相似度检索并行执行，并在 Weaviate 服务端融合为一个排序结果，是"兼顾关键词召回与语义召回"的默认选择。

alpha：混合权重因子，取值范围[0.0, 1.0]，控制两种打分对最终结果的贡献：

• alpha = 0.0：仅使用关键词（BM25）打分；
• alpha = 1.0：仅使用向量相似度打分；
• 中间值按比例混合，值越大越偏向向量分，值越小越偏向 BM25 分。

默认alpha = 0.7，这也是 Weaviate 服务端的默认值。

max_vector_distance：可选阈值，在混合前将向量部分限制在最大向量距离内——距离超过该阈值的候选会从向量部分剔除。适合在保留关键词召回的同时剪除低质量向量匹配；设为None则使用 Weaviate 默认行为（不设显式截断）。

run与run_async签名一致，同时需要query（文本）与query_embedding（向量），并支持运行时覆盖top_k、alpha、max_vector_distance、filters：

run( query: str, query_embedding: list[float], filters: dict[str, Any] | None = None, top_k: int | None = None, alpha: float | None = None, max_vector_distance: float | None = None, ) -> dict[str, list[Document]]

独立使用示例：

from haystack_integrations.document_stores.weaviate.document_store import ( WeaviateDocumentStore, ) from haystack_integrations.components.retrievers.weaviate import WeaviateHybridRetriever document_store = WeaviateDocumentStore(url="http://localhost:8080") retriever = WeaviateHybridRetriever(document_store=document_store) # using a fake vector to keep the example simple retriever.run(query="How many languages are there?", query_embedding=[0.1] * 768)

filter_policy：REPLACE 与 MERGE 的底层实现

filter_policy是三个检索器共有的参数。FilterPolicy枚举定义在 haystack/document_stores/types/filter_policy.py：

• REPLACE（默认）：运行时过滤器直接替换初始化时设定的过滤器，便于针对不同查询动态调整过滤条件。
• MERGE：运行时过滤器与初始化过滤器合并，进一步收窄搜索范围；键冲突时运行时值覆盖初始化值。

其底层实现为同文件中的apply_filter_policy(filter_policy, init_filters, runtime_filters, default_logical_operator="AND")。源码逻辑（filter_policy.py）表明：仅当策略为MERGE且运行时与初始化过滤器均非空时才会执行合并；合并过程按过滤器形态（比较型ComparisonFilter与逻辑型LogicalFilter）的四种组合分别调用对应的组合函数，将两个过滤器以默认AND逻辑操作符合并成一个新的过滤器树；其余情况直接返回runtime_filters or init_filters。

在 Pipeline 中的实战：混合检索 RAG

将WeaviateHybridRetriever嵌入 Haystack Pipeline 的典型做法是：文本经 Document Embedder 向量化后写入 Weaviate，查询侧用 Text Embedder 生成查询向量，两者在 Pipeline 中按组件连接。完整示例需要sentence-transformers-haystack包：

pip install sentence-transformers-haystack

from haystack import Document, Pipeline from haystack.document_stores.types import DuplicatePolicy from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder, ) from haystack_integrations.document_stores.weaviate.document_store import ( WeaviateDocumentStore, ) from haystack_integrations.components.retrievers.weaviate import ( WeaviateHybridRetriever, ) document_store = WeaviateDocumentStore(url="http://localhost:8080") documents = [ Document(content="There are over 7,000 languages spoken around the world today."), Document( content="Elephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors." ), Document( content="In certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves." ), ] document_embedder = SentenceTransformersDocumentEmbedder() documents_with_embeddings = document_embedder.run(documents) document_store.write_documents( documents_with_embeddings.get("documents"), policy=DuplicatePolicy.OVERWRITE, ) query_pipeline = Pipeline() query_pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder()) query_pipeline.add_component("retriever", WeaviateHybridRetriever(document_store=document_store)) query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding") result = query_pipeline.run( { "text_embedder": {"text": "How many languages are there?"}, "retriever": {"query": "How many languages are there?"}, } ) print(result["retriever"]["documents"])

值得注意的是，WeaviateHybridRetriever在 Pipeline 中最常见的位置是：①RAG 管道中位于 Text Embedder 之后、PromptBuilder之前；②混合检索管道的末位组件；③抽取式问答管道中位于 Text Embedder 之后、Extractive Reader 之前。核心运行变量是query（字符串）与query_embedding（浮点列表），输出变量为documents列表。

实战注意事项与最佳实践

1. 集合 schema 先行：默认default集合不含meta属性，生产环境务必为你的元数据结构创建自定义collection_settings，避免过滤/统计能力受限。
2. 写文档用OVERWRITE：这是唯一走 batch API 的策略，吞吐性能最优；需要严格去重语义时再考虑FAIL/SKIP（代价是失去 batch 加速）。
3. distance与certainty二选一：同时传入会抛ValueError；两者分别对应 Weaviate 向量搜索的两种阈值语义。
4. 混合检索调alpha是核心杠杆：偏关键词召回调低、偏语义召回调高；max_vector_distance可在保留关键词召回的同时剪除低质量向量候选。
5. contains过滤大小写敏感：需要不敏感匹配时先归一化再构造过滤器。
6. int 元数据会变 float：标量 int 元数据经 Weaviate 取回时变为 float 是 weaviate-client 协议所致，业务侧做类型断言时需兼容。
7. 清理索引用recreate_index=True：清空集合时 drop-and-recreate 比逐批删除更快，是官方推荐路径。
8. 资源释放：Pipeline 运行完毕或应用退出前，调用 Document Store 与检索器的close()（同步场景）或close_async()（异步场景）释放底层连接资源。
9. 认证凭据走环境变量：四种AuthCredentials默认从WEAVIATE_*环境变量读取密钥，序列化时不落盘敏感信息；需要自定义变量名时显式传入Secret.from_env_var(...)。
10. 异步能力贯穿始终：Document Store 与三个检索器均提供*_async方法族，配合 Haystack 的异步 Pipeline 可在高并发场景获得更好的吞吐。

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack