Article

记忆库 Zep

更新于：2026-07-18

第一章：Zep 概述

1.1 什么是 Zep

概念名称说明注意事项
ZepZep 是一个专为 AI 应用设计的会话记忆管理框架，提供长期记忆存储、上下文感知检索、自动摘要生成等功能，帮助 LLM 应用维护跨会话的用户状态和历史信息。Zep 不是通用数据库，而是面向对话式 AI 优化的记忆系统；需配合 LLM 使用。
开源项目Zep 由 The Zep Team 开发并开源（GitHub 可查），支持自托管部署，也提供云服务选项。自托管版本功能完整，但需自行维护基础设施；云版可能有使用限制或费用。
架构定位Zep 通常作为 LLM 应用的”记忆层”，位于应用逻辑与向量/关系数据库之间，抽象了记忆的写入、读取与管理复杂性。不应将 Zep 直接暴露给前端；建议通过后端 API 封装调用。

1.2 Zep 的核心功能

功能名称说明注意事项
会话（Session）管理支持创建、查询、更新和删除独立的用户会话，每个会话拥有唯一 ID 并可携带元数据（如 user_id、设备信息等）。会话 ID 应由应用层生成并保持一致性，避免重复或冲突。
消息（Message）记录自动按时间顺序存储用户与 AI 的交互消息，支持角色（user / assistant / system）标记和任意元数据附加。消息不可修改，仅可追加；删除会话将清除其所有消息。
自动记忆摘要（Auto-summarization）基于滑动窗口或 token 阈值，自动调用 LLM 对历史消息生成摘要，减少上下文长度压力。需配置有效的 LLM API 密钥（如 OpenAI）；摘要策略可自定义。
向量语义检索（Memory Search）将关键记忆条目嵌入为向量，支持基于自然语言查询的相关记忆召回（例如：“用户上次提到喜欢什么？”）。默认使用内置嵌入模型（如 all-MiniLM-L6-v2），也可集成外部向量数据库。
元数据与标签（Metadata & Tags）支持为会话、消息、记忆添加结构化元数据（JSON）和非结构化标签，便于过滤与分析。元数据不参与向量检索，仅用于精确匹配或业务逻辑判断。
多租户支持通过 session_id 和 user_id 实现天然的多用户隔离，适用于 SaaS 或多用户 AI 产品。应用层需确保 user_id 的真实性与安全性，防止越权访问。

1.3 Zep 的典型应用场景

场景名称说明注意事项
长期记忆聊天机器人构建能记住用户偏好、历史对话、个人事实的 AI 助手（如”你上次说你喜欢咖啡”）。需合理设置记忆保留策略，避免隐私泄露或信息过载。
客户支持自动化在客服对话中自动调取用户过往问题、订单状态、投诉记录，提升响应准确性。敏感信息（如身份证、银行卡）应脱敏后再存入 Zep。
游戏 NPC 记忆系统使游戏中的非玩家角色（NPC）记住玩家行为、选择和关系变化，增强沉浸感。可结合标签系统标记”敌对""友好”等状态，便于快速检索。
教育辅导助手记录学生的学习进度、错题历史、知识薄弱点，实现个性化教学。需定期清理过期记忆，避免模型被陈旧数据误导。
企业内部知识助手员工与 AI 问答时，自动关联其部门、项目、权限等上下文，返回定制化答案。应与企业 IAM 系统集成，确保数据访问合规。

第二章：环境搭建与初始化

2.1 安装 Zep SDK

步骤名称操作细节注意事项
安装 Python SDK使用 pip 安装官方 zep-python 包：pip install zep-python确保 Python 版本 ≥ 3.8；建议在虚拟环境中安装。
安装 Node.js SDK使用 npm 安装：npm install @getzep/zep-js需 Node.js ≥ 16；TypeScript 用户可直接使用类型定义。
验证安装在 Python 中执行 import zep 或在 Node.js 中 require('@getzep/zep-js')，无报错即成功。若提示模块未找到，请检查环境是否激活或全局/局部安装冲突。

注：目前 Zep 官方主要维护 Python 和 JavaScript/Node.js SDK，其他语言可通过 REST API 调用。

2.2 启动本地 Zep 服务（Docker / Binary）

启动方式操作细节注意事项
使用 Docker（推荐）执行命令：docker run -p 8000:8000 ghcr.io/getzep/zep:latest，服务默认监听 http://localhost:8000首次运行会自动拉取镜像；确保 Docker 已启动；生产环境应挂载数据卷持久化存储。
使用预编译二进制（Linux/macOS）从 GitHub Releases 下载对应平台的 zep 二进制文件，赋予执行权限后运行：./zep serve --port 8000需手动管理进程；日志输出到 stdout；不适用于 Windows（除非 WSL）。
配置环境变量（可选）可通过 -e OPENAI_API_KEY=... 传入 LLM 密钥以启用自动摘要功能敏感密钥建议通过 .env 文件或密钥管理服务注入，避免硬编码。
验证服务状态访问 http://localhost:8000/healthz，返回 {"status":"ok"} 表示服务正常若端口被占用，可修改 -p 参数（如 -p 9000:8000）并同步更新客户端配置。

