Haystack 与 Valkey 集成实战从 ValkeyDocumentStore 向量检索到 ValkeyEmbeddingRetriever 的完整 API 指南【免费下载链接】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 开源仓库中的 Valkey 集成 API 参考文档docs-website/reference/integrations-api/valkey.md为核心系统讲解ValkeyEmbeddingRetriever、ValkeyDocumentStore与过滤工具三个模块的完整接口、参数语义与底层原理。读完本文你将能够基于 ValkeyRedis 兼容的高性能内存数据库搭建具备 HNSW 向量检索、元数据过滤、同步/异步双通道能力的 Haystack 检索与 RAG 管线并理解过滤器从 Haystack 格式到 Valkey Search 查询语法的转换机制。一、Valkey 集成概览为什么在 Haystack 中选择 ValkeyValkey 是一个与 Redis 兼容的高性能内存数据结构存储在 Haystack 生态中通过ValkeyDocumentStore接入。该集成面向需要语义检索 元数据过滤双能力的高性能检索场景核心能力包括基于HNSW 算法的向量相似度检索对 **TagField标签字段与 NumericField数值字段**的元数据过滤可配置的距离度量L2、余弦cosine、内积ip批量写入等高效的文档管理操作同步与异步两套完整 API单机standalone与集群cluster两种连接模式。从官方用户指南docs-website/docs/document-stores/valkeydocumentstore.mdx可以确认Valkey 默认以内存方式运行以获得最大性能同时可配置持久化选项保证数据可靠性文档存储连接到运行了search 模块的 Valkey 服务器为 RAG 等检索场景提供向量相似度搜索能力。安装与本地启动安装集成包pip install valkey-haystack文档中的示例使用 Sentence Transformers 嵌入器已迁移至独立包需一并安装pip install sentence-transformers-haystack开发与测试环境可用 Docker 快速启动带 search 模块的 Valkey 服务器docker run -d -p 6379:6379 valkey/valkey-bundle:latest启动后即可通过nodes_list[(localhost, 6379)]连接。二、ValkeyEmbeddingRetriever基于稠密向量的语义检索组件ValkeyEmbeddingRetriever位于haystack_integrations.components.retrievers.valkey.embedding_retriever模块是一个从ValkeyDocumentStore中按向量相似度检索文档的组件。它使用稠密嵌入寻找语义相近的文档支持按元数据字段过滤并支持可配置的相似度阈值。关键特性基于HNSW 算法的向量相似度检索支持tag 与 numeric 字段的元数据过滤可配置的top-k结果数量过滤策略管理用于运行时过滤的应用方式。完整使用示例以下示例来自 API 参考文档valkey.md演示了写文档 → 建查询管线 → 语义检索的完整链路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.这里有两个值得注意的细节管线连接text_embedder.embedding输出的查询向量被连接到retriever.query_embedding输入——这是嵌入检索器在 Haystack 管线中的标准连接模式DuplicatePolicy.OVERWRITE写入时采用覆盖策略其枚举定义在仓库的 haystack/document_stores/types/policy.py 中完整取值包括NONE、SKIP、OVERWRITE、FAIL。本文档涉及的 Valkey 存储仅支持NONE与OVERWRITE。init初始化参数__init__( *, document_store: ValkeyDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数类型默认值说明document_storeValkeyDocumentStore必填底层 Valkey 文档存储实例filtersdict[str, Any] \| NoneNone应用于检索结果的过滤器top_kint10返回的文档最大数量filter_policystr \| FilterPolicyFilterPolicy.REPLACE决定过滤器如何应用见下文过滤策略异常若document_store不是ValkeyDocumentStore实例抛出ValueError。关于filter_policy仓库源码 haystack/document_stores/types/filter_policy.py 中定义了两种策略FilterPolicy.REPLACE默认运行时过滤器直接替换初始化时的过滤器FilterPolicy.MERGE运行时过滤器与初始化过滤器合并同一字段上运行时值覆盖初始化值。底层由apply_filter_policy()函数统一处理它会根据初始化/运行时过滤器是比较型含field、operator、value三键还是逻辑型含operator与conditions自动选择组合方式例如两个逻辑型过滤器同运算符时合并conditions同字段冲突时保留运行时值并记录告警日志。run / run_async同步与异步检索run( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]参数类型说明query_embeddinglist[float]查询的向量表示filtersdict[str, Any] \| None应用于检索结果的运行时过滤器其应用方式取决于初始化时选择的filter_policytop_kint \| None返回文档的最大数量覆盖初始化值返回dict[str, list[Document]]即与query_embedding相似的文档列表键为documents。run_async提供完全相同的签名与语义用于异步场景。序列化与资源释放to_dict() - dict[str, Any]将组件序列化为字典含文档存储信息、过滤器、top_k、过滤策略等支持 YAML/JSON 管线描述from_dict(data: dict[str, Any]) - ValkeyEmbeddingRetriever从字典反序列化组件close() - None释放底层文档存储的同步资源close_async() - None释放底层文档存储的异步资源。在管线中的典型位置根据官方检索器指南valkeyembeddingretriever.mdxValkeyEmbeddingRetriever在管线中的常见位置为RAG 管线中位于 Text Embedder 之后、ChatPromptBuilder/PromptBuilder之前语义搜索管线中的最后一个组件抽取式问答管线中位于 Text Embedder 之后、TransformersExtractiveReader之前。必填初始化变量为document_store必填运行变量为query_embedding输出变量为documents。三、ValkeyDocumentStore基于 Valkey Search 的向量文档存储ValkeyDocumentStore位于haystack_integrations.document_stores.valkey.document_store模块继承自 Haystack 的DocumentStore协议协议定义见 haystack/document_stores/types/init.py。它提供带向量的文档持久化存储并借助 Valkey Search 模块支持向量相似度检索专为需要语义检索与元数据过滤的高性能应用设计。核心特性基于 HNSW 算法的向量相似度检索对 tag 与 numeric 字段的元数据过滤可配置的距离度量L2、余弦、内积批量操作高效管理文档同步与异步双通道 API集群与单机两种模式支持。可过滤的元数据字段文档存储默认支持以下可过滤的 Document 元数据字段配合metadata_fields参数配置字段索引类型用途meta_categoryTagField精确字符串匹配meta_statusTagField状态过滤meta_priorityNumericField数值比较meta_scoreNumericField分数过滤meta_timestampNumericField日期/时间过滤初始化与参数详解__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搜索索引名称文档中另有说明写作haystack_document以版本为准distance_metricLiteral[l2, cosine, ip]cosine向量相似度距离度量embedding_dimint768文档嵌入维度metadata_fieldsdict[str, type[str] \| type[int]] \| NoneNone元数据字段名到类型的映射用于过滤。支持str精确匹配与int数值比较例如{category: str, priority: int}。不配置则不索引任何可过滤元数据字段其中username与password使用 Haystack 的Secret机制源自haystack.utils支持从环境变量VALKEY_USERNAME、VALKEY_PASSWORD惰性读取避免密钥硬编码。文档读写write_documents / write_documents_asyncwrite_documents( documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE ) - int行为将文档连同嵌入与元数据写入 Valkey若搜索索引不存在则自动创建没有嵌入的文档会被赋予一个 dummy 向量用于索引。文档要求content文本、embedding向量可选缺失时用 dummy 向量、meta可选的受支持字段如 category、status、priority、score、timestamp。policy重复文档处理策略仅支持DuplicatePolicy.NONE与DuplicatePolicy.OVERWRITE默认NONE。返回成功写入的文档数。异常ValkeyDocumentStoreError写入出错、ValueError文档列表含非法对象。write_documents_async为异步版本采用批处理batch_size参数控制每批数量以提升性能索引同样在不存在时自动创建。检索filter_documents纯元数据过滤无向量检索filter_documents(filters: dict[str, Any] | None None) - list[Document]由于 Valkey Search 强制要求向量查询此方法在内部使用dummy 向量执行查询再从结果中移除相似度分数从而实现对meta.category字符串等值、meta.status字符串等值、meta.priority/meta.score/meta.timestamp数值比较的纯元数据过滤。返回文档的score均为None。filter_documents_async为异步版本。# Filter by category docs document_store.filter_documents( filters{field: meta.category, operator: , value: news} ) # Filter by numeric range docs document_store.filter_documents( filters{field: meta.priority, operator: , value: 5} )统计与计数方法签名说明count_documents() - int查询 Valkey Search 索引获取文档总数索引不存在时返回 0。异常ValkeyDocumentStoreErrorcount_documents_async() - int上述方法的异步版本count_documents_by_filter(filters: dict[str, Any]) - int统计匹配过滤条件的文档数。异常FilterError、ValkeyDocumentStoreErrorcount_documents_by_filter_async(filters: dict[str, Any]) - int上述方法的异步版本删除操作方法签名说明delete_documents(document_ids: list[str]) - None按 ID 同时从 Valkey 数据库与搜索索引中删除部分文档未找到时记录告警并继续。异常ValkeyDocumentStoreErrordelete_documents_async(document_ids: list[str]) - None上述方法的异步版本delete_by_filter(filters: dict[str, Any]) - int删除匹配过滤条件的所有文档返回删除数量。异常FilterError、ValkeyDocumentStoreErrordelete_by_filter_async(filters: dict[str, Any]) - int上述方法的异步版本delete_all_documents() - None删除全部文档通过直接丢弃整个搜索索引实现最高效但不可逆后续写入会重建索引索引不存在时静默成功。异常ValkeyDocumentStoreErrordelete_all_documents_async() - None上述方法的异步版本更新操作update_by_filter(filters: dict[str, Any], meta: dict[str, Any]) - int更新所有匹配过滤条件文档的元数据meta中的键值对会与已有 meta 合并。返回更新文档数。异常FilterError过滤结构非法、ValkeyDocumentStoreError更新或写入失败。update_by_filter_async为异步版本。元数据字段信息与探查方法签名说明get_metadata_fields_info() - dict[str, dict[str, str]]返回已配置可过滤元数据字段的名称与类型字段名不带meta.前缀keyword表示 tag 字段long表示数值字段get_metadata_field_min_max(metadata_field: str) - dict[str, Any]返回数值字段的最小/最大值形如{min: ..., max: ...}无值时可为 None。异常ValueError字段未配置或非数值、ValkeyDocumentStoreErrorget_metadata_field_min_max_async(metadata_field: str) - dict[str, Any]上述方法的异步版本get_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone) - tuple[list[Any], int]返回字段的唯一值列表支持大小写不敏感的子串搜索search_term与分页from_/size可附加filters限定文档范围返回值保留原始类型如 int、bool返回(当前页唯一值列表, 唯一值总数)。异常ValkeyDocumentStoreErrorget_metadata_field_unique_values_async同上上述方法的异步版本get_metadata_field_unique_values特别适合为 UI 筛选器生成候选值列表例如按 category 字段展示去重后的所有可选分类。序列化与资源管理to_dict() - dict[str, Any]将存储序列化为字典连接信息、索引配置、距离度量等。from_dict(data: dict[str, Any]) - ValkeyDocumentStore从字典反序列化存储实例。close() - None释放关联的同步资源。close_async() - None释放关联的异步资源。四、过滤工具模块Haystack 过滤器到 Valkey Search 查询的转换haystack_integrations.document_stores.valkey.filters模块提供过滤工具将 Haystack 的过滤器格式转换为 Valkey Search 查询语法同时支持基于 tag 的精确匹配与基于数值的范围过滤并支持逻辑运算符组合。支持的过滤操作TagField标签字段、!、in、not in精确字符串匹配NumericField数值字段、!、、、、、in、not in数值比较逻辑运算符AND、OR用于组合多个条件。过滤语法示例# Simple equality filter filters {field: meta.category, operator: , value: tech} # Numeric range filter filters {field: meta.priority, operator: , value: 5} # List membership filter filters {field: meta.status, operator: in, value: [active, pending]} # Complex logical filter filters { operator: AND, conditions: [ {field: meta.category, operator: , value: tech}, {field: meta.priority, operator: , value: 3} ] }这段语法与 Haystack 核心的过滤器约定保持一致单条件为{field, operator, value}三键结构组合条件为{operator, conditions}结构。在 Haystack 核心仓库中对应的判读逻辑位于 haystack/document_stores/types/filter_policy.py 的is_comparison_filter()与is_logical_filter()辅助函数分别校验三键结构field/operator/value与逻辑结构operator/conditions这为跨存储Valkey、Elasticsearch、InMemory 等复用同一套过滤器语义提供了统一基础。五、实战在 RAG 管线中组合使用 Valkey 组件将上述 API 组合起来即可构建一个完整可运行的 RAG 查询管线。以下示例取自官方文档存储指南valkeydocumentstore.mdx展示了 Valkey 检索结果如何注入提示词并交给 LLM 生成回答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}, }, )配套索引管线数据入库侧可构建索引管线MarkdownToDocument转换器 →DocumentSplitter切分 →SentenceTransformersDocumentEmbedder嵌入 →DocumentWriter写入ValkeyDocumentStore全部由 Haystack 内置组件haystack/components 目录与集成嵌入器协作完成from haystack import Pipeline from haystack.components.converters import MarkdownToDocument from haystack.components.writers import DocumentWriter from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersDocumentEmbedder, ) from haystack.components.preprocessors import DocumentSplitter 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]}})六、性能特性与适用边界根据官方文档Valkey 作为内存存储具备以下特性内存存储读写操作快速高吞吐每秒可处理大量操作低延迟文档操作响应时间极小可扩展支持集群模式以横向扩展。需要说明的适用前提与限制均以当前仓库文档为准上述数据均来自官方文档陈述具体性能数值取决于部署硬件与数据集规模建议结合自身场景实测本文介绍的向量检索依赖运行了search 模块的 Valkey 服务器如valkey/valkey-bundle镜像普通 Valkey/Redis 实例不具备该能力文档写入的重复策略仅支持DuplicatePolicy.NONE与DuplicatePolicy.OVERWRITE使用SKIP或FAIL将不受支持filter_documents因 Valkey Search 要求向量查询内部使用 dummy 向量并剥离分数属于实现层面的折中结果score恒为None。七、进一步阅读Valkey 集成完整 API 参考docs-website/reference/integrations-api/valkey.md本文主体来源ValkeyDocumentStore 使用指南docs-website/docs/document-stores/valkeydocumentstore.mdxValkeyEmbeddingRetriever 使用指南docs-website/docs/pipeline-components/retrievers/valkeyembeddingretriever.mdx过滤策略FilterPolicy与apply_filter_policy实现haystack/document_stores/types/filter_policy.py重复策略DuplicatePolicy定义haystack/document_stores/types/policy.py文档存储协议DocumentStorehaystack/document_stores/types/init.py【免费下载链接】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 集成实战：从 ValkeyDocumentStore 向量检索到 ValkeyEmbeddingRetriever 的完整 API 指南
Haystack 与 Valkey 集成实战从 ValkeyDocumentStore 向量检索到 ValkeyEmbeddingRetriever 的完整 API 指南【免费下载链接】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 开源仓库中的 Valkey 集成 API 参考文档docs-website/reference/integrations-api/valkey.md为核心系统讲解ValkeyEmbeddingRetriever、ValkeyDocumentStore与过滤工具三个模块的完整接口、参数语义与底层原理。读完本文你将能够基于 ValkeyRedis 兼容的高性能内存数据库搭建具备 HNSW 向量检索、元数据过滤、同步/异步双通道能力的 Haystack 检索与 RAG 管线并理解过滤器从 Haystack 格式到 Valkey Search 查询语法的转换机制。一、Valkey 集成概览为什么在 Haystack 中选择 ValkeyValkey 是一个与 Redis 兼容的高性能内存数据结构存储在 Haystack 生态中通过ValkeyDocumentStore接入。该集成面向需要语义检索 元数据过滤双能力的高性能检索场景核心能力包括基于HNSW 算法的向量相似度检索对 **TagField标签字段与 NumericField数值字段**的元数据过滤可配置的距离度量L2、余弦cosine、内积ip批量写入等高效的文档管理操作同步与异步两套完整 API单机standalone与集群cluster两种连接模式。从官方用户指南docs-website/docs/document-stores/valkeydocumentstore.mdx可以确认Valkey 默认以内存方式运行以获得最大性能同时可配置持久化选项保证数据可靠性文档存储连接到运行了search 模块的 Valkey 服务器为 RAG 等检索场景提供向量相似度搜索能力。安装与本地启动安装集成包pip install valkey-haystack文档中的示例使用 Sentence Transformers 嵌入器已迁移至独立包需一并安装pip install sentence-transformers-haystack开发与测试环境可用 Docker 快速启动带 search 模块的 Valkey 服务器docker run -d -p 6379:6379 valkey/valkey-bundle:latest启动后即可通过nodes_list[(localhost, 6379)]连接。二、ValkeyEmbeddingRetriever基于稠密向量的语义检索组件ValkeyEmbeddingRetriever位于haystack_integrations.components.retrievers.valkey.embedding_retriever模块是一个从ValkeyDocumentStore中按向量相似度检索文档的组件。它使用稠密嵌入寻找语义相近的文档支持按元数据字段过滤并支持可配置的相似度阈值。关键特性基于HNSW 算法的向量相似度检索支持tag 与 numeric 字段的元数据过滤可配置的top-k结果数量过滤策略管理用于运行时过滤的应用方式。完整使用示例以下示例来自 API 参考文档valkey.md演示了写文档 → 建查询管线 → 语义检索的完整链路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.这里有两个值得注意的细节管线连接text_embedder.embedding输出的查询向量被连接到retriever.query_embedding输入——这是嵌入检索器在 Haystack 管线中的标准连接模式DuplicatePolicy.OVERWRITE写入时采用覆盖策略其枚举定义在仓库的 haystack/document_stores/types/policy.py 中完整取值包括NONE、SKIP、OVERWRITE、FAIL。本文档涉及的 Valkey 存储仅支持NONE与OVERWRITE。init初始化参数__init__( *, document_store: ValkeyDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数类型默认值说明document_storeValkeyDocumentStore必填底层 Valkey 文档存储实例filtersdict[str, Any] \| NoneNone应用于检索结果的过滤器top_kint10返回的文档最大数量filter_policystr \| FilterPolicyFilterPolicy.REPLACE决定过滤器如何应用见下文过滤策略异常若document_store不是ValkeyDocumentStore实例抛出ValueError。关于filter_policy仓库源码 haystack/document_stores/types/filter_policy.py 中定义了两种策略FilterPolicy.REPLACE默认运行时过滤器直接替换初始化时的过滤器FilterPolicy.MERGE运行时过滤器与初始化过滤器合并同一字段上运行时值覆盖初始化值。底层由apply_filter_policy()函数统一处理它会根据初始化/运行时过滤器是比较型含field、operator、value三键还是逻辑型含operator与conditions自动选择组合方式例如两个逻辑型过滤器同运算符时合并conditions同字段冲突时保留运行时值并记录告警日志。run / run_async同步与异步检索run( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]参数类型说明query_embeddinglist[float]查询的向量表示filtersdict[str, Any] \| None应用于检索结果的运行时过滤器其应用方式取决于初始化时选择的filter_policytop_kint \| None返回文档的最大数量覆盖初始化值返回dict[str, list[Document]]即与query_embedding相似的文档列表键为documents。run_async提供完全相同的签名与语义用于异步场景。序列化与资源释放to_dict() - dict[str, Any]将组件序列化为字典含文档存储信息、过滤器、top_k、过滤策略等支持 YAML/JSON 管线描述from_dict(data: dict[str, Any]) - ValkeyEmbeddingRetriever从字典反序列化组件close() - None释放底层文档存储的同步资源close_async() - None释放底层文档存储的异步资源。在管线中的典型位置根据官方检索器指南valkeyembeddingretriever.mdxValkeyEmbeddingRetriever在管线中的常见位置为RAG 管线中位于 Text Embedder 之后、ChatPromptBuilder/PromptBuilder之前语义搜索管线中的最后一个组件抽取式问答管线中位于 Text Embedder 之后、TransformersExtractiveReader之前。必填初始化变量为document_store必填运行变量为query_embedding输出变量为documents。三、ValkeyDocumentStore基于 Valkey Search 的向量文档存储ValkeyDocumentStore位于haystack_integrations.document_stores.valkey.document_store模块继承自 Haystack 的DocumentStore协议协议定义见 haystack/document_stores/types/init.py。它提供带向量的文档持久化存储并借助 Valkey Search 模块支持向量相似度检索专为需要语义检索与元数据过滤的高性能应用设计。核心特性基于 HNSW 算法的向量相似度检索对 tag 与 numeric 字段的元数据过滤可配置的距离度量L2、余弦、内积批量操作高效管理文档同步与异步双通道 API集群与单机两种模式支持。可过滤的元数据字段文档存储默认支持以下可过滤的 Document 元数据字段配合metadata_fields参数配置字段索引类型用途meta_categoryTagField精确字符串匹配meta_statusTagField状态过滤meta_priorityNumericField数值比较meta_scoreNumericField分数过滤meta_timestampNumericField日期/时间过滤初始化与参数详解__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搜索索引名称文档中另有说明写作haystack_document以版本为准distance_metricLiteral[l2, cosine, ip]cosine向量相似度距离度量embedding_dimint768文档嵌入维度metadata_fieldsdict[str, type[str] \| type[int]] \| NoneNone元数据字段名到类型的映射用于过滤。支持str精确匹配与int数值比较例如{category: str, priority: int}。不配置则不索引任何可过滤元数据字段其中username与password使用 Haystack 的Secret机制源自haystack.utils支持从环境变量VALKEY_USERNAME、VALKEY_PASSWORD惰性读取避免密钥硬编码。文档读写write_documents / write_documents_asyncwrite_documents( documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE ) - int行为将文档连同嵌入与元数据写入 Valkey若搜索索引不存在则自动创建没有嵌入的文档会被赋予一个 dummy 向量用于索引。文档要求content文本、embedding向量可选缺失时用 dummy 向量、meta可选的受支持字段如 category、status、priority、score、timestamp。policy重复文档处理策略仅支持DuplicatePolicy.NONE与DuplicatePolicy.OVERWRITE默认NONE。返回成功写入的文档数。异常ValkeyDocumentStoreError写入出错、ValueError文档列表含非法对象。write_documents_async为异步版本采用批处理batch_size参数控制每批数量以提升性能索引同样在不存在时自动创建。检索filter_documents纯元数据过滤无向量检索filter_documents(filters: dict[str, Any] | None None) - list[Document]由于 Valkey Search 强制要求向量查询此方法在内部使用dummy 向量执行查询再从结果中移除相似度分数从而实现对meta.category字符串等值、meta.status字符串等值、meta.priority/meta.score/meta.timestamp数值比较的纯元数据过滤。返回文档的score均为None。filter_documents_async为异步版本。# Filter by category docs document_store.filter_documents( filters{field: meta.category, operator: , value: news} ) # Filter by numeric range docs document_store.filter_documents( filters{field: meta.priority, operator: , value: 5} )统计与计数方法签名说明count_documents() - int查询 Valkey Search 索引获取文档总数索引不存在时返回 0。异常ValkeyDocumentStoreErrorcount_documents_async() - int上述方法的异步版本count_documents_by_filter(filters: dict[str, Any]) - int统计匹配过滤条件的文档数。异常FilterError、ValkeyDocumentStoreErrorcount_documents_by_filter_async(filters: dict[str, Any]) - int上述方法的异步版本删除操作方法签名说明delete_documents(document_ids: list[str]) - None按 ID 同时从 Valkey 数据库与搜索索引中删除部分文档未找到时记录告警并继续。异常ValkeyDocumentStoreErrordelete_documents_async(document_ids: list[str]) - None上述方法的异步版本delete_by_filter(filters: dict[str, Any]) - int删除匹配过滤条件的所有文档返回删除数量。异常FilterError、ValkeyDocumentStoreErrordelete_by_filter_async(filters: dict[str, Any]) - int上述方法的异步版本delete_all_documents() - None删除全部文档通过直接丢弃整个搜索索引实现最高效但不可逆后续写入会重建索引索引不存在时静默成功。异常ValkeyDocumentStoreErrordelete_all_documents_async() - None上述方法的异步版本更新操作update_by_filter(filters: dict[str, Any], meta: dict[str, Any]) - int更新所有匹配过滤条件文档的元数据meta中的键值对会与已有 meta 合并。返回更新文档数。异常FilterError过滤结构非法、ValkeyDocumentStoreError更新或写入失败。update_by_filter_async为异步版本。元数据字段信息与探查方法签名说明get_metadata_fields_info() - dict[str, dict[str, str]]返回已配置可过滤元数据字段的名称与类型字段名不带meta.前缀keyword表示 tag 字段long表示数值字段get_metadata_field_min_max(metadata_field: str) - dict[str, Any]返回数值字段的最小/最大值形如{min: ..., max: ...}无值时可为 None。异常ValueError字段未配置或非数值、ValkeyDocumentStoreErrorget_metadata_field_min_max_async(metadata_field: str) - dict[str, Any]上述方法的异步版本get_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone) - tuple[list[Any], int]返回字段的唯一值列表支持大小写不敏感的子串搜索search_term与分页from_/size可附加filters限定文档范围返回值保留原始类型如 int、bool返回(当前页唯一值列表, 唯一值总数)。异常ValkeyDocumentStoreErrorget_metadata_field_unique_values_async同上上述方法的异步版本get_metadata_field_unique_values特别适合为 UI 筛选器生成候选值列表例如按 category 字段展示去重后的所有可选分类。序列化与资源管理to_dict() - dict[str, Any]将存储序列化为字典连接信息、索引配置、距离度量等。from_dict(data: dict[str, Any]) - ValkeyDocumentStore从字典反序列化存储实例。close() - None释放关联的同步资源。close_async() - None释放关联的异步资源。四、过滤工具模块Haystack 过滤器到 Valkey Search 查询的转换haystack_integrations.document_stores.valkey.filters模块提供过滤工具将 Haystack 的过滤器格式转换为 Valkey Search 查询语法同时支持基于 tag 的精确匹配与基于数值的范围过滤并支持逻辑运算符组合。支持的过滤操作TagField标签字段、!、in、not in精确字符串匹配NumericField数值字段、!、、、、、in、not in数值比较逻辑运算符AND、OR用于组合多个条件。过滤语法示例# Simple equality filter filters {field: meta.category, operator: , value: tech} # Numeric range filter filters {field: meta.priority, operator: , value: 5} # List membership filter filters {field: meta.status, operator: in, value: [active, pending]} # Complex logical filter filters { operator: AND, conditions: [ {field: meta.category, operator: , value: tech}, {field: meta.priority, operator: , value: 3} ] }这段语法与 Haystack 核心的过滤器约定保持一致单条件为{field, operator, value}三键结构组合条件为{operator, conditions}结构。在 Haystack 核心仓库中对应的判读逻辑位于 haystack/document_stores/types/filter_policy.py 的is_comparison_filter()与is_logical_filter()辅助函数分别校验三键结构field/operator/value与逻辑结构operator/conditions这为跨存储Valkey、Elasticsearch、InMemory 等复用同一套过滤器语义提供了统一基础。五、实战在 RAG 管线中组合使用 Valkey 组件将上述 API 组合起来即可构建一个完整可运行的 RAG 查询管线。以下示例取自官方文档存储指南valkeydocumentstore.mdx展示了 Valkey 检索结果如何注入提示词并交给 LLM 生成回答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}, }, )配套索引管线数据入库侧可构建索引管线MarkdownToDocument转换器 →DocumentSplitter切分 →SentenceTransformersDocumentEmbedder嵌入 →DocumentWriter写入ValkeyDocumentStore全部由 Haystack 内置组件haystack/components 目录与集成嵌入器协作完成from haystack import Pipeline from haystack.components.converters import MarkdownToDocument from haystack.components.writers import DocumentWriter from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersDocumentEmbedder, ) from haystack.components.preprocessors import DocumentSplitter 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]}})六、性能特性与适用边界根据官方文档Valkey 作为内存存储具备以下特性内存存储读写操作快速高吞吐每秒可处理大量操作低延迟文档操作响应时间极小可扩展支持集群模式以横向扩展。需要说明的适用前提与限制均以当前仓库文档为准上述数据均来自官方文档陈述具体性能数值取决于部署硬件与数据集规模建议结合自身场景实测本文介绍的向量检索依赖运行了search 模块的 Valkey 服务器如valkey/valkey-bundle镜像普通 Valkey/Redis 实例不具备该能力文档写入的重复策略仅支持DuplicatePolicy.NONE与DuplicatePolicy.OVERWRITE使用SKIP或FAIL将不受支持filter_documents因 Valkey Search 要求向量查询内部使用 dummy 向量并剥离分数属于实现层面的折中结果score恒为None。七、进一步阅读Valkey 集成完整 API 参考docs-website/reference/integrations-api/valkey.md本文主体来源ValkeyDocumentStore 使用指南docs-website/docs/document-stores/valkeydocumentstore.mdxValkeyEmbeddingRetriever 使用指南docs-website/docs/pipeline-components/retrievers/valkeyembeddingretriever.mdx过滤策略FilterPolicy与apply_filter_policy实现haystack/document_stores/types/filter_policy.py重复策略DuplicatePolicy定义haystack/document_stores/types/policy.py文档存储协议DocumentStorehaystack/document_stores/types/init.py【免费下载链接】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），仅供参考