Haystack 集成 Valkey 指南：基于 HNSW 向量检索的 ValkeyDocumentStore 与 ValkeyEmbeddingRetriever 完整实战-平芜编程栈

Haystack 集成 Valkey 指南：基于 HNSW 向量检索的 ValkeyDocumentStore 与 ValkeyEmbeddingRetriever 完整实战

【免费下载链接】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

Valkey 是一个高性能、内存优先的数据结构存储服务（Redis 的社区分支），Haystack 通过valkey-haystack集成包为其提供文档存储与向量检索能力。本文以仓库内 version-2.18 的 Valkey 集成 API 参考为主体骨架，结合当前仓库的 ValkeyDocumentStore 指南、ValkeyEmbeddingRetriever 指南与 Haystack 核心源码，系统讲解其安装初始化、索引写入、向量检索、元数据过滤、异步操作与序列化等完整用法。读完本文，你将能够在 Haystack Pipeline 中独立搭建一套基于 Valkey 的 RAG 或语义检索应用，并理解底层 HNSW 索引、过滤策略与距离度量的工作原理。

一、集成概览：Valkey 在 Haystack 生态中的定位

从当前仓库的 Choosing a Document Store 集成总表可以看到，Valkey 被归类为In-memory Key-Value Store（内存键值存储），其引擎类型为“内存键值存储（Redis 分支），通过glide客户端提供 HNSW 向量检索”，支持异步操作与 Embedding 检索器，且为开源核心集成。

这一类存储的典型特征与适用场景是：

内存架构带来极低的读写延迟，适合需要亚毫秒级响应的实时检索场景（如聊天机器人）；
向量检索（HNSW）构建在已有的缓存/会话存储基础设施之上，适合技术栈中已经引入 Valkey 作为缓存层的团队；
数据默认是临时的（ephemeral），持久化需要显式配置；
不适合超大规模语料——当语料体量很大、内存成本显著上升时，需要评估其他方案。

在 Haystack 中，ValkeyDocumentStore负责连接运行着Valkey Search 模块的 Valkey 服务端，并提供向量相似度检索能力；ValkeyEmbeddingRetriever则作为检索组件在 Pipeline 中消费这个文档存储。二者共同构成 RAG、语义搜索等场景的存取闭环。

核心能力清单

根据 API 参考文档，ValkeyEmbeddingRetriever与ValkeyDocumentStore提供的关键特性如下：

层面	能力
检索算法	基于 HNSW 算法的向量相似度搜索
元数据过滤	支持 Tag 字段（字符串精确匹配）与数值字段（范围比较）
距离度量	可配置 L2、余弦（cosine）、内积（ip）三种
批量操作	批量写入、批量删除，面向高吞吐文档管理
同步/异步	全部核心操作均提供同步与异步两个版本
部署模式	支持单机（standalone）与集群（cluster）两种连接模式
序列化	`to_dict`/`from_dict`，可与 Haystack Pipeline YAML/JSON 序列化体系对接

二、安装与本地启动

安装集成包

ValkeyDocumentStore与ValkeyEmbeddingRetriever位于独立的valkey-haystack集成包中（参考 ValkeyEmbeddingRetriever 指南中的包名说明）：

pip install valkey-haystack

指南中的示例大量使用 Sentence Transformers 嵌入模型，而相关 embedder 已迁移至独立的sentence-transformers-haystack包，运行示例前需一并安装：

pip install sentence-transformers-haystack

本地快速启动 Valkey 服务端

Valkey 需要运行Search 模块才能提供向量检索能力。开发与测试阶段，可以用 Docker 一行命令启动官方 bundle 镜像：

docker run -d -p 6379:6379 valkey/valkey-bundle:latest

启动后即可通过nodes_list=[("localhost", 6379)]连接。更高级的配置与集群搭建，请参考 Valkey 官方文档。

三、ValkeyDocumentStore：初始化与全部构造参数

基本初始化

from haystack_integrations.document_stores.valkey import ValkeyDocumentStore document_store = ValkeyDocumentStore( nodes_list=[("localhost", 6379)], index_name="my_documents", embedding_dim=768, distance_metric="cosine", )

构造参数全解析

根据 API 参考中的__init__签名，完整参数如下：

