长期记忆

使用场景

• 用户偏好记忆：记住用户的语言偏好、常用设置等，在后续对话中自动应用

• 历史交互回顾：跨会话检索用户过往对话内容，避免重复提问

• 个性化推荐：基于用户历史行为和偏好，提供个性化的推荐和建议

前提条件

• 已创建记忆存储实例，可参考创建和管理记忆存储。

• 长期记忆需要配置大语言模型（LLM）和向量模型（Embedder），即 llmConfig 和 embedderConfig 均非空

限制说明

添加记忆

1. 进入记忆存储实例详情页，选择长期记忆。

2. 点击添加记忆按钮，打开添加用户记忆表单。

3. 填写记忆信息：

   字段	说明	示例
   用户ID	必填，标识记忆所属的用户	test-user-001
   记忆内容	必填，记忆的具体内容	居住在杭州
   元数据	可选，自定义键值对。点击添加元数据动态增加	{"category": "food"}

4. 点击保存，等待创建完成。

创建成功后，记忆出现在记忆列表中，显示用户 ID、创建时间、记忆内容等信息。每个记忆卡片提供查看详情、编辑、查看历史、删除等操作按钮。

查看记忆列表

记忆列表默认展示最多 200 条记忆。每条记忆卡片包含以下信息：

• 第一行：User ID（粗体），搜索模式下额外显示匹配度分数标签

• 第二行：记忆内容原文

• 第三行：元信息格式为 {Agent ID} · {Run ID} · {日期}，Agent ID 和 Run ID 为可选，无值时不显示

支持通过复选框批量选择记忆条目。当有选中项时，列表右上角出现批量删除按钮。

搜索和筛选记忆

当记忆数量超过 200 条时，使用搜索功能查找特定记忆。搜索时需同时提供关键词和至少一个 ID（User ID / Agent ID / Run ID）。

1. 在记忆列表页面顶部的搜索框中输入关键词。

2. 展开高级筛选面板，填写至少一个筛选条件（用户 ID、Agent ID 或 Run ID）。

3. 搜索结果按向量相似度从高到低排序，每条结果显示匹配度分数。

查看记忆详情

1. 在记忆列表中，点击记忆卡片右侧的查看按钮（Eye 图标）。

2. 详情抽屉展示以下信息：

   • Memory ID、User ID、Agent ID（无值显示 -）、Run ID（无值显示 -）、创建时间

   • 记忆内容区域，保留原始换行格式

   • 元数据区域（如有），以 JSON 格式展示

   • 标签区域（如 metadata.tags 存在），以蓝色圆角标签展示

修改记忆

1. 在记忆列表中，点击目标记忆右侧的编辑按钮。

2. 在编辑表单中修改记忆内容字段。

3. 点击保存更改。

编辑表单底部展示 User ID、Agent ID、Run ID、创建时间和更新时间等只读信息。

删除记忆

查看记忆历史

记忆的版本历史记录了每次新增、更新、删除操作的变更轨迹。

1. 在记忆详情抽屉中，点击查看历史按钮。

2. 历史列表按时间倒序排列，每条记录包含：

   • 版本信息：版本号 + 「当前版本」标签（最新版本）+ 事件类型（新增/更新/删除）

   • 时间戳：格式为 YYYY/MM/DD HH:mm:ss

   • 操作者：actor_id + role（如有）

   • 记忆内容：当前版本的记忆文本

   • 变更详情（如有变更）：旧值（红色背景）和新值（绿色背景）对比

MCP 集成

记忆存储支持通过 MCP（Model Context Protocol）服务将记忆能力集成到智能体中。在 MCP 集成中，可以启动/关闭 MCP 服务、获取服务 URL、查看动态配置，并将服务接入智能体。

代码集成

通过 AgentRun SDK 可集成记忆存储功能。以下示例演示如何通过 MemoryCollection 获取记忆存储配置，并转换为 mem0 Memory 客户端进行记忆操作。

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 对话链中，实现基于历史记忆的个性化响应。

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}")

开启长期记忆失败

MCP 服务启动失败

添加记忆失败

搜索记忆失败