2.3 初始化客户端连接

方法名称语法用途代码示例注意事项
Python: ZepClientZepClient(base_url: str, api_key: Optional[str] = None)创建与 Zep 服务的连接客户端from zep_python import ZepClient
client = ZepClient("http://localhost:8000")		base_url 必须包含协议（http/https）；若服务启用了 API Key 认证，需传入 api_key。
Node.js: ZepClientnew ZepClient(baseURL: string, config?: { apiKey?: string })初始化 JS 客户端实例import { ZepClient } from "@getzep/zep-js";
const client = new ZepClient("http://localhost:8000");		baseURL 末尾不要加 /；若使用 TypeScript，类型提示完整。
客户端健康检查client.ahealth()（Python）
await client.ahealth()（Node.js）		异步验证连接是否正常status = await client.ahealth()
print(status)		返回值为字典/对象，包含 status 字段；网络不通时会抛出异常，建议加 try-catch。
设置默认超时通过底层 HTTP 客户端配置（如 aiohttp / axios）设置请求超时避免因网络问题长时间阻塞import aiohttp
timeout = aiohttp.ClientTimeout(total=30)
session = aiohttp.ClientSession(timeout=timeout)
client = ZepClient("http://localhost:8000", http_client=session)		SDK 默认无超时限制，生产环境务必设置合理超时（如 10–30 秒）。

第三章：会话（Session）管理

3.1 创建会话

方法名称语法用途代码示例注意事项
Python: client.memory.aadd_sessionaadd_session(session_id: str, metadata: Optional[Dict] = None, user_id: Optional[str] = None)创建新会话或初始化已有会话await client.memory.aadd_session(
session_id="sess_12345",
user_id="user_abc",
metadata={"device": "mobile", "version": "1.0"}
)		session_id 必须全局唯一；重复调用不会覆盖元数据，仅首次生效（除非使用更新接口）。
Node.js: client.memory.addSessionaddSession(sessionId: string, options?: { userId?: string; metadata?: Record<string, any> })创建或注册会话await client.memory.addSession("sess_12345", {
userId: "user_abc",
metadata: { device: "mobile", version: "1.0" }
});		若 sessionId 已存在，Zep 不会报错，但元数据不会自动合并或更新。
自动创建机制在首次向某 session_id 添加消息时，Zep 会自动创建会话简化流程，无需显式创建# 直接添加消息，会话自动创建
await client.message.aadd_message("sess_12345", {...})		自动创建的会话无 user_id 和元数据，建议显式创建以保证上下文完整性。

3.2 获取与更新会话元数据

方法名称语法用途代码示例注意事项
Python: client.memory.aget_sessionaget_session(session_id: str)获取会话完整信息（含元数据、用户ID、创建时间等）session = await client.memory.aget_session("sess_12345")
print(session.user_id, session.metadata)		若会话不存在，抛出 NotFoundError；需异常处理。
Node.js: client.memory.getSessiongetSession(sessionId: string)获取会话详情const session = await client.memory.getSession("sess_12345");
console.log(session.userId, session.metadata);		返回对象包含 id, userId, metadata, createdAt, updatedAt 等字段。
Python: client.memory.aupdate_sessionaupdate_session(session_id: str, metadata: Dict, user_id: Optional[str] = None)更新会话的元数据和/或用户IDawait client.memory.aupdate_session(
session_id="sess_12345",
metadata={"device": "desktop", "last_seen": "2026-03-17"},
user_id="user_abc"
)		元数据为全量替换，非增量合并；若只想更新部分字段，需先读取原 metadata 再合并。
Node.js: client.memory.updateSessionupdateSession(sessionId: string, options: { userId?: string; metadata: Record<string, any> })更新会话信息await client.memory.updateSession("sess_12345", {
metadata: { device: "desktop", last_seen: "2026-03-17" }
});		同样为全量覆盖 metadata；userId 可单独更新，不影响 metadata。

3.3 列出与删除会话

方法名称语法用途代码示例注意事项
Python: client.memory.alist_sessionsalist_sessions(user_id: Optional[str] = None, limit: int = 10, cursor: Optional[str] = None)分页列出会话（可按用户过滤）sessions = await client.memory.alist_sessions(
user_id="user_abc",
limit=20
)
for s in sessions:
print(s.id)		支持游标分页（cursor）；默认返回最近创建的会话；limit 最大值由服务端配置决定（通常 ≤ 100）。
Node.js: client.memory.listSessionslistSessions(options?: { userId?: string; limit?: number; cursor?: string })获取会话列表const { sessions } = await client.memory.listSessions({
userId: "user_abc",
limit: 20
});
sessions.forEach(s => console.log(s.id));		返回对象包含 sessions 数组和 cursor（用于下一页）。
Python: client.memory.adelete_sessionadelete_session(session_id: str)删除指定会话及其所有消息、记忆await client.memory.adelete_session("sess_12345")不可逆操作；删除后无法恢复；关联的记忆向量也会被清除。
Node.js: client.memory.deleteSessiondeleteSession(sessionId: string)删除会话await client.memory.deleteSession("sess_12345");删除操作异步执行，但立即生效；后续对该 session_id 的操作将视为新会话。
批量删除（间接支持）通过遍历 list_sessions 结果逐个调用 delete_session清理过期或无效会话sessions = await client.memory.alist_sessions(limit=100)
for s in sessions:
if is_expired(s):
await client.memory.adelete_session(s.id)		Zep 当前不支持原生批量删除 API，需应用层循环处理；注意速率限制。

