Graphiti 知识图谱框架开发指南:面向 AI Agent 的实时时序知识图谱构建与工程实践
2026/9/11 8:00:58 网站建设 项目流程

Graphiti 知识图谱框架开发指南:面向 AI Agent 的实时时序知识图谱构建与工程实践

【免费下载链接】graphitiBuild Real-Time Knowledge Graphs for AI Agents项目地址: https://gitcode.com/GitHub_Trending/grap/graphiti

导读

本文基于 Graphiti 仓库根目录的 CLAUDE.md 开发指南,结合graphiti_coreservermcp_server的实际源码,系统讲解 Graphiti 这一面向 AI Agent 的时序知识图谱 Python 框架:从双时态数据模型与混合检索原理、到开发环境搭建与 Makefile 工作流、再到 FastAPI 服务与 MCP Server 的工程化用法。读完本文,你将掌握 Graphiti 的核心 API(add_episode/search)、自定义实体(Pydantic 模型)、多 LLM Provider 选型与测试策略,能够在真实项目中完成知识图谱的构建、检索与 Agent 记忆集成。

Graphiti 是什么:为 AI Agent 设计的时序知识图谱框架

Graphiti 是一个用于构建时间感知(temporally-aware)知识图谱的 Python 框架,其设计目标是支撑 AI Agent 的长期记忆与动态事实推理。与传统的批量建图(batch recomputation)方式不同,Graphiti 支持无需批量重算的实时增量更新,非常适合对话历史、事件流等持续变化的数据环境。

根据 CLAUDE.md 中的项目概述,其核心特性可以归纳为四点:

  1. 双时态数据模型(Bi-temporal data model):显式记录事件的"发生时间"与"知识录入时间",使图谱既能回答"当前事实是什么",也能回答"某个历史时刻的事实是什么";
  2. 混合检索(Hybrid retrieval):同时组合语义向量检索(embedding)、关键词检索(BM25)与图遍历(graph traversal)三种手段召回事实;
  3. 自定义实体定义:通过 Pydantic 模型声明自定义实体类型,将领域结构约束注入图谱;
  4. 多后端存储:原生支持 Neo4j 与 FalkorDB 作为图存储后端,并附带可选的开源 OpenTelemetry 分布式追踪支持。

从仓库的pyproject.toml可以看到核心库名为graphiti-core(版本 0.30.1),依赖pydantic>=2.11.5neo4j>=5.26.0openai>=1.91.0tenacitynumpypython-dotenv等,支持 Python 3.10+。除 Neo4j 与 FalkorDB 外,pyproject.toml 还声明了kuzuneptunefalkordblite等可选依赖——其中kuzu标注为已废弃(上游项目停止维护,将在未来版本移除),这是选型时值得注意的信息。

代码架构全景:core / server / mcp_server 三层结构

CLAUDE.md 将仓库划分为三个主要部分,与实际目录结构完全对应:

核心库graphiti_core/

模块职责关键文件
主入口Graphiti类编排建图、检索、社区维护等全部能力graphiti.py
图存储Neo4j / FalkorDB / Kuzu / Neptune 驱动层driver/
LLM 集成OpenAI、Anthropic、Gemini、Groq 等客户端llm_client/
嵌入各 Provider 的 Embedding 客户端embedder/
图元素节点与边的核心数据结构nodes.py、edges.py
搜索可配置策略的混合检索实现search/
提示词实体抽取、去重、摘要的 LLM 提示词prompts/
工具维护操作、批量处理、日期时间处理utils/

服务层server/

一个基于 FastAPI 的 REST API 服务:

  • FastAPI 服务入口:graph_service/main.py
  • 路由:routers/ 提供 ingest(数据摄入)与 retrieve(检索)两组端点
  • DTO:dto/ 定义 API 契约的数据传输对象

MCP Server 层mcp_server/

面向 AI 助手的 Model Context Protocol 服务:

  • MCP 实现:graphiti_mcp_server.py,将知识图谱暴露为可供 AI 助手调用的 MCP 工具
  • Docker 支持:docker/ 提供容器化部署方案,可搭配 Neo4j 一键起服务

核心原理一:双时态数据模型与增量建图

