Haystack 集成 Vespa：VespaDocumentStore、VespaEmbeddingRetriever 与 VespaKeywordRetriever 完整指南-酒店常州论坛

Haystack 集成 Vespa：VespaDocumentStore、VespaEmbeddingRetriever 与 VespaKeywordRetriever 完整指南

【免费下载链接】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 通过vespa-haystack集成包将开源大模型编排框架与 Vespa（一个支持结构化数据、文本与向量在规模上检索的大数据 serving 引擎）连接起来，提供VespaDocumentStore文档存储、VespaEmbeddingRetriever稠密向量检索器和VespaKeywordRetriever词法检索器三大组件。本文基于仓库中 version-2.23 的 Vespa 集成 API 参考文档，结合源码与配套文档，完整讲解每个构造参数、运行方法、认证方式与真实管线示例，帮助你快速搭建基于 Vespa 的语义搜索、关键词搜索与 RAG 应用。

集成概览：Haystack 如何对接已有 Vespa 应用

Vespa 是一个开源的“大数据 serving 引擎”，同时支持结构化数据、全文文本与向量（tensor）检索，并内建 rank profile 机制用于结果排序。Haystack 的 Vespa 集成通过 pyvespa HTTP 客户端与一个已经部署好的Vespa 应用通信，支持词法检索、稠密向量检索以及元数据过滤。

与大多数 Haystack Document Store 不同，VespaDocumentStore不会替你创建或部署 Vespa 应用与 schema。你需要自己配置 Vespa 的字段（fields）与 rank profile，将其部署到自托管环境或 Vespa Cloud，然后再把 Document Store 指向运行中的端点。整个集成由三个核心模块组成，分别对应 API 参考文档中的三个小节：

模块路径	类	职责
`haystack_integrations.document_stores.vespa.document_store`	`VespaDocumentStore`	读写文档、按 id/过滤条件查询与删除、元数据字段管理、序列化
`haystack_integrations.components.retrievers.vespa.embedding_retriever`	`VespaEmbeddingRetriever`	使用近邻搜索（nearest-neighbor search）按稠密向量检索
`haystack_integrations.components.retrievers.vespa.keyword_retriever`	`VespaKeywordRetriever`	使用 YQL`userQuery()`按词法匹配检索

对应的人类可读文档见 vespadocumentstore.mdx、vespaembeddingretriever.mdx 与 vespakeywordretriever.mdx，最新版 API 参考见 reference/integrations-api/vespa.md。

安装与前置条件

安装集成包：

pip install vespa-haystack

如需在本地运行 Vespa，参见 Vespa 官方快速开始指南；若要部署托管应用，可使用 Vespa Cloud。运行示例中如果用到 Sentence Transformers 嵌入器，还需安装：

pip install sentence-transformers-haystack

必备的 Vespa Schema

在使用VespaDocumentStore之前，需要有一个已部署且 schema 与 Document Store 配置兼容的 Vespa 应用。默认情况下，集成期望 schema 中包含：

名为content的文本字段，用于存放文档正文（Document body）；
名为embedding的tensor 字段，用于稠密向量（使用 embedding 检索时）；
名为bm25的rank profile，用于词法检索（VespaKeywordRetriever使用），典型实现为bm25(content)；
名为semantic的rank profile，使用closeness(field, embedding)对近邻候选打分（VespaEmbeddingRetriever使用）。

字段名与 rank profile 名称都可以通过 Document Store 与 Retriever 的构造参数自定义（见下文各参数说明）。编写 schema 与 rank profile 的细节参见 Vespa 官方 schema 文档。

VespaDocumentStore：连接、写入与查询

VespaDocumentStore是"由已有 Vespa 应用支撑的文档存储"，位于haystack_integrations.document_stores.vespa.document_store模块。它的 HTTP 客户端是惰性创建的，即首次使用时才建立连接。

构造参数详解

构造函数签名如下：