第四章：消息（Message）操作

4.1 添加消息到会话

方法名称语法用途代码示例注意事项
Python: client.message.aadd_messageaadd_message(session_id: str, message: Message)向指定会话追加一条消息from zep_python import Message
msg = Message(
role="user",
content="What's the weather today?",
metadata={"location": "Beijing"}
)
await client.message.aadd_message("sess_12345", msg)		消息按添加顺序存储；session_id 对应的会话若不存在，Zep 会自动创建（但无 user_id）。
Node.js: client.message.addMessageaddMessage(sessionId: string, message: { role: string; content: string; metadata?: Record<string, any> })添加消息（JS 对象形式）await client.message.addMessage("sess_12345", {
role: "assistant",
content: "It's sunny in Beijing.",
metadata: { source: "weather_api_v2" }
});		role 必须为预定义值（见 4.3 节）；content 不能为空字符串。
批量添加消息（Python）aadd_messages(session_id: str, messages: List[Message])一次性添加多条消息（保持顺序）msgs = [
Message(role="user", content="Hi"),
Message(role="assistant", content="Hello!")
]
await client.message.aadd_messages("sess_12345", msgs)		原子性不保证；部分失败可能导致部分消息写入；建议用于初始化历史数据。
批量添加消息（Node.js）addMessages(sessionId: string, messages: Array<{ role: string; content: string; ... }>)批量写入 JS 消息数组await client.message.addMessages("sess_12345", [
{ role: "user", content: "Tell me a joke" },
{ role: "assistant", content: "Why don't skeletons fight? They don't have the guts!" }
]);		同上，非事务性操作；网络中断可能导致部分丢失。

4.2 获取会话中的消息历史

方法名称语法用途代码示例注意事项
Python: client.message.aget_session_messagesaget_session_messages(session_id: str, limit: int = 10, cursor: Optional[str] = None)分页获取消息历史（按时间倒序）messages = await client.message.aget_session_messages(
session_id="sess_12345",
limit=5
)
for m in messages:
print(m.role, m.content)		默认返回最新 limit 条；cursor 用于翻页（从响应中获取）；最大 limit 通常 ≤ 100。
Node.js: client.message.getSessionMessagesgetSessionMessages(sessionId: string, options?: { limit?: number; cursor?: string })获取消息列表const { messages } = await client.message.getSessionMessages("sess_12345", { limit: 5 });
messages.forEach(m => console.log(m.role, m.content));		返回对象含 messages 数组和 cursor 字段（用于下一页）。
获取全部消息（应用层实现）循环调用分页接口直至 cursor 为空获取完整对话历史all_msgs = []
cursor = None
while True:
resp = await client.message.aget_session_messages("sess_12345", limit=50, cursor=cursor)
all_msgs.extend(resp.messages)
cursor = resp.cursor
if not cursor:
break		大量消息时性能较差；建议仅在必要时使用（如导出、调试）。
消息字段说明每条消息包含：uuid, role, content, metadata, created_at, token_count提供完整上下文信息# 示例访问字段
msg = messages[0]
print(f"[{msg.created_at}] {msg.role}: {msg.content}")		token_count 由 Zep 自动估算（基于 tiktoken），可用于上下文长度控制。

4.3 消息类型与角色定义

概念名称说明注意事项
role: “user”表示消息由最终用户发出内容通常为自然语言提问或指令；是触发记忆写入的主要来源。
role: “assistant”表示消息由 AI 助手生成内容为模型回复；可用于后续摘要或作为记忆依据。
role: “system”表示系统级指令或上下文提示（如”你是一个 helpful assistant”）通常在会话开始时设置；部分 LLM 会特殊处理此角色；Zep 会正常存储但不参与自动摘要。
role: “tool”（实验性）表示工具调用结果（如函数返回值）需配合支持工具调用的 LLM（如 OpenAI Function Calling）；Zep 当前按普通消息处理。
消息内容格式content 必须为字符串；支持 Markdown、JSON 等，但 Zep 不解析其结构若需结构化数据，建议存入 metadata 字段（如 metadata={"tool_name": "get_weather", "result": {...}}）。
元数据（metadata）用途存储与消息相关的业务上下文，如来源、意图分类、情感标签等不参与向量检索；可用于过滤或分析；大小建议 < 1KB。
时间戳（created_at）消息创建时间，ISO 8601 格式（如 "2026-03-17T09:42:00Z"）由 Zep 服务端生成，不可修改；用于排序和时效性判断。