Graphiti 将"知识入库"建模为一系列episode(片段)。每一次调用add_episode,框架都会执行:节点抽取(extract_nodes)→ 边抽取(extract_edges)→ 去重(dedupe)→ 嵌入生成 → 边失效处理 → 数据库写入,整个流程的入口与实现细节均可在 graphiti.py 的add_episode方法中验证。

add_episode的核心参数(来自 graphiti.py):

参数含义
name片段名称
episode_body片段正文内容
source_description片段来源描述
reference_time片段的参考时间(即事件发生时间,双时态的"业务时间"维度)
source片段类型,默认EpisodeType.message
group_id图分区的 id,用于隔离不同租户/业务域的数据
entity_types自定义实体类型字典,映射实体类型名到 Pydantic 模型
excluded_entity_types需要排除的实体类型列表(可包含'Entity'排除默认实体类型)
update_communities是否用新节点信息更新社区
custom_extraction_instructions注入实体/边抽取提示词的自定义指令
saga将片段关联到某个 saga(叙事弧),跨片段维持故事连续性

文档特别强调(graphiti.py):add_episode适合作为后台任务运行(如 FastAPI 的 BackgroundTasks 或 Celery 队列),且每个 episode 必须顺序添加并 await 完成后再添加下一个,这一点对并发环境下的数据一致性至关重要。

核心原理二:混合检索(embedding + BM25 + 图遍历)

Graphiti 的检索不是单一的向量相似度,而是多路召回 + 重排的组合。从 search_config.py 源码看,检索配置由四个子配置组成,各自拥有独立的搜索方法与重排器:

搜索方法(SearchMethod)枚举

枚举值含义
cosine_similarity语义向量相似度检索
bm25关键词全文检索
bfs广度优先图遍历(沿边关系扩散)

重排器(Reranker)枚举

枚举值含义
rrfReciprocal Rank Fusion(倒数排名融合)
node_distance基于与中心节点距离重排
episode_mentions基于片段提及次数重排
mmr最大边际相关性(兼顾相关性与多样性)
cross_encoder交叉编码器精排

框架预置了多套开箱即用的检索配置(search_config_recipes.py):

  • COMBINED_HYBRID_SEARCH_RRF:对 edges、nodes、communities 三类元素分别做bm25 + cosine_similarity混合召回,统一用 RRF 融合,是最推荐的通用配置;
  • COMBINED_HYBRID_SEARCH_CROSS_ENCODER:额外加入 BFS 图遍历,并用 cross-encoder 精排,精度更高但成本更大;
  • EDGE_HYBRID_SEARCH_RRF/EDGE_HYBRID_SEARCH_NODE_DISTANCE等:仅针对边(事实)的轻量配置,也是Graphiti.search()默认使用的方案。

在 search.py 的search函数中可以看到完整的执行流程:先校验group_ids,若配置中任一子配置使用相似度检索或 MMR 重排,则先对查询文本做嵌入得到query_vector;随后按各子配置并行执行多路召回;最后执行配置的重排器并返回带reranker_scores的结果对象SearchResults(包含 edges、nodes、episodes、communities 四类结果及各自得分)。

两种搜索 API 的选择

graphiti.py 提供了两个搜索入口:

  • graphiti.search(query, num_results=10, ...):基础版。无中心节点时使用EDGE_HYBRID_SEARCH_RRF,提供中心节点时自动切换为EDGE_HYBRID_SEARCH_NODE_DISTANCE,返回list[EntityEdge](即事实边列表);
  • graphiti.search_(...):进阶版,返回包含节点、边、片段、社区的完整SearchResults,支持自定义SearchConfig,官方注释建议"需要更强健结果时使用"。

开发环境与工程命令:从安装到 CI 检查

CLAUDE.md 给出了三组开发命令,均基于uv包管理器。

主项目命令(在仓库根目录执行)

# 安装依赖(含开发依赖) uv sync --extra dev # 格式化代码(ruff 的 import 排序 + 格式化) make format # 代码检查(ruff + pyright 类型检查) make lint # 运行测试 make test # 一键跑完 format、lint、test 全部检查 make check