__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（str | None）：Vespa 端点基础 URL。省略时使用VESPA_URL环境变量。
port（int，默认8080）：Vespa HTTP 端口。
cert/key（Secret | None）：mTLS 认证用的数据面证书与私钥文件路径，以 Secret 形式传入。
vespa_cloud_secret_token（Secret | None）：Vespa Cloud 数据面 token（Bearer token 认证）。省略时，若设置了VESPA_CLOUD_SECRET_TOKEN环境变量则自动使用，这与 pyvespa 的行为一致。
additional_headers（dict[str, str] | None）：发送给 Vespa 应用的额外请求头。
content_cluster_name（str，默认"content"）：Vespa content cluster 名称。
schema（str，默认"doc"）：要读写数据的 Vespa schema 名称。
namespace（str | None）：Vespa namespace，省略时默认与 schema 名相同。
groupname（str | None）：可选的 Vespa group 名称。
content_field（str，默认"content"）：Vespa 中存放文档文本的字段名。
embedding_field（str，默认"embedding"）：Vespa 中存放稠密向量的字段名。
id_field（str，默认"id"）：查询响应中存放文档 id 的可选字段名。Vespa 文档 id 始终通过data_id写入；若 schema 或 summary 中缺少该字段，集成会回退到解析 Vespa 文档路径（document path）来获取 id。
metadata_fields（list[str] | None）：可选的元数据字段白名单，决定哪些元数据会被写入（feed）与读回。
query_limit（int，默认400）：批量查询最多返回的文档数，默认 400 以保持在 Vespa 常见查询命中数限制内，除非显式覆盖。

认证方式

VespaDocumentStore支持 pyvespa 提供的三种认证方式：

无认证：本地开发时针对未加固的 Vespa 端点。
mTLS：通过cert与key参数传入数据面证书与私钥（Secret 管理）。
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"

app 属性：底层 pyvespa 客户端

app: Any

该属性返回底层 pyvespaVespaHTTP 客户端。它由当前 store 的url、port与认证设置（cert、key、vespa_cloud_secret_token、additional_headers）构建，因此构造函数或环境变量中配置的 mTLS、Bearer token 与自定义请求头都会生效。当需要绕过 Haystack 抽象直接调用 pyvespa 的底层能力时，可以通过document_store.app访问。

序列化：to_dict

to_dict() -> dict[str, Any]

将文档存储序列化为字典。它复用__init__的参数名与 Haystack 的default_to_dict工具，保证嵌套序列化与 Haystack 默认组件序列化保持一致。这意味着VespaDocumentStore可以像其他 Haystack 组件一样被 YAML/字典管线描述引用并反序列化重建。

写入与计数

count_documents() -> int # 返回 Vespa 中文档总数 count_documents_by_filter(filters: dict[str, Any]) -> int # 返回匹配过滤条件的文档数 write_documents(documents: list[Document], policy: DuplicatePolicy = DuplicatePolicy.NONE) -> int

write_documents的policy参数使用 Haystack 的DuplicatePolicy枚举（定义见 haystack/document_stores/types/policy.py），可选值如下：

策略	行为
`DuplicatePolicy.NONE`	默认策略，具体行为取决于 Document Store 实现
`DuplicatePolicy.SKIP`	若同 id 文档已存在则跳过不写
`DuplicatePolicy.OVERWRITE`	若同 id 文档已存在则覆盖（此时返回值恒等于输入文档数）
`DuplicatePolicy.FAIL`	若同 id 文档已存在则抛出`DuplicateError`

一个完整的写入示例：

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())

删除与更新

delete_documents(document_ids: list[str]) -> None delete_all_documents() -> None delete_by_filter(filters: dict[str, Any]) -> int update_by_filter(filters: dict[str, Any], meta: dict[str, Any]) -> int