第五章：记忆（Memory）存储与检索

5.1 自动记忆摘要生成

概念/方法名称说明注意事项
自动摘要机制Zep 监控会话消息长度（按 token 计），当超过阈值时自动调用配置的 LLM（如 OpenAI）对历史消息生成摘要，并作为”记忆”存入会话。需在 Zep 服务启动时配置 LLM_API_KEY 和模型（如 gpt-3.5-turbo）；否则摘要功能禁用。
摘要触发条件默认在消息总 token 数 > 70% 上下文窗口（如 2800/4096）时触发；可通过 summary_threshold 等参数调整。摘要基于最近 N 条消息生成，非全量历史；旧摘要会被新摘要覆盖或合并。
摘要存储位置生成的摘要以特殊记忆条目形式存储，type="summary"，可通过 client.memory.aget_memory 获取。摘要内容包含关键事实（如”用户喜欢咖啡”），但不包含原始消息细节。
查看当前摘要（Python）memory = await client.memory.aget_memory("sess_12345")
print(memory.summary.content)		若未触发摘要，memory.summary 可能为 None。
查看当前摘要（Node.js）const memory = await client.memory.getMemory("sess_12345");
console.log(memory.summary?.content);		使用可选链防止空指针错误。
自定义摘要提示词通过 Zep 服务配置文件（如 zep.yaml）修改 summary_instruction 字段示例："Summarize the conversation in 3 bullet points focusing on user preferences."

5.2 手动添加记忆条目

方法名称语法用途代码示例注意事项
Python: client.memory.aadd_memoryaadd_memory(session_id: str, memory: Memory)手动注入结构化记忆from zep_python import Memory, Message
mem = Memory(
facts=["User prefers dark mode", "User works at Alibaba"],
metadata={"source": "user_settings"}
)
await client.memory.aadd_memory("sess_12345", mem)		facts 是核心记忆内容，每条应为独立、原子的事实；将被嵌入向量用于检索。
Node.js: client.memory.addMemoryaddMemory(sessionId: string, memory: { facts?: string[]; metadata?: Record<string, any> })添加记忆（JS 对象）await client.memory.addMemory("sess_12345", {
facts: ["User dislikes spam emails"],
metadata: { source: "feedback_form" }
});		不支持直接传入 messages；若需关联消息，应先存消息再提取事实。
从消息提取事实（推荐流程）应用层调用 LLM 从对话中抽取事实，再调用 aadd_memory实现精准记忆注入# 假设 LLM 返回 ["Lives in Hangzhou"]
await client.memory.aadd_memory("sess_12345", Memory(facts=["Lives in Hangzhou"]))		避免将整段对话存为 fact；应提炼为简洁陈述句。
内存限制单次 facts 列表建议 ≤ 10 条；总记忆条目数无硬限，但影响检索性能过多低质量记忆会降低检索准确率；建议定期清理。

5.3 基于向量的语义检索

方法名称语法用途代码示例注意事项
Python: client.memory.asearch_memoryasearch_memory(session_id: str, query: str, limit: int = 3)根据自然语言查询，从会话记忆中语义检索相关事实results = await client.memory.asearch_memory(
session_id="sess_12345",
query="What does the user like?",
limit=2
)
for r in results:
print(r.fact, r.score)		返回结果按相关性得分 score（0~1）降序排列；limit 最大通常为 10。
Node.js: client.memory.searchMemorysearchMemory(sessionId: string, query: string, options?: { limit?: number })语义搜索记忆const results = await client.memory.searchMemory("sess_12345", "User's job?", { limit: 1 });
results.forEach(r => console.log(r.fact, r.score));		score 越高表示越相关；低于 0.3 的结果通常可忽略。
检索范围仅搜索该 session_id 下的手动 facts 和自动摘要内容不跨会话检索；若需全局记忆，需自行聚合多个会话结果。
嵌入模型默认使用 Sentence Transformers 的 all-MiniLM-L6-v2（本地运行）无需外部 API；若需更高精度，可配置外部嵌入服务（如 OpenAI embeddings）。
查询优化建议使用完整问句（如 “What is the user’s favorite color?”）比关键词（“color”）效果更好避免模糊查询；Zep 不支持布尔逻辑或过滤器（如 “AND”）。

5.4 记忆的更新与删除