查看根目录 Makefile 可以还原这些命令的真实实现:format实际执行ruff check --select I --fixruff formatlint执行ruff checkpyright ./graphiti_core;而test通过环境变量DISABLE_FALKORDB=1 DISABLE_KUZU=1 DISABLE_NEPTUNE=1禁用可选后端,并运行pytest -m "not integration"——这意味着默认的make test只跑单元测试,需要数据库的集成测试不会被执行。

Server 与 MCP Server 开发

# Server(在 server/ 目录) cd server/ uv sync --extra dev uvicorn graph_service.main:app --reload # 开发模式热重载启动 make format && make lint && make test # MCP Server(在 mcp_server/ 目录) cd mcp_server/ uv sync docker-compose up

配置:环境变量与数据库准备

环境变量

CLAUDE.md 列出的关键环境变量:

  • OPENAI_API_KEY:LLM 推理与嵌入所必需(默认的OpenAIClientOpenAIEmbedder都依赖它,见 graphiti.py 的默认客户端初始化逻辑);
  • USE_PARALLEL_RUNTIME:可选布尔值,启用 Neo4j 并行运行时(仅企业版可用);
  • Provider 专属密钥:ANTHROPIC_API_KEYGOOGLE_API_KEYGROQ_API_KEYVOYAGE_API_KEY

另外,Graphiti.__init__还支持max_coroutines参数,用于覆盖环境变量SEMAPHORE_LIMIT控制的并发协程上限(graphiti.py)。

数据库版本要求

  • Neo4j:要求 5.26+,可通过 Neo4j Desktop 获取。数据库名默认硬编码为neo4j,可在驱动构造时传database参数覆盖——这在 neo4j_driver.py 中得到验证:Neo4jDriver(uri, user, password, database='neo4j')
  • FalkorDB:要求 1.1.2+(对应 pyproject.toml 中falkordb>=1.1.2,<2.0.0的版本约束),数据库名默认default_db,同样可通过database参数覆盖。

初始化 Graphiti

from graphiti_core.graphiti import Graphiti graphiti = Graphiti( uri='bolt://localhost:7687', user='neo4j', password='your_password', store_raw_episode_content=True, ) # 用完务必关闭驱动连接,释放资源 await graphiti.close()

构造函数(graphiti.py)的默认行为:未显式传入llm_client/embedder/cross_encoder时,分别默认使用OpenAIClientOpenAIEmbedderOpenAIRerankerClient;未传入graph_driver时默认创建Neo4jDriver,因此只要准备好OPENAI_API_KEY和一个 Neo4j 实例即可跑通最小闭环。若需要自定义这些组件,将实例通过参数注入即可;cross_encoder用于检索阶段的精排,仓库 cross_encoder/ 中提供了 OpenAI、Gemini、BGE 等 reranker 客户端实现。

LLM Provider 选型与模型清单

CLAUDE.md 明确指出:代码库支持多个 LLM Provider,但对支持结构化输出的服务(OpenAI、Gemini)效果最佳,其他 Provider(尤其小模型)可能因 schema 校验问题导致实体抽取失败。

模型清单(截至文档标注时间 November 2025,实际可用性以各 Provider 官方为准):

  • OpenAI:GPT-5 系列推理模型gpt-5-minigpt-5-nano(要求temperature=0,代码中已自动处理);GPT-4.1 系列gpt-4.1gpt-4.1-minigpt-4.1-nano;旧代gpt-4ogpt-4o-mini
  • Anthropic:Claude 4.5 家族(claude-sonnet-4-5-latest、固定版claude-sonnet-4-5-20250929claude-haiku-4-5-latest);Claude 3.7 家族(latest 与 20250219 固定版);Claude 3.5 家族;
  • Google Gemini:Gemini 2.5 家族(gemini-2.5-pro旗舰、gemini-2.5-flash高效);Gemini 2.0 家族(gemini-2.0-flash实验版);Gemini 1.5 家族(生产稳定版gemini-1.5-progemini-1.5-flash)。

各 Provider 客户端分别位于 llm_client/ 目录(openai_client.pyanthropic_client.pygemini_client.pygroq_client.pyazure_openai_client.py等),对应可选依赖anthropicgroqgoogle-genai等。

测试体系:单元测试与集成测试的划分

