数据截至 (上游 commit 10374d6044f9)
Graphiti — 架构与原理
30 秒导读: Graphiti 是给 AI agent 用的「时序上下文图谱」引擎。你把对话、文档、JSON 一段段喂给它(每段叫一个 episode),它用 LLM 把里面的人、物、关系抽成图谱;关键是每条事实都带「何时开始为真 / 何时被推翻」的时间窗——新信息进来不是覆盖旧的, 而是让旧事实「过期」。检索时它同时用向量、关键词(BM25)和图遍历三路召回再融合排序。它是 Zep 这家公司开源的记忆引擎内核。
1. 这是什么(零基础也能懂)
一句话定义: Graphiti 把流式进来的原始数据,增量地变成一张「实体 + 带时间有效期的事实」的知识图谱,并支持跨时间、跨语义的查询。
解决什么问题 / 给谁用。 假设你在做一个会「记事」的 AI 助手。用户三月说「我喜欢 Adidas」,六月又说「现在改穿 Nike 了」。
- 普通做法(把聊天记录塞进向量库做 RAG)会同时检索出这两条,模型分不清哪条还成立。
- Graphiti 的做法:它知道「喜欢 Adidas」这条事实在六月被「改穿 Nike」推翻了,于是给前者打上
invalid_at = 六月,查询「现在喜欢什么」时只返回有效的那条;但「三月时喜欢什么」依然能查到。
给谁用:做有状态 agent、个性化助手、需要「记住事实且事实会变」的开发者。
它能做什么(功能):
- 从对话 / 纯文本 / JSON 三种来源增量抽取实体和关系。
- 自动判断「这个新实体是不是已有实体」(去重)、「这条新事实是否推翻了旧事实」(时序失效)。
- 混合检索:语义向量 + 全文 BM25 + 图遍历(BFS),多种重排器。
- 允许你用 Pydantic model 预定义实体/边的类型(prescribed ontology),也允许结构从数据中自然涌现(learned)。
- 可选地把节点聚类成「社 区」并生成社区摘要。
用起来什么样: 最小用法就是反复调用 add_episode 喂数据,再调用 search 取事实。
# 示意,非源码(API 来自 graphiti_core/graphiti.py 的 Graphiti 类)
graphiti = Graphiti(uri, user, password) # 默认连 Neo4j、用 OpenAI
await graphiti.build_indices_and_constraints() # 初始化索引
# 喂一段 episode(就是一段原始数据)
await graphiti.add_episode(
name="chat-1",
episode_body="Kendra loves Adidas shoes.", # 原文
source_description="user message",
reference_time=datetime(2026, 3, 1), # 这段话「何时发生」
)
# 之后检索:返回的是 EntityEdge(事实/关系)列表
edges = await graphiti.search("What shoes does Kendra like?")
一句话直觉 / 类比: 把它想成给 agent 用的「会自动维护时间线的维基」——每个条目(事实)不是被新版本覆盖,而是记录「这一版从何时到何时为真」,所以你既能问「现在」也能问「去年此时」。
2. 顶层全景(它大概怎么转)
Graphiti 有两条核心数据流:写入(add_episode:原文 → 图谱)和读取(search:查询 → 排好序的事实)。先看写入主线。
一段 episode 进来(add_episode)
│
▼
① 取上下文:拉最近 N 条历史 episode 当背景
│
▼
② 抽实体(LLM):从原文抽出「节点」候选
│
▼
③ 实体去重(resolve):新节点 vs 图中已有节点
│ 先精确名→再 MinHash 模糊→再 LLM 兜底
▼
④ 抽关系(LLM):在已确定的节点之间抽「事实/边」
│
▼
⑤ 事实去重 + 时序判定(resolve):
│ 是重复吗?推翻了哪条旧事实?给旧事实盖 invalid_at
▼
⑥ 抽节点属性 / 摘要(LLM)
│
▼
⑦ 批量落库:节点、实体边、episodic 边一起写进图数据库
这条主线的代码主入口是 Graphiti.add_episode(graphiti_core/graphiti.py:980),它把上面每一步委托给 utils/maintenance/ 下的函数。
主要部件 一句话职责:
| 部件 | 干什么 | 在哪个文件 |
|---|---|---|
Graphiti | 顶层门面,编排写入/读取/社区/三元组 | graphiti_core/graphiti.py |
extract_nodes / extract_edges | 调 LLM 从 episode 原文抽实体、抽关系 | utils/maintenance/node_operations.py、edge_operations.py |
resolve_extracted_nodes | 实体去重(决定新节点是否=已有节点) | utils/maintenance/node_operations.py:627 |
dedup_helpers | 去重的确定性引擎:精确名 + MinHash/LSH 模糊匹配 | utils/maintenance/dedup_helpers.py |
resolve_extracted_edges | 事实去重 + 矛盾检测 + 时序失效 | utils/maintenance/edge_operations.py:325 |
search | 混合检索:多路召回 + 重排 | search/search.py:98 |
| 数据模型 | EntityNode / EntityEdge / EpisodicNode 等 | graphiti_core/nodes.py、edges.py |
GraphDriver | 抽象图数据库(Neo4j / FalkorDB / Kuzu / Neptune) | driver/driver.py |
| LLM / Embedder / CrossEncoder | 可插拔的模型客户端 | llm_client/、embedder/、cross_encoder/ |
三类节点 + 关键边(数据模型骨架):
| 图元素 | 含义 | 关键字段 |
|---|---|---|
EpisodicNode(节点) | 原始数据本体,provenance 源头 | content、valid_at、source、entity_edges |
EntityNode(节点) | 抽出的实体(人/物/概念) | name、name_embedding、summary、labels、attributes |
EntityEdge(边) | 一条事实/关系(实体→关系→实体) | fact、fact_embedding、valid_at、invalid_at、expired_at |
MENTIONS(边) | 哪个 episode 提到了哪个实体(溯源) | 由 EpisodicEdge 表示 |
CommunityNode | 一簇实体的聚类摘要 | name、summary |
EntityEdge 的三个时间字段是整个项目的灵魂,见 edges.py:271-279;EpisodicNode 的字段见 nodes.py:318-332。
3. 阅读地图(建议顺序)
这是个大项目,文档拆成四章,由浅入深:
- 01-ingestion.md — 写入主线:跟着一条 episode 端到端走一遍
add_episode,看抽点、抽边、落库怎么串起来。先读这章建立全局。 - 02-resolution-temporal.md — 去重与时序:Graphiti 最硬核的两块——实体怎么去重(MinHash/LSH 三级降级,省 LLM 调用),事实怎么「过期而不删」(双时间轴 + 矛盾检测)。
- 03-search.md — 混合检索:查 询时如何多路召回再用 RRF / MMR / 节点距离 / cross-encoder 重排。
- 04-internals.md — 深入与边界:社区检测(标签传播)、可借鉴的巧妙设计、刻意不做的事、横向对比、代码地图。
agent 提示:只想知道「事实如何随时间失效」→ 直接读第 2 章;只想接搜索 → 第 3 章;想改 ingestion 流程 → 第 1 章。