端到端上下文工程平台，基于 Graphiti 知识图谱引擎，以 sub-200ms 延迟提供关系感知的上下文
Zep 是一个端到端的上下文工程平台，其核心创新在于使用时序知识图谱（Temporal Knowledge Graph）而非简单的向量存储来管理 Agent 记忆。
与传统的向量数据库方案不同，Zep 能够理解实体之间的关系，追踪事实的时间演变，并以极低的延迟（sub-200ms）检索最相关的上下文。
记忆不只是"相似文本的检索"，而是对谁、什么、何时的结构化理解。Zep 让 Agent 拥有真正的时间感知能力。
Zep Cloud 是主要产品（托管服务），Community Edition 已弃用。Graphiti 引擎本身开源可独立使用。
Zep 的知识图谱引擎基于 Graphiti 框架构建，这是专门为 LLM Agent 设计的时序知识图谱系统。
从对话、文档、业务数据中自动抽取实体和关系，无需手动 schema 定义。支持多源数据融合。
Graph RAG 不仅找相似文本，更理解实体间的关系链路，提供更精准的上下文。
专为实时对话优化，图谱查询和上下文组装在 200ms 内完成，不影响对话流畅性。
统一组装聊天历史、业务数据、文档、应用事件等多维信息，为 Agent 提供完整上下文。
Zep 的核心差异化能力是时间维度：每个事实都有 valid_at 和 invalid_at 时间戳，记录信息的有效期。
用户说："我在 Google 工作，担任高级工程师。"
用户说："我刚跳槽到 OpenAI 了，做 ML 研究。"
→ 当前回答: OpenAI, ML 研究
→ 历史记录: 曾在 Google 担任高级工程师（2024.01 - 2024.06）
# 时间感知查询示例
# Zep 能区分"当前事实"和"历史事实"
facts = zep.graph.search("user employment")
# 返回:
# { "entity": "User", "relation": "works_at", "target": "OpenAI",
# "valid_at": "2024-06-20", "invalid_at": null } ← 当前有效
# { "entity": "User", "relation": "works_at", "target": "Google",
# "valid_at": "2024-01-15", "invalid_at": "2024-06-20" } ← 已过期两者都是 Agent 记忆方案，但底层架构理念截然不同：
from zep_cloud.client import Zep
from zep_cloud.types import Message
client = Zep(api_key="your-api-key")
# 创建用户
client.user.add(user_id="user-1", first_name="Alice")
# 添加记忆（自动提取实体和关系）
client.memory.add(
user_id="user-1",
messages=[
Message(role="user", content="我刚搬到东京，在索尼做游戏开发"),
Message(role="assistant", content="恭喜你搬到东京！索尼的游戏部门很棒。"),
]
)
# 检索记忆（包含知识图谱上下文）
memory = client.memory.get(
user_id="user-1",
lastn=10, # 最近10轮对话
)
print(memory.context) # 包含图谱推理的完整上下文from graphiti_core import Graphiti
# 连接 Neo4j 数据库
graphiti = Graphiti(
neo4j_uri="bolt://localhost:7687",
neo4j_user="neo4j",
neo4j_password="password"
)
# 初始化图谱索引
await graphiti.build_indices_and_constraints()
# 添加事实（自动抽取实体和关系）
await graphiti.add_episode(
name="conversation-1",
episode_body="张三是 ABC 公司的 CTO，他上周刚宣布公司完成了 B 轮融资。",
source="conversation",
reference_time="2024-06-15"
)
# 搜索（支持时间过滤和关系感知）
results = await graphiti.search(
query="ABC 公司的融资情况",
num_results=5
)
for edge in results:
print(f"{edge.source_node} → {edge.name} → {edge.target_node}")
print(f" 有效时间: {edge.valid_at} - {edge.invalid_at or '至今'}")