操作类型方法/机制语法与说明代码示例注意事项
更新记忆Zep 不支持直接更新已有记忆条目需先删除旧记忆，再添加新记忆# 假设已知某记忆的 uuid（实际中难获取）
# 更佳做法：清空后重建
await client.memory.adelete_memory("sess_12345")
await client.memory.aadd_memory("sess_12345", new_mem)		Zep 的记忆设计为追加式（append-only）；无原地更新 API。
删除全部记忆client.memory.adelete_memory(session_id: str)清除指定会话的所有手动记忆和摘要await client.memory.adelete_memory("sess_12345")不影响消息历史；仅清除 facts 和 summary；向量索引同步清除。
删除单条记忆当前不支持无 API 可删除特定 fact—若需精细控制，建议应用层维护记忆 ID 映射，或采用”标记失效”策略（如添加 {"deleted": true} 元数据）。
通过会话删除间接清除删除会话（adelete_session）会级联删除其所有记忆见第三章 3.3 节await client.memory.adelete_session("sess_12345") # 同时删消息+记忆最彻底的清理方式；适用于用户注销或会话过期场景。
记忆版本控制建议应用层可为 facts 添加版本号或时间戳元数据Memory(facts=["Prefers email"], metadata={"version": "2026-03-17"})检索后由应用决定是否采纳旧记忆；Zep 本身无版本概念。

第六章：元数据与标签系统

6.1 为会话/消息/记忆添加元数据

对象类型添加方式语法与说明代码示例注意事项
会话元数据创建或更新会话时传入 metadata 参数metadata 为任意 JSON-serializable 字典# Python 创建时添加
await client.memory.aadd_session(
session_id="sess_123",
metadata={"plan": "premium", "locale": "zh-CN"}
)

# 更新时覆盖
await client.memory.aupdate_session(
session_id="sess_123",
metadata={"plan": "enterprise", "last_active": "2026-03-17"}
)		元数据为全量替换，非合并；建议先读取再更新部分字段。
消息元数据在 Message 对象中设置 metadata 字段每条消息可携带独立元数据// Node.js 示例
await client.message.addMessage("sess_123", {
role: "user",
content: "I need help with billing",
metadata: { intent: "billing_support", confidence: 0.95 }
});		消息元数据不可修改；若需变更，需重新添加新消息。
记忆元数据在 Memory 对象中设置 metadata用于标记记忆来源、可信度等mem = Memory(
facts=["User subscribed to newsletter"],
metadata={"source": "web_form", "verified": True}
)
await client.memory.aadd_memory("sess_123", mem)		记忆元数据随整个记忆条目一起被删除；不参与向量检索。
元数据限制所有元数据存储于 Zep 内部数据库（默认 SQLite / PostgreSQL）单字段值建议 < 1KB，总大小 < 10KB—过大的元数据会影响查询性能；避免存储二进制或长文本。
查询支持Zep 当前不支持基于元数据的过滤查询（如”查找 plan=premium 的会话”）需应用层自行维护索引或使用外部数据库关联—元数据主要用于业务上下文传递，非检索条件；规划中未来可能支持。

6.2 使用标签进行分类与过滤

概念/操作说明注意事项
标签（Tags）定义标签是附加到会话、消息或记忆上的字符串列表，用于非结构化分类（如 ["urgent", "finance"]）标签区分大小写；建议使用小写+下划线命名（如 "customer_support"）。
会话标签在创建/更新会话时通过 metadata 间接实现（Zep 无原生 tag 字段）推荐约定：metadata["tags"] = ["tag1", "tag2"]
消息标签同样通过 message.metadata["tags"] 存储await client.message.addMessage("sess_123", {
role: "user",
content: "Cancel my subscription",
metadata: { tags: ["cancellation", "high_priority"] }
});
记忆标签在 Memory.metadata["tags"] 中定义Memory(
facts=["User requested data deletion"],
metadata={"tags": ["gdpr", "data_privacy"]}
)
标签 vs 元数据标签适合多值分类（一对多），元数据适合键值属性（一对一）示例：
- 标签：["bug_report", "ui_issue"]
- 元数据：{"priority": "high", "assigned_to": "team_a"}
过滤与检索限制Zep 不提供基于标签的内置搜索 API需在应用层加载对象后过滤
最佳实践- 预定义标签枚举集
- 避免动态生成无限标签
- 定期清理废弃标签		可结合配置中心管理有效标签列表

第七章：高级功能

7.1 配置自定义记忆摘要策略

配置项 / 方法说明代码/配置示例注意事项
摘要触发阈值控制何时触发自动摘要（基于累计 token 数）在 zep.yaml 中配置：
memory:
summary_threshold: 2500 # 默认约 2800		值应小于 LLM 上下文窗口（如 4096）；过低会导致频繁调用 LLM，过高则失去摘要意义。
摘要窗口大小指定用于生成摘要的消息数量（滑动窗口）memory:
summary_window_size: 10 # 默认 10 条消息		窗口越大，摘要越全面，但成本越高；建议 5–15 条。
自定义摘要提示词修改 LLM 生成摘要的指令memory:
summary_instruction: "Summarize key user facts in bullet points, focusing on preferences and personal details."		提示词直接影响摘要质量；可加入语言要求（如”用中文总结”）。
LLM 模型选择指定用于摘要的模型（需支持 Chat Completions）启动服务时设置环境变量：
ZEP_EMBEDDING_MODEL="text-embedding-3-small"
ZEP_LLM_MODEL="gpt-4o"
（具体变量名依版本而定）		必须提供有效的 OPENAI_API_KEY 或兼容 API 的密钥；本地模型需通过 Ollama 等代理。
禁用自动摘要完全关闭自动摘要功能memory:
auto_summarize: false		适用于仅使用手动记忆的场景；节省 LLM 调用成本。
动态策略（应用层）应用根据用户类型切换摘要策略# 示例：为 VIP 用户启用更频繁摘要
threshold = 2000 if is_vip(user_id) else 2800
# 但 Zep 当前不支持 per-session 策略		Zep 摘要策略是全局配置，无法按会话动态调整；需自行实现摘要逻辑并调用 aadd_memory。

