Graphiti:AI Agent 的时序知识图谱
AI 导读
Graphiti:AI Agent 的时序知识图谱 作者:Maurice | 灵阙学院 Agent 记忆的困境 AI Agent 面临一个根本性的记忆问题:LLM 的上下文窗口是有限的,而 Agent 需要处理的信息是无限的。当前主流的 Agent 记忆方案各有局限: 方案 优势 局限 上下文窗口 即时可用、高保真 容量有限、成本高 向量数据库 语义检索、大容量 无结构、无时序、易冲突 对话摘要...
Graphiti:AI Agent 的时序知识图谱
作者:Maurice | 灵阙学院
Agent 记忆的困境
AI Agent 面临一个根本性的记忆问题:LLM 的上下文窗口是有限的,而 Agent 需要处理的信息是无限的。当前主流的 Agent 记忆方案各有局限:
| 方案 | 优势 | 局限 |
|---|---|---|
| 上下文窗口 | 即时可用、高保真 | 容量有限、成本高 |
| 向量数据库 | 语义检索、大容量 | 无结构、无时序、易冲突 |
| 对话摘要 | 压缩高效 | 信息丢失、不可逆 |
| 全文搜索 | 精确匹配 | 无语义理解 |
核心矛盾在于:Agent 需要结构化的、可更新的、带时序的记忆,而现有方案要么没有结构(向量),要么不可更新(摘要),要么没有时序(知识图谱)。
Graphiti 的设计理念
Zep(YC S23)开源的 Graphiti 框架将知识图谱重新设计为 Agent 的"活记忆":
核心创新
- 时序化实体关系:每条边和节点都带有时间戳,支持"此刻正确"的查询
- 增量式更新:新信息作为"事件"(Episode)注入,自动更新图谱而不覆盖历史
- 双重检索:同时支持语义搜索(Embedding)和图遍历(Structure),取最佳结果
- 冲突解决:当新信息与旧信息矛盾时,通过 LLM 判断更新策略
与传统 KG 的区别
传统知识图谱:
(Alice, works_at, Company_A) // 静态事实,覆盖式更新
Graphiti 时序图谱:
(Alice, works_at, Company_B, valid_from: 2024-03-01) // 当前事实
(Alice, works_at, Company_A, valid_from: 2021-06-15, valid_to: 2024-02-28) // 历史事实
(Episode_42: "Alice told me she just started at Company B", timestamp: 2024-03-01) // 来源
架构概览
Agent 对话/事件
|
v
[Episode 注入管线]
|
├── 1. 实体抽取(LLM)
├── 2. 关系抽取(LLM)
├── 3. 实体消歧(Embedding + LLM)
├── 4. 冲突检测(图查询)
└── 5. 图更新(写入 Neo4j)
[检索管线]
|
├── 语义路径:Query Embedding → 向量相似度 → Top-K 节点/边
├── 图路径:关键词 → 实体匹配 → N跳邻域扩展
└── 融合:合并去重 → Rerank → 返回上下文
数据模型
Graphiti 使用三种核心数据结构:
EntityNode(实体节点):
name: 实体名称entity_type: 实体类型(Person, Organization, Concept 等)summary: LLM 生成的实体描述created_at/expired_at: 生命周期
EntityEdge(实体关系):
source_node_id/target_node_id: 两端节点name: 关系名称fact: 自然语言描述的事实valid_at/invalid_at: 有效时间episodes: 支持该关系的 Episode 列表
EpisodicNode(事件节点):
content: 原始文本source: 来源(对话、文档、API等)timestamp: 发生时间valid_at: 事件的有效时间
工作流程详解
Episode 注入
当 Agent 收到新信息(用户对话、工具返回、外部事件)时:
# 1. 创建 Episode
episode = EpisodicNode(
content="用户说:我们刚完成了 Series B 融资,领投方是红杉资本",
source="user_message",
timestamp=datetime.now()
)
# 2. LLM 抽取实体和关系
# 抽取结果:
# 实体: [我们的公司(Organization), Series B(FundingRound), 红杉资本(Organization)]
# 关系: [我们的公司 --完成--> Series B, 红杉资本 --领投--> Series B]
# 3. 实体消歧
# "我们的公司" → 匹配到已知实体 "Acme Corp"(基于对话上下文)
# "红杉资本" → 匹配到已知实体 "Sequoia Capital" 或创建新实体
# 4. 冲突检测
# 检查是否已有 "Acme Corp --融资轮次--> Series A" 的记录
# 如果有,不覆盖,而是标记 Series A 关系为 expired,新增 Series B 关系
# 5. 写入图谱
检索与召回
Agent 需要记忆时,Graphiti 执行双重检索:
# 语义检索:找与查询语义相近的实体和关系
semantic_results = search_by_embedding(
query="Acme Corp 的融资历史",
limit=10
)
# 图遍历:从关键实体出发,扩展邻域
graph_results = traverse(
start_node="Acme Corp",
max_hops=2,
edge_filter=lambda e: e.name in ["融资", "投资", "领投"]
)
# 融合结果
combined = merge_and_rerank(semantic_results, graph_results)
与 Agent 框架集成
LangGraph 集成
from graphiti_core import Graphiti
from langgraph.graph import StateGraph
# 初始化 Graphiti
graphiti = Graphiti(neo4j_uri, neo4j_user, neo4j_pass)
async def memory_node(state):
"""在 Agent 工作流中加入记忆节点"""
# 1. 检索相关记忆
context = await graphiti.search(
query=state["current_message"],
num_results=10
)
# 2. 注入上下文
state["memory_context"] = context
return state
async def update_memory_node(state):
"""对话结束后更新记忆"""
await graphiti.add_episode(
name=f"conversation_{state['turn_id']}",
episode_body=state["full_conversation"],
source_description="agent_conversation"
)
return state
MCP Server 集成
Graphiti 可以包装为 MCP Server,让任何支持 MCP 的 Agent 使用:
{
"tools": [
{
"name": "memory_search",
"description": "Search agent memory for relevant context",
"parameters": {
"query": "string",
"time_range": "optional date range",
"entity_types": "optional filter"
}
},
{
"name": "memory_add",
"description": "Add new information to agent memory",
"parameters": {
"content": "string",
"source": "string"
}
}
]
}
实际效果
对比实验(Zep 官方数据)
在 MemoryBench 基准测试中,Graphiti 相比其他记忆方案的表现:
| 指标 | 向量RAG | 摘要记忆 | Graphiti |
|---|---|---|---|
| 事实准确率 | 72% | 65% | 89% |
| 时序一致性 | 41% | 53% | 91% |
| 多跳推理 | 38% | 29% | 78% |
| 冲突处理 | 12% | 45% | 85% |
时序一致性的差距尤其显著:当信息经历过多次更新时,向量检索容易返回过时的版本。
适用场景
- 长期客服 Agent:记住每个客户的偏好变化和交互历史
- 研究助手:追踪项目进展、文献发现、实验结果的演变
- 合规顾问:维护法规变更的时序链,自动识别影响面
- 项目管理:记录需求变更、决策历程、风险演化
不适用场景
- 短期单次交互:上下文窗口足够,无需持久化记忆
- 超大规模数据:数十亿三元组需要分布式图数据库,Graphiti 目前基于 Neo4j 单机
- 纯数值时序:传感器数据更适合时序数据库
部署与运维
最小部署
# 1. 启动 Neo4j
docker run -d --name neo4j \
-p 7474:7474 -p 7687:7687 \
-e NEO4J_AUTH=neo4j/password \
neo4j:latest
# 2. 安装 Graphiti
pip install graphiti-core
# 3. 初始化
python -c "
from graphiti_core import Graphiti
g = Graphiti('bolt://localhost:7687', 'neo4j', 'password')
import asyncio
asyncio.run(g.build_indices_and_constraints())
"
生产建议
- Neo4j 高可用:使用 Neo4j Cluster 部署,确保读写分离
- LLM 成本控制:实体抽取使用 Flash 模型,复杂推理使用 Pro 模型
- Episode 批处理:非实时场景下批量注入 Episode,减少 LLM 调用
- 定期压缩:对过期的历史关系做归档,保持图谱精简
- 备份策略:定期导出 Neo4j 快照,防止数据丢失
与 GraphRAG 的关系
| 维度 | GraphRAG | Graphiti |
|---|---|---|
| 核心目标 | 增强文档检索 | Agent 长期记忆 |
| 数据来源 | 静态文档集合 | 动态交互流 |
| 更新方式 | 全量重建 | 增量注入 |
| 时序支持 | 无 | 原生支持 |
| 查询模式 | Local/Global Search | 语义+图遍历融合 |
| 适用场景 | 知识库问答 | Agent 对话记忆 |
两者可以互补:用 GraphRAG 处理静态文档知识,用 Graphiti 处理动态交互记忆。
Maurice | maurice_wen@proton.me