Haystack FalkorDB 集成实战指南用图数据库构建 GraphRAG 文档存储与检索【免费下载链接】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/haystackFalkorDB 是一个面向 GraphRAG 负载优化、内置 ANN 向量搜索的高性能图数据库。Haystack 通过falkordb-haystack集成包提供FalkorDBDocumentStore、FalkorDBCypherRetriever与FalkorDBEmbeddingRetriever三个核心组件让开发者可以在 Haystack 的 Pipeline 中原生组合图遍历、多跳查询与向量相似度检索。读完本文你将掌握 FalkorDB 文档存储的完整初始化参数与元数据操作 API、两种检索器的调用方式与安全要点并能独立搭建一个可运行的 GraphRAG / RAG 查询流水线。FalkorDB 集成在 Haystack 中的定位从 Haystack 官方文档的 选择文档存储指南 可以看到FalkorDB 属于核心集成Core Integration分类为「图数据库」引擎类型为「支持 ANN 向量搜索的 OpenCypher 图数据库」开源协议为 SSPL当前不支持异步提供的检索器为Embedding与Cypher两种。这意味着与纯粹面向向量的存储如 Qdrant、Chroma相比FalkorDB 的差异化价值在于文档不仅可以按向量相似度召回还可以借助图结构执行多跳multi-hop遍历——这正是 GraphRAG 场景中「从文档节点沿关系边探索关联实体与概念」的典型需求。该集成与neo4j-haystack参考集成共享相同的节点属性扁平化存储布局但无需安装 APOC 插件即可完成向量检索所有批量写入均通过 OpenCypher 的UNWINDMERGE实现安全的 upsert详见 API 参考 中FalkorDBDocumentStore一节。安装与启动 FalkorDB首先用 Docker 启动 FalkorDB 服务默认端口 6379与 Redis 一致docker run -d -p 6379:6379 falkordb/falkordb:latest安装 Haystack 集成包pip install falkordb-haystack安装完成后即可导入haystack_integrations.document_stores.falkordb与haystack_integrations.components.retrievers.falkordb两个模块。如果希望完整运行本文的管道示例还需按检索器类型安装对应的嵌入组件包# 运行 FalkorDBEmbeddingRetriever 的管道示例Sentence Transformers 嵌入器 pip install sentence-transformers-haystack # 运行 FalkorDBCypherRetriever 的管道示例Transformers 生成器 pip install transformers-haystackFalkorDBDocumentStore图数据库中的文档存储FalkorDBDocumentStore继承自 Haystack 的DocumentStore基类是整套集成的数据层。文档以图节点形式存储默认节点标签为Document位于指定的命名图graph内每个文档的属性包括meta字段都扁平化地与id、content存放在同一层级不加任何前缀——这一布局与neo4j-haystack参考集成完全一致方便在图查询中直接以属性形式访问元数据。初始化参数详解FalkorDBDocumentStore的构造函数全部为关键字参数默认值如下对应 API 参考 的__init__签名参数类型默认值说明hoststrlocalhostFalkorDB 服务器主机名portint6379FalkorDB 服务监听端口graph_namestrhaystack使用的图名称每个图都是相互隔离的命名空间usernamestr \| NoneNoneFalkorDB 认证用户名可选passwordSecret \| NoneNone持有 FalkorDB 密码的haystack.utils.Secret在首次连接时才惰性解析node_labelstrDocument文档节点在图中的标签embedding_dimint768向量维度用于创建向量索引embedding_fieldstrembedding存放嵌入向量的节点属性名similaritySimilarityFunctioncosine向量索引的相似度函数仅接受cosine或euclidean否则抛出ValueErrorwrite_batch_sizeint100每个UNWIND批次写入的文档数量recreate_graphboolFalse为True时在初始化阶段删除并重建现有图含全部数据常用于测试verify_connectivityboolFalse为True时在__init__中立即执行连通性探测服务器不可达则直接抛错embedding_dim与similarity在创建向量索引时生效因此必须在写入带向量的文档之前确定password使用Secret对象而不是明文推荐从环境变量注入见下文「认证」。写入与计数最基础的用法是初始化存储、写入文档并统计数量from haystack import Document from haystack_integrations.document_stores.falkordb import FalkorDBDocumentStore store FalkorDBDocumentStore(hostlocalhost, port6379) store.write_documents( [ Document(contentHello, GraphRAG!, meta{year: 2024}), ] ) print(store.count_documents()) # 1write_documents(documents, policyDuplicatePolicy.NONE)是批量写入入口底层使用UNWINDMERGE构造批处理 upsertwrite_batch_size控制每个批次的大小。policy参数决定遇到id已存在的文档时的行为DuplicatePolicy.NONE默认等价于 FAIL遇到重复 ID 抛出DuplicateDocumentErrorDuplicatePolicy.OVERWRITE覆盖已有文档DuplicatePolicy.SKIP跳过重复文档。此外该方法还会在documents含非Document元素时抛出ValueError在其他数据库错误时抛出DocumentStoreError。官方文档中的写入示例也常配合recreate_graphTrue使用确保每次运行从空图开始见 FalkorDBDocumentStore 指南。认证通过 Secret 传递密码连接受密码保护的 FalkorDB 实例时通过haystack.utils.Secret注入凭据避免在代码中硬编码from haystack.utils import Secret from haystack_integrations.document_stores.falkordb import FalkorDBDocumentStore document_store FalkorDBDocumentStore( hostlocalhost, port6379, passwordSecret.from_env_var(FALKORDB_PASSWORD), )Secret.from_env_var(FALKORDB_PASSWORD)会在首次建立连接时从环境变量解析密码值惰性求值这是官方推荐的安全写法见 认证小节。相似度函数选择向量索引支持两种相似度函数通过similarity参数指定cosine默认余弦相似度适合已归一化的嵌入向量euclidean欧氏距离当向量模长本身携带语义信息时更合适。document_store FalkorDBDocumentStore( hostlocalhost, port6379, embedding_dim768, similarityeuclidean, )需要说明的是similarity的值必须在合法集合内否则__init__直接抛出ValueError。如果你希望在写入文档前为它们计算真实嵌入可先通过 Document Embedder如SentenceTransformersDocumentEmbedder批量生成向量再把带embedding的文档写入存储。过滤、删除与更新完整的元数据操作 API除读写外FalkorDBDocumentStore还实现了覆盖检索、删除、更新、统计的一整套元数据操作接口全部可通过 Haystack 标准过滤器filter dict驱动方法签名要点行为filter_documents(filtersNone)filters: dict \| None返回匹配过滤器的文档传None返回全部过滤器语法遵循 Haystack 元数据过滤规范见 元数据过滤文档畸形过滤器抛出ValueErrordelete_documents(document_ids)document_ids: list[str]基于单条UNWIND查询按 ID 删除delete_all_documents()无参清空图中的全部文档delete_by_filter(filters)filters: dict删除匹配过滤器的文档返回删除数量update_by_filter(filters, meta)meta: dict更新匹配文档的元数据字段键可带或不带meta.前缀返回更新数量count_documents()无参返回图中文档节点总数count_documents_by_filter(filters)filters: dict返回匹配过滤器的文档数量count_unique_metadata_by_filter(filters, metadata_fields)metadata_fields: list[str]返回每个元数据字段在匹配文档中的唯一值数量键名不含meta.前缀get_metadata_fields_info()无参返回每个元数据字段的类型信息形如{field: {type: str}}类型取值为str/int/float/boolget_metadata_field_min_max(metadata_field)字段名可带或不带前缀返回{min: ..., max: ...}字段无非空值时为Noneget_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone)支持分页返回(values, total_count)元组values保留原始类型并支持search_term大小写不敏感子串过滤total_count为匹配过滤器的唯一值总数不受分页影响get_metadata_field_unique_values有一个值得注意的细节Python 中比较相等的不同类型值会被视为不同条目例如整数1、布尔True、字符串1会作为三个独立值返回唯一例外是 Cypher 的DISTINCT会把整数值浮点如1.0与数值相等的整数1折叠为同一个值带小数部分的浮点如1.5不受影响。close()方法用于释放底层文档存储持有的同步资源通常在管道或应用生命周期结束时调用Haystack 也支持通过组件资源生命周期机制自动管理。FalkorDBCypherRetriever任意 OpenCypher 查询检索FalkorDBCypherRetriever是面向高级用户的「动力型」检索器用于对 FalkorDB 执行任意 OpenCypher 查询从而在图遍历、多跳查询等 GraphRAG 场景中直接复用图结构。查询结果必须是能够精确映射为 HaystackDocument的节点或字典。安全警告务必先读原始 Cypher 查询只能来自可信来源。绝不要将未经清洗的用户输入直接拼接进查询字符串应始终使用parameters参数化查询详见 Cypher 检索器指南。这一点与 SQL 注入防御同理Cypher 字符串中的$param_name占位符由run(parameters{...})传入的值安全替换。初始化与运行from haystack_integrations.document_stores.falkordb import FalkorDBDocumentStore from haystack_integrations.components.retrievers.falkordb import FalkorDBCypherRetriever store FalkorDBDocumentStore(hostlocalhost, port6379) retriever FalkorDBCypherRetriever( document_storestore, custom_cypher_queryMATCH (d:Document)-[:RELATES_TO]-(:Concept {name: $concept}) RETURN d, ) res retriever.run(parameters{concept: GraphRAG}) print(res[documents])__init__签名__init__( document_store: FalkorDBDocumentStore, custom_cypher_query: str | None None, ) - Nonedocument_store必须传入FalkorDBDocumentStore实例否则抛出ValueErrorcustom_cypher_query初始化时设定的静态 OpenCypher 查询可在运行时被run()的query参数覆盖。run签名run( query: str | None None, parameters: dict[str, Any] | None None ) - dict[str, list[Document]]query可选 OpenCypher 查询字符串一旦传入即覆盖初始化时的custom_cypher_queryparameters查询参数字典对应 Cypher 串中的$param_name返回{documents: [...]}若初始化与运行时均未提供查询字符串抛出ValueError。在 GraphRAG 管道中使用以「按主题多跳检索 对话生成」为例官方文档给出了完整管道示例。先用FalkorDBCypherRetriever执行带参数的图查询再把命中的文档交给ChatPromptBuilder拼装提示词最后用TransformersChatGenerator生成回答from haystack import Document, Pipeline from haystack.components.builders import ChatPromptBuilder from haystack.dataclasses import ChatMessage from haystack_integrations.document_stores.falkordb import FalkorDBDocumentStore from haystack_integrations.components.retrievers.falkordb import FalkorDBCypherRetriever from haystack_integrations.components.generators.transformers import ( TransformersChatGenerator, ) document_store FalkorDBDocumentStore( hostlocalhost, port6379, recreate_graphTrue, ) document_store.write_documents( [ Document(contentThere are over 7,000 languages spoken around the world today., meta{topic: linguistics}), Document(contentElephants have been observed to recognize themselves in mirrors., meta{topic: biology}), ], ) prompt_template [ ChatMessage.from_user( Given these documents, answer the question. Documents: {% for doc in documents %} {{ doc.content }} {% endfor %} Question: {{ question }}, ), ] pipeline Pipeline() pipeline.add_component( retriever, FalkorDBCypherRetriever( document_storedocument_store, custom_cypher_queryMATCH (d:Document {topic: $topic}) RETURN d, ), ) pipeline.add_component(prompt_builder, ChatPromptBuilder(templateprompt_template)) pipeline.add_component( llm, TransformersChatGenerator(modelHuggingFaceTB/SmolLM2-135M-Instruct), ) pipeline.connect(retriever.documents, prompt_builder.documents) pipeline.connect(prompt_builder.prompt, llm.messages) result pipeline.run( { retriever: {parameters: {topic: linguistics}}, prompt_builder: {question: How many languages are there?}, }, ) print(result[llm][replies][0].text)该组件在管道中最常见的位置是「查询构建组件之后、PromptBuilder之前」必备初始化变量为document_store必备运行变量为query或初始化时设置custom_cypher_query输出变量为documents列表见 Cypher 检索器文档。注意这里的图是「扁平」的文档节点上的topic属性就是写入时meta的扁平化字段因此 Cypher 可以直接用{topic: $topic}匹配——这正体现了「meta 平铺」布局对图查询的友好性。FalkorDBEmbeddingRetriever原生向量相似度检索FalkorDBEmbeddingRetriever使用 FalkorDB 的原生向量索引将查询嵌入与文档嵌入比对并返回最相似的文档。与 Cypher 检索器不同它不需要你手写查询语句适合标准 RAG 与语义搜索管道。独立使用from haystack.dataclasses import Document from haystack_integrations.document_stores.falkordb import FalkorDBDocumentStore from haystack_integrations.components.retrievers.falkordb import FalkorDBEmbeddingRetriever store FalkorDBDocumentStore(hostlocalhost, port6379) store.write_documents( [ Document(contentGraphRAG is powerful., embedding[0.1, 0.2, 0.3]), Document(contentFalkorDB is fast., embedding[0.8, 0.9, 0.1]), ] ) retriever FalkorDBEmbeddingRetriever(document_storestore) res retriever.run(query_embedding[0.1, 0.2, 0.3]) print(res[documents][0].content) # GraphRAG is powerful.__init__签名__init__( document_store: FalkorDBDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: FilterPolicy FilterPolicy.REPLACE, ) - Nonefilters初始化时可选的 Haystack 过滤器用于收窄搜索空间top_k最大召回文档数默认 10filter_policy运行时过滤器与初始化过滤器如何组合的策略默认FilterPolicy.REPLACE运行时覆盖初始化过滤器传入非FalkorDBDocumentStore时抛出ValueError。run签名run( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]query_embedding查询嵌入向量浮点列表必备运行变量filters运行时过滤器按filter_policy与初始化过滤器合并top_k不传则使用初始化时的默认值返回{documents: [...]}。在 RAG 管道中使用嵌入器 检索器官方文档给出的完整语义检索管道如下先用SentenceTransformersDocumentEmbedder为文档批量生成嵌入并以DuplicatePolicy.OVERWRITE写入再构建「文本嵌入器 → 向量检索器」的查询管道from haystack import Document, Pipeline from haystack.document_stores.types import DuplicatePolicy from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder, ) from haystack_integrations.document_stores.falkordb import FalkorDBDocumentStore from haystack_integrations.components.retrievers.falkordb import ( FalkorDBEmbeddingRetriever, ) document_store FalkorDBDocumentStore( hostlocalhost, port6379, embedding_dim384, recreate_graphTrue, ) documents [ Document(contentThere are over 7,000 languages spoken around the world today.), Document(contentElephants have been observed to recognize themselves in mirrors.), Document(contentBioluminescent waves can be seen in the Maldives and Puerto Rico.), ] document_embedder SentenceTransformersDocumentEmbedder( modelsentence-transformers/all-MiniLM-L6-v2, ) documents_with_embeddings document_embedder.run(documents) document_store.write_documents( documents_with_embeddings[documents], policyDuplicatePolicy.OVERWRITE, ) query_pipeline Pipeline() query_pipeline.add_component( text_embedder, SentenceTransformersTextEmbedder(modelsentence-transformers/all-MiniLM-L6-v2), ) query_pipeline.add_component( retriever, FalkorDBEmbeddingRetriever(document_storedocument_store, top_k3), ) query_pipeline.connect(text_embedder.embedding, retriever.query_embedding) result query_pipeline.run( {text_embedder: {text: How many languages are there?}}, ) print(result[retriever][documents][0].content)注意FalkorDBDocumentStore(embedding_dim384)与all-MiniLM-L6-v2的输出维度384 维必须一致因为向量索引在存储初始化时创建。该组件在管道中最常见的两种位置是① RAG 管道中位于 Text Embedder 之后、PromptBuilder之前② 语义搜索管道中的最后一个组件见 Embedding 检索器文档。序列化与资源管理三个组件均实现了完整的序列化协议便于在 Haystack 中保存、传输与重建to_dict() - dict[str, Any]将检索器/存储序列化为字典from_dict(data: dict[str, Any]) - 对应组件由to_dict产出的字典重建实例FalkorDBCypherRetriever、FalkorDBEmbeddingRetriever、FalkorDBDocumentStore各自实现close() - None释放底层同步资源。这些方法遵循 Haystack 组件统一的反序列化安全规范是 Pipeline 序列化、断点调试与远端部署的基础设施。总结三种组件如何协同falkordb-haystack集成围绕「图数据库 向量索引」双引擎设计形成了完整的分层能力数据层FalkorDBDocumentStore负责文档的批量 upsertUNWINDMERGE、扁平化元数据存储、向量索引创建与全套过滤/删除/统计 API图查询层FalkorDBCypherRetriever通过参数化 OpenCypher 实现任意图遍历与多跳检索是 GraphRAG 的核心入口但必须遵守「查询只来自可信来源」的安全红线语义召回层FalkorDBEmbeddingRetriever借助原生向量索引完成相似度检索配合top_k、filters与filter_policy精准控制召回范围适合标准 RAG 与语义搜索。从官方 检索器索引表 可以确认这两类检索器Embedding 与 Cypher正是 FalkorDB 区别于纯向量库的核心组合。需要进一步深入时可查阅本文引用的 FalkorDB API 参考、文档存储指南、Cypher 检索器指南 与 Embedding 检索器指南 四份文档它们分别覆盖了完整签名、安装步骤与可运行的管道示例。【免费下载链接】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），仅供参考
