agentic_huge_data_base / wiki
页面 Graphiti · 7.2 Index 与 Constraint 管理·DeepWiki 中文全文译文
DeepWiki 中文译文 / Graphiti / 7.2

7.2 · Index 与 Constraint 管理（Index and Constraint Management）

时序知识图谱与动态事实记忆 · 本章是 Graphiti DeepWiki 中文译文的独立章节页，保留原始链接、源码锚点、模块标签和章节层级。

项目Graphiti 章节7.2 状态全文译文 模块图谱与关系、检索、召回与索引、测试、发布与运维、工作流与编排
项目要点页2.5 参考项目 项目章节目录Graphiti DeepWiki 原始章节Index and Constraint Management 上一章7.1 下一章7.3
源码线索
  • graphiti_core/driver/driver.py
  • graphiti_core/driver/falkordb_driver.py
  • graphiti_core/driver/neo4j_driver.py
  • graphiti_core/graph_queries.py
  • graphiti_core/helpers.py
  • graphiti_core/models/edges/edge_db_queries.py
  • graphiti_core/models/nodes/node_db_queries.py
模块标签
  • 图谱与关系
  • 检索、召回与索引
  • 测试、发布与运维
  • 工作流与编排
  • 模型调用与提供方适配

中文译文

Index 与 Constraint 管理（中文译文）

原始 DeepWiki 页面：https://deepwiki.com/getzep/graphiti/7.2-index-and-constraint-management
翻译时间：2026-05-27T08:44:57.721Z
翻译模型：deepseek-chat
原文字符数：13137
项目：Graphiti (graphiti)

---

索引与约束管理

相关源文件

以下文件用于生成此维基页面：

  • graphiti_core/driver/driver.py
  • graphiti_core/driver/falkordb_driver.py
  • graphiti_core/driver/neo4j_driver.py
  • graphiti_core/graph_queries.py
  • graphiti_core/helpers.py
  • graphiti_core/models/edges/edge_db_queries.py
  • graphiti_core/models/nodes/node_db_queries.py

本文档描述了 Graphiti 如何在不同图数据库提供商之间创建和管理数据库索引与约束。索引对查询性能至关重要，能够通过节点属性、标签、时间字段和全文搜索实现高效查找。

目的与范围

Graphiti 中的索引和约束服务于以下几个目的：

  1. 范围索引：支持对 uuid、created_at、valid_at、group_id 和时间字段等属性的快速查找。
  2. 全文索引：为 name、summary 和 fact 等字段提供 BM25 文本搜索能力。
  3. 向量索引：支持嵌入向量字段上的余弦相似度搜索（取决于数据库提供商）。
  4. 约束：在支持的数据库中强制实施唯一性和数据完整性。

每个数据库提供商（Neo4j、FalkorDB、Kuzu、Neptune）在创建和管理这些结构时具有不同的能力和语法。

来源： graphiti_core/driver/driver.py:128-129, graphiti_core/driver/driver.py:53-56

索引类型及其用途

Graphiti 使用两种主要的索引类别，这些类别从集中的 graph_queries.py 模块中获取。

索引类型用途应用对象性能影响
范围索引属性相等性和范围查询uuid、group_id、时间字段（created_at、valid_at、invalid_at、expired_at）对节点/边查找和时间过滤至关重要
全文索引带分词的 BM25 文本搜索name（实体、社区）、summary、fact（RELATES_TO）、content（情节性）支持结合语义和关键词检索的混合搜索
按节点和边类型的索引覆盖范围
graph TB
    subgraph "节点索引"
        EntityNode["EntityNode"]
        EpisodicNode["EpisodicNode"]
        CommunityNode["CommunityNode"]
        SagaNode["SagaNode"]

        EntityNode --> |"范围: uuid, group_id, created_at"| RangeIdx1["范围索引"]
        EntityNode --> |"全文: name, summary"| FulltextIdx1["全文索引"]

        EpisodicNode --> |"范围: uuid, group_id, created_at, valid_at"| RangeIdx2["范围索引"]
        EpisodicNode --> |"全文: content, name"| FulltextIdx2["全文索引"]

        CommunityNode --> |"范围: uuid, group_id"| RangeIdx3["范围索引"]
        CommunityNode --> |"全文: name, summary"| FulltextIdx3["全文索引"]

        SagaNode --> |"范围: uuid, group_id, name"| RangeIdx4["范围索引"]
    end

    subgraph "边索引"
        EntityEdge["EntityEdge (RELATES_TO)"]
        EpisodicEdge["EpisodicEdge (MENTIONS)"]

        EntityEdge --> |"范围: uuid, group_id, created_at, valid_at, invalid_at, expired_at"| RangeIdx5["范围索引"]
        EntityEdge --> |"全文: name, fact"| FulltextIdx4["全文索引"]

        EpisodicEdge --> |"范围: uuid, group_id"| RangeIdx6["范围索引"]
    end

来源： graphiti_core/graph_queries.py:28-82, graphiti_core/graph_queries.py:85-140

索引创建工作流

所有驱动程序遵循一致的初始化模式，在驱动构造期间异步创建索引。

