企业官网建设流程全解析

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/haystack

FalkorDB 是一个面向 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 的UNWIND+MERGE实现安全的 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-haystack

FalkorDBDocumentStore：图数据库中的文档存储

FalkorDBDocumentStore继承自 Haystack 的DocumentStore基类，是整套集成的数据层。文档以图节点形式存储（默认节点标签为Document），位于指定的命名图（graph）内；每个文档的属性，包括meta字段，都扁平化地与id、content存放在同一层级，不加任何前缀——这一布局与neo4j-haystack参考集成完全一致，方便在图查询中直接以属性形式访问元数据。

初始化参数详解

FalkorDBDocumentStore的构造函数全部为关键字参数，默认值如下（对应 API 参考 的__init__签名）：

embedding_dim与similarity在创建向量索引时生效，因此必须在写入带向量的文档之前确定；password使用Secret对象而不是明文，推荐从环境变量注入（见下文「认证」）。

写入与计数

最基础的用法是初始化存储、写入文档并统计数量：

from haystack import Document from haystack_integrations.document_stores.falkordb import FalkorDBDocumentStore store = FalkorDBDocumentStore(host="localhost", port=6379) store.write_documents( [ Document(content="Hello, GraphRAG!", meta={"year": 2024}), ] ) print(store.count_documents()) # 1

write_documents(documents, policy=DuplicatePolicy.NONE)是批量写入入口，底层使用UNWIND+MERGE构造批处理 upsert：write_batch_size控制每个批次的大小。policy参数决定遇到id已存在的文档时的行为：

• DuplicatePolicy.NONE（默认，等价于 FAIL）：遇到重复 ID 抛出DuplicateDocumentError；
• DuplicatePolicy.OVERWRITE：覆盖已有文档；
• DuplicatePolicy.SKIP：跳过重复文档。

此外该方法还会在documents含非Document元素时抛出ValueError，在其他数据库错误时抛出DocumentStoreError。官方文档中的写入示例也常配合recreate_graph=True使用，确保每次运行从空图开始（见 FalkorDBDocumentStore 指南）。

认证：通过 Secret 传递密码

连接受密码保护的 FalkorDB 实例时，通过haystack.utils.Secret注入凭据，避免在代码中硬编码：

from haystack.utils import Secret from haystack_integrations.document_stores.falkordb import FalkorDBDocumentStore document_store = FalkorDBDocumentStore( host="localhost", port=6379, password=Secret.from_env_var("FALKORDB_PASSWORD"), )

Secret.from_env_var("FALKORDB_PASSWORD")会在首次建立连接时从环境变量解析密码值（惰性求值），这是官方推荐的安全写法（见 认证小节）。

相似度函数选择

向量索引支持两种相似度函数，通过similarity参数指定：

• "cosine"（默认）：余弦相似度，适合已归一化的嵌入向量；
• "euclidean"：欧氏距离，当向量模长本身携带语义信息时更合适。

document_store = FalkorDBDocumentStore( host="localhost", port=6379, embedding_dim=768, similarity="euclidean", )

需要说明的是，similarity的值必须在合法集合内，否则__init__直接抛出ValueError。如果你希望在写入文档前为它们计算真实嵌入，可先通过 Document Embedder（如SentenceTransformersDocumentEmbedder）批量生成向量，再把带embedding的文档写入存储。

过滤、删除与更新：完整的元数据操作 API

除读写外，FalkorDBDocumentStore还实现了覆盖检索、删除、更新、统计的一整套元数据操作接口，全部可通过 Haystack 标准过滤器（filter dict）驱动：

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(host="localhost", port=6379) retriever = FalkorDBCypherRetriever( document_store=store, custom_cypher_query="MATCH (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, ) -> None

• document_store：必须传入FalkorDBDocumentStore实例，否则抛出ValueError；
• custom_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_query；
• parameters：查询参数字典，对应 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( host="localhost", port=6379, recreate_graph=True, ) document_store.write_documents( [ Document(content="There are over 7,000 languages spoken around the world today.", meta={"topic": "linguistics"}), Document(content="Elephants 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_store=document_store, custom_cypher_query="MATCH (d:Document {topic: $topic}) RETURN d", ), ) pipeline.add_component("prompt_builder", ChatPromptBuilder(template=prompt_template)) pipeline.add_component( "llm", TransformersChatGenerator(model="HuggingFaceTB/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(host="localhost", port=6379) store.write_documents( [ Document(content="GraphRAG is powerful.", embedding=[0.1, 0.2, 0.3]), Document(content="FalkorDB is fast.", embedding=[0.8, 0.9, 0.1]), ] ) retriever = FalkorDBEmbeddingRetriever(document_store=store) 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, ) -> None

• filters：初始化时可选的 Haystack 过滤器，用于收窄搜索空间；
• top_k：最大召回文档数，默认 10；
• filter_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( host="localhost", port=6379, embedding_dim=384, recreate_graph=True, ) documents = [ Document(content="There are over 7,000 languages spoken around the world today."), Document(content="Elephants have been observed to recognize themselves in mirrors."), Document(content="Bioluminescent waves can be seen in the Maldives and Puerto Rico."), ] document_embedder = SentenceTransformersDocumentEmbedder( model="sentence-transformers/all-MiniLM-L6-v2", ) documents_with_embeddings = document_embedder.run(documents) document_store.write_documents( documents_with_embeddings["documents"], policy=DuplicatePolicy.OVERWRITE, ) query_pipeline = Pipeline() query_pipeline.add_component( "text_embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"), ) query_pipeline.add_component( "retriever", FalkorDBEmbeddingRetriever(document_store=document_store, top_k=3), ) 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_dim=384)与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集成围绕「图数据库 + 向量索引」双引擎设计，形成了完整的分层能力：

1. 数据层：FalkorDBDocumentStore负责文档的批量 upsert（UNWIND+MERGE）、扁平化元数据存储、向量索引创建与全套过滤/删除/统计 API；
2. 图查询层：FalkorDBCypherRetriever通过参数化 OpenCypher 实现任意图遍历与多跳检索，是 GraphRAG 的核心入口，但必须遵守「查询只来自可信来源」的安全红线；
3. 语义召回层：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