delete_documents按 id 列表删除文档。
delete_all_documents删除该 store 的 schema、namespace 与 content cluster 下的全部文档，底层通过 pyvespaVespa.delete_all_docs（Document V1 批量删除）实现。
delete_by_filter删除所有匹配过滤条件的文档并返回删除数量。
update_by_filter将meta中的元数据值合并（merge）进匹配过滤条件的文档，返回更新数量。

查询与元数据字段信息

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

get_documents_by_id按 id 获取文档。
filter_documents获取匹配过滤条件的文档，filters=None时返回全部。
get_metadata_fields_info基于配置的字段，尽力返回元数据字段信息，供上层组件（如评估、展示）使用。

元数据字段白名单：metadata_fields

Vespa 是强 schema 约束的：任何要写入或读回的元数据字段，都必须存在于已部署的 schema 中。因此需要用metadata_fields声明一个白名单，列出要发给 Vespa 写入、以及读取时请求返回的元数据键。不在白名单中的元数据键会保留在内存中的 Document 对象上，但不会被存储到 Vespa。

元数据过滤：Filters 到 YQL 的翻译

VespaDocumentStore支持比较运算符==、!=、>、>=、<、<=、in、not in，以及逻辑运算符AND、OR、NOT。过滤器会尽量被翻译为 Vespa 的 YQL。

VespaEmbeddingRetriever：稠密向量语义检索

VespaEmbeddingRetriever位于haystack_integrations.components.retrievers.vespa.embedding_retriever，使用 Vespa 的 nearest-neighbor search 找到与查询向量最接近的文档，并通过可配置的 rank profile 打分。

构造参数详解

__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，例如VespaDocumentStore(url="http://localhost", schema="doc", namespace="doc")，需要与你部署的 Vespa schema 对齐。若传入的不是VespaDocumentStore实例，构造函数抛出ValueError。
filters（dict[str, Any] | None）：可选的静态 Haystack 元数据过滤器，例如{"field": "meta.category", "operator": "==", "value": "news"}；运行时可通过run的filters参数覆盖。语法约定同 Haystack 元数据过滤与 Vespa 查询语言。
top_k（int，默认10）：每次查询默认返回的最大文档数。
ranking（str | None）：近邻检索后使用的 Vespa rank profile，例如semantic（用closeness(field, embedding)打分的 profile）。默认为semantic；传None则使用 schema 默认 profile。
query_tensor_name（str，默认"query_embedding"）：YQL 中以及 rank profile 里input.query(...)使用的查询 tensor 名称。例如query_embedding与默认semanticprofile 匹配。
target_hits（int | None）：可选的近邻targetHits值，例如10或100，表示在第一阶段排序（first-phase ranking）之前，每个 content node 考虑多少个近邻。增大该值可以提升召回但增加计算开销。

run 方法

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

query_embedding（list[float]）：查询的稠密向量。
filters（dict[str, Any] | None）：抓取文档时应用的过滤器。
top_k（int | None）：返回的最大文档数。

返回dict[str, list[Document]]，即{"documents": [...]}。

单独使用

该 Retriever 需要VespaDocumentStore以及已索引的文档。设置VESPA_URL环境变量（或在 Document Store 中传url=...）即可连接：

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) ## 使用假向量简化示例 retriever.run(query_embedding=[0.1] * 768)

在语义搜索管线中使用

在管线中使用时，需要同时有查询向量与文档向量。索引管线中加入 Document Embedder，查询管线中加入 Text Embedder：

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") query = "semantic vector search" result = query_pipeline.run({"text_embedder": {"text": query}}) print(result["retriever"]["documents"][0])

VespaKeywordRetriever：词法关键词检索

VespaKeywordRetriever位于haystack_integrations.components.retrievers.vespa.keyword_retriever，向 Vespa 应用发送 YQLuserQuery()查询，并用可配置的 rank profile（默认bm25，通常使用 Vespa 的 BM25 ranking feature）对结果排序。

构造参数详解

__init__( *, document_store: VespaDocumentStore, filters: dict[str, Any] | None = None, top_k: int = 10, ranking: str | None = DEFAULT_BM25_RANKING ) -> None

