GraphRAG-SDK：实践指南

作者：袖梨 2026-08-21

实际评估GraphRAG-SDK时，我先确认它解决的具体问题：使用 GraphRAG SDK 大规模构建快速、准确的 GenAI 应用程序。把它放到日常自动化流程中，最容易暴露的是输入边界、依赖和失败处理如果不清楚就很难稳定复用，不能只看演示是否顺利。落地前可以用一项范围明确的真实任务完成最小试跑，用配置时间、输出质量、异常信息和维护痕迹判断它是否真的省事。我会把它列入愿意先做小范围验证并复查原始文档的团队的候选清单，而不是仅凭项目介绍直接纳入生产。

GraphRAG-SDK

该最简单、最准确的 GraphRAG 框架构建在 FalkorDB

上

Benchmark-leading accuracy · FalkorDB-fast · Multi-tenant · Graph traversal · 5- 分钟 setup

大多数 GraphRAG 系统都在演示中工作，并在生产限制下崩溃。 GraphRAG SDK 是根据实际部署构建的，围绕一个简单的想法：检索工具比模型更重要。结果是一个模块化的、领先基准的框架，具有可预测的成本和合理的默认值，可让您在 5 分钟内从原始文档到引用的答案。

基准测试

排名	系统	小说（多文档）	医疗（单文档）	整体
1	FalkorDB GraphRAG SDK ◄	66.09	76.87	71.48
2	G-推理者	58.94	73.30	66.12
3	AutoPrunedRetriever	63.72	67.00	65.36
4	HippoRAG2	56.48	64.85	60.67
5	快速-GraphRAG	52.02	64.12	58.07
6	RAG（重新排序）（向量 RAG）	48.35	62.43	55.39
7	LightRAG	45.09	62.59	53.84
8	HippoRAG	44.75	59.08	51.92
9	MS-GraphRAG（本地）	50.93	45.16	48.05

这些是如何计算的。 每个数据集，ACC 是四个数据集的未加权平均值任务类别分数，匹配 GraphRAG-Bench 排行榜约定：

Dataset ACC = (Fact Retrieval + Complex Reasoning + Contextual Summarize + Creative Generation) / 4 Overall = (Novel ACC + Medical ACC) / 2

总体而言，这是我们自己对两个数据集的总结；排行榜分别排名单独的数据集。小说有 20 个文档和 2,010 个问题，医学 1 个语料库和 2,062 个问题。 FalkorDB 于 2026 年 8 月得分为 gpt-4o-mini (Azure OpenAI) 图构建和生成的温度为 0.7，text-embedding-3-large 1024 维，启用文本到密码检索，以及基准测试自己的 generation_eval.py 未修改为判断。竞争对手的号码来自已发布排行榜，未更改。请参阅基准页面按类别结果，完整的 15 个系统比较、配置和再现说明。

向量匹配相似的块。该图遍历关系。每个答案都引用其来源。

快速入门

1.安装并启动FalkorDB

pip install graphrag-sdk[litellm]
docker run -d -p 6379:6379 -p 3000:3000 --name falkordb falkordb/falkordb:latest
export OPENAI_API_KEY="sk-..."

对于 PDF 摄取，请安装 pdf 额外组件：pip install graphrag-sdk[litellm,pdf]。在图形更新插入之前，摄取会清理 IDs 和字符串属性中不支持的控制字符，这有助于避免嘈杂的 PDFs 上的 FalkorDB Cypher 解析错误。

2. 摄取文档

import asyncio
from graphrag_sdk import GraphRAG, ConnectionConfig, LiteLLM, LiteLLMEmbedder

async def main():
    async with GraphRAG(
        connection=ConnectionConfig(host="localhost", graph_name="my_graph"),  # graph_name = per-tenant isolation
        llm=LiteLLM(model="openai/gpt-5.5"),
        embedder=LiteLLMEmbedder(model="openai/text-embedding-3-large", dimensions=256),
    ) as rag:
        # Ingest raw text (pass a file path with the `pdf` extra installed for PDFs)
        result = await rag.ingest(
            text="Alice Johnson is a software engineer at Acme Corp in London.",
            document_id="my_doc",
        )
        print(f"Nodes: {result.nodes_created}, Edges: {result.relationships_created}")

        # Finalize: deduplicate entities, backfill embeddings, create indexes
        await rag.finalize()

        # Full RAG: retrieve + generate
        answer = await rag.completion("Where does Alice work?")
        print(answer.answer)

asyncio.run(main())

3. 定义模式（可选）

from graphrag_sdk import GraphSchema, EntityType, RelationType

schema = GraphSchema(
    entities=[
        EntityType(label="Person", description="A human being"),
        EntityType(label="Organization", description="A company or institution"),
        EntityType(label="Location", description="A geographic location"),
    ],
    relations=[
        RelationType(label="WORKS_AT", description="Is employed by", patterns=[("Person", "Organization")]),
        RelationType(label="LOCATED_IN", description="Is situated in", patterns=[("Organization", "Location")]),
    ],
)