7.2 集成外部向量数据库（如 Pinecone、Weaviate）

集成方式说明配置/代码示例注意事项
Zep 内置向量引擎默认使用本地 Sentence Transformers + FAISS/SQLite无需额外配置；开箱即用适合中小规模应用；不支持分布式或高并发场景。
外部向量库支持状态截至当前版本（v0.x），Zep 不直接支持对接 Pinecone、Weaviate 等外部向量数据库—所有向量操作由 Zep 内部管理；官方暂未开放向量存储插件接口。
间接集成方案应用层双写：同时将记忆写入 Zep 和外部向量库# 1. 写入 Zep（用于会话管理）
await client.memory.aadd_memory(session_id, mem)

# 2. 单独写入 Pinecone
pinecone_index.upsert([(f"{session_id}_fact1", embed("User likes tea"), {"session": session_id})])		需自行维护嵌入一致性（使用相同模型）；增加系统复杂度。
检索时融合结果分别从 Zep 和外部库检索，合并排序zep_results = await client.memory.asearch_memory(session_id, query)
pinecone_results = pinecone_index.query(embed(query), filter={"session": session_id})
combined = merge_and_rank(zep_results, pinecone_results)		需处理不同 score 范围（如 Zep 01，Pinecone 余弦相似度 -11）；建议归一化。
未来展望Zep 团队已在 GitHub 讨论中提出向量后端插件化计划关注 Zep GitHub Issues生产环境若需企业级向量库，建议评估 LangChain + 自定义记忆层替代方案。

⚠️ 重要提示：目前 Zep 的向量存储是封闭的，无法直接替换或扩展其底层向量引擎。如需深度集成 Pinecone/Weaviate，建议将 Zep 仅用于会话和消息管理，而将”长期记忆”功能完全交由外部向量库实现。

7.3 事件监听与回调机制

机制类型说明实现方式注意事项
Zep 内置事件系统当前版本（v0.x）未提供 WebSocket、Hook 或回调 API—无法直接监听”新消息""新记忆”等事件。
轮询模拟事件应用定期调用 aget_session_messages 检查新消息last_msg_time = None
while True:
msgs = await client.message.aget_session_messages(session_id, limit=1)
if msgs and (not last_msg_time or msgs[0].created_at > last_msg_time):
handle_new_message(msgs[0])
last_msg_time = msgs[0].created_at
await asyncio.sleep(2)		效率低、延迟高；仅适用于低频场景。
日志钩子（实验性）通过解析 Zep 服务日志捕获事件（不推荐）监听容器 stdout 中包含 “msg added” 的行极不稳定；日志格式可能变更；违反封装原则。
自定义中间层在应用调用 Zep 前插入业务逻辑async def add_message_with_hook(session_id, msg):
# 1. 执行业务回调
await on_before_message_save(session_id, msg)
# 2. 调用 Zep
await client.message.aadd_message(session_id, msg)
# 3. 触发后续动作
await on_after_message_saved(session_id, msg)		推荐做法：将 Zep SDK 封装在应用服务内部，由应用控制事件流。
Webhook 规划Zep 社区已提出 Webhook 支持需求（GitHub Issue #XXX）—可关注更新；当前需自行实现事件总线（如 Kafka、Redis Pub/Sub）。

💡 总结：Zep 目前无原生事件监听机制。生产系统应通过封装客户端调用的方式，在应用层实现事件触发与回调逻辑。

第八章：生产部署与监控

8.1 部署 Zep 服务（Kubernetes / Docker Compose）

部署方式操作细节配置/命令示例注意事项
Docker Compose（开发/测试）使用官方 docker-compose.yml 快速启动# docker-compose.yml
services:
zep:
image: ghcr.io/getzep/zep:latest
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=sk-xxx
volumes:
- zep_data:/root/.zep
volumes:
zep_data:

