记忆存储为Agent提供持久化记忆能力，支持跨会话保存和检索用户交互历史、偏好信息等上下文数据。通过向量检索技术，Agent能够基于历史记忆提供个性化响应，实现真正的会话连续性。支持通过MCP工具集成到快速创建Agent，也可与LangChain等开源框架集成。
快速开始
创建记忆存储实例后，即可通过控制台或SDK进行记忆的创建、搜索和管理。
创建记忆存储实例
登录AgentRun控制台，在顶部菜单栏点击其他。
在左侧目录选择记忆存储，进入记忆存储列表页面。
点击添加记忆存储按钮，打开创建表单。
填写基础信息：
配置向量数据库：
一键配置（推荐）：自动创建和配置OTS表格存储实例，无需手动设置
自定义配置：选择已有的阿里云OTS实例
选择OTS实例：从下拉列表中选择当前地域下已有的表格存储（OTS）实例。若列表中无实例，可点击右侧按钮
，并参考创建OTS实例进行自定义创建。表名：定义用于存储向量数据的表名称。
向量维度：设置向量数据的维度。
设置网络策略：
公网模式：通过公网访问
私网模式：需配置VPC ID、安全组、子网ID
点击开始部署，等待约10秒，状态变为运行中。
部署完成后，可在实例详情页进行记忆管理、搜索记忆、查看可观测性指标，或配置MCP集成和代码集成。
记忆管理
创建记忆
在记忆存储实例详情页，点击记忆管理标签，进入记忆列表页面。页面默认展示200个记忆，更多记忆请使用搜索功能。
点击创建记忆按钮，打开添加用户记忆表单对话框。
填写记忆信息：
用户ID（必填）：标识记忆所属用户，如
test-user-001记忆内容（必填）：记忆的具体内容，如
居住在杭州元数据（可选）：点击添加元数据，添加自定义字段。
点击保存按钮，等待创建完成（首次调用会存在冷启动）。
创建成功后，记忆会出现在记忆列表中，显示用户ID、创建时间、记忆内容等信息。每个记忆卡片提供查看详情、编辑、查看历史、删除等操作按钮。
搜索记忆
在记忆存储实例详情页，点击搜索记忆标签，进入记忆搜索页面。
填写搜索条件：
搜索查询（必填）：在搜索查询输入框中输入搜索关键词，如
杭州用户ID、Agent ID或Run ID（至少填写一个）：用于筛选特定用户或会话的记忆。例如，填写用户ID
test-user-001可搜索该用户匹配关键词的记忆。
点击搜索按钮，等待搜索结果加载（约5秒）。
查看搜索结果：结果列表显示匹配的记忆内容、用户ID、相似度、ID、Hash、创建时间等信息。
搜索结果按相似度从高到低排序，相似度越高表示与查询越相关。
删除记忆
在记忆列表页面，找到目标记忆，点击删除按钮，确认删除操作。
修改记忆
在记忆列表页面，找到目标记忆，点击编辑按钮，修改记忆内容后保存。
可观测性
在记忆存储实例详情页，点击可观测标签，可实时查看记忆操作的关键指标数据。
创建记忆操作指标
创建记忆次数：统计创建记忆操作的总次数
创建记忆平均时延：统计创建记忆操作的综合响应时间
创建记忆失败率：统计创建记忆操作失败的比例
搜索记忆操作指标
搜索记忆次数：统计搜索记忆操作的总次数
记忆召回平均时延：统计搜索记忆操作的综合响应时间
记忆召回失败率：统计搜索记忆操作失败的比例
监控图表
可观测页面提供以下监控图表：
每秒请求次数（QPS）：显示总QPS、2xx QPS、4xx QPS、5xx QPS、429 QPS
平均访问延迟：显示请求延时趋势
行数统计：显示总行数、失败行数
流量统计：显示外网流入、外网流出、内网流入、内网流出、免费外网流出
CapacityUnit统计：显示读CU、写CU、内部读CU、内部写CU
请求状态统计：显示执行状态分类（请求成功、参数错误、请求限流、服务端错误）及其统计值和百分比
支持切换时间范围（如1天、7天）和操作类型（创建记忆操作、搜索记忆操作）进行筛选查看。
生态集成
MCP集成
记忆存储支持通过MCP工具集成到快速创建Agent中使用。
在记忆存储实例详情页，点击集成与案例标签，选择MCP集成子标签。
查看MCP服务状态：
如果显示"未启动"，点击启动服务配置按钮启动MCP服务
服务启动后，状态显示为"处理中"
在快速创建Agent页面，在记忆配置项中选择已创建的记忆存储实例，并启用MCP选项。
启用后，Agent可在运行时通过MCP工具访问记忆存储，实现记忆的创建、搜索和管理，为Agent提供持久化记忆能力。
代码集成
通过AgentRun SDK可直接集成记忆存储功能。
from agentrun.memory_collection import MemoryCollection
# 获取记忆存储META信息
memory_collection = MemoryCollection.get_by_name("mem-OGzJ")
print(f"获取成功： {memory_collection}")
# 转换为mem0_memory客户端
memory = MemoryCollection.to_mem0_memory("mem-OGzJ")
# 使用mem0ai Memory客户端进行操作
user_id = "user123"
# 添加记忆
result = memory.add(
"我喜欢吃苹果和香蕉",
user_id=user_id,
metadata={"category": "food"},
)
for idx, res in enumerate(result.get("results", []), 1):
print(f" {idx}. ID: {res.get('id')}, 事件： {res.get('event')}")
# 搜索记忆
search_results = memory.search("用户喜欢吃什么水果？", user_id=user_id)
for idx, result in enumerate(search_results.get("results", []), 1):
print(
f" {idx}. 内容： {result.get('memory')}, 相似度： {result.get('score', 0):.4f}"
)LangChain集成
记忆存储可与LangChain等开源Agent开发框架集成，实现基于记忆的对话系统。该示例展示了如何将记忆存储集成到LangChain对话链中，实现基于历史记忆的个性化响应。
from typing import List, Dict
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from agentrun.integration.langchain import model
from agentrun.memory_collection import MemoryCollection
# 从模型名称创建
llm = model("qwen3-max")
# 从记忆存储名称创建
mem0 = MemoryCollection.to_mem0_memory("mem-OGzJ")
prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="""You are a helpful travel agent AI. Use the provided context to personalize your responses and remember user preferences and past interactions.
Provide travel recommendations, itinerary suggestions, and answer questions about destinations.
If you don't have specific information, you can make general suggestions based on common travel knowledge."""),
MessagesPlaceholder(variable_name="context"),
HumanMessage(content="{input}")
])
def retrieve_context(query: str, user_id: str) -> List[Dict]:
"""Retrieve relevant context from Mem0"""
try:
memories = mem0.search(query, user_id=user_id)
memory_list = memories['results']
print(f"Memory search successfully: {memory_list}")
serialized_memories = ' '.join([mem["memory"] for mem in memory_list])
context = [
{
"role": "system",
"content": f"Relevant information: {serialized_memories}"
},
{
"role": "user",
"content": query
}
]
return context
except Exception as e:
print(f"Error retrieving memories: {e}")
# Return empty context if there's an error
return [{"role": "user", "content": query}]
def generate_response(input: str, context: List[Dict]) -> str:
"""Generate a response using the language model"""
chain = prompt | llm
response = chain.invoke({
"context": context,
"input": input
})
return response.content
def save_interaction(user_id: str, user_input: str, assistant_response: str):
"""Save the interaction to Mem0"""
try:
interaction = [
{
"role": "user",
"content": user_input
},
{
"role": "assistant",
"content": assistant_response
}
]
result = mem0.add(interaction, user_id=user_id)
print(f"Memory saved successfully: {len(result.get('results', []))} memories added")
except Exception as e:
print(f"Error saving interaction: {e}")
def chat_turn(user_input: str, user_id: str) -> str:
# Retrieve context
context = retrieve_context(user_input, user_id)
# Generate response
response = generate_response(user_input, context)
# Save interaction
save_interaction(user_id, user_input, response)
return response
if __name__ == "__main__":
print("Welcome to your personal Travel Agent Planner! How can I assist you with your travel plans today?")
user_id = "alice"
while True:
user_input = input("You: ")
if user_input.lower() in ['quit', 'exit', 'bye']:
print("Travel Agent: Thank you for using our travel planning service. Have a great trip!")
break
response = chat_turn(user_input, user_id)
print(f"Travel Agent: {response}")