11 · 示例与 Use Cases（Examples and Use Cases）

示例与 Use Cases（中文译文）

原始 DeepWiki 页面：https://deepwiki.com/getzep/graphiti/11-examples-and-use-cases

翻译时间：2026-05-27T08:44:54.307Z

翻译模型：deepseek-chat

原文字符数：10434

项目：Graphiti (graphiti)

---

示例与用例

相关源文件

以下文件被用作生成此维基页面的上下文：

• examples/ecommerce/runner.py
• examples/quickstart/quickstart_falkordb.py
• examples/quickstart/quickstart_neo4j.py
• examples/wizard_of_oz/runner.py
• graphiti_core/decorators.py
• mcp_server/tests/test_async_operations.py
• mcp_server/tests/test_comprehensive_integration.py

本页面提供了 Graphiti 的实用示例和常见使用模式，涵盖了基本操作、提供方配置和集成模式，旨在帮助您快速上手并理解如何将 Graphiti 应用于实际场景。

如需详细的分步教程，请参阅：

• 快速入门教程（快速入门教程）—— 使用 Neo4j、FalkorDB 或 Amazon Neptune 进行基本设置和首次操作。
• AI 智能体集成模式（AI 智能体集成模式）—— 通过模型上下文协议（MCP）将 Graphiti 与 AI 智能体集成。
• 播客处理示例（播客处理示例）—— 处理转录文本，进行实体提取和关系构建。

如需了解高级功能用法，请参考：

• 自定义实体类型（自定义实体类型和属性）—— 使用 Pydantic 模型定义自定义实体模式。
• 搜索配置（搜索配置与调优）—— 调优搜索方法和重排序策略。
• 提供方配置（提供方配置）—— 各种大语言模型（LLM）和数据库提供方的详细设置。

常见用例

Graphiti 支持以下几种主要用例：

用例	描述	关键特性
AI 智能体记忆	为对话智能体提供持久化知识图谱	片段链、时间查询、混合搜索
文档处理	从文档中提取实体和关系	批量处理、JSON 入库、结构化提取
实时数据集成	从实时数据流持续更新图谱	增量更新、去重、时间追踪
企业知识管理	从业务数据构建可搜索的知识库	社区检测、自定义实体类型、多源集成
研究与分析	追踪随时间演变的关系	时间点查询、失效处理、历史分析

来源： examples/quickstart/quickstart_neo4j.py:73-77、examples/ecommerce/runner.py:60-72

---

基本使用模式

使用默认配置初始化 Graphiti

最简单的 Graphiti 设置使用 Neo4j 和 OpenAI，并通过环境变量进行配置。Graphiti 类负责编排底层驱动和客户端。

from graphiti_core import Graphiti

# 使用默认 Neo4j 连接进行初始化
graphiti = Graphiti(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password"
)

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

来源： examples/quickstart/quickstart_neo4j.py:47-67、examples/ecommerce/runner.py:88-90

片段入库工作流

下图展示了在入库过程中，自然语言输入如何映射到内部代码实体空间，以及高层调用如何对应到内部操作。

flowchart TB
    subgraph "自然语言空间"
        TEXT["episode_body: str"]
        META["name, source, reference_time"]
    end

    subgraph "Graphiti 核心 API"
        ADD_EP["Graphiti.add_episode()"]
        ADD_BULK["Graphiti.add_episode_bulk()"]
    end

    subgraph "代码实体空间（处理）"
        EXTRACT["LLMClient.generate_response()<br/>（节点/边提取）"]
        RESOLVE["EntityNodeOperations.resolve_nodes()<br/>EntityEdgeOperations.resolve_edges()"]
        EMBED["EmbedderClient.create()<br/>（向量生成）"]
        PERSIST["EpisodeNodeOperations.create()<br/>EntityEdgeOperations.create_many()"]
    end

    subgraph "数据库驱动空间"
        DRIVER["GraphDriver（Neo4j/Falkor/Kuzu）"]
        NODES["EntityNode<br/>EpisodicNode"]
        EDGES["EntityEdge<br/>EpisodicEdge"]
    end

    TEXT --> ADD_EP
    META --> ADD_EP
    ADD_EP --> ADD_BULK
    ADD_BULK --> EXTRACT
    EXTRACT --> RESOLVE
    RESOLVE --> EMBED
    EMBED --> PERSIST
    PERSIST --> DRIVER
    DRIVER --> NODES
    DRIVER --> EDGES

来源： examples/ecommerce/runner.py:75-84、examples/wizard_of_oz/runner.py:84-90、graphiti_core/decorators.py:29-53

添加片段

Graphiti 支持多种片段类型，包括原始文本、JSON 和消息列表。使用 EpisodeType 指定格式。

from datetime import datetime, timezone
from graphiti_core.nodes import EpisodeType

# 添加文本片段
await graphiti.add_episode(
    name="用户偏好",
    episode_body="John 喜欢在他的应用中使用深色模式。他在 Acme Corp 工作。",
    source_description="用户对话",
    source=EpisodeType.text,
    reference_time=datetime.now(timezone.utc)
)