运行：docker compose up -d		适用于单机测试；数据通过 volume 持久化；不要用于生产。
Kubernetes（生产推荐）编写 Deployment + Service + PersistentVolume# zep-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: zep
spec:
replicas: 2
template:
spec:
containers:
- name: zep
image: ghcr.io/getzep/zep:latest
ports:
- containerPort: 8000
env:
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: zep-secrets
key: openai-api-key
volumeMounts:
- name: zep-storage
mountPath: /root/.zep
volumes:
- name: zep-storage
persistentVolumeClaim:
claimName: zep-pvc		- 使用 Secret 管理密钥
- PVC 确保数据持久化
- 建议启用 liveness/readiness 探针
外部数据库支持Zep 支持 PostgreSQL 替代默认 SQLite在环境变量中设置：
DATABASE_URL=postgresql://user:pass@host:5432/zep		生产环境必须使用 PostgreSQL；SQLite 不支持并发写入和高可用。
服务暴露通过 Ingress（K8s）或 Nginx 反向代理暴露 APIlocation / {
proxy_pass http://zep-service:8000;
proxy_set_header Host $host;
}		启用 TLS；限制 IP 访问；避免直接暴露 8000 端口到公网。
健康检查端点Zep 提供 /healthz 用于探活# K8s readiness probe
readinessProbe:
httpGet:
path: /healthz
port: 8000
initialDelaySeconds: 5		返回 {"status":"ok"} 表示就绪；可用于自动扩缩容判断。

8.2 性能调优与缓存配置

调优方向配置项 / 方法说明注意事项
LLM 调用缓存启用摘要结果缓存，避免重复生成目前 Zep 无内置 LLM 响应缓存建议在应用层缓存摘要输入（如消息哈希）+ 输出，减少 LLM 调用。
嵌入缓存避免重复计算相同文本的向量Zep 内部对 facts 和 query 自动缓存嵌入（基于文本内容）缓存存储于内存或数据库；重启后失效；无法配置 TTL。
数据库连接池配置 PostgreSQL 连接池参数通过 DATABASE_URL 附加参数：
?pool_size=20&max_overflow=30		默认连接数较低；高并发场景需调大，避免 Too many connections 错误。
消息分页优化限制单次 get_session_messages 的 limit应用层设置 limit ≤ 50大 limit 会导致内存激增和响应延迟；前端应实现滚动加载。
摘要频率控制调高 summary_threshold 减少 LLM 调用例如设为 3500（接近上下文上限）平衡记忆完整性与成本；可结合业务逻辑动态决定是否触发摘要。
资源限制（K8s）为 Pod 设置 CPU/Memory 请求与限制resources:
requests:
memory: "1Gi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "1000m"		Zep 内存占用随会话数线性增长；建议监控 RSS 并设置 OOM 安全边界。

8.3 日志与监控集成（Prometheus / Grafana）

监控组件集成方式配置/示例注意事项
Prometheus 指标暴露Zep 内置 /metrics 端点（需启用）默认开启；访问 http://zep:8000/metrics 可见：
zep_requests_total{method="POST",endpoint="/memory/add"} 12
zep_latency_seconds_bucket{...} ...		指标包括：请求计数、延迟分布、错误率、LLM 调用次数等。
Prometheus 抓取配置在 prometheus.yml 中添加 jobscrape_configs:
- job_name: 'zep'
static_configs:
- targets: ['zep-service:8000']		确保网络可达；若在 K8s，可使用 ServiceMonitor（Prometheus Operator）。
Grafana 仪表盘导入官方或自定义 Dashboard JSON关键面板：
- QPS & 错误率
- P99 延迟
- 活跃会话数
- LLM 调用成本估算		可基于 zep_requests_total 和 zep_latency_seconds 构建。
结构化日志输出Zep 默认输出 JSON 格式日志（INFO 级别）示例：
{"level":"info","ts":1710666420,"msg":"message added","session_id":"sess_123"}		可被 Fluentd / Loki / ELK 直接解析；无需额外格式化。
日志级别调整通过环境变量控制日志详细程度LOG_LEVEL=debug（默认 info）生产环境建议保持 info；debug 会输出敏感内容（如完整消息）。
告警规则示例在 Prometheus 中配置告警groups:
- name: zep-alerts
rules:
- alert: ZepHighErrorRate
expr: rate(zep_requests_total{status=~"5.."}[5m]) / rate(zep_requests_total[5m]) > 0.05
for: 2m		常见告警：错误率 >5%、P99 延迟 >2s、LLM 调用失败等。

第九章：实战案例

9.1 构建一个带长期记忆的聊天机器人

步骤名称操作细节代码逻辑示例注意事项
1. 初始化会话用户首次访问时创建会话，绑定用户标识session_id = f"sess_{user_id}"
await client.memory.aadd_session(
session_id=session_id,
user_id=user_id,
metadata={"source": "web_chat"}
)		session_id 应全局唯一；建议包含用户 ID 前缀便于追踪。
2. 接收用户消息并存储将用户输入作为 role="user" 消息存入 Zepawait client.message.aadd_message(
session_id,
Message(role="user", content=user_input)
)		存储后立即可用于后续检索或摘要。
3. 检索相关记忆根据用户问题语义搜索历史记忆memory_results = await client.memory.asearch_memory(
session_id, query=user_input, limit=3
)
relevant_facts = "\n".join([r.fact for r in memory_results if r.score > 0.4])		过滤低分结果（如 score < 0.4）避免噪声干扰。
4. 获取近期对话上下文读取最近 N 条消息构建 LLM 上下文recent_msgs = await client.message.aget_session_messages(
session_id, limit=6
)
chat_history = [{"role": m.role, "content": m.content} for m in reversed(recent_msgs)]		按时间倒序排列，确保最新消息在最后。
5. 调用 LLM 生成回复将记忆 + 历史 + 当前问题拼接为 promptsystem_prompt = f"You are a helpful assistant. Known facts:\n{relevant_facts}"
messages = [{"role": "system", "content": system_prompt}] + chat_history
response = openai.chat.completions.create(model="gpt-4o", messages=messages)
assistant_reply = response.choices[0].message.content		避免上下文超长；可结合自动摘要替代部分历史。
6. 存储 AI 回复并提取新事实保存回复，并（可选）调用 LLM 提取新记忆# 存回复
await client.message.aadd_message(session_id, Message(role="assistant", content=assistant_reply))

