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 | maurice_wen@proton.me