__init__( nodes_list: list[tuple[str, int]] | None = None, *, cluster_mode: bool = False, use_tls: bool = False, username: Secret | None = Secret.from_env_var("VALKEY_USERNAME", strict=False), password: Secret | None = Secret.from_env_var("VALKEY_PASSWORD", strict=False), request_timeout: int = 500, retry_attempts: int = 3, retry_base_delay_ms: int = 1000, retry_exponent_base: int = 2, batch_size: int = 100, index_name: str = "default", distance_metric: Literal["l2", "cosine", "ip"] = "cosine", embedding_dim: int = 768, metadata_fields: dict[str, type[str] | type[int]] | None = None ) -> None

逐项说明：

参数	类型	默认值	说明
`nodes_list`	`list[tuple[str, int]] \\| None`	`[("localhost", 6379)]`	Valkey 节点 (host, port) 列表
`cluster_mode`	`bool`	`False`	是否以集群模式连接
`use_tls`	`bool`	`False`	连接是否启用 TLS
`username`	`Secret \\| None`	读自`VALKEY_USERNAME`环境变量	认证用户名
`password`	`Secret \\| None`	读自`VALKEY_PASSWORD`环境变量	认证密码
`request_timeout`	`int`	`500`	请求超时（毫秒）
`retry_attempts`	`int`	`3`	失败操作的重试次数
`retry_base_delay_ms`	`int`	`1000`	指数退避的基准延迟（毫秒）
`retry_exponent_base`	`int`	`2`	指数退避计算的指数底数
`batch_size`	`int`	`100`	异步操作的批量处理文档数
`index_name`	`str`	`"default"`	搜索索引名称
`distance_metric`	`"l2" \\| "cosine" \\| "ip"`	`"cosine"`	向量相似度距离度量
`embedding_dim`	`int`	`768`	文档嵌入向量维度
`metadata_fields`	`dict[str, type[str] \\| type[int]] \\| None`	`None`	可过滤元数据字段映射，如`{"category": str, "priority": int}`

几点值得强调：

凭据安全：用户名与密码使用 Haystack 的Secret类型封装，默认从VALKEY_USERNAME/VALKEY_PASSWORD环境变量读取，避免硬编码敏感信息；也可在代码中显式传入。
元数据字段声明：metadata_fields决定哪些 meta 字段会被建立索引用于过滤。支持的类型为str（精确匹配，索引为 keyword）与int（数值比较，索引为 long）。如果传入None，则不会有任何元数据字段被索引用于过滤。
维度一致性：embedding_dim必须与嵌入模型实际输出的向量维度一致，否则检索时会出现维度不匹配错误。
重试策略：retry_attempts、retry_base_delay_ms、retry_exponent_base三个参数共同构成指数退避重试机制，提升对瞬时故障的容错能力。

支持的元数据过滤字段

文档默认声明了以下可过滤字段（对应 Search 模块中的 TagField / NumericField 类型）：

字段	索引类型	典型用途
`meta_category`	TagField	字符串精确匹配
`meta_status`	TagField	状态过滤
`meta_priority`	NumericField	数值比较
`meta_score`	NumericField	分数过滤
`meta_timestamp`	NumericField	日期/时间过滤

四、写入文档：索引 Pipeline 与 write_documents

手动写入

write_documents会将文档连同其嵌入向量与元数据一起存入 Valkey；搜索索引若不存在会自动创建；缺少嵌入向量的文档会被赋予一个哑向量（dummy vector）以便完成索引（详见 API 参考）。

from haystack import Document from haystack_integrations.document_stores.valkey import ValkeyDocumentStore document_store = ValkeyDocumentStore( nodes_list=[("localhost", 6379)], index_name="my_documents", embedding_dim=768, distance_metric="cosine", ) documents = [ Document( content="First document", embedding=[0.1, 0.2, 0.3], meta={"category": "news", "priority": 1}, ), Document( content="Second document", embedding=[0.4, 0.5, 0.6], meta={"category": "blog", "priority": 2}, ), ] count = document_store.write_documents(documents) print(f"Wrote {count} documents")

write_documents的签名与约束：

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

documents：每个文档应包含content（正文）、embedding（向量，可选，缺失时用哑向量）、meta（可选元数据，使用受支持的 category / status / priority / score / timestamp 字段）；
policy：重复文档处理策略，仅支持DuplicatePolicy.NONE与DuplicatePolicy.OVERWRITE（默认NONE）。DuplicatePolicy与FilterPolicy同从haystack.document_stores.types导入（见 types/init.py）；
返回成功写入的文档数量；
异常：写入失败抛ValkeyDocumentStoreError，文档对象非法抛ValueError。

异步版本write_documents_async采用batch_size参数指定的批量处理以提升性能，签名与同步版本一致。