# 添加 JSON 格式片段
await graphiti.add_episode(
    name="产品数据",
    episode_body='{"name": "SuperLight Wool Runners", "material": "wool"}',
    source_description="产品目录",
    source=EpisodeType.json,
    reference_time=datetime.now(timezone.utc)
)

来源： examples/quickstart/quickstart_falkordb.py:92-138、examples/ecommerce/runner.py:102-120

搜索图谱

搜索系统使用混合检索来查找相关节点和边，通常会使用 center_node_uuid 基于图谱邻近度进行重排序。

flowchart LR
    subgraph "搜索输入"
        QUERY["query: str"]
        CONFIG["SearchConfig"]
        FILTERS["SearchFilters"]
    end

    subgraph "Graphiti 搜索引擎"
        SEARCH["Graphiti.search()"]
        OPS["SearchOperations.search()"]
        RERANK["CrossEncoderClient.rerank()"]
    end

    subgraph "代码实体结果"
        EDGES["List[EntityEdge]"]
        NODES["List[EntityNode]"]
    end

    QUERY --> SEARCH
    CONFIG --> SEARCH
    FILTERS --> SEARCH
    SEARCH --> OPS
    OPS --> RERANK
    RERANK --> EDGES
    RERANK --> NODES

来源： examples/quickstart/quickstart_neo4j.py:138-151、graphiti_core/decorators.py:76-80

from graphiti_core.search.search_config_recipes import NODE_HYBRID_SEARCH_RRF

# 基本边搜索（事实）
results = await graphiti.search(
    "谁是加州总检察长？",
    num_results=5
)
for result in results:
    print(f"事实：{result.fact}")

# 使用预定义配置进行节点搜索
node_results = await graphiti.search(
    "Gavin Newsom",
    search_config=NODE_HYBRID_SEARCH_RRF
)

来源： examples/quickstart/quickstart_neo4j.py:138-151、examples/quickstart/quickstart_neo4j.py:186-193

---

提供方配置示例

数据库后端选项

Graphiti 通过 GraphDriver 接口抽象了数据库操作。Graphiti 类可以直接使用驱动进行初始化。

# Neo4j 驱动（默认）
from graphiti_core.driver.neo4j_driver import Neo4jDriver
driver = Neo4jDriver(uri="bolt://localhost:7687", user="neo4j", password="password")

# FalkorDB 驱动
from graphiti_core.driver.falkordb_driver import FalkorDriver
falkor_driver = FalkorDriver(host="localhost", port=6379)
graphiti = Graphiti(graph_driver=falkor_driver)

来源： examples/quickstart/quickstart_falkordb.py:75-78、examples/quickstart/quickstart_neo4j.py:67-67

大语言模型（LLM）提供方选项

Graphiti 通过 LLMClient 接口支持多种大语言模型（LLM）提供方。

Anthropic 示例

from graphiti_core.llm_client.anthropic_client import AnthropicClient
from graphiti_core.llm_client.config import LLMConfig

llm_client = AnthropicClient(LLMConfig(api_key=os.environ.get('ANTHROPIC_API_KEY')))
client = Graphiti(neo4j_uri, neo4j_user, neo4j_password, llm_client)

来源： examples/wizard_of_oz/runner.py:61-62

---

MCP 服务器集成

MCP（模型上下文协议）服务器将 Graphiti 的能力暴露给 AI 智能体。它包含顺序队列管理，确保在 group_id 内按顺序处理片段。

关键工具：

• add_memory：将片段加入处理队列。
• search_memory_nodes：对实体执行混合搜索。
• search_memory_facts：对关系/边执行混合搜索。
• get_episodes：检索特定组的最新片段。
• get_status：检查图谱和处理队列的状态。

graph TD
    A[AI 智能体] -->|调用工具| B{MCP 服务器}
    B -->|add_memory(group_id, ...)| C[队列服务]
    C -->|按 group_id 顺序处理| D[Graphiti 核心]
    D -->|add_episode()| E[图数据库]
    B -->|search_memory_nodes(group_id, query)| D
    B -->|search_memory_facts(group_id, query)| D
    B -->|get_episodes(group_id)| D
    D -->|返回结果| B
    B -->|返回结果| A

来源： mcp_server/tests/test_async_operations.py:23-58、mcp_server/tests/test_comprehensive_integration.py:151-171

---

运行示例

快速入门示例展示了不同数据库后端下的核心功能：

# 运行 Neo4j 快速入门
python examples/quickstart/quickstart_neo4j.py

# 运行 FalkorDB 快速入门
python examples/quickstart/quickstart_falkordb.py

这些示例涵盖：

• 使用 build_indices_and_constraints() 进行图谱初始化。
• 使用文本和 JSON 内容进行片段入库。
• 结合语义相似度和 BM25 的混合搜索。
• 使用 center_node_uuid 进行图谱感知的重排序。

来源： examples/quickstart/quickstart_neo4j.py:57-67、examples/quickstart/quickstart_falkordb.py:65-79

---

如需更详细的教程和集成模式，请参阅子页面：

• 快速入门教程 —— 分步设置。
• AI 智能体集成模式 —— MCP 工作流。
• 播客处理示例 —— 批量处理。