自然语言空间到代码实体空间：初始化
graph TD
    User["用户应用程序"]
    DriverInit["Neo4jDriver.__init__ / FalkorDriver.__init__"]
    BuildIndices["GraphDriver.build_indices_and_constraints"]
    GetQueries["get_range_indices / get_fulltext_indices"]
    DB["图数据库 (Neo4j/FalkorDB/Kuzu)"]

    User -- "实例化" --> DriverInit
    DriverInit -- "asyncio.create_task" --> BuildIndices
    BuildIndices -- "获取 Cypher 查询" --> GetQueries
    BuildIndices -- "在数据库上执行" --> DB

来源： graphiti_core/driver/neo4j_driver.py:92-101, graphiti_core/driver/falkordb_driver.py:160-168

序列图：异步索引构建
sequenceDiagram
    participant User
    participant Driver as "GraphDriver (Neo4j/Falkor/Kuzu)"
    participant Loop as "asyncio 事件循环"
    participant GraphQueries as "graph_queries 模块"
    participant DB as "数据库"

    User->>Driver: __init__(uri, credentials)
    Driver->>Driver: 初始化连接
    Driver->>Driver: 实例化操作类

    Note over Driver,Loop: 尝试调度索引创建
    Driver->>Loop: get_running_loop()

    alt 事件循环正在运行
        Loop-->>Driver: 当前循环
        Driver->>Loop: create_task(build_indices_and_constraints())

        Loop->>Driver: build_indices_and_constraints()
        opt delete_existing=True
            Driver->>DB: delete_all_indexes()
        end

        Driver->>GraphQueries: get_range_indices(provider)
        GraphQueries-->>Driver: 范围索引查询列表

        Driver->>GraphQueries: get_fulltext_indices(provider)
        GraphQueries-->>Driver: 全文索引查询列表

        Driver->>DB: 执行每个索引查询
        DB-->>Driver: 成功（或已存在）

    else 无事件循环
        Loop-->>Driver: RuntimeError
        Note over Driver: 索引创建推迟到首次查询时
    end

    Driver-->>User: 驱动实例

来源： graphiti_core/driver/neo4j_driver.py:92-101, graphiti_core/driver/falkordb_driver.py:160-168

各提供商实现细节

Neo4j 实现

Neo4j 驱动程序使用 CREATE INDEX IF NOT EXISTS 语句创建索引。它使用 semaphore_gather 来限制并发索引创建任务的数量。

graph TB
    BuildIndices["build_indices_and_constraints()"]
    DeleteOpt["delete_existing=True?"]
    DeleteAll["delete_all_indexes()"]
    GetRange["get_range_indices(NEO4J)"]
    GetFulltext["get_fulltext_indices(NEO4J)"]
    ExecuteQueries["semaphore_gather(*queries)"]
    ExecuteIndex["_execute_index_query(query)"]
    CheckError["ClientError?"]
    CheckEquiv["EquivalentSchemaRuleAlreadyExists?"]
    LogIgnore["记录日志并忽略（竞态条件）"]
    Raise["抛出异常"]
    Success["返回 None"]

    BuildIndices --> DeleteOpt
    DeleteOpt -->|是| DeleteAll
    DeleteOpt -->|否| GetRange
    DeleteAll --> GetRange
    GetRange --> GetFulltext
    GetFulltext --> ExecuteQueries
    ExecuteQueries --> ExecuteIndex
    ExecuteIndex --> CheckError
    CheckError -->|否| Success
    CheckError -->|是| CheckEquiv
    CheckEquiv -->|是| LogIgnore
    CheckEquiv -->|否| Raise
    LogIgnore --> Success

_execute_index_query() 方法会捕获当多个并发的 CREATE INDEX IF NOT EXISTS 查询发生竞态时出现的 EquivalentSchemaRuleAlreadyExists 错误。由于所需索引已存在，因此可以安全地忽略此错误。

来源： graphiti_core/driver/neo4j_driver.py:206-216, graphiti_core/driver/neo4j_driver.py:191-204, graphiti_core/helpers.py:123-133

FalkorDB 实现

FalkorDB 使用 CALL db.idx.fulltext.createNodeIndex 进行全文搜索。它在初始化期间包含自动重复检测。

graph TB
    BuildIndices["build_indices_and_constraints()"]
    DeleteOpt["delete_existing=True?"]
    DeleteAll["delete_all_indexes()"]
    GetIndices["get_range_indices(FALKORDB) +<br/>get_fulltext_indices(FALKORDB)"]
    LoopQueries["对每个查询"]
    ExecuteQuery["execute_query(query)"]
    CheckError["异常?"]
    CheckAlreadyIndexed["错误信息包含 'already indexed'?"]
    LogInfo["记录日志：索引已存在"]
    RaiseError["记录错误并抛出"]
    Success["继续"]

    BuildIndices --> DeleteOpt
    DeleteOpt -->|是| DeleteAll
    DeleteOpt -->|否| GetIndices
    DeleteAll --> GetIndices
    GetIndices --> LoopQueries
    LoopQueries --> ExecuteQuery
    ExecuteQuery --> CheckError
    CheckError -->|否| Success
    CheckError -->|是| CheckAlreadyIndexed
    CheckAlreadyIndexed -->|是| LogInfo
    CheckAlreadyIndexed -->|否| RaiseError
    LogInfo --> Success