CLAUDE.md 定义了清晰的测试分层:

  • 单元测试:位于 tests/,使用 pytest,不依赖外部数据库;
  • 集成测试:文件名以_int后缀标记(如 test_graphiti_int.py),需要真实数据库连接;
  • 评测脚本:tests/evals/ 提供端到端图构建评估。

常用 pytest 命令:

# 运行单个测试文件 pytest tests/test_specific_file.py # 运行单个测试方法 pytest tests/test_file.py::test_method_name # 只跑集成测试(需要数据库) pytest tests/ -k "_int" # 只跑单元测试 pytest tests/ -k "not _int" # 并行执行(依赖 pytest-xdist) pytest -n auto

pytest-xdist已列入 dev 依赖(pyproject.toml),支持并行测试加速。测试中会利用环境变量禁用可选后端(见上文 Makefile 分析),保证核心逻辑测试的轻量可复现。

代码规范:Ruff + Pyright 的强制约束

CLAUDE.md 与 pyproject.toml 中的配置完全一致:

  • Ruff:负责格式化与 lint,line-length = 100引号风格为单引号,启用了 pycodestyle(E)、Pyflakes(F)、pyupgrade(UP)、flake8-bugbear(B)、flake8-simplify(SIM)、isort(I)等规则组,并忽略 E501(行长交由 formatter 处理);
  • Pyright:强制类型检查。核心库为typeCheckingMode = "basic",而server/采用更严格的typeCheckingMode = "standard"(可在 server/pyproject.toml 确认)。

代码库大量使用typing_extensions.TypedDict而非typing.TypedDict(pyproject 中甚至将后者设为 banned API,原因是 Pydantic 在 Python < 3.12 下的兼容性要求),这是贡献代码时需要遵循的细节。

工程实践:将 Graphiti 用作 Agent 记忆层(MCP 用法)

CLAUDE.md 最后给出了 MCP Server 的使用准则,其详细模式记录在 mcp_server/docs/cursor_rules.md。核心工作流是"先检索、后存储、再遵循"的记忆闭环:

  1. 任务开始前先搜索:使用search_nodes工具查找已有的偏好(Preference)与流程(Procedure),用search_facts发现实体间的关系事实;节点搜索时指定PreferenceProcedureRequirement等实体类型过滤,让结果更聚焦;
  2. 立即存储新知识:用户表达需求或偏好时,立刻用add_memory落库;长需求拆分为较短的逻辑块存储;只记录变更或新增内容,避免重复;发现的用户做事方式记录为 procedure,实体间关系记录为 facts;
  3. 工作中遵守既有知识:严格按检索到的 procedure 分步执行,尊重已发现的偏好,用事实信息辅助决策,并保持与历史知识的一致性;
  4. 最佳实践:提出建议前先确认是否已有既定知识;复杂任务同时搜索节点与事实;探索关联信息时用center_node_uuid以某节点为中心展开检索;更具体的匹配优先于泛化信息。

这套准则的本质是"知识图谱即 Agent 的记忆"——把用户的偏好、流程、事实持续写入时序图谱,检索时就能让 Agent 做出符合用户习惯的个性化响应。若希望将 Graphiti 作为 FastAPI 服务暴露,可参考 server/graph_service/main.py 与 server/graph_service/zep_graphiti.py 了解服务化封装方式;examples/ 目录下还提供了 quickstart、ecommerce、podcast、langgraph-agent 等可直接运行的示例(如 examples/quickstart/quickstart_neo4j.py),适合作为上手 Graphiti 的起点。

结语

Graphiti 通过"双时态建模 + 增量更新 + 混合检索"的组合,为 AI Agent 提供了一个可持续演进的记忆基础设施:add_episode让知识实时流入,search让事实精准召回,Pydantic 自定义实体让领域约束得以固化,而 server / MCP Server 两层封装则让图谱能力可以被 REST 接口和 AI 助手直接消费。开发者在接入时,只需按本指南完成环境搭建(uv sync --extra dev)、数据库准备(Neo4j 5.26+ 或 FalkorDB 1.1.2+)、Provider 配置(推荐结构化输出能力强的 OpenAI / Gemini),即可基于仓库自带的示例与测试快速进入工程化开发。

【免费下载链接】graphitiBuild Real-Time Knowledge Graphs for AI Agents项目地址: https://gitcode.com/GitHub_Trending/grap/graphiti

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询