document_store（VespaDocumentStore）：配置好的VespaDocumentStore，例如VespaDocumentStore(url="http://localhost", schema="doc", namespace="doc")，需与已部署 schema 与端点匹配。非VespaDocumentStore实例会抛ValueError。
filters（dict[str, Any] | None）：可选的静态 Haystack 元数据过滤器，例如{"field": "meta.category", "operator": "==", "value": "news"}；运行时可通过run覆盖。
top_k（int，默认10）：每次查询默认返回的最大文档数。
ranking（str | None）：词法匹配使用的 Vespa rank profile，例如bm25（使用bm25(content)）。默认bm25；传None使用 schema 默认 profile。

该 Retriever 期望底层 Vespa 应用提供：

存放文档正文的文本字段（默认content，可通过 Document Store 的content_field配置），该字段需要在 Vespa schema 中为文本匹配建立索引；
对词法匹配打分的rank profile（默认bm25，可通过ranking配置）。

run 方法

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

query（str）：查询文本。
filters（dict[str, Any] | None）：抓取文档时应用的过滤器。
top_k（int | None）：返回的最大文档数。

返回dict[str, list[Document]]。

单独使用

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")

在 RAG 管线中使用

运行以下代码需要满足：设置OPENAI_API_KEY环境变量；设置VESPA_URL（或给 Document Store 传url=...）；已部署包含content文本字段、category元数据字段与bm25rank 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, ) ## 创建 RAG 查询管线 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"])

实战配置建议与注意事项

综合 API 参考文档、组件文档与源码行为，以下几个要点直接影响检索效果与可维护性：

schema 先行，配置对齐：schema、namespace、content_field、embedding_field、metadata_fields必须与已部署的 Vespa schema 一一对应。字段不匹配是连接成功后最常见的报错来源。
rank profile 与 tensor 名称联动：VespaEmbeddingRetriever的query_tensor_name必须与 rank profile 中input.query(...)引用的名字一致（默认query_embedding），否则近邻打分无法生效；ranking=None可用于回退到 schema 默认 profile。
target_hits 是召回与开销的权衡旋钮：target_hits决定每个 content node 在 first-phase ranking 前考虑的近邻数量，增大它通常提升召回率，但会带来更高的计算与内存开销；文档参考给出10或100这类量级作为起点。
query_limit 与批量操作：query_limit默认 400 是为了留在 Vespa 常见查询命中数限制内；批量删除（delete_all_documents）基于 Document V1 批量删除实现，删除范围受 schema、namespace 与 content cluster 约束。
过滤的客户端求值边界：日期类过滤在 YQL 无法直接表达时于 Python 端求值，因此涉及大量日期过滤的查询会有一定客户端开销，建议在 Vespa schema 侧尽量用可索引字段表达过滤条件。
认证三选一：本地无认证、数据面 mTLS（cert/key）、Vespa Cloud Bearer token（vespa_cloud_secret_token或VESPA_CLOUD_SECRET_TOKEN环境变量）。HTTP 客户端是惰性构建的，认证配置错误会在首次实际请求时才暴露，建议初始化后立即执行一次count_documents()做连通性验证。

总结

本文围绕 Vespa 集成 API 参考文档完整梳理了vespa-haystack的三个核心组件：负责连接、写入、删除、更新与过滤查询的VespaDocumentStore；基于 nearest-neighbor 与可配置 rank profile 的VespaEmbeddingRetriever；以及基于 YQLuserQuery()与 BM25 的VespaKeywordRetriever。配套的用户文档（VespaDocumentStore、VespaEmbeddingRetriever、VespaKeywordRetriever）提供了完整的可运行示例，而本仓库中的 DuplicatePolicy 定义与元数据过滤规范则为写入策略与过滤行为提供了底层依据。你可以在此基础上，把 Vespa 作为大规模生产环境下的向量与文本双模检索后端，构建语义搜索、混合检索与 RAG 应用。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析