Graphiti：AI Agent 的时序知识图谱

作者：Maurice | 灵阙学院

Agent 记忆的困境

AI Agent 面临一个根本性的记忆问题：LLM 的上下文窗口是有限的，而 Agent 需要处理的信息是无限的。当前主流的 Agent 记忆方案各有局限：

方案	优势	局限
上下文窗口	即时可用、高保真	容量有限、成本高
向量数据库	语义检索、大容量	无结构、无时序、易冲突
对话摘要	压缩高效	信息丢失、不可逆
全文搜索	精确匹配	无语义理解

核心矛盾在于：Agent 需要结构化的、可更新的、带时序的记忆，而现有方案要么没有结构（向量），要么不可更新（摘要），要么没有时序（知识图谱）。

Graphiti 的设计理念

Zep（YC S23）开源的 Graphiti 框架将知识图谱重新设计为 Agent 的"活记忆"：

核心创新

1. 时序化实体关系：每条边和节点都带有时间戳，支持"此刻正确"的查询
2. 增量式更新：新信息作为"事件"（Episode）注入，自动更新图谱而不覆盖历史
3. 双重检索：同时支持语义搜索（Embedding）和图遍历（Structure），取最佳结果
4. 冲突解决：当新信息与旧信息矛盾时，通过 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%

时序一致性的差距尤其显著：当信息经历过多次更新时，向量检索容易返回过时的版本。

适用场景

1. 长期客服 Agent：记住每个客户的偏好变化和交互历史
2. 研究助手：追踪项目进展、文献发现、实验结果的演变
3. 合规顾问：维护法规变更的时序链，自动识别影响面
4. 项目管理：记录需求变更、决策历程、风险演化

不适用场景

1. 短期单次交互：上下文窗口足够，无需持久化记忆
2. 超大规模数据：数十亿三元组需要分布式图数据库，Graphiti 目前基于 Neo4j 单机
3. 纯数值时序：传感器数据更适合时序数据库

部署与运维

最小部署

# 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())
"

生产建议

1. Neo4j 高可用：使用 Neo4j Cluster 部署，确保读写分离
2. LLM 成本控制：实体抽取使用 Flash 模型，复杂推理使用 Pro 模型
3. Episode 批处理：非实时场景下批量注入 Episode，减少 LLM 调用
4. 定期压缩：对过期的历史关系做归档，保持图谱精简
5. 备份策略：定期导出 Neo4j 快照，防止数据丢失

与 GraphRAG 的关系

维度	GraphRAG	Graphiti
核心目标	增强文档检索	Agent 长期记忆
数据来源	静态文档集合	动态交互流
更新方式	全量重建	增量注入
时序支持	无	原生支持
查询模式	Local/Global Search	语义+图遍历融合
适用场景	知识库问答	Agent 对话记忆

两者可以互补：用 GraphRAG 处理静态文档知识，用 Graphiti 处理动态交互记忆。

---

Maurice | [email protected]