索引 Pipeline 实战

更规范的做法是用 Pipeline 串联转换、切分、嵌入与写入（参考 ValkeyDocumentStore 指南的索引示例）：

from haystack import Pipeline from haystack.components.converters import MarkdownToDocument from haystack.components.writers import DocumentWriter from haystack.components.preprocessors import DocumentSplitter from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersDocumentEmbedder, ) from haystack_integrations.document_stores.valkey import ValkeyDocumentStore document_store = ValkeyDocumentStore( nodes_list=[("localhost", 6379)], index_name="my_documents", embedding_dim=768, distance_metric="cosine", ) indexing = Pipeline() indexing.add_component("converter", MarkdownToDocument()) indexing.add_component("splitter", DocumentSplitter(split_by="sentence", split_length=2)) indexing.add_component("embedder", SentenceTransformersDocumentEmbedder()) indexing.add_component("writer", DocumentWriter(document_store)) indexing.connect("converter", "splitter") indexing.connect("splitter", "embedder") indexing.connect("embedder", "writer") indexing.run({"converter": {"sources": ["filename.md"]}})

五、ValkeyEmbeddingRetriever：Pipeline 中的向量检索

初始化与参数

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

参数	类型	默认值	说明
`document_store`	`ValkeyDocumentStore`	必填	底层文档存储实例
`filters`	`dict[str, Any] \\| None`	`None`	作用于检索结果的元数据过滤条件
`top_k`	`int`	`10`	返回的文档最大数量
`filter_policy`	`str \\| FilterPolicy`	`FilterPolicy.REPLACE`	运行时过滤条件的应用策略

若document_store不是ValkeyDocumentStore实例，构造时抛出ValueError。

检索方法 run / run_async

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

query_embedding：查询文本对应的嵌入向量（由 Text Embedder 产出）；
filters：运行时过滤条件，具体如何生效取决于初始化时选择的filter_policy；
top_k：返回文档数量上限；
返回{"documents": [...]}，即与query_embedding语义相近的文档列表。

run_async为异步版本，签名与返回结构完全一致。

单独使用

from haystack_integrations.document_stores.valkey import ValkeyDocumentStore from haystack_integrations.components.retrievers.valkey import ValkeyEmbeddingRetriever document_store = ValkeyDocumentStore( nodes_list=[("localhost", 6379)], index_name="my_documents", embedding_dim=768, distance_metric="cosine", ) retriever = ValkeyEmbeddingRetriever(document_store=document_store) # 用假向量保持示例简洁 retriever.run(query_embedding=[0.1] * 768)

在 Pipeline 中与 Embedder 配合

API 参考中的完整示例展示了“Text Embedder → Retriever”的查询链路：

from haystack.document_stores.types import DuplicatePolicy from haystack import Document from haystack import Pipeline # Requires: pip install sentence-transformers-haystack from haystack_integrations.components.embedders.sentence_transformers import SentenceTransformersTextEmbedder from haystack_integrations.components.embedders.sentence_transformers import SentenceTransformersDocumentEmbedder from haystack_integrations.components.retrievers.valkey import ValkeyEmbeddingRetriever from haystack_integrations.document_stores.valkey import ValkeyDocumentStore document_store = ValkeyDocumentStore(index_name="my_index", embedding_dim=768) 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..."), Document(content="In certain places, 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", ValkeyEmbeddingRetriever(document_store=document_store)) query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding") query = "How many languages are there?" res = query_pipeline.run({"text_embedder": {"text": query}}) assert res["retriever"]["documents"][0].content == "There are over 7,000 languages spoken around the world today."

六、完整的 RAG Pipeline 示例

将检索结果接入提示词构建与生成模型，即可构成端到端 RAG 链路（参考 ValkeyDocumentStore 指南）：

