Haystack × Vespa 集成实战指南:VespaDocumentStore 与语义/关键词双检索器全解析
Haystack × Vespa 集成实战指南:VespaDocumentStore 与语义/关键词双检索器全解析
本篇技术指南围绕 Haystack 的 Vespa 官方集成展开,系统讲解 VespaDocumentStore 文档存储以及 VespaEmbeddingRetriever(稠密向量检索)与 VespaKeywordRetriever(词法检索)两个检索组件的完整 API、配置参数与底层原理。读完本文,你将掌握如何把 Haystack 的 RAG、语义搜索与问答流水线对接到一个已部署的 Vespa 应用上,包括 schema 前置约束、认证方式、元数据过滤、重复文档策略以及端到端 Pipeline 的搭建方法。
一、Vespa 集成概览:连接已有 Vespa 应用而非代管
Vespa 是一个开源的大数据 serving 引擎,在同一套集群上同时支持结构化检索、文本检索与向量检索。Haystack 通过 vespa-haystack 集成包,借助 pyvespa 这一官方 HTTP 客户端将 Haystack 组件连接到运行中的 Vespa 应用。
与 Haystack 中大多数文档存储不同,VespaDocumentStore 不会替你创建或部署 Vespa 应用与 schema。你需要自己按需配置好字段与 rank profile,完成部署(自托管或 Vespa Cloud),再把文档存储指向这个运行中的端点。这种设计让集成层保持轻量,Vespa 侧的全部索引、分片、布隆过滤与排序逻辑都由你自己的应用配置决定。
集成的核心组件分布在三个模块中,完整签名见仓库 API 参考 vespa.md:
| 模块 | 组件 | 能力 |
|---|---|---|
haystack_integrations.document_stores.vespa.document_store |
VespaDocumentStore |
文档读写、计数、删除、更新、过滤、序列化 |
haystack_integrations.components.retrievers.vespa.embedding_retriever |
VespaEmbeddingRetriever |
基于最近邻的稠密向量检索 |
haystack_integrations.components.retrievers.vespa.keyword_retriever |
VespaKeywordRetriever |
基于 YQL userQuery() 的词法检索 |
haystack_integrations.document_stores.vespa.filters |
过滤器工具 | 将 Haystack 过滤语法转换为 Vespa YQL 查询条件 |
说明:当前仓库的 version-3.1 参考文档 与本文所依的 2.22 版本 API 完全一致,说明该集成自 2.22 起 API 保持稳定,本文内容对后续版本同样适用。
二、安装与前置条件
2.1 安装集成包
pip install vespa-haystack
若需要跑本文的向量检索示例,还需安装对应的 embedding 组件(以下使用 Sentence Transformers 系列):
pip install sentence-transformers-haystack
本地运行 Vespa 可参考 Vespa 官方 quick start;需要托管式服务则使用 Vespa Cloud。
2.2 Schema 前置约束
使用任何 Vespa 组件之前,你部署的 Vespa 应用必须有一个与文档存储配置兼容的 schema。集成默认期望以下字段与 rank profile 存在:
- 文本字段
content:存放 Document 正文(对应文档存储的content_field); - 张量字段
embedding:存放稠密向量,供向量检索使用(对应embedding_field); - rank profile
bm25:词法检索使用,通常以bm25(content)打分(VespaKeywordRetriever默认使用); - rank profile
semantic:以closeness(field, embedding)打分,并接收名为query_embedding的 query input 张量(VespaEmbeddingRetriever默认使用)。
字段名与 rank profile 名都可以通过文档存储与检索器的构造函数自定义。文本字段需要在 schema 中按文本匹配索引(indexed for text matching),向量字段则需按张量字段声明。
三、VespaDocumentStore 深度解析
VespaDocumentStore 的定位是"由已有 Vespa 应用支撑的文档存储",其全部初始化参数、方法与序列化行为在参考文档中都有完整定义。
3.1 构造函数与全部参数
__init__(
*,
url: str | None = None,
port: int = 8080,
cert: Secret | None = None,
key: Secret | None = None,
vespa_cloud_secret_token: Secret | None = None,
additional_headers: dict[str, str] | None = None,
content_cluster_name: str = "content",
schema: str = "doc",
namespace: str | None = None,
groupname: str | None = None,
content_field: str = "content",
embedding_field: str = "embedding",
id_field: str = "id",
metadata_fields: list[str] | None = None,
query_limit: int = DEFAULT_QUERY_LIMIT
) -> None
各参数语义如下:
| 参数 | 默认值 | 说明 |
|---|---|---|
url |
None |
Vespa 端点基础 URL;省略时读取 VESPA_URL 环境变量 |
port |
8080 |
Vespa HTTP 端口 |
cert |
None |
mTLS 认证用的数据面证书文件路径(Secret) |
key |
None |
mTLS 认证用的数据面密钥文件路径(Secret) |
vespa_cloud_secret_token |
None |
Vespa Cloud 数据面 token 认证;省略时若设置了 VESPA_CLOUD_SECRET_TOKEN 环境变量则使用之(与 pyvespa 行为一致) |
additional_headers |
None |
发送给 Vespa 应用的额外 HTTP 头 |
content_cluster_name |
"content" |
Vespa content cluster 名称 |
schema |
"doc" |
读写的 Vespa schema 名称 |
namespace |
None |
Vespa namespace,省略时默认取 schema 名称 |
groupname |
None |
可选的 Vespa group 名称 |
content_field |
"content" |
存放文档正文的 Vespa 字段 |
embedding_field |
"embedding" |
存放稠密向量的 Vespa 字段 |
id_field |
"id" |
查询响应中承载文档 id 的字段。Vespa 文档 id 始终通过 data_id 写入;若 schema 或 summary 中缺少该字段,集成会回退到解析 Vespa 文档路径 |
metadata_fields |
None |
元数据字段 allowlist,声明要写入并从 Vespa 读回的元数据键 |
query_limit |
400 |
批量查询最多返回的文档数。默认 400 以避免超出 Vespa 常见的查询 hit 上限,除非显式覆盖 |
Vespa 是严格 schema 约束的:任何你要写入或读回的元数据字段都必须已存在于部署的 schema 中。metadata_fields 就是这层约束的显式声明——不在 allowlist 中的元数据键只保留在内存里的 Document 对象上,不会被写入 Vespa。
3.2 app 属性:底层 pyvespa 客户端
app: Any
app 返回底层 pyvespa 的 Vespa HTTP 客户端。它由当前 store 的 url、port 以及认证设置(cert、key、vespa_cloud_secret_token、additional_headers)构建,因此构造函数(或环境变量)中的 mTLS、bearer token 与自定义请求头都会被应用到该客户端。参考文档明确说明 HTTP 客户端是首次使用时惰性创建的(见 vespadocumentstore.mdx),因此初始化文档存储本身不会触发网络连接。
3.3 写入与计数
write_documents(documents: list[Document], policy: DuplicatePolicy = DuplicatePolicy.NONE) -> int
写入文档到 Vespa,policy 指定重复文档处理策略(来自 haystack.document_stores.types.DuplicatePolicy),返回实际写入的文档数。配套的计数方法有:
count_documents() -> int:返回 Vespa 中的文档总数;count_documents_by_filter(filters: dict[str, Any]) -> int:返回命中给定 Haystack 元数据过滤器的文档数。
3.4 删除与更新
delete_documents(document_ids: list[str]) -> None:按 id 删除文档;delete_all_documents() -> None:删除当前 store 的 schema、namespace 与 content cluster 下的全部文档,底层通过 pyvespaVespa.delete_all_docs(Document V1 批量删除)实现;delete_by_filter(filters: dict[str, Any]) -> int:删除所有匹配过滤条件的文档,返回删除数量;update_by_filter(filters: dict[str, Any], meta: dict[str, Any]) -> int:将meta中的元数据值合并(merge)进匹配过滤条件的文档,返回更新的文档数。
3.5 查询与过滤
get_documents_by_id(document_ids: list[str]) -> list[Document]
filter_documents(filters: dict[str, Any] | None = None) -> list[Document]
get_documents_by_id 按 id 取回文档;filter_documents 按过滤器取回匹配文档。此外还有 get_metadata_fields_info() -> dict[str, dict[str, str]],基于已配置字段返回尽力而为(best-effort)的元数据字段信息。
3.6 序列化
to_dict() -> dict[str, Any]
to_dict 使用与 __init__ 相同的参数名以及 default_to_dict,保证嵌套序列化与 Haystack 默认组件序列化保持对齐,这意味着 VespaDocumentStore 可以无缝嵌入 Haystack 的 YAML Pipeline 序列化体系中(参见仓库的 marshal/yaml.py 与 marshal/protocol.py)。
3.7 认证与端点配置
文档存储支持的认证方式完全继承自 pyvespa:
- 无认证:面向本地开发、未加防护的 Vespa 端点;
- mTLS:通过
cert与key参数传入数据面证书与密钥(两者均以Secret形式传入,参见 secret-management.mdx); - Bearer token:面向 Vespa Cloud token 端点,通过
vespa_cloud_secret_token参数或VESPA_CLOUD_SECRET_TOKEN环境变量提供。
端点 URL 可以显式传 url 参数,或通过环境变量配置:
export VESPA_URL="http://localhost"
Vespa Cloud token 认证的典型配置:
export VESPA_URL="https://my-app.my-tenant.aws-us-east-1c.z.vespa-app.cloud"
export VESPA_CLOUD_SECRET_TOKEN="my-secret-token"
3.8 基础写入示例
from haystack import Document
from haystack_integrations.document_stores.vespa import VespaDocumentStore
document_store = VespaDocumentStore(
url="http://localhost",
schema="doc",
namespace="doc",
content_field="content",
embedding_field="embedding",
metadata_fields=["category"],
)
document_store.write_documents(
[
Document(
content="Haystack integrates with Vespa for search.",
meta={"category": "docs"},
),
Document(
content="Vespa supports lexical and vector retrieval.",
meta={"category": "docs"},
),
],
)
print(document_store.count_documents())
四、元数据过滤:从 Haystack 语法到 Vespa YQL
VespaDocumentStore 支持比较运算符 ==、!=、>、>=、<、<=、in、not in,以及逻辑运算符 AND、OR、NOT。过滤器在可行时会被翻译为 Vespa 的 YQL where 子句;当 YQL 无法直接表达比较(例如日期类型值的过滤)时,集成会在 Python 端客户端侧求值。过滤器的完整语法约定见仓库的 metadata-filtering.mdx。
单个比较过滤器示例:
filters = {"field": "meta.category", "operator": "==", "value": "news"}
嵌套逻辑过滤器示例(AND 组合多个比较条件、OR 组合可选条件):
filters = {
"operator": "AND",
"conditions": [
{"field": "meta.type", "operator": "==", "value": "article"},
{"field": "meta.date", "operator": ">=", "value": 1420066800},
{"field": "meta.date", "operator": "<", "value": 1609455600},
{"field": "meta.rating", "operator": ">=", "value": 3},
{
"operator": "OR",
"conditions": [
{"field": "meta.genre", "operator": "in", "value": ["economy", "politics"]},
{"field": "meta.publisher", "operator": "==", "value": "nytimes"},
],
},
],
}
过滤器既可以静态配置在检索器构造函数上(每次检索默认生效),也可以在 run() 调用时动态覆盖,还可以在 Pipeline.run() 中通过 retriever 输入直接传入。
五、VespaEmbeddingRetriever:稠密向量检索
VespaEmbeddingRetriever 使用 Vespa 的最近邻(nearest-neighbor)搜索,找出与查询向量最接近的文档 embedding,并用可配置的 rank profile 打分。
5.1 构造函数
__init__(
*,
document_store: VespaDocumentStore,
filters: dict[str, Any] | None = None,
top_k: int = 10,
ranking: str | None = DEFAULT_SEMANTIC_RANKING,
query_tensor_name: str = "query_embedding",
target_hits: int | None = None
) -> None
| 参数 | 默认值 | 说明 |
|---|---|---|
document_store |
(必填) | 配置好的 VespaDocumentStore,例如 VespaDocumentStore(url="http://localhost", schema="doc", namespace="doc"),须与你的 Vespa schema 对齐 |
filters |
None |
可选的静态 Haystack 元数据过滤器,除非在 run() 中被覆盖 |
top_k |
10 |
每次查询默认返回的最大文档数 |
ranking |
"semantic" |
最近邻检索后使用的 Vespa rank profile。semantic 通常以 closeness(field, embedding) 打分;传 None 则使用 schema 默认 profile |
query_tensor_name |
"query_embedding" |
YQL 与 rank profile 中 input.query(...) 使用的查询张量名,须与 semantic profile 中声明的输入张量一致 |
target_hits |
None |
可选的最近邻 targetHits 值(如 10 或 100):决定在进入 first-phase ranking 前每个 content 节点考虑多少个近邻,用于调节召回与性能的平衡 |
异常:若 document_store 不是 VespaDocumentStore 实例,抛出 ValueError。
5.2 run 方法
run(
query_embedding: list[float],
filters: dict[str, Any] | None = None,
top_k: int | None = None,
) -> dict[str, list[Document]]
query_embedding:查询的稠密向量(float 列表),为必填;filters:取文档时应用的过滤器,覆盖构造时的静态配置;top_k:返回的最大文档数,覆盖构造时的默认值。
返回 {"documents": [...]} 结构。在 Pipeline 中使用时,请确保查询侧有 Text Embedder 提供 query_embedding(例如 SentenceTransformersTextEmbedder),索引侧有 Document Embedder(例如 SentenceTransformersDocumentEmbedder)负责生成文档向量。
5.3 独立使用示例
from haystack_integrations.document_stores.vespa import VespaDocumentStore
from haystack_integrations.components.retrievers.vespa import VespaEmbeddingRetriever
document_store = VespaDocumentStore(schema="doc", namespace="doc")
retriever = VespaEmbeddingRetriever(document_store=document_store)
# 使用假向量保持示例简单(768 维)
retriever.run(query_embedding=[0.1] * 768)
5.4 在 Pipeline 中使用(语义搜索 / RAG 前置检索)
from haystack import Document, Pipeline
from haystack_integrations.components.embedders.sentence_transformers import (
SentenceTransformersDocumentEmbedder,
SentenceTransformersTextEmbedder,
)
from haystack.components.writers import DocumentWriter
from haystack_integrations.document_stores.vespa import VespaDocumentStore
from haystack_integrations.components.retrievers.vespa import VespaEmbeddingRetriever
document_store = VespaDocumentStore(
schema="doc",
namespace="doc",
content_field="content",
embedding_field="embedding",
metadata_fields=["category"],
)
documents = [
Document(content="Haystack integrates with Vespa for search.", meta={"category": "docs"}),
Document(content="Vespa supports lexical and vector retrieval.", meta={"category": "docs"}),
Document(content="Cats sleep most of the day.", meta={"category": "animals"}),
]
indexing = Pipeline()
indexing.add_component("embedder", SentenceTransformersDocumentEmbedder())
indexing.add_component("writer", DocumentWriter(document_store=document_store))
indexing.connect("embedder", "writer")
indexing.run({"embedder": {"documents": documents}})
query_pipeline = Pipeline()
query_pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder())
query_pipeline.add_component(
"retriever",
VespaEmbeddingRetriever(
document_store=document_store,
top_k=2,
query_tensor_name="query_embedding",
),
)
query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding")
result = query_pipeline.run({"text_embedder": {"text": "semantic vector search"}})
print(result["retriever"]["documents"][0])
六、VespaKeywordRetriever:词法检索
VespaKeywordRetriever 对 Vespa 应用执行 YQL userQuery(),并用可配置的 rank profile(默认 bm25)对结果排序。它要求底层 Vespa 应用暴露:按文本匹配索引的正文文本字段(默认 content,可通过 content_field 配置),以及词法匹配的 rank profile(默认 bm25)。
6.1 构造函数
__init__(
*,
document_store: VespaDocumentStore,
filters: dict[str, Any] | None = None,
top_k: int = 10,
ranking: str | None = DEFAULT_BM25_RANKING
) -> None
| 参数 | 默认值 | 说明 |
|---|---|---|
document_store |
(必填) | 配置好的 VespaDocumentStore,须与已部署的 schema 和端点匹配 |
filters |
None |
可选的静态 Haystack 元数据过滤器,除非在 run() 中覆盖 |
top_k |
10 |
每次查询默认返回的最大文档数 |
ranking |
"bm25" |
词法匹配使用的 Vespa rank profile,bm25 通常以 bm25(content) 打分;传 None 使用 schema 默认 profile |
异常:若 document_store 不是 VespaDocumentStore 实例,抛出 ValueError。
6.2 run 方法
run(
query: str, filters: dict[str, Any] | None = None, top_k: int | None = None
) -> dict[str, list[Document]]
query:查询文本(必填);filters:取文档时应用的过滤器;top_k:返回的最大文档数。
返回 {"documents": [...]} 结构。
6.3 独立使用示例
from haystack_integrations.document_stores.vespa import VespaDocumentStore
from haystack_integrations.components.retrievers.vespa import VespaKeywordRetriever
document_store = VespaDocumentStore(schema="doc", namespace="doc")
retriever = VespaKeywordRetriever(document_store=document_store)
retriever.run(query="my nice query")
6.4 在 RAG Pipeline 中使用
以下端到端示例来自 vespakeywordretriever.mdx,运行前需满足:设置 OPENAI_API_KEY、设置 VESPA_URL(或传 url 参数)、部署含 content 文本字段、category 元数据字段与 bm25 rank profile 的 Vespa schema。
from haystack import Document, Pipeline
from haystack.components.builders.answer_builder import AnswerBuilder
from haystack.components.builders.chat_prompt_builder import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.document_stores.types import DuplicatePolicy
from haystack_integrations.document_stores.vespa import VespaDocumentStore
from haystack_integrations.components.retrievers.vespa import VespaKeywordRetriever
prompt_template = [
ChatMessage.from_system("You are a helpful assistant."),
ChatMessage.from_user(
"Given these documents, answer the question.\nDocuments:\n"
"{% for doc in documents %}{{ doc.content }}{% endfor %}\n"
"Question: {{question}}\nAnswer:",
),
]
document_store = VespaDocumentStore(
schema="doc",
namespace="doc",
content_field="content",
metadata_fields=["category"],
)
documents = [
Document(content="Haystack integrates with Vespa for search.", meta={"category": "docs"}),
Document(content="Vespa supports lexical and vector retrieval.", meta={"category": "docs"}),
Document(content="This note is about something else entirely.", meta={"category": "misc"}),
]
document_store.write_documents(documents=documents, policy=DuplicatePolicy.OVERWRITE)
retriever = VespaKeywordRetriever(
document_store=document_store,
filters={"field": "meta.category", "operator": "==", "value": "docs"},
)
rag_pipeline = Pipeline()
rag_pipeline.add_component(name="retriever", instance=retriever)
rag_pipeline.add_component(
instance=ChatPromptBuilder(
template=prompt_template,
required_variables={"question", "documents"},
),
name="prompt_builder",
)
rag_pipeline.add_component(instance=OpenAIChatGenerator(), name="llm")
rag_pipeline.add_component(instance=AnswerBuilder(), name="answer_builder")
rag_pipeline.connect("retriever", "prompt_builder.documents")
rag_pipeline.connect("prompt_builder.prompt", "llm.messages")
rag_pipeline.connect("llm.replies", "answer_builder.replies")
rag_pipeline.connect("retriever", "answer_builder.documents")
question = "How does Haystack work with Vespa?"
result = rag_pipeline.run(
{
"retriever": {"query": question},
"prompt_builder": {"question": question},
"answer_builder": {"query": question},
},
)
print(result["answer_builder"])
注意这里 DuplicatePolicy.OVERWRITE 与文档存储默认的 DuplicatePolicy.NONE 的区别:前者在文档 id 冲突时覆盖写入,后者默认不处理重复。该策略枚举定义在 document_store.py 中(haystack.document_stores.types)。
七、混合检索:语义与词法协同
由于 VespaEmbeddingRetriever 与 VespaKeywordRetriever 都接收同一个 VespaDocumentStore,你可以很容易地在一条查询 Pipeline 中并行接入两个检索器,再用 Haystack 的 JoinDocuments 组件做结果融合,实现典型的混合检索(hybrid search)——Vespa 自身的 schema 同时支持文本与张量字段,这正好是其核心优势场景:
from haystack import Pipeline
from haystack.components.joiners import DocumentJoiner
from haystack_integrations.document_stores.vespa import VespaDocumentStore
from haystack_integrations.components.retrievers.vespa import (
VespaEmbeddingRetriever,
VespaKeywordRetriever,
)
document_store = VespaDocumentStore(
schema="doc",
namespace="doc",
content_field="content",
embedding_field="embedding",
metadata_fields=["category"],
)
hybrid_pipeline = Pipeline()
hybrid_pipeline.add_component(
"text_embedder", SentenceTransformersTextEmbedder()
)
hybrid_pipeline.add_component(
"semantic", VespaEmbeddingRetriever(document_store=document_store, top_k=5)
)
hybrid_pipeline.add_component(
"lexical", VespaKeywordRetriever(document_store=document_store, top_k=5)
)
hybrid_pipeline.add_component("join", DocumentJoiner())
hybrid_pipeline.connect("text_embedder.embedding", "semantic.query_embedding")
hybrid_pipeline.connect("semantic.documents", "join.documents")
hybrid_pipeline.connect("lexical.documents", "join.documents")
提示:上面的代码示意了组合方式,实际接入时需根据所用 Embedder 的具体类名与版本调整导入路径;两个检索器的
documents输出均为list[Document],可直接作为DocumentJoiner的输入。
八、使用要点与最佳实践
- schema 先行:Vespa 侧的一切(字段、rank profile、索引方式)都必须在你部署的应用中预先定义好,
VespaDocumentStore只负责连接与读写。字段名不一致通常不会在初始化时报错,而是在查询或写入时暴露问题,建议先核对 vespadocumentstore.mdx 的字段约定。 metadata_fields是硬约束:不在 allowlist 中的元数据不会被持久化到 Vespa。写入前先确认元数据键与部署 schema 的字段一一对应。target_hits是性能开关:它控制每个 content 节点在 first-phase ranking 前考察的近邻数量,直接影响召回率与查询延迟的权衡;值越大召回越全但越慢。query_limit默认 400:批量查询(如count_documents、filter_documents)默认最多返回 400 条,超出时需显式调大该参数。ranking=None降级:若不想为检索器单独建 rank profile,传None即可回退到 schema 默认 profile,便于快速验证。- 认证优先级:
url/vespa_cloud_secret_token构造函数参数优先于环境变量,环境变量仅作为省略时的后备;mTLS 与 token 认证都可与additional_headers叠加使用。 - 类型校验:两个检索器在初始化时都会校验
document_store类型,非VespaDocumentStore实例会抛出ValueError。
九、进一步阅读(仓库内资料)
- VespaDocumentStore 用户指南:安装、schema 前置条件、认证与初始化示例
- VespaEmbeddingRetriever 组件指南:稠密检索的 Pipeline 用法与字段要求
- VespaKeywordRetriever 组件指南:词法检索与完整 RAG 示例
- Vespa API 参考(当前版):完整类与方法签名
- 元数据过滤概念:过滤语法、比较运算符与嵌套逻辑条件
- Secret 管理概念:
cert/key/token 等凭据的推荐管理方式 - Haystack 默认序列化:理解
to_dict与 Pipeline 序列化的衔接机制