graphiti核心功能全解析：构建与查询时序感知知识图谱

在动态环境中运行的AI代理需要处理不断变化的数据和关系，传统的知识图谱框架往往难以应对这种实时性和时序性的挑战。graphiti作为专为AI代理设计的时序感知知识图谱框架，通过独特的架构设计和功能实现，解决了动态数据管理、历史上下文追踪和高效查询等关键问题。本文将全面解析graphiti的核心功能，展示如何利用其构建和查询时序感知知识图谱。

为什么选择graphiti？

传统的检索增强生成（RAG）方法依赖批量处理和静态数据汇总，难以应对频繁变化的数据。graphiti通过以下创新特性解决了这些挑战：

• 实时增量更新：无需批量重新计算即可立即整合新数据片段
• 双时态数据模型：显式跟踪事件发生和摄入时间，支持精确的时间点查询
• 高效混合检索：结合语义嵌入、关键词（BM25）和图遍历，实现低延迟查询
• 自定义实体定义：通过简单的Pydantic模型支持开发人员定义的实体
• 可扩展性：通过并行处理高效管理大型数据集，适合企业环境

graphiti与传统GraphRAG的核心差异如下表所示：

方面	GraphRAG	graphiti
主要用途	静态文档摘要	动态数据管理
数据处理	批处理导向	连续增量更新
知识结构	实体集群和社区摘要	事件数据、语义实体、社区
检索方法	顺序LLM摘要	混合语义、关键词和基于图的搜索
适应性	低	高
时间处理	基本时间戳跟踪	显式双时态跟踪
矛盾处理	LLM驱动的摘要判断	时间边缘失效
查询延迟	秒到数十秒	通常亚秒级延迟

核心功能解析

1. 多数据库支持与灵活驱动架构

graphiti采用灵活的驱动架构，支持多种主流图数据库，包括Neo4j、FalkorDB、Kuzu和Amazon Neptune。这种设计使开发人员能够根据项目需求和现有基础设施选择最合适的数据库后端。

驱动架构的核心实现位于graphiti_core/driver/目录，提供了统一的接口抽象和针对不同数据库的具体实现：

• Neo4jDriver：支持Neo4j图数据库
• FalkorDriver：支持Redis生态的FalkorDB
• KuzuDriver：支持Kuzu嵌入式图数据库
• NeptuneDriver：支持AWS Neptune服务

初始化graphiti实例时，可以指定特定的数据库驱动：

from graphiti_core import Graphiti
from graphiti_core.driver.falkordb_driver import FalkorDriver

# 创建FalkorDB驱动
driver = FalkorDriver(
    host="localhost",
    port=6379,
    database="my_custom_graph"  # 自定义数据库名称
)

# 使用自定义驱动初始化graphiti
graphiti = Graphiti(graph_driver=driver)

2. 实体与关系管理

graphiti提供全面的实体和关系管理功能，能够自动从文本中提取实体和关系，并构建结构化的知识图谱。核心实现位于graphiti_core/nodes.py和graphiti_core/edges.py。

主要节点类型包括：

• EntityNode：表示知识图谱中的实体
• EpisodicNode：表示事件或情节数据
• CommunityNode：表示实体集群或社区

主要边类型包括：

• EntityEdge：实体之间的关系
• EpisodicEdge：连接事件和实体的边
• CommunityEdge：社区与实体之间的关系

添加实体和关系的代码示例：

# 添加 episode
results = await graphiti.add_episode(
    name="用户对话",
    episode_body="Alice喜欢使用graphiti构建知识图谱",
    source_description="聊天记录",
    reference_time=datetime.now(),
    group_id="user-sessions-123"
)

# 结果包含提取的节点和边
print(f"提取的实体: {[node.name for node in results.nodes]}")
print(f"提取的关系: {[edge.fact for edge in results.edges]}")

3. 时间感知知识管理

graphiti的核心优势在于其时间感知能力，能够跟踪实体和关系随时间的变化。这一功能通过双时态模型实现，记录事件的发生时间和摄入时间。

时间感知功能主要通过以下组件实现：

• graphiti_core/utils/datetime_utils.py：时间处理工具
• graphiti_core/edges.py：边的时间属性管理
• graphiti_core/graph_queries.py：时间感知查询

查询特定时间点知识图谱状态的示例：

# 检索特定时间点的实体关系
results = await graphiti.search(
    query="Alice使用什么工具?",
    search_config=SearchConfig(
        time=datetime(2024, 1, 1),  # 查询2024年1月1日的知识状态
        limit=10
    )
)