async with GraphRAG(
    connection=ConnectionConfig(host="localhost", graph_name="my_graph"),
    llm=LiteLLM(model="openai/gpt-5.5"),
    embedder=LiteLLMEmbedder(model="openai/text-embedding-3-large", dimensions=256),
    schema=schema,
) as rag:
    ...  # ingest / completion as above

→ 完整演练：Getting Started

→ 编写自己的管道：Custom Strategies

增量更新 (v1.1.0)

重新同步各个文档而不重建图表。规范的 CI 用例正在更新 PR 合并上的图表 - 添加、修改和批量删除文件：

async with GraphRAG(connection=ConnectionConfig(...), llm=..., embedder=...) as graph:
    result = await graph.apply_changes(
        added=["docs/new_feature.md"],
        modified=["docs/api.md"],
        deleted=["docs/removed_page.md"],
    )
    await graph.finalize()  # once per batch — finalize is O(graph size)

    # Per-file outcomes are wrapped in BatchEntry — the batch never raises.
    for entry in result.added + result.modified + result.deleted:
        if not entry.is_success:
            print(f"failed: {entry.error_type}: {entry.error}")

包装器背后的三个原语：

方法	何时使用
`update(source, document_id=...)`	文档内容已更改。 SHA-256 哈希短路无操作更新（仅触摸 PRs 成本约为 1 个 Cypher 查询）。传递 `if_missing="ingest"` 以获得更新插入语义。
`delete_document(document_id)`	文档已删除。清理因删除而孤立的实体；保留其他文档仍引用的实体。
`apply_changes(added=..., modified=..., deleted=...)`	异质批次。每个文件的错误都会被收集，而不是引发。不调用 `finalize()` — 调用者驱动该节奏。

在文件模式下，document_id 默认为 os.path.normpath(source) 所以 update("docs/x.md") 与原始 ingest("docs/x.md") 匹配没有额外的管道。请参见 examples/07_incremental_updates.py。

成本模型。 finalize() 运行跨文档重复数据删除，这扫描完整的实体表 - 其成本是 O(图形大小)，而不是 O（改变尺寸）。在 finalize() 中嵌入回填是 O(change 大小）（仅触及 nodes/edges 缺失的嵌入）。对于CI使用情况下，通过 apply_changes 批处理所有 PR 更改并调用 finalize 在运行结束时一次，不是每个文件 — 每个文件 finalize 将 dedup 常量乘以文件数感动。

碰撞安全。 update() 使用幂等前滚切换：新内容写入 __pending__ 文档，然后单个原子Cypher语句标记为ready_to_commit=true，则直播文档被替换。标记丢弃待处理之前发生崩溃重试时；标记向前滚动完成后发生崩溃。要么方式，重试相同的 update() 调用是安全的并且收敛于正确的最终状态。

并发性。 apply_changes 公开两个旋钮：max_concurrency （添加，默认3）和update_concurrency（修改，默认1）。更新默认为 1，因为孤儿清理正确性低于并发更新取决于管道排序不变量；提高仅当您验证了并发更新时，该默认值才是安全的永远不能共享一个实体。集成测试 test_concurrent_updates_preserve_shared_entity 是绊线保护默认值。

摄取和回收管道

面积	步骤	成本
食入	提取实体和关系	LLM
食入	解析和删除重复实体	LLM
食入	嵌入和索引	LLM
检索	矢量搜索	DB
检索	全文检索	DB
检索	文本到密码（实验）	LLM
检索	密码查询	DB
检索	关系拓展	DB
检索	余弦重新排序	本地

每个答案都可以通过 MENTIONS 边追溯到其源块。将 return_context=True 传递给 completion() 以获取检索轨迹和答案。

示例

工作启动器 - 克隆、插入源代码、发布。

#	示例	你将构建什么
1	快速入门	您的第一个摄取和查询循环不到 30 行
2	PDF 与架构	具有您自己的实体和关系类型的 PDF 问答机器人
3	自定义策略	明确地制定摄取策略
4	定制提供商	在干净的接口后面插入任何 LLM 或嵌入器
5	笔记本演示	显示出处轨迹的交互式演练
7	增量更新	`update`、`delete_document` 和 `apply_changes` 用于 CI- 驱动的图形同步

文档

完整文档：https://docs.falkordb.com/graphrag

指南	描述
入门	从安装到首次查询的分步教程
架构	管道设计、图模式、检索策略
配置	连接、提供程序和调整参考
策略	所有 ABCs 和内置实现
提供商	LLM 和嵌入器配置指南
基准	方法、结果和复制说明
API 参考号	完整的 API 文档