别再为AI Agent的记忆问题头疼了!这个开源项目让LLM真正拥有“持久记忆”
从ChatGPT到Claude,从GPT-4到Gemini,大语言模型的能力已经强大到令人惊叹。然而,有一个问题始终困扰着所有AI应用开发者:如何让AI Agent拥有持续的记忆能力?
当用户与AI进行多轮对话时,模型往往会“忘记”之前的上下文;当Agent需要执行复杂任务时,它无法记住中间步骤的结果;当需要跨会话复用信息时,传统方案显得笨拙且难以维护。
今天要介绍的这个开源项目 agentmemory,正是为了解决这些痛点而生。它为AI Agent提供了一套完整的记忆管理系统,让LLM真正拥有“持久记忆”。
为什么值得关注 / 为什么值得关注
解决Agent开发的核心痛点
在传统的AI应用开发中,记忆管理是一个被严重低估但又极其重要的问题。当我们构建一个需要多轮交互的AI助手时,通常会面临以下挑战:
上下文窗口限制:虽然Claude 200K上下文看起来很长,但在处理复杂任务时,Token数量的消耗速度远超想象。每一轮对话都需要重新传递历史消息,很快就会触及限制。更糟糕的是,模型在超长上下文中往往会出现“迷失在中间”的问题,对早期信息注意力下降。
信息检索效率低下:当Agent需要从历史记忆中查找特定信息时,简单的字符串匹配或关键词搜索已经无法满足需求。我们需要的是语义级别的检索——能够理解“查找上周讨论的项目方案”这样的模糊查询。
记忆持久化困难:重启服务后所有对话历史消失,用户无法继续之前的任务。不同Agent之间无法共享记忆,导致重复劳动和上下文断层。
记忆质量参差不齐:系统会记住大量无用信息,而真正重要的决策依据却可能被淹没。缺乏有效的记忆组织和摘要机制。
agentmemory的解决方案
agentmemory正是针对这些痛点设计的记忆管理框架。它提供了一套完整的技术栈,包括:
分层记忆架构:将记忆分为工作记忆、情景记忆、语义记忆和程序记忆四个层次。不同层级的记忆有不同的生命周期和访问频率,确保系统既有即时响应能力,又有长期知识沉淀。
向量语义检索:基于 embeddings 技术实现语义级别的相似度搜索。即使查询表述与存储内容不同,也能准确匹配相关记忆。
灵活的存储后端:支持 SQLite、PostgreSQL、MongoDB 等多种数据库作为底层存储,可以根据部署环境灵活选择。
会话管理:提供完整的会话生命周期管理,支持跨会话恢复、对话摘要和历史回溯。
易于集成:简洁的API设计,可以快速集成到现有项目中,无论是LangChain、AutoGen还是其他Agent框架。
实际应用场景
想象一下这些场景:智能客服Agent能够记住用户的所有偏好和历史问题,每次对话都是对上次交流的延续;个人助理Agent了解你的工作习惯和日程安排,主动提供恰到好处的帮助;代码审查Agent记得整个代码库的历史变更,能够追踪问题的演进过程;研究助手Agent整合多篇论文的关键发现,形成连贯的知识网络。
这些不再是想象,agentmemory让它们成为可能。
环境搭建 / Getting Started
系统要求
在开始之前,确保你的开发环境满足以下要求:
Python 版本需要 3.8 或更高版本,推荐使用 3.10 或 3.11 以获得最佳性能和兼容性。项目依赖主要包括 openai 或其他 LLM 提供商的 SDK,以及向量数据库相关的库。
安装步骤
首先,推荐创建一个独立的虚拟环境来管理项目依赖,避免与全局 Python 环境产生冲突。使用 pip 进行安装是最简单的方式:
# 创建虚拟环境
python -m venv agentmemory_env
# 激活虚拟环境
# Linux/Mac
source agentmemory_env/bin/activate
# Windows
agentmemory_env\Scripts\activate
# 安装 agentmemory
pip install agentmemory
# 或者安装最新开发版本
pip install git+https://github.com/rohitg00/agentmemory.git
如果遇到安装问题,可以尝试手动安装依赖:
# 克隆仓库
git clone https://github.com/rohitg00/agentmemory.git
cd agentmemory
# 安装依赖
pip install -e .
# 或者单独安装可选依赖
pip install agentmemory[postgres] # PostgreSQL支持
pip install agentmemory[mongodb] # MongoDB支持
pip install agentmemory[all] # 所有可选依赖
依赖配置
安装完成后,需要配置 LLM API 密钥。agentmemory 支持多种 LLM 提供商:
# 使用环境变量配置
import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
# 或者使用其他提供商
os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-key"
os.environ["GOOGLE_API_KEY"] = "your-google-key"
验证安装
安装完成后,运行以下代码验证环境配置是否正确:
# 验证安装
import agentmemory
# 检查版本
print(f"agentmemory 版本: {agentmemory.__version__}")
# 测试基本功能
memory = agentmemory.create_memory_store()
result = memory.add("测试记忆", metadata={"type": "test"})
print(f"记忆添加成功: {result}")
# 验证检索功能
results = memory.search("测试")
print(f"检索结果: {results}")
如果以上代码运行成功,说明环境配置正确,可以开始正式使用 agentmemory 了。
核心功能详解 / 核心功能详解
1. 记忆存储系统
agentmemory 的核心是它的记忆存储系统,采用分层架构设计,兼顾性能和功能完整性。
工作记忆(Working Memory):这是 Agent 当前任务执行时的临时记忆空间,类似于人类的工作记忆区域。信息在这里被快速处理和转换,但不会长期保存。当任务完成后,工作记忆中的信息要么被转移到其他层级,要么被清除。工作记忆的容量有限,通常只保留最近几条关键信息。
# 工作记忆示例
working_mem = agentmemory.WorkingMemory()
# 存储当前任务的中间结果
working_mem.update("current_task", {
"goal": "分析销售数据",
"step": 3,
"findings": ["Q1增长15%", "客户流失率下降"]
})
# 读取当前状态
state = working_mem.get("current_task")
print(f"当前任务进度: {state['step']}/{state['goal']}")
# 清理工作记忆
working_mem.clear()
情景记忆(Episodic Memory):存储具体的事件和经历,类似人类的情景记忆。每次用户与 Agent 的交互、每个完成的任务、每次重要的决策都被记录在这里。情景记忆支持时间线检索,可以回溯任意时间点之前的对话内容。
# 情景记忆示例
episodic = agentmemory.EpisodicMemory()
# 记录一次完整的交互会话
session_id = episodic.record_session(
user_id="user_001",
events=[
{"role": "user", "content": "帮我分析这份销售报告", "time": "2024-01-15 10:30"},
{"role": "assistant", "content": "好的,请上传报告文件", "time": "2024-01-15 10:30"},
{"role": "user", "content": "[上传文件]", "time": "2024-01-15 10:32"},
{"role": "assistant", "content": "报告分析完成,发现三个关键问题", "time": "2024-01-15 10:35"}
],
summary="用户请求分析销售报告,助手完成了初步分析"
)
# 检索历史会话
past_sessions = episodic.retrieve(
query="之前的销售报告分析",
user_id="user_001",
limit=5
)
print(f"找到 {len(past_sessions)} 条相关历史记录")
语义记忆(Semantic Memory):存储结构化的知识和概念,与具体事件无关。这是 Agent 对世界的认知模型,包括事实、规则、分类等。例如“Python是一种编程语言”这样的知识就存储在语义记忆中。
# 语义记忆示例
semantic = agentmemory.SemanticMemory()
# 存储实体和关系
semantic.add_entity(
entity_type="project",
name="营销自动化项目",
properties={
"start_date": "2024-01-01",
"budget": 500000,
"team_size": 8,
"status": "进行中"
}
)
semantic.add_relation(
from_entity="营销自动化项目",
relation_type="has_member",
to_entity="张三"
)
semantic.add_relation(
from_entity="营销自动化项目",
relation_type="depends_on",
to_entity="CRM系统升级"
)
# 查询实体信息
project = semantic.get_entity("营销自动化项目")
print(f"项目状态: {project['properties']['status']}")
# 查询关系网络
dependencies = semantic.get_relations(
from_entity="营销自动化项目",
relation_type="depends_on"
)
print(f"依赖项目: {[d['to_entity'] for d in dependencies]}")
程序记忆(Procedural Memory):存储操作步骤和技能,即“如何做”的知识。当 Agent 学习到一个新技能或流程后,这个知识就存储在程序记忆中,便于后续复用和迁移。
# 程序记忆示例
procedural = agentmemory.ProceduralMemory()
# 存储一个操作流程
procedural.store_skill(
skill_name="代码审查流程",
description="执行代码审查的标准步骤",
steps=[
{"order": 1, "action": "检查代码格式和风格", "tool": "linter"},
{"order": 2, "action": "验证测试覆盖率", "tool": "coverage"},
{"order": 3, "action": "分析复杂度", "tool": "radon"},
{"order": 4, "action": "检查安全漏洞", "tool": "bandit"},
{"order": 5, "action": "生成审查报告", "tool": "reporter"}
],
prerequisites=["需要安装审查工具", "有代码仓库访问权限"],
estimated_time="15-30分钟"
)
# 检索并执行流程
skill = procedural.get_skill("代码审查流程")
for step in skill["steps"]:
print(f"步骤{step['order']}: {step['action']}")
2. 向量检索引擎
这是 agentmemory 最强大的功能之一。它基于 embeddings 技术,将文本转换为向量表示,从而支持语义级别的相似度搜索。
工作原理:当你存储一段文字时,系统会自动生成它的向量表示(embedding)。当你搜索时,你的查询同样被转换为向量,然后通过计算向量间的相似度来找到最相关的结果。
# 向量检索示例
store = agentmemory.VectorStore(
dimension=1536, # OpenAI embedding dimension
backend="chromadb" # 使用 ChromaDB 作为后端
)
# 存储文档
documents = [
{
"id": "doc1",
"content": "Python是一种高级编程语言,以其简洁的语法和强大的功能著称。",
"metadata": {"source": "wiki", "language": "zh"}
},
{
"id": "doc2",
"content": "机器学习是人工智能的一个分支,专注于开发能够从数据中学习的算法。",
"metadata": {"source": "textbook", "language": "zh"}
},
{
"id": "doc3",
"content": "深度学习是机器学习的子集,使用多层神经网络来处理复杂模式。",
"metadata": {"source": "textbook", "language": "zh"}
}
]
store.add_documents(documents)
# 语义搜索
query = "与神经网络相关的内容"
results = store.search(query, top_k=3)
for result in results:
print(f"文档ID: {result['id']}")
print(f"相似度: {result['score']:.4f}")
print(f"内容: {result['content']}")
print("---")
# 输出可能如下:
# 文档ID: doc3
# 相似度: 0.8921
# 内容: 深度学习是机器学习的子集,使用多层神经网络来处理复杂模式。
# ---
# 文档ID: doc2
# 相似度: 0.7543
# 内容: 机器学习是人工智能的一个分支,专注于开发能够从数据中学习的算法。
3. 会话管理系统
会话管理是构建多轮对话应用的基础。agentmemory 提供了完整的会话生命周期管理。
# 会话管理示例
session_manager = agentmemory.SessionManager()
# 创建新会话
session = session_manager.create_session(
user_id="user_001",
agent_id="assistant_01",
metadata={
"topic": "项目规划",
"context": "用户正在讨论Q2季度计划"
}
)
session_id = session["id"]
print(f"创建会话: {session_id}")
# 在会话中添加消息
session_manager.add_message(
session_id=session_id,
role="user",
content="我们需要制定Q2季度的技术目标"
)
session_manager.add_message(
session_id=session_id,
role="assistant",
content="好的,请问您希望从哪些维度来设定目标?比如功能交付、技术债务、技术创新等?"
)
session_manager.add_message(
session_id=session_id,
role="user",
content="重点关注API重构和监控体系的建设"
)
# 获取会话历史
history = session_manager.get_session_history(
session_id=session_id,
include_metadata=True
)
print(f"会话包含 {len(history['messages'])} 条消息")
# 生成会话摘要(对于长对话非常有用)
summary = session_manager.summarize_session(
session_id=session_id,
llm_provider="openai"
)
print(f"会话摘要: {summary}")
# 标记会话状态
session_manager.update_session_status(
session_id=session_id,
status="completed",
outcome="完成目标设定讨论"
)
# 检索历史会话
past_sessions = session_manager.find_sessions(
user_id="user_001",
query="技术目标",
limit=10
)
4. 记忆组织和检索
除了基本的存储和检索,agentmemory 还提供了高级的记忆组织功能。
记忆分类和标签:对记忆进行多维度分类,便于精确检索。
# 记忆分类示例
memory = agentmemory.MemoryStore()
# 添加带标签的记忆
memory.store(
content="完成用户登录模块的重构,修复了3个安全漏洞",
tags=["重构", "安全", "登录模块"],
category="开发任务",
importance=5, # 1-5分
entities=["用户模块", "安全审计"],
timestamp=datetime.now()
)
# 按标签检索
security_memories = memory.get_by_tags(["安全"])
print(f"安全相关记忆: {len(security_memories)}条")
# 按分类检索
dev_memories = memory.get_by_category("开发任务")
print(f"开发任务记忆: {len(dev_memories)}条")
# 按重要性筛选
important = memory.get_by_importance(min_score=4)
print(f"重要记忆: {len(important)}条")
记忆关联:建立不同记忆之间的关联,形成知识网络。
# 记忆关联示例
memory = agentmemory.MemoryStore()
# 存储两条相关记忆
mem1_id = memory.store("用户反馈系统存在延迟问题")
mem2_id = memory.store("数据库查询优化可以提升系统性能")
# 建立关联
memory.link(mem1_id, mem2_id, relation_type="solves")
memory.link(mem2_id, mem1_id, relation_type="caused_by")
# 检索关联记忆
related = memory.get_related(mem1_id)
print(f"相关记忆: {[r['content'] for r in related]}")
# 获取关联图谱
graph = memory.get_memory_graph()
print(f"图谱包含 {len(graph['nodes'])} 个节点和 {len(graph['edges'])} 条边")
实战教程 / Step-by-Step Tutorial
下面我们通过一个完整的实战案例,从零开始构建一个具有持久记忆能力的AI助手。
项目背景
我们将构建一个个人知识管理助手,它能够:
– 接收用户输入的知识片段并存储
– 回答用户关于已存储知识的问题
– 记住用户的偏好和习惯
– 跨会话保持上下文连贯性
第一步:基础配置和初始化
首先创建项目结构并初始化系统:
# config.py
import os
# LLM 配置
class Config:
LLM_PROVIDER = "openai" # 或 "anthropic", "google"
LLM_MODEL = "gpt-4"
EMBEDDING_MODEL = "text-embedding-ada-002"
# 向量数据库配置
VECTOR_DB_PATH = "./data/vectors"
# 会话存储配置
SESSION_DB_PATH = "./data/sessions.db"
# 记忆配置
MAX_WORKING_MEMORY = 50 # 工作记忆容量
SESSION_SUMMARY_THRESHOLD = 20 # 自动摘要的阈值
@classmethod
def from_env(cls):
# 从环境变量加载配置
if os.getenv("LLM_PROVIDER"):
cls.LLM_PROVIDER = os.getenv("LLM_PROVIDER")
if os.getenv("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")
return cls
# ---
# main.py
from config import Config
import agentmemory
# 初始化配置
Config.from_env()
# 初始化各个存储组件
memory_store = agentmemory.MemoryStore(
vector_db_path=Config.VECTOR_DB_PATH,
embedding_model=Config.EMBEDDING_MODEL
)
session_manager = agentmemory.SessionManager(
db_path=Config.SESSION_DB_PATH
)
working_memory = agentmemory.WorkingMemory(
max_size=Config.MAX_WORKING_MEMORY
)
print("系统初始化完成!")
第二步:构建知识存储功能
实现知识的接收和存储逻辑:
# knowledge_manager.py
from datetime import datetime
from typing import List, Dict, Optional
class KnowledgeManager:
"""
知识管理器 - 负责知识的存储、检索和组织
"""
def __init__(self, memory_store):
self.store = memory_store
self.categories = [
"技术文档", "产品需求", "会议记录",
"用户反馈", "设计方案", "其他"
]
def add_knowledge(
self,
content: str,
category: str,
tags: List[str],
title: Optional[str] = None
) -> Dict:
"""
添加新知识到存储系统
参数:
content: 知识内容
category: 分类
tags: 标签列表
title: 可选的标题
返回:
存储结果
"""
# 验证分类
if category not in self.categories:
raise ValueError(f"未知的分类: {category}")
# 创建记忆对象
memory_data = {
"content": content,
"title": title or content[:50],
"category": category,
"tags": tags,
"created_at": datetime.now().isoformat(),
"source": "user_input"
}
# 存储到记忆系统
memory_id = self.store.store(
content=content,
metadata=memory_data
)
# 自动提取实体和关键词
extracted = self._extract_entities(content)
# 更新语义记忆
for entity in extracted.get("entities", []):
self._update_entity_memory(entity, memory_id)
return {
"id": memory_id,
"status": "success",
"extracted": extracted
}
def _extract_entities(self, content: str) -> Dict:
"""
从内容中提取实体和关系
这里简化处理,实际可以使用NER模型
"""
# 简化版:基于规则的实体提取
entities = []
keywords = []
# 实际项目中这里应该调用LLM或NER模型
# 简化版本不做实际提取
return {
"entities": entities,
"keywords": keywords
}
def _update_entity_memory(self, entity: str, memory_id: str):
"""
更新语义记忆中的实体
"""
# 确保实体存在并建立关联
pass
def search_knowledge(
self,
query: str,
category: Optional[str] = None,
tags: Optional[List[str]] = None,
limit: int = 10
) -> List[Dict]:
"""
语义搜索知识库
"""
# 基础搜索
results = self.store.search(query, top_k=limit)
# 过滤分类
if category:
results = [
r for r in results
if r.get("metadata", {}).get("category") == category
]
# 过滤标签
if tags:
results = [
r for r in results
if any(tag in r.get("metadata", {}).get("tags", [])
for tag in tags)
]
return results
def get_category_summary(self, category: str) -> Dict:
"""
获取某个分类的知识统计
"""
all_knowledge = self.store.get_by_category(category)
return {
"category": category,
"total_count": len(all_knowledge),
"recent": all_knowledge[-5:], # 最近5条
"tags": self._aggregate_tags(all_knowledge)
}
def _aggregate_tags(self, knowledge_list: List[Dict]) -> Dict:
"""
聚合所有知识的标签
"""
tag_counts = {}
for k in knowledge_list:
for tag in k.get("tags", []):
tag_counts[tag] = tag_counts.get(tag, 0) + 1
return tag_counts
# 使用示例
km = KnowledgeManager(memory_store)
# 添加知识
result = km.add_knowledge(
content="agentmemory是一个用于AI Agent记忆管理的开源库,支持向量存储、会话管理等功能。",
category="技术文档",
tags=["AI", "Agent", "记忆管理", "开源"],
title="agentmemory项目介绍"
)
print(f"知识添加成功: {result['id']}")
# 搜索知识
results = km.search_knowledge("关于AI Agent的库")
print(f"找到 {len(results)} 条相关知识")
第三步:构建会话管理功能
实现用户交互和会话管理:
# conversation_manager.py
from typing import Dict, List, Optional
from datetime import datetime
class ConversationManager:
"""
对话管理器 - 处理用户交互和会话状态
"""
def __init__(
self,
session_manager,
knowledge_manager,
working_memory,
llm_provider: str = "openai"
):
self.sessions = session_manager
self.knowledge = knowledge_manager
self.working = working_memory
self.llm = self._init_llm(llm_provider)
def _init_llm(self, provider: str):
"""
初始化 LLM 客户端
"""
if provider == "openai":
from openai import OpenAI
return OpenAI()
# 可以添加其他提供商
return None
def start_conversation(self, user_id: str, context: Dict = None) -> str:
"""
开始新的对话会话
"""
# 检查是否有未完成的会话
existing = self.sessions.find_sessions(
user_id=user_id,
status="active",
limit=1
)
if existing:
# 恢复已有会话
session_id = existing[0]["id"]
self._load_session_context(session_id)
return session_id
# 创建新会话
session = self.sessions.create_session(
user_id=user_id,
agent_id="knowledge_assistant",
metadata={
"started_at": datetime.now().isoformat(),
"context": context or {}
}
)
# 初始化工作记忆
self.working.update(f"session_{session['id']}", {
"user_id": user_id,
"turn_count": 0,
"pending_tasks": []
})
return session["id"]
def _load_session_context(self, session_id: str):
"""
加载会话上下文到工作记忆
"""
history = self.sessions.get_session_history(session_id)
# 加载最近的对话
recent = history["messages"][-5:] if history["messages"] else []
context = {
"recent_history": recent,
"session_id": session_id
}
self.working.update(f"session_{session_id}", context)
def process_message(
self,
session_id: str,
user_message: str
) -> Dict:
"""
处理用户消息并生成回复
"""
# 更新工作记忆
state = self.working.get(f"session_{session_id}")
state["turn_count"] += 1
# 记录消息
self.sessions.add_message(
session_id=session_id,
role="user",
content=user_message,
timestamp=datetime.now().isoformat()
)
# 判断用户意图
intent = self._classify_intent(user_message)
# 根据意图处理
response = None
if intent == "query":
response = self._handle_query(session_id, user_message)
elif intent == "add_knowledge":
response = self._handle_add_knowledge(session_id, user_message)
elif intent == "summary":
response = self._handle_summary(session_id)
else:
response = self._handle_general(session_id, user_message)
# 记录助手回复
self.sessions.add_message(
session_id=session_id,
role="assistant",
content=response["content"],
timestamp=datetime.now().isoformat()
)
# 检查是否需要摘要
if state["turn_count"] >= 20:
self._auto_summarize(session_id)
state["turn_count"] = 0
return response
def _classify_intent(self, message: str) -> str:
"""
简单意图分类
实际项目中可以使用更复杂的分类器
"""
message_lower = message.lower()
if any(kw in message_lower for kw in ["搜索", "查找", "查询", "告诉我关于", "有什么"]):
return "query"
elif any(kw in message_lower for kw in ["添加", "记录", "存储", "记住"]):
return "add_knowledge"
elif any(kw in message_lower for kw in ["总结", "摘要", "概括"]):
return "summary"
else:
return "general"
def _handle_query(self, session_id: str, query: str) -> Dict:
"""
处理知识查询
"""
# 搜索相关知识
results = self.knowledge.search_knowledge(query, limit=5)
if not results:
return {
"content": "抱歉,我在知识库中没有找到相关内容。您可以尝试其他关键词,或者告诉我新的信息。",
"type": "query",
"results": []
}
# 构建回复
response_text = f"我在知识库中找到了 {len(results)} 条相关信息:\n\n"
for i, r in enumerate(results, 1):
content = r["content"]
metadata = r.get("metadata", {})
response_text += f"{i}. {metadata.get('title', content[:50])}...\n"
response_text += f" 分类: {metadata.get('category', '未分类')}\n"
response_text += f" 相关度: {r.get('score', 0):.2f}\n\n"
response_text += "您想了解哪一条的详细内容吗?"
return {
"content": response_text,
"type": "query",
"results": results
}
def _handle_add_knowledge(
self,
session_id: str,
message: str
) -> Dict:
"""
处理添加知识
"""
# 简化处理:假设消息包含知识内容
# 实际项目中可以使用 LLM 提取结构化信息
result = self.knowledge.add_knowledge(
content=message,
category="其他",
tags=["用户添加"]
)
return {
"content": f"已将知识添加到存储系统,ID: {result['id']}",
"type": "add_knowledge",
"knowledge_id": result["id"]
}
def _handle_summary(self, session_id: str) -> Dict:
"""
生成会话摘要
"""
summary = self.sessions.summarize_session(
session_id=session_id,
llm_provider="openai"
)
return {
"content": f"本次会话摘要:\n\n{summary}",
"type": "summary"
}
def _handle_general(
self,
session_id: str,
message: str
) -> Dict:
"""
处理一般对话
"""
# 获取上下文
state = self.working.get(f"session_{session_id}")
recent = state.get("recent_history", [])
# 构建上下文提示
context_prompt = "以下是最近的对话历史:\n"
for msg in recent[-5:]:
context_prompt += f"{msg['role']}: {msg['content']}\n"
# 调用 LLM 生成回复
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "你是一个友好的知识管理助手。"},
{"role": "user", "content": context_prompt + "\n用户新消息: " + message}
]
)
return {
"content": response.choices[0].message.content,
"type": "general"
}
def _auto_summarize(self, session_id: str):
"""
自动生成会话摘要
"""
summary = self.sessions.summarize_session(
session_id=session_id,
llm_provider="openai"
)
# 存储摘要到知识库
self.knowledge.add_knowledge(
content=f"会话摘要: {summary}",
category="会话记录",
tags=["自动摘要", "会话历史"]
)
print(f"自动生成会话摘要: {summary[:50]}...")
# 使用示例
conv_manager = ConversationManager(
session_manager=session_manager,
knowledge_manager=km,
working_memory=working_memory
)
# 开始对话
session_id = conv_manager.start_conversation(
user_id="user_001",
context={"platform": "web", "language": "zh"}
)
print(f"开始新会话: {session_id}")
# 处理用户消息
response = conv_manager.process_message(
session_id=session_id,
user_message="添加一条知识:agentmemory支持多种向量数据库后端"
)
print(f"助手回复: {response['content']}")
第四步:构建Web服务接口
为了方便使用,我们将整个系统封装成REST API:
# api.py
from flask import Flask, request, jsonify
from flask_cors import CORS
app = Flask(__name__)
CORS(app)
# 初始化系统组件
from conversation_manager import ConversationManager
from knowledge_manager import KnowledgeManager
from config import Config
# 全局实例
knowledge_mgr = None
conv_mgr = None
sessions = {} # 简单内存存储,实际应使用数据库
@app.before_request
def init_services():
"""延迟初始化服务"""
global knowledge_mgr, conv_mgr
if knowledge_mgr is None:
Config.from_env()
from agentmemory import MemoryStore, SessionManager, WorkingMemory
memory_store = MemoryStore(
vector_db_path=Config.VECTOR_DB_PATH,
embedding_model=Config.EMBEDDING_MODEL
)
session_manager = SessionManager(db_path=Config.SESSION_DB_PATH)
working_memory = WorkingMemory(max_size=Config.MAX_WORKING_MEMORY)
knowledge_mgr = KnowledgeManager(memory_store)
conv_mgr = ConversationManager(
session_manager=session_manager,
knowledge_manager=knowledge_mgr,
working_memory=working_memory
)
@app.route("/api/chat", methods=["POST"])
def chat():
"""
处理聊天请求
POST /api/chat
{
"user_id": "user_001",
"message": "你好",
"session_id": "optional_session_id"
}
"""
data = request.json
user_id = data.get("user_id")
message = data.get("message")
session_id = data.get("session_id")
if not user_id or not message:
return jsonify({"error": "缺少必要参数"}), 400
# 获取或创建会话
if not session_id:
session_id = conv_mgr.start_conversation(user_id)
elif str(session_id) not in sessions:
session_id = conv_mgr.start_conversation(user_id)
# 处理消息
response = conv_mgr.process_message(session_id, message)
# 存储会话关联
sessions[str(session_id)] = {
"user_id": user_id,
"last_active": datetime.now().isoformat()
}
return jsonify({
"session_id": session_id,
"response": response["content"],
"type": response["type"]
})
@app.route("/api/knowledge", methods=["POST"])
def add_knowledge():
"""
添加知识
POST /api/knowledge
{
"content": "知识内容",
"category": "技术文档",
"tags": ["AI", "ML"]
}
"""
data = request.json
result = knowledge_mgr.add_knowledge(
content=data.get("content"),
category=data.get("category", "其他"),
tags=data.get("tags", []),
title=data.get("title")
)
return jsonify(result)
@app.route("/api/knowledge/search", methods=["GET"])
def search_knowledge():
"""
搜索知识
GET /api/knowledge/search?q=关键词&category=分类
"""
query = request.args.get("q", "")
category = request.args.get("category")
limit = int(request.args.get("limit", 10))
results = knowledge_mgr.search_knowledge(
query=query,
category=category,
limit=limit
)
return jsonify({
"query": query,
"count": len(results),
"results": results
})
@app.route("/api/session/<session_id>/summary", methods=["GET"])
def get_session_summary(session_id):
"""
获取会话摘要
"""
summary = conv_mgr.sessions.summarize_session(session_id)
return jsonify({"session_id": session_id, "summary": summary})
if __name__ == "__main__":
app.run(debug=True, port=5000)
第五步:运行完整示例
将所有代码整合,运行一个完整的示例:
# run_demo.py
"""
完整演示:运行知识管理助手
"""
import os
from datetime import datetime
# 设置 API Key
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
# 导入组件
from knowledge_manager import KnowledgeManager
from conversation_manager import ConversationManager
from config import Config
# 初始化
Config.from_env()
from agentmemory import MemoryStore, SessionManager, WorkingMemory
print("=" * 50)
print("个人知识管理助手 - 演示程序")
print("=" * 50)
# 初始化存储
memory_store = MemoryStore(
vector_db_path="./data/demo_vectors",
embedding_model=Config.EMBEDDING_MODEL
)
session_manager = SessionManager(
db_path="./data/demo_sessions.db"
)
working_memory = WorkingMemory(max_size=50)
# 创建管理器
knowledge_mgr = KnowledgeManager(memory_store)
conv_mgr = ConversationManager(
session_manager=session_manager,
knowledge_manager=knowledge_mgr,
working_memory=working_memory
)
# ---
# 添加一些初始知识
print("\n[1] 初始化知识库...")
initial_knowledge = [
{
"content": "agentmemory 是一个开源的 AI Agent 记忆管理框架,支持向量存储、会话管理、语义检索等功能。",
"category": "技术文档",
"tags": ["AI", "Agent", "记忆管理", "开源"]
},
{
"content": "向量数据库是用于存储和检索向量嵌入的专用数据库,支持语义相似度搜索。",
"category": "技术文档",
"tags": ["向量数据库", "embedding", "语义搜索"]
},
{
"content": "RAG(检索增强生成)是一种结合检索和生成的技术,可以提高 LLM 回答的准确性。",
"category": "技术文档",
"tags": ["RAG", "LLM", "检索增强"]
}
]
for k in initial_knowledge:
result = knowledge_mgr.add_knowledge(**k)
print(f" - 添加: {k['title'] if 'title' in k else k['content'][:30]}...")
# ---
# 开始对话
print("\n[2] 开始对话...")
session_id = conv_mgr.start_conversation(
user_id="demo_user",
context={"demo": True}
)
print(f" 会话ID: {session_id}")
# 对话流程
dialogues = [
"你好,请介绍一下你自己",
"查找关于向量数据库的知识",
"我想了解 RAG 技术的应用场景",
"添加一条新知识:上下文学习是一种让 LLM 从提示中学习任务的方法"
]
for i, msg in enumerate(dialogues, 1):
print(f"\n[用户消息 {i}]: {msg}")
response = conv_mgr.process_message(session_id, msg)
# 格式化输出回复
reply_preview = response["content"][:100]
print(f"[助手回复]: {reply_preview}...")
print(f"[回复类型]: {response['type']}")
# ---
# 查询知识统计
print("\n[3] 知识库统计...")
tech_summary = knowledge_mgr.get_category_summary("技术文档")
print(f" 技术文档总数: {tech_summary['total_count']}")
print(f" 标签统计: {tech_summary['tags']}")
# ---
# 获取会话历史
print("\n[4] 会话历史...")
history = session_manager.get_session_history(session_id)
print(f" 本次会话共 {len(history['messages'])} 条消息")
# 生成摘要
print("\n[5] 生成会话摘要...")
summary = conv_mgr.sessions.summarize_session(session_id)
print(f" 摘要: {summary}")
print("\n" + "=" * 50)
print("演示完成!")
print("=" * 50)
运行这个脚本,你会看到完整的知识添加、对话处理和检索流程。系统会创建本地数据库文件存储向量和会话信息。
常见使用场景 / Common Use Cases
场景一:智能客服系统
在客服场景中,agentmemory 可以帮助系统记住用户的历史问题和偏好:
# customer_service.py
class CustomerServiceAgent:
"""
智能客服 Agent - 利用记忆系统提供个性化服务
"""
def __init__(self):
self.sessions = session_manager
self.memory = memory_store
self.user_profiles = {} # 用户画像存储
def identify_user(self, user_id: str) -> Dict:
"""
识别用户并加载其画像
"""
# 查找用户历史记录
past_sessions = self.sessions.find_sessions(
user_id=user_id,
limit=10
)
# 分析用户特征
profile = self._analyze_user_patterns(past_sessions)
# 加载相关记忆
user_memories = self.memory.search(
f"用户 {user_id} 的偏好",
top_k=5
)
return {
"user_id": user_id,
"total_sessions": len(past_sessions),
"profile": profile,
"relevant_memories": user_memories
}
def handle_complaint(
self,
session_id: str,
complaint: str
) -> Dict:
"""
处理用户投诉
"""
# 记录投诉
self.memory.store(
content=f"用户投诉: {complaint}",
metadata={
"type": "complaint",
"session_id": session_id,
"timestamp": datetime.now().isoformat()
}
)
# 分类处理
category = self._classify_complaint(complaint)
if category == "质量问题":
# 触发质量检查流程
return self._handle_quality_issue(complaint)
elif category == "服务态度":
# 触发服务改进流程
return self._handle_service_issue(complaint)
else:
# 通用处理
return self._general_response(complaint)
def _analyze_user_patterns(self, sessions: List) -> Dict:
"""
分析用户行为模式
"""
patterns = {
"常见问题": [],
"购买习惯": "中等频率",
"服务评价": "积极"
}
# 简化分析逻辑
return patterns
场景二:个人AI助手
构建一个能记住你所有习惯的个人助手:
# personal_assistant.py
class PersonalAssistant:
"""
个人AI助手 - 记住你的所有习惯和偏好
"""
def __init__(self, user_id: str):
self.user_id = user_id
self.semantic = semantic_memory
self.episodic = episodic_memory
self.working = working_memory
def learn_preference(
self,
preference_type: str,
value: str
):
"""
学习用户偏好
"""
# 存储偏好
self.semantic.add_entity(
entity_type="user_preference",
name=f"{self.user_id}_{preference_type}",
properties={
"value": value,
"learned_at": datetime.now().isoformat(),
"confidence": 0.8
}
)
# 建立关联
self.semantic.add_relation(
from_entity=self.user_id,
relation_type=f"prefers_{preference_type}",
to_entity=f"{self.user_id}_{preference_type}"
)
def apply_preferences(self, task: str) -> str:
"""
在执行任务时应用用户偏好
"""
# 获取所有偏好
preferences = self.semantic.get_relations(
from_entity=self.user_id,
relation_type="prefers_*"
)
# 构建个性化提示
prompt_modifiers = []
for pref in preferences:
pref_type = pref["relation_type"].replace("prefers_", "")
value = pref["to_entity"]
prompt_modifiers.append(f"{pref_type}: {value}")
return f"任务要求: {task}\n用户偏好: {', '.join(prompt_modifiers)}"
场景三:代码审查Agent
帮助开发团队追踪代码质量和历史问题:
# code_review_agent.py
class CodeReviewAgent:
"""
代码审查 Agent - 利用记忆追踪代码质量和改进历史
"""
def __init__(self):
self.memory = memory_store
def review_code(
self,
repo: str,
pr_id: str,
changes: List[Dict]
) -> Dict:
"""
审查代码变更
"""
results = []
for change in changes:
# 检查是否有过类似问题
similar = self.memory.search(
f"类似的代码问题 {change.get('file')}",
top_k=3
)
# 执行审查
review_result = self._perform_review(change, similar)
results.append(review_result)
# 存储审查结果
if review_result["issues"]:
self.memory.store(
content=f"代码问题: {review_result['issues']}",
metadata={
"type": "code_issue",
"file": change.get("file"),
"repo": repo,
"pr_id": pr_id
}
)
return self._generate_review_report(results)
def track_debt(self, repo: str) -> Dict:
"""
追踪技术债务
"""
# 获取所有相关问题
issues = self.memory.search(
f"技术债务 {repo}",
top_k=100
)
# 按严重程度分类
by_severity = {}
for issue in issues:
severity = issue.get("severity", "medium")
if severity not in by_severity:
by_severity[severity] = []
by_severity[severity].append(issue)
return {
"total_issues": len(issues),
"by_severity": {
k: len(v) for k, v in by_severity.items()
},
"critical": by_severity.get("critical", [])
}
场景四:研究助手
帮助研究人员整理和关联文献资料:
# research_assistant.py
class ResearchAssistant:
"""
研究助手 - 管理和关联学术文献
"""
def __init__(self):
self.semantic = semantic_memory
self.memory = memory_store
def add_paper(
self,
title: str,
authors: List[str],
abstract: str,
citations: List[str]
):
"""
添加论文到知识库
"""
# 存储论文实体
paper_id = self.semantic.add_entity(
entity_type="paper",
name=title,
properties={
"authors": authors,
"abstract": abstract,
"added_at": datetime.now().isoformat()
}
)
# 存储摘要到向量库
self.memory.store(
content=abstract,
metadata={
"type": "paper_abstract",
"title": title,
"paper_id": paper_id
}
)
# 建立引用关系
for cited_title in citations:
self.semantic.add_relation(
from_entity=title,
relation_type="cites",
to_entity=cited_title
)
return paper_id
def find_related_work(self, topic: str) -> List[Dict]:
"""
查找相关研究工作
"""
# 语义搜索
results = self.memory.search(topic, top_k=10)
# 获取相关论文的详细信息
related_papers = []
for r in results:
paper_info = self.semantic.get_entity(
r["metadata"]["title"]
)
if paper_info:
related_papers.append(paper_info)
return related_papers
def build_knowledge_graph(self, field: str) -> Dict:
"""
构建特定领域的知识图谱
"""
# 获取领域内所有论文
papers = self.semantic.get_entities_by_type("paper")
field_papers = [
p for p in papers
if field.lower() in p["properties"]["abstract"].lower()
]
# 构建引用网络
edges = []
for paper in field_papers:
citations = self.semantic.get_relations(
from_entity=paper["name"],
relation_type="cites"
)
for c in citations:
edges.append({
"from": paper["name"],
"to": c["to_entity"]
})
return {
"nodes": field_papers,
"edges": edges,
"paper_count": len(field_papers)
}
最佳实践 / Tips and Best Practices
1. 记忆管理策略
定期清理无用记忆:记忆系统不是越大越好,定期清理可以提高检索效率和降低成本。
# memory_cleanup.py
class MemoryCleanupManager:
"""
记忆清理管理器 - 定期清理过时或无用的记忆
"""
def __init__(self, memory_store):
self.store = memory_store
def cleanup_by_age(self, days: int = 90):
"""
删除指定天数之前的记忆
"""
cutoff_date = datetime.now() - timedelta(days=days)
# 查找过期记忆
old_memories = self.store.get_by_date_range(
start_date=None,
end_date=cutoff_date
)
deleted_count = 0
for memory in old_memories:
# 检查是否标记为重要
if memory.get("importance", 0) < 3:
self.store.delete(memory["id"])
deleted_count += 1
return deleted_count
def cleanup_duplicates(self):
"""
合并重复的记忆
"""
# 使用向量相似度检测重复
all_memories = self.store.get_all()
duplicates = []
for i, mem1 in enumerate(all_memories):
for mem2 in all_memories[i+1:]:
similarity = self._calculate_similarity(
mem1["content"],
mem2["content"]
)
if similarity > 0.95:
duplicates.append((mem1["id"], mem2["id"]))
# 合并重复
for keep_id, remove_id in duplicates:
self._merge_memories(keep_id, remove_id)
return len(duplicates)
def cleanup_low_relevance(self, min_score: float = 0.3):
"""
删除低访问频率的记忆
"""
all_memories = self.store.get_all()
removed = 0
for memory in all_memories:
access_count = memory.get("access_count", 0)
relevance_score = memory.get("relevance_score", 1.0)
# 如果访问频率低且相关性低,考虑删除
if access_count < 5 and relevance_score < min_score:
self.store.delete(memory["id"])
removed += 1
return removed
合理使用分层记忆:根据信息的时效性和重要性选择合适的存储层级。
# memory_layer_usage.py
"""
记忆层级选择指南
"""
MEMORY_TIER_GUIDE = """
根据以下标准选择记忆存储层级:
1. 工作记忆 (WorkingMemory)
- 当前任务需要的即时信息
- 临时计算结果
- 即将使用的数据
- 示例: 用户当前查询的解析结果
2. 情景记忆 (EpisodicMemory)
- 完整的交互会话
- 事件和经历
- 需要回溯的历史
- 示例: 昨天的客服对话记录
3. 语义记忆 (SemanticMemory)
- 事实和概念
- 结构化知识
- 实体和关系
- 示例: 产品的功能特性
4. 程序记忆 (ProceduralMemory)
- 操作流程和步骤
- 技能和方法
- 可复用的模式
- 示例: 代码审查的标准流程
"""
# 选择示例
def select_memory_tier(info_type: str) -> str:
"""
根据信息类型推荐存储层级
"""
tier_map = {
"temp_result": "working",
"conversation": "episodic",
"fact": "semantic",
"procedure": "procedural",
"preference": "semantic",
"context": "episodic",
"knowledge": "semantic"
}
return tier_map.get(info_type, "working")
2. 性能优化建议
批量操作:减少API调用次数,提高效率。
# batch_operations.py
class BatchKnowledgeManager:
"""
批量知识管理器 - 优化大量数据的存储和检索
"""
def __init__(self, knowledge_manager):
self.km = knowledge_manager
self.batch_buffer = []
self.batch_size = 50
def add_batch(self, knowledge_list: List[Dict]):
"""
批量添加知识,自动分批处理
"""
self.batch_buffer.extend(knowledge_list)
results = []
while len(self.batch_buffer) >= self.batch_size:
batch = self.batch_buffer[:self.batch_size]
batch_results = self._process_batch(batch)
results.extend(batch_results)
self.batch_buffer = self.batch_buffer[self.batch_size:]
# 处理剩余部分
if self.batch_buffer:
results.extend(self._process_batch(self.batch_buffer))
self.batch_buffer = []
return results
def _process_batch(self, batch: List[Dict]) -> List[Dict]:
"""
处理一批知识
"""
results = []
for item in batch:
result = self.km.add_knowledge(**item)
results.append(result)
return results
def flush(self):
"""
强制刷新缓冲区
"""
if self.batch_buffer:
return self._process_batch(self.batch_buffer)
return []
异步处理:对于不需要即时返回的操作使用异步处理。
# async_memory.py
import asyncio
from concurrent.futures import ThreadPoolExecutor
class AsyncMemoryManager:
"""
异步记忆管理器 - 提高系统响应速度
"""
def __init__(self, memory_store, max_workers: int = 4):
self.store = memory_store
self.executor = ThreadPoolExecutor(max_workers=max_workers)
async def async_store(self, content: str, metadata: Dict):
"""
异步存储记忆
"""
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
self.executor,
self.store.store,
content,
metadata
)
return result
async def async_search(self, query: str, top_k: int = 10):
"""
异步搜索记忆
"""
loop = asyncio.get_event_loop()
results = await loop.run_in_executor(
self.executor,
self.store.search,
query,
top_k
)
return results
async def batch_store(self, items: List[Dict]):
"""
批量异步存储
"""
tasks = [
self.async_store(item["content"], item.get("metadata", {}))
for item in items
]
return await asyncio.gather(*tasks)
3. 安全和隐私
敏感信息处理:确保敏感信息得到适当保护。
# privacy_manager.py
import re
class PrivacyManager:
"""
隐私管理器 - 处理敏感信息保护
"""
def __init__(self):
# 敏感信息模式
self.patterns = {
"email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
"phone": r'\b\d{11}\b',
"id_card": r'\b\d{17}[\dXx]\b',
"credit_card": r'\b\d{16}\b'
}
def mask_sensitive(self, content: str) -> str:
"""
脱敏处理
"""
masked = content
# 邮箱脱敏
masked = re.sub(
self.patterns["email"],
"[邮箱已脱敏]",
masked
)
# 手机号脱敏
masked = re.sub(
self.patterns["phone"],
"****",
masked
)
# 身份证脱敏
masked = re.sub(
self.patterns["id_card"],
"***********",
masked
)
return masked
def check_consent(self, user_id: str, data_type: str) -> bool:
"""
检查用户是否授权存储特定类型的数据
"""
# 这里应该连接真实的用户权限系统
consent_db = {
"user_001": ["conversation", "preferences"],
"user_002": ["conversation", "preferences", "behavior"]
}
user_consents = consent_db.get(user_id, [])
return data_type in user_consents
4. 错误处理和恢复
# error_handling.py
import logging
from tenacity import retry, stop_after_attempt, wait_exponential
logger = logging.getLogger(__name__)
class ResilientMemoryManager:
"""
弹性记忆管理器 - 具备错误恢复能力
"""
def __init__(self, memory_store):
self.store = memory_store
self.retry_config = {
"max_attempts": 3,
"wait_min": 1,
"wait_max": 10
}
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10)
)
def safe_store(self, content: str, metadata: Dict) -> str:
"""
带重试机制的存储操作
"""
try:
return self.store.store(content, metadata)
except Exception as e:
logger.error(f"存储失败: {e}")
# 尝试降级存储
return self._fallback_store(content, metadata)
def _fallback_store(self, content: str, metadata: Dict) -> str:
"""
降级存储策略
"""
# 存储到本地文件作为备份
backup_path = f"./backup/{datetime.now().strftime('%Y%m%d%H%M%S')}.json"
backup_data = {
"content": content,
"metadata": metadata,
"failed_at": datetime.now().isoformat()
}
with open(backup_path, 'w', encoding='utf-8') as f:
json.dump(backup_data, f, ensure_ascii=False)
return f"backup:{backup_path}"
总结与相关资源 / 总结与相关资源
核心要点回顾
通过本文的详细介绍,我们已经全面了解了 agentmemory 这个开源项目的强大功能:
为什么值得关注:agentmemory 解决了 AI Agent 开发中的核心痛点——记忆管理问题。它提供了完整的分层记忆架构,包括工作记忆、情景记忆、语义记忆和程序记忆,让 AI 系统能够像人类一样拥有持续性和层次性的记忆能力。
核心功能:向量语义检索让搜索超越关键词匹配,进入语义理解的新阶段;会话管理系统支持完整的多轮对话生命周期管理;灵活的存储后端让系统可以根据部署环境选择最优方案。
实战应用:通过知识管理助手的完整案例,我们展示了如何从零开始构建一个具有持久记忆能力的 AI 应用,涵盖了从基础配置、核心功能实现到 Web 服务封装的完整流程。
最佳实践:合理的记忆管理策略、批量操作和异步处理可以显著提升系统性能;隐私保护和安全处理是不可忽视的重要环节;健壮的错误处理机制确保系统在各种异常情况下都能稳定运行。
项目优势总结
相比其他记忆管理方案,agentmemory 具有以下优势:
一体化解决方案:不需要组合多个工具,agentmemory 提供了开箱即用的完整功能集。
高度可定制:模块化设计让每个组件都可以根据需求进行定制和扩展。
生产就绪:内置的错误处理、日志记录和监控功能让系统可以直接用于生产环境。
活跃的社区支持:作为开源项目,有持续的社区贡献和功能更新。
扩展阅读和相关项目
如果你对 agentmemory 感兴趣,以下相关项目也值得关注:
向量数据库:Pinecone、Weaviate、ChromaDB、Milvus – 用于存储和检索向量嵌入的专业数据库系统。
Agent框架:LangChain、AutoGen、LlamaIndex – 用于构建复杂 AI Agent 的开发框架,可以与 agentmemory 无缝集成。
知识图谱:Neo4j、NetworkX – 用于构建复杂的实体关系网络,增强知识的结构性存储。
LLM工具:LiteLLM、PortKey – 提供统一的 LLM 调用接口,方便切换不同的模型提供商。
获取帮助和参与贡献
想要深入了解或参与项目开发,可以访问以下资源:
官方文档:详细的 API 参考和使用指南,帮助你快速上手。
GitHub Issues:报告问题或提出功能建议,与开发者和社区直接交流。
Discord/Slack 社区:加入讨论群组,与其他使用者交流使用心得和最佳实践。
Pull Requests:欢迎提交代码贡献,一起完善这个项目。
开始你的 Agent 记忆之旅
现在你已经掌握了 agentmemory 的核心概念和使用方法,是时候开始实践了。建议从简单的示例开始,逐步尝试更复杂的功能。在这个过程中,你会发现记忆管理对于构建真正智能的 AI 应用有多么重要。
记住,一个好的 AI Agent 不仅需要强大的推理能力,更需要持久和高效的记忆系统。agentmemory 正是为此而生,让你的 AI 应用真正“记住”一切。
祝你开发愉快!
本文档基于 agentmemory 开源项目编写,版本和功能可能会随项目更新而变化。建议始终参考最新的官方文档。
评论区