4. 混合检索引擎

graphiti提供强大的混合检索能力，结合语义搜索、关键词搜索和图遍历，实现高效准确的信息检索。检索功能主要在graphiti_core/search/目录中实现。

主要检索策略包括：

• EDGE_HYBRID_SEARCH_RRF：使用 reciprocal rank fusion融合语义和关键词结果
• EDGE_HYBRID_SEARCH_NODE_DISTANCE：考虑节点距离的混合搜索
• COMBINED_HYBRID_SEARCH_CROSS_ENCODER：使用交叉编码器重排序的混合搜索

使用预定义搜索配置的示例：

from graphiti_core.search.search_config_recipes import EDGE_HYBRID_SEARCH_RRF

# 使用混合搜索配置搜索关系
results = await graphiti.search(
    query="谁喜欢使用graphiti?",
    search_config=EDGE_HYBRID_SEARCH_RRF
)

# 处理搜索结果
for result in results.items:
    print(f"事实: {result.fact}, 相似度: {result.score}")

5. LLM集成与扩展

graphiti与主流LLM提供商无缝集成，支持实体提取、关系识别和查询理解。LLM集成通过graphiti_core/llm_client/模块实现，支持OpenAI、Anthropic、Google Gemini等。

使用Google Gemini的示例：

from graphiti_core.llm_client.gemini_client import GeminiClient
from graphiti_core.embedder.gemini import GeminiEmbedder
from graphiti_core.cross_encoder.gemini_reranker_client import GeminiRerankerClient

# 使用Gemini初始化graphiti
graphiti = Graphiti(
    "bolt://localhost:7687",
    "neo4j",
    "password",
    llm_client=GeminiClient(config=LLMConfig(api_key="your-key", model="gemini-2.0-flash")),
    embedder=GeminiEmbedder(config=GeminiEmbedderConfig(api_key="your-key")),
    cross_encoder=GeminiRerankerClient(config=LLMConfig(api_key="your-key"))
)

6. MCP服务器集成

graphiti提供MCP（Model Context Protocol）服务器，允许AI助手通过MCP协议与graphiti的知识图谱功能交互。MCP服务器实现位于mcp_server/目录。

MCP服务器的主要功能：

• 事件管理（添加、检索、删除）
• 实体管理和关系处理
• 语义和混合搜索功能
• 用于组织相关数据的组管理

启动MCP服务器的命令：

cd mcp_server && python graphiti_mcp_server.py

快速开始

环境准备

1. 安装graphiti：

pip install graphiti-core

2. 设置环境变量：

# 必要的LLM和嵌入配置
export OPENAI_API_KEY=your_openai_api_key

# 可选的数据库连接参数
export NEO4J_URI=bolt://localhost:7687
export NEO4J_USER=neo4j
export NEO4J_PASSWORD=password

基本使用流程

完整的快速入门示例可在examples/quickstart/目录找到，以下是核心步骤：

1. 初始化graphiti：

from graphiti_core import Graphiti

# 初始化graphiti
graphiti = Graphiti(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password"
)

# 构建索引和约束
await graphiti.build_indices_and_constraints()

2. 添加知识片段：

from datetime import datetime

# 添加episode
await graphiti.add_episode(
    name="示例对话",
    episode_body="Alice正在使用graphiti构建时序知识图谱，她发现这个框架非常适合AI代理应用。",
    source_description="示例数据",
    reference_time=datetime.now()
)

3. 查询知识图谱：

# 搜索关系
results = await graphiti.search(query="谁在使用graphiti?")

# 打印结果
for item in results.items:
    print(f"事实: {item.fact}, 相关实体: {item.source_node_name} -> {item.target_node_name}")

总结与展望

graphiti为构建动态环境中的AI代理提供了强大的时序感知知识图谱框架。通过实时增量更新、双时态数据模型和高效混合检索等核心功能，graphiti解决了传统RAG方法在动态数据管理方面的局限性。

目前graphiti正在积极开发中，未来将重点关注：

• 增强检索能力，提供更强大和可配置的选项
• 扩展测试覆盖范围，确保可靠性并捕获边缘情况
• 优化性能，支持更大规模的知识图谱应用

要深入了解graphiti的更多功能，请参考官方文档和示例代码库。

加入Zep Discord服务器的**#graphiti**频道获取支持和最新更新！