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/haystackValkey 是一个高性能、内存优先的数据结构存储服务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_namemy_documents, embedding_dim768, distance_metriccosine, )构造参数全解析根据 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, strictFalse), password: Secret | None Secret.from_env_var(VALKEY_PASSWORD, strictFalse), 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_listlist[tuple[str, int]] \| None[(localhost, 6379)]Valkey 节点 (host, port) 列表cluster_modeboolFalse是否以集群模式连接use_tlsboolFalse连接是否启用 TLSusernameSecret \| None读自VALKEY_USERNAME环境变量认证用户名passwordSecret \| None读自VALKEY_PASSWORD环境变量认证密码request_timeoutint500请求超时毫秒retry_attemptsint3失败操作的重试次数retry_base_delay_msint1000指数退避的基准延迟毫秒retry_exponent_baseint2指数退避计算的指数底数batch_sizeint100异步操作的批量处理文档数index_namestrdefault搜索索引名称distance_metricl2 \| cosine \| ipcosine向量相似度距离度量embedding_dimint768文档嵌入向量维度metadata_fieldsdict[str, type[str] \| type[int]] \| NoneNone可过滤元数据字段映射如{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_categoryTagField字符串精确匹配meta_statusTagField状态过滤meta_priorityNumericField数值比较meta_scoreNumericField分数过滤meta_timestampNumericField日期/时间过滤四、写入文档索引 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_namemy_documents, embedding_dim768, distance_metriccosine, ) documents [ Document( contentFirst document, embedding[0.1, 0.2, 0.3], meta{category: news, priority: 1}, ), Document( contentSecond document, embedding[0.4, 0.5, 0.6], meta{category: blog, priority: 2}, ), ] count document_store.write_documents(documents) print(fWrote {count} documents)write_documents的签名与约束write_documents(documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE) - intdocuments每个文档应包含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_namemy_documents, embedding_dim768, distance_metriccosine, ) indexing Pipeline() indexing.add_component(converter, MarkdownToDocument()) indexing.add_component(splitter, DocumentSplitter(split_bysentence, split_length2)) 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]}})五、ValkeyEmbeddingRetrieverPipeline 中的向量检索初始化与参数__init__( *, document_store: ValkeyDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数类型默认值说明document_storeValkeyDocumentStore必填底层文档存储实例filtersdict[str, Any] \| NoneNone作用于检索结果的元数据过滤条件top_kint10返回的文档最大数量filter_policystr \| FilterPolicyFilterPolicy.REPLACE运行时过滤条件的应用策略若document_store不是ValkeyDocumentStore实例构造时抛出ValueError。检索方法 run / run_asyncrun( 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_policytop_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_namemy_documents, embedding_dim768, distance_metriccosine, ) retriever ValkeyEmbeddingRetriever(document_storedocument_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_namemy_index, embedding_dim768) documents [ Document(contentThere are over 7,000 languages spoken around the world today.), Document(contentElephants have been observed to behave in a way that indicates...), Document(contentIn 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), policyDuplicatePolicy.OVERWRITE, ) query_pipeline Pipeline() query_pipeline.add_component(text_embedder, SentenceTransformersTextEmbedder()) query_pipeline.add_component(retriever, ValkeyEmbeddingRetriever(document_storedocument_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_namemy_documents, embedding_dim768, distance_metriccosine, ) prompt_template [ ChatMessage.from_system( Answer the question based on the provided context. If the context does not include an answer, reply with I dont 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_storedocument_store), ) query_pipeline.add_component( prompt_builder, ChatPromptBuilder( templateprompt_template, required_variables[query, documents], ), ) query_pipeline.add_component( generator, OpenAIChatGenerator( api_keySecret.from_token(YOUR_OPENAI_API_KEY), modelgpt-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逻辑过滤器通过operatorAND/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()并路由到 Retrieverpipeline.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.pyREPLACEreplace运行时过滤器直接替换初始化过滤器。对应源码中apply_filter_policy的兜底逻辑——非 MERGE 情况下返回runtime_filters or init_filtersMERGEmerge运行时过滤器与初始化过滤器合并字段冲突时运行时值覆盖初始化值。合并时按四种组合分别处理同样见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/countfilter_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出错抛ValkeyDocumentStoreErrorcount_documents_by_filter(filters)统计匹配过滤器的文档数过滤器非法抛FilterError计数失败抛ValkeyDocumentStoreErrordelete_documents(document_ids)按 ID 从数据库与索引中删除未找到的文档记录 warning 后继续delete_by_filter(filters)按过滤器删除返回删除数量过滤器非法抛FilterErrordelete_all_documents()通过丢弃整个搜索索引清空全部数据不可逆下次写入时索引自动重建update_by_filter(filters, meta)更新匹配文档的元数据与现有 meta 合并返回更新数量过滤器非法抛FilterErrorcount_unique_metadata_by_filter(filters, metadata_fields)统计每个元数据字段的唯一值数量未配置的字段抛ValueError元数据字段内省 API为便于构建过滤 UI 或调试文档存储还提供以下内省方法get_metadata_fields_info()返回已配置元数据字段名与类型keyword表示 taglong表示 numeric字段名不带meta.前缀get_metadata_field_min_max(metadata_field)返回数值字段的最小/最大值{min: ..., max: ...}字段非数值或未配置抛ValueErrorget_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone)分页返回字段唯一值search_term对值的字符串表示做不区分大小写的子串匹配返回(values, total_count)元组。适合实现“按分类浏览”“搜索词联想”等功能。上述方法同样全部提供_async异步版本。九、序列化与资源管理to_dict / from_dictValkeyEmbeddingRetriever与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_asyncValkeyEmbeddingRetriever.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.pyValkey 检索器与其他检索器如 in_memory/embedding_retriever.py行为保持一致降低了跨文档存储迁移的学习成本。同步/异步双轨文档存储与检索器的几乎所有操作都提供_async版本配合batch_size批量写入可满足高吞吐索引与低延迟查询并存的生产需求。延伸阅读ValkeyEmbeddingRetriever 组件指南检索器在 Pipeline 中的定位RAG 检索段、语义搜索末端、抽取式 QA 前段与独立使用示例ValkeyDocumentStore 存储指南索引 Pipeline、RAG Pipeline 与性能特性说明Metadata Filtering 概念文档Haystack 统一过滤器语法Comparison / Logic的完整说明Choosing a Document StoreValkey 与其他内存键值/向量存储类别的选型对比FilterPolicy 核心实现REPLACE / MERGE 语义与合并算法的权威定义。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 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/haystackValkey 是一个高性能、内存优先的数据结构存储服务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_namemy_documents, embedding_dim768, distance_metriccosine, )构造参数全解析根据 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, strictFalse), password: Secret | None Secret.from_env_var(VALKEY_PASSWORD, strictFalse), 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_listlist[tuple[str, int]] \| None[(localhost, 6379)]Valkey 节点 (host, port) 列表cluster_modeboolFalse是否以集群模式连接use_tlsboolFalse连接是否启用 TLSusernameSecret \| None读自VALKEY_USERNAME环境变量认证用户名passwordSecret \| None读自VALKEY_PASSWORD环境变量认证密码request_timeoutint500请求超时毫秒retry_attemptsint3失败操作的重试次数retry_base_delay_msint1000指数退避的基准延迟毫秒retry_exponent_baseint2指数退避计算的指数底数batch_sizeint100异步操作的批量处理文档数index_namestrdefault搜索索引名称distance_metricl2 \| cosine \| ipcosine向量相似度距离度量embedding_dimint768文档嵌入向量维度metadata_fieldsdict[str, type[str] \| type[int]] \| NoneNone可过滤元数据字段映射如{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_categoryTagField字符串精确匹配meta_statusTagField状态过滤meta_priorityNumericField数值比较meta_scoreNumericField分数过滤meta_timestampNumericField日期/时间过滤四、写入文档索引 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_namemy_documents, embedding_dim768, distance_metriccosine, ) documents [ Document( contentFirst document, embedding[0.1, 0.2, 0.3], meta{category: news, priority: 1}, ), Document( contentSecond document, embedding[0.4, 0.5, 0.6], meta{category: blog, priority: 2}, ), ] count document_store.write_documents(documents) print(fWrote {count} documents)write_documents的签名与约束write_documents(documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE) - intdocuments每个文档应包含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_namemy_documents, embedding_dim768, distance_metriccosine, ) indexing Pipeline() indexing.add_component(converter, MarkdownToDocument()) indexing.add_component(splitter, DocumentSplitter(split_bysentence, split_length2)) 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]}})五、ValkeyEmbeddingRetrieverPipeline 中的向量检索初始化与参数__init__( *, document_store: ValkeyDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数类型默认值说明document_storeValkeyDocumentStore必填底层文档存储实例filtersdict[str, Any] \| NoneNone作用于检索结果的元数据过滤条件top_kint10返回的文档最大数量filter_policystr \| FilterPolicyFilterPolicy.REPLACE运行时过滤条件的应用策略若document_store不是ValkeyDocumentStore实例构造时抛出ValueError。检索方法 run / run_asyncrun( 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_policytop_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_namemy_documents, embedding_dim768, distance_metriccosine, ) retriever ValkeyEmbeddingRetriever(document_storedocument_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_namemy_index, embedding_dim768) documents [ Document(contentThere are over 7,000 languages spoken around the world today.), Document(contentElephants have been observed to behave in a way that indicates...), Document(contentIn 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), policyDuplicatePolicy.OVERWRITE, ) query_pipeline Pipeline() query_pipeline.add_component(text_embedder, SentenceTransformersTextEmbedder()) query_pipeline.add_component(retriever, ValkeyEmbeddingRetriever(document_storedocument_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_namemy_documents, embedding_dim768, distance_metriccosine, ) prompt_template [ ChatMessage.from_system( Answer the question based on the provided context. If the context does not include an answer, reply with I dont 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_storedocument_store), ) query_pipeline.add_component( prompt_builder, ChatPromptBuilder( templateprompt_template, required_variables[query, documents], ), ) query_pipeline.add_component( generator, OpenAIChatGenerator( api_keySecret.from_token(YOUR_OPENAI_API_KEY), modelgpt-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逻辑过滤器通过operatorAND/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()并路由到 Retrieverpipeline.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.pyREPLACEreplace运行时过滤器直接替换初始化过滤器。对应源码中apply_filter_policy的兜底逻辑——非 MERGE 情况下返回runtime_filters or init_filtersMERGEmerge运行时过滤器与初始化过滤器合并字段冲突时运行时值覆盖初始化值。合并时按四种组合分别处理同样见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/countfilter_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出错抛ValkeyDocumentStoreErrorcount_documents_by_filter(filters)统计匹配过滤器的文档数过滤器非法抛FilterError计数失败抛ValkeyDocumentStoreErrordelete_documents(document_ids)按 ID 从数据库与索引中删除未找到的文档记录 warning 后继续delete_by_filter(filters)按过滤器删除返回删除数量过滤器非法抛FilterErrordelete_all_documents()通过丢弃整个搜索索引清空全部数据不可逆下次写入时索引自动重建update_by_filter(filters, meta)更新匹配文档的元数据与现有 meta 合并返回更新数量过滤器非法抛FilterErrorcount_unique_metadata_by_filter(filters, metadata_fields)统计每个元数据字段的唯一值数量未配置的字段抛ValueError元数据字段内省 API为便于构建过滤 UI 或调试文档存储还提供以下内省方法get_metadata_fields_info()返回已配置元数据字段名与类型keyword表示 taglong表示 numeric字段名不带meta.前缀get_metadata_field_min_max(metadata_field)返回数值字段的最小/最大值{min: ..., max: ...}字段非数值或未配置抛ValueErrorget_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone)分页返回字段唯一值search_term对值的字符串表示做不区分大小写的子串匹配返回(values, total_count)元组。适合实现“按分类浏览”“搜索词联想”等功能。上述方法同样全部提供_async异步版本。九、序列化与资源管理to_dict / from_dictValkeyEmbeddingRetriever与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_asyncValkeyEmbeddingRetriever.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.pyValkey 检索器与其他检索器如 in_memory/embedding_retriever.py行为保持一致降低了跨文档存储迁移的学习成本。同步/异步双轨文档存储与检索器的几乎所有操作都提供_async版本配合batch_size批量写入可满足高吞吐索引与低延迟查询并存的生产需求。延伸阅读ValkeyEmbeddingRetriever 组件指南检索器在 Pipeline 中的定位RAG 检索段、语义搜索末端、抽取式 QA 前段与独立使用示例ValkeyDocumentStore 存储指南索引 Pipeline、RAG Pipeline 与性能特性说明Metadata Filtering 概念文档Haystack 统一过滤器语法Comparison / Logic的完整说明Choosing a Document StoreValkey 与其他内存键值/向量存储类别的选型对比FilterPolicy 核心实现REPLACE / MERGE 语义与合并算法的权威定义。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考