# 提取新事实（简化版）
if "my name is" in user_input.lower():
name = extract_name(user_input)
await client.memory.aadd_memory(session_id, Memory(facts=[f"User's name is {name}"]))		自动事实抽取需谨慎；建议人工规则 + LLM 结合，避免幻觉写入记忆。

9.2 多用户会话隔离与上下文管理

关键机制实现方式代码/设计示例注意事项
会话 ID 设计每个用户每个设备/会话使用唯一 IDsession_id = f"user_{user_id}_device_{device_id}_{timestamp}"避免多个终端共享同一会话导致上下文污染。
用户 ID 绑定所有会话显式关联 user_idawait client.memory.aadd_session(
session_id="sess_abc",
user_id="user_123", # ← 关键字段
metadata={...}
)		user_id 是实现多租户隔离的核心；Zep 不验证其真实性，需应用层保证。
按用户列出会话使用 alist_sessions(user_id=...) 获取用户所有会话sessions = await client.memory.alist_sessions(user_id="user_123")可用于”切换历史对话”功能；前端展示会话列表。
跨会话记忆聚合（谨慎）若业务需要全局用户画像，聚合该用户所有会话的记忆sessions = await client.memory.alist_sessions(user_id="user_123")
all_facts = []
for s in sessions:
mem = await client.memory.aget_memory(s.id)
if mem and mem.facts:
all_facts.extend(mem.facts)		默认不推荐；会话应保持独立；仅在明确需求（如 CRM）下使用。
权限与安全应用层校验当前用户是否有权访问目标 session_id# 请求 /chat/sess_abc
if get_user_id(request) != get_session_owner("sess_abc"):
raise PermissionDenied()		Zep 无内置 RBAC；必须由应用实现访问控制，防止越权读写。
会话清理策略定期删除过期会话（如 90 天未活跃）# 定时任务
sessions = await client.memory.alist_sessions(limit=1000)
for s in sessions:
if is_inactive(s, days=90):
await client.memory.adelete_session(s.id)		避免存储无限增长；符合 GDPR 等数据最小化原则。

9.3 与 LangChain / LlamaIndex 集成

集成框架集成方式代码示例注意事项
LangChain 集成使用 ZepMemory 作为 LangChain 的记忆组件from langchain_community.chat_message_histories import ZepChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

def get_session_history(session_id: str):
return ZepChatMessageHistory(
session_id=session_id,
url="http://localhost:8000",
api_key=None
)

chain_with_history = RunnableWithMessageHistory(
chain,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)		需安装 langchain-community；ZepChatMessageHistory 自动处理消息读写。
LlamaIndex 集成将 Zep 作为自定义记忆后端（需封装）# LlamaIndex 无官方 Zep 支持
# 需实现 BaseMemory 接口
class ZepMemory(BaseMemory):
def put(self, key: str, val: Any):
# 转换为 Zep fact 并存储
asyncio.run(zep_client.memory.aadd_memory(...))
def get(self, key: str) -> Any:
# 从 Zep 搜索并返回
results = asyncio.run(zep_client.memory.asearch_memory(...))
return results[0].fact if results else ""		非开箱即用；需自行处理异步与同步转换；建议优先用 LangChain。
混合上下文构建在 LLM 调用前融合 Zep 记忆 + 向量库检索# 1. 从 Zep 获取长期记忆
zep_facts = await client.memory.asearch_memory(session_id, query)

# 2. 从 LlamaIndex/Pinecone 获取知识库片段
kb_chunks = vector_index.similarity_search(query)

# 3. 合并到 prompt
context = "Personal facts:\n" + "\n".join(zep_facts) + "\n\nKnowledge base:\n" + "\n".join(kb_chunks)		实现”个性化 + 通用知识”双重上下文；注意总 token 限制。
自动摘要复用利用 Zep 自动生成的摘要替代原始历史memory = await client.memory.aget_memory(session_id)
summary = memory.summary.content if memory.summary else ""
messages = [{"role": "system", "content": f"Conversation summary: {summary}"}, ...]		显著降低上下文长度；适合长周期对话。
依赖版本兼容性确保 SDK 与 LangChain 版本匹配检查 langchain-community >= 0.0.30 是否支持当前 Zep APILangChain 接口可能变更；建议锁定版本并测试。