FalkorDB 的 execute_query() 方法会捕获包含 'already indexed' 的异常，并将其记录为信息性消息而非错误。

来源： graphiti_core/driver/falkordb_driver.py:299-304, graphiti_core/driver/falkordb_driver.py:228-236, graphiti_core/graph_queries.py:86-121

Kuzu 实现

Kuzu 使用 CALL CREATE_FTS_INDEX 创建全文索引。对于 Kuzu，范围索引目前不会通过驱动程序显式创建，因为它使用不同的索引策略。

来源： graphiti_core/graph_queries.py:123-129, graphiti_core/graph_queries.py:51-52

删除和重建索引

每个驱动程序都提供 delete_all_indexes() 方法用于清理操作。

Neo4j 索引删除

Neo4j 使用系统调用来删除所有索引： CALL db.indexes() YIELD name DROP INDEX name

来源： graphiti_core/driver/neo4j_driver.py:186-189

FalkorDB 索引删除

FalkorDB 需要检查索引元数据，并为 RANGE 和 FULLTEXT 类型生成特定于提供商的 DROP 命令。

graph TB
    DeleteAll["delete_all_indexes()"]
    CallDBIndexes["execute_query('CALL db.indexes()')"]
    CheckResult["结果存在?"]
    ParseRecords["对每条记录"]
    ExtractMeta["提取 label, entitytype, field, index_type"]
    CheckRange["索引类型: RANGE?"]
    CheckFulltext["索引类型: FULLTEXT?"]
    CheckEntity["实体类型?"]
    DropRange["DROP INDEX ON :label(field)"]
    DropFTNode["DROP FULLTEXT INDEX FOR (n:label) ON (n.field)"]
    DropFTEdge["DROP FULLTEXT INDEX FOR ()-[e:label]-() ON (e.field)"]
    GatherDrop["asyncio.gather(*drop_tasks)"]

    DeleteAll --> CallDBIndexes
    CallDBIndexes --> CheckResult
    CheckResult -->|否| Return["返回"]
    CheckResult -->|是| ParseRecords
    ParseRecords --> ExtractMeta
    ExtractMeta --> CheckRange
    CheckRange -->|是| DropRange
    CheckRange -->|否| CheckFulltext
    CheckFulltext -->|是| CheckEntity
    CheckEntity -->|NODE| DropFTNode
    CheckEntity -->|RELATIONSHIP| DropFTEdge
    DropRange --> GatherDrop
    DropFTNode --> GatherDrop
    DropFTEdge --> GatherDrop
    GatherDrop --> Return

来源： graphiti_core/driver/falkordb_driver.py:267-297

索引名称的环境变量

索引名称可以通过环境变量自定义：

环境变量默认值用途
ENTITY_INDEX_NAMEentities实体节点全文索引的名称
EPISODE_INDEX_NAMEepisodes情节节点全文索引的名称
COMMUNITY_INDEX_NAMEcommunities社区节点全文索引的名称
ENTITY_EDGE_INDEX_NAMEentity_edges实体边全文索引的名称

来源： graphiti_core/driver/driver.py:53-56

全文查询构建与消毒

由于 FalkorDB 使用 RedisSearch 后端，因此需要特殊的全文查询语法。驱动程序提供了 build_fulltext_query() 方法，将用户查询转换为 RedisSearch 格式。

代码实体空间：全文搜索处理
graph LR
    UserQuery["搜索查询"]
    Sanitize["lucene_sanitize / helpers.py"]
    FalkorBuild["FalkorDriver.build_fulltext_query"]
    Stopwords["STOPWORDS / driver/falkordb/__init__.py"]
    DB["FalkorDB 查询引擎"]

    UserQuery --> Sanitize
    Sanitize --> FalkorBuild
    FalkorBuild -- "过滤" --> Stopwords
    FalkorBuild -- "RedisSearch 语法" --> DB
helpers.py 中的 lucene_sanitize 函数会在将特殊字符传递给基于 Lucene 的搜索引擎之前对其进行转义（+ - &&! ( ) { } [ ] ^ " ~ * ? : \ /）。

来源： graphiti_core/helpers.py:79-113, graphiti_core/driver/falkordb_driver.py:388-422

索引管理最佳实践

  1. 异步初始化：驱动程序在 __init__() 期间使用 asyncio.create_task 调度索引创建，以避免阻塞主线程。
  2. 并发控制：通过 semaphore_gather 使用 SEMAPHORE_LIMIT（默认 20）来防止在批量索引创建期间压垮数据库。
  3. 竞态条件韧性：驱动程序会捕获并忽略"已存在"错误，以允许在多节点环境中安全地并发执行。

来源： graphiti_core/driver/neo4j_driver.py:92-101, graphiti_core/helpers.py:38, graphiti_core/helpers.py:123-133, graphiti_core/driver/neo4j_driver.py:196-204