Graphiti 知识图谱框架开发指南:面向 AI Agent 的实时时序知识图谱构建与工程实践
【免费下载链接】graphitiBuild Real-Time Knowledge Graphs for AI Agents项目地址: https://gitcode.com/GitHub_Trending/grap/graphiti
导读
本文基于 Graphiti 仓库根目录的 CLAUDE.md 开发指南,结合graphiti_core、server、mcp_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 中的项目概述,其核心特性可以归纳为四点:
- 双时态数据模型(Bi-temporal data model):显式记录事件的"发生时间"与"知识录入时间",使图谱既能回答"当前事实是什么",也能回答"某个历史时刻的事实是什么";
- 混合检索(Hybrid retrieval):同时组合语义向量检索(embedding)、关键词检索(BM25)与图遍历(graph traversal)三种手段召回事实;
- 自定义实体定义:通过 Pydantic 模型声明自定义实体类型,将领域结构约束注入图谱;
- 多后端存储:原生支持 Neo4j 与 FalkorDB 作为图存储后端,并附带可选的开源 OpenTelemetry 分布式追踪支持。
从仓库的pyproject.toml可以看到核心库名为graphiti-core(版本 0.30.1),依赖pydantic>=2.11.5、neo4j>=5.26.0、openai>=1.91.0、tenacity、numpy、python-dotenv等,支持 Python 3.10+。除 Neo4j 与 FalkorDB 外,pyproject.toml 还声明了kuzu、neptune、falkordblite等可选依赖——其中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)枚举:
| 枚举值 | 含义 |
|---|---|
rrf | Reciprocal 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 --fix和ruff format;lint执行ruff check与pyright ./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 推理与嵌入所必需(默认的OpenAIClient与OpenAIEmbedder都依赖它,见 graphiti.py 的默认客户端初始化逻辑);USE_PARALLEL_RUNTIME:可选布尔值,启用 Neo4j 并行运行时(仅企业版可用);- Provider 专属密钥:
ANTHROPIC_API_KEY、GOOGLE_API_KEY、GROQ_API_KEY、VOYAGE_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时,分别默认使用OpenAIClient、OpenAIEmbedder、OpenAIRerankerClient;未传入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-mini、gpt-5-nano(要求temperature=0,代码中已自动处理);GPT-4.1 系列gpt-4.1、gpt-4.1-mini、gpt-4.1-nano;旧代gpt-4o、gpt-4o-mini; - Anthropic:Claude 4.5 家族(
claude-sonnet-4-5-latest、固定版claude-sonnet-4-5-20250929、claude-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-pro、gemini-1.5-flash)。
各 Provider 客户端分别位于 llm_client/ 目录(openai_client.py、anthropic_client.py、gemini_client.py、groq_client.py、azure_openai_client.py等),对应可选依赖anthropic、groq、google-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 autopytest-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。核心工作流是"先检索、后存储、再遵循"的记忆闭环:
- 任务开始前先搜索:使用
search_nodes工具查找已有的偏好(Preference)与流程(Procedure),用search_facts发现实体间的关系事实;节点搜索时指定Preference、Procedure、Requirement等实体类型过滤,让结果更聚焦; - 立即存储新知识:用户表达需求或偏好时,立刻用
add_memory落库;长需求拆分为较短的逻辑块存储;只记录变更或新增内容,避免重复;发现的用户做事方式记录为 procedure,实体间关系记录为 facts; - 工作中遵守既有知识:严格按检索到的 procedure 分步执行,尊重已发现的偏好,用事实信息辅助决策,并保持与历史知识的一致性;
- 最佳实践:提出建议前先确认是否已有既定知识;复杂任务同时搜索节点与事实;探索关联信息时用
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),仅供参考