from haystack import Pipeline from haystack.utils import Secret from haystack.dataclasses import ChatMessage from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, ) from haystack.components.builders import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack_integrations.document_stores.valkey import ValkeyDocumentStore from haystack_integrations.components.retrievers.valkey import ValkeyEmbeddingRetriever document_store = ValkeyDocumentStore( nodes_list=[("localhost", 6379)], index_name="my_documents", embedding_dim=768, distance_metric="cosine", ) prompt_template = [ ChatMessage.from_system( "Answer the question based on the provided context. If the context does not include an answer, reply with 'I don't know'.", ), ChatMessage.from_user( "Query: {{query}}\n" "Documents:\n{% for doc in documents %}{{ doc.content }}\n{% endfor %}\n" "Answer:", ), ] query_pipeline = Pipeline() query_pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder()) query_pipeline.add_component( "retriever", ValkeyEmbeddingRetriever(document_store=document_store), ) query_pipeline.add_component( "prompt_builder", ChatPromptBuilder( template=prompt_template, required_variables=["query", "documents"], ), ) query_pipeline.add_component( "generator", OpenAIChatGenerator( api_key=Secret.from_token("YOUR_OPENAI_API_KEY"), model="gpt-4o", ), ) query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding") query_pipeline.connect("retriever.documents", "prompt_builder.documents") query_pipeline.connect("prompt_builder.prompt", "generator.messages") query = "What is Valkey?" results = query_pipeline.run( { "text_embedder": {"text": query}, "prompt_builder": {"query": query}, }, )

七、元数据过滤：Haystack 过滤器格式与 FilterPolicy

过滤器基本语法

Haystack 的过滤器分为Comparison（比较）与Logic（逻辑）两种字典结构（详见 Metadata Filtering 概念文档）。比较过滤器必须包含field、operator、value三个键：

operator 可选值: ==, !=, >, >=, <, <=, in, not in

逻辑过滤器通过operator（AND/OR/NOT）与conditions（子条件列表）组合多个条件。Valkey 集成支持的操作集合（见 API 参考的 filters 模块）为：

TagField 过滤器：==、!=、in、not in（字符串精确匹配）；
NumericField 过滤器：==、!=、>、>=、<、<=、in、not in（数值比较）；
逻辑运算符：AND、OR组合条件。

过滤器示例：

# 简单等值过滤 filters = {"field": "meta.category", "operator": "==", "value": "tech"} # 数值范围过滤 filters = {"field": "meta.priority", "operator": ">=", "value": 5} # 列表成员过滤 filters = {"field": "meta.status", "operator": "in", "value": ["active", "pending"]} # 复杂逻辑过滤 filters = { "operator": "AND", "conditions": [ {"field": "meta.category", "operator": "==", "value": "tech"}, {"field": "meta.priority", "operator": ">=", "value": 3}, ], }

在 Pipeline 中，过滤器可直接传给Pipeline.run()并路由到 Retriever：

pipeline.run( data={ "retriever": { "query_embedding": query_embedding, "filters": { "operator": "AND", "conditions": [ {"field": "meta.years", "operator": "==", "value": "2019"}, {"field": "meta.companies", "operator": "in", "value": ["BMW", "Mercedes"]}, ], }, }, }, )

FilterPolicy：初始化过滤器与运行时过滤器的关系

ValkeyEmbeddingRetriever的filter_policy参数（默认FilterPolicy.REPLACE）决定了初始化过滤器与每次run()传入的运行时过滤器如何协同。该策略由 Haystack 核心定义，源码见 filter_policy.py：

REPLACE（replace）：运行时过滤器直接替换初始化过滤器。对应源码中apply_filter_policy的兜底逻辑——非 MERGE 情况下返回runtime_filters or init_filters；
MERGE（merge）：运行时过滤器与初始化过滤器合并，字段冲突时运行时值覆盖初始化值。合并时按四种组合分别处理（同样见apply_filter_policy与各combine_*辅助函数）：
- 两个比较过滤器 → 以默认逻辑运算符（AND）组合为逻辑过滤器；同字段时初始化过滤器被忽略；
- 初始化比较 + 运行时逻辑 → 运算符一致时并入条件列表，同字段则忽略初始化过滤器；
- 初始化逻辑 + 运行时比较 → 并入条件列表，运行时同字段覆盖初始化条件；
- 两个逻辑过滤器 → 运算符一致时合并 conditions，不一致时仅保留运行时过滤器。

Haystack 检索组件的这一行为可参考 in_memory/embedding_retriever.py 中run对apply_filter_policy的调用模式：filters = apply_filter_policy(self.filter_policy, self.filters, filters)。Valkey 检索器遵循同一约定，因此初始化时的filters会作为静态检索范围，而运行时filters支持逐次查询动态变化（如按用户会话、租户或权限动态收窄范围）。

八、纯元数据操作：filter_documents 与 delete/update/count

filter_documents：不依赖向量的元数据过滤

由于 Valkey Search 要求向量查询，filter_documents内部使用哑向量执行查询，再从结果中移除相似度分数（详见 API 参考）：

docs = document_store.filter_documents( filters={"field": "meta.category", "operator": "==", "value": "news"}, ) docs = document_store.filter_documents( filters={"field": "meta.priority", "operator": ">=", "value": 5}, )

返回的文档score为None。异常时抛ValkeyDocumentStoreError。异步版本filter_documents_async行为一致。

文档管理操作一览

以下方法均提供同步与异步两个版本（异步以_async结尾，并通过batch_size分批处理）：

方法	功能	关键约束 / 异常
`count_documents()`	查询索引中文档总数	索引不存在时返回 0；出错抛`ValkeyDocumentStoreError`
`count_documents_by_filter(filters)`	统计匹配过滤器的文档数	过滤器非法抛`FilterError`；计数失败抛`ValkeyDocumentStoreError`
`delete_documents(document_ids)`	按 ID 从数据库与索引中删除	未找到的文档记录 warning 后继续
`delete_by_filter(filters)`	按过滤器删除	返回删除数量；过滤器非法抛`FilterError`
`delete_all_documents()`	通过丢弃整个搜索索引清空全部数据	不可逆！下次写入时索引自动重建
`update_by_filter(filters, meta)`	更新匹配文档的元数据（与现有 meta 合并）	返回更新数量；过滤器非法抛`FilterError`
`count_unique_metadata_by_filter(filters, metadata_fields)`	统计每个元数据字段的唯一值数量	未配置的字段抛`ValueError`

元数据字段内省 API

为便于构建过滤 UI 或调试，文档存储还提供以下内省方法：

get_metadata_fields_info()：返回已配置元数据字段名与类型（"keyword"表示 tag，"long"表示 numeric），字段名不带meta.前缀；
get_metadata_field_min_max(metadata_field)：返回数值字段的最小/最大值（{"min": ..., "max": ...}），字段非数值或未配置抛ValueError；
get_metadata_field_unique_values(metadata_field, search_term=None, from_=0, size=10, filters=None)：分页返回字段唯一值，search_term对值的字符串表示做不区分大小写的子串匹配，返回(values, total_count)元组。适合实现“按分类浏览”“搜索词联想”等功能。

上述方法同样全部提供_async异步版本。

九、序列化与资源管理

to_dict / from_dict

ValkeyEmbeddingRetriever与ValkeyDocumentStore均实现to_dict()/from_dict()方法，用于将组件序列化为字典、再反序列化还原。这使得它们可以无缝嵌入 Haystack 的 YAML/JSON Pipeline 描述体系（包括Pipeline.dumps序列化与Pipeline.loads反序列化），实现 Pipeline 定义的版本化与共享。

from_dict会将字典中的字符串形式的filter_policy解析回FilterPolicy枚举（Haystack 核心中通过FilterPolicy.from_str完成，非法字符串会抛ValueError，见 filter_policy.py）。

close / close_async

ValkeyEmbeddingRetriever.close()/close_async()：释放底层 Document Store 的同步/异步资源；
ValkeyDocumentStore.close()/close_async()：释放文档存储自身关联的同步/异步连接资源。

在长生命周期服务中，应用退出前应调用对应close方法释放连接；异步场景（如 FastAPI 应用）则使用close_async。

十、源码级原理小结

回顾本集成在 Haystack 生态中的实现要点：

HNSW 向量索引：文档写入后，ValkeyDocumentStore依赖 Valkey Search 模块自动创建/复用 HNSW 索引；检索器通过向量相似度查询命中语义相近文档，无需启动时全量扫描。
距离度量选择：distance_metric支持l2、cosine、ip（内积），默认cosine。余弦距离对向量模长不敏感，适合绝大多数文本嵌入模型；ip适合已归一化向量以追求更高吞吐。
哑向量机制：为保证“无嵌入也可索引/过滤”与纯元数据查询（filter_documents）的统一，存储层内部使用哑向量补齐索引位，并在纯过滤场景剥离相似度分数。
过滤策略统一：FilterPolicy.REPLACE / MERGE由 Haystack 核心统一实现（filter_policy.py），Valkey 检索器与其他检索器（如 in_memory/embedding_retriever.py）行为保持一致，降低了跨文档存储迁移的学习成本。
同步/异步双轨：文档存储与检索器的几乎所有操作都提供_async版本，配合batch_size批量写入，可满足高吞吐索引与低延迟查询并存的生产需求。

Haystack 集成 Valkey 指南：基于 HNSW 向量检索的 ValkeyDocumentStore 与 ValkeyEmbeddingRetriever 完整实战