使用Mem0构建具备持久记忆能力的AI智能体

项目概述

Mem0是一个提供持久记忆能力的AI服务，通过与OpenAI Agents SDK集成，开发者可以创建具备记忆功能的AI智能体。本文将详细介绍如何使用Mem0为AI智能体添加记忆能力，使其能够记住过往对话并根据上下文提供更精准的响应。

核心优势

Mem0为AI智能体带来的核心价值包括：

1. 上下文感知：智能体可以记住用户偏好和历史对话
2. 长期记忆：信息存储不受单次会话限制
3. 精准检索：基于语义搜索快速定位相关记忆
4. 多用户隔离：支持为不同用户维护独立的记忆空间

环境准备

安装依赖

首先需要安装必要的Python包：

pip install mem0ai pydantic openai-agents

配置API密钥

将Mem0 API密钥设置为环境变量：

export MEM0_API_KEY="your_mem0_api_key"

或者在Python代码中直接设置：

import os
os.environ["MEM0_API_KEY"] = "your_mem0_api_key"

架构设计

Mem0集成架构包含三个核心组件：

1. 上下文管理器：定义用户上下文信息
2. 记忆工具集：实现记忆的增删改查功能
3. 记忆智能体：配置了记忆工具的主智能体

详细实现步骤

1. 导入依赖库

from __future__ import annotations
import os
import asyncio
from pydantic import BaseModel
from mem0 import AsyncMemoryClient
from agents import (
    Agent,
    ItemHelpers,
    MessageOutputItem,
    RunContextWrapper,
    Runner,
    ToolCallItem,
    ToolCallOutputItem,
    TResponseInputItem,
    function_tool,
)

2. 定义记忆上下文模型

使用Pydantic定义用户上下文模型：

class Mem0Context(BaseModel):
    user_id: str | None = None

3. 初始化Mem0客户端

client = AsyncMemoryClient(api_key=os.getenv("MEM0_API_KEY"))

4. 实现记忆工具

添加记忆

@function_tool
async def add_to_memory(
    context: RunContextWrapper[Mem0Context],
    content: str,
) -> str:
    """
    将信息存入Mem0
    参数:
        content: 需要存储的内容
    """
    messages = [{"role": "user", "content": content}]
    user_id = context.context.user_id or "default_user"
    await client.add(messages, user_id=user_id)
    return f"已存储信息: {content}"

搜索记忆

@function_tool
async def search_memory(
    context: RunContextWrapper[Mem0Context],
    query: str,
) -> str:
    """
    在Mem0中搜索记忆
    参数:
        query: 搜索查询词
    """
    user_id = context.context.user_id or "default_user"
    memories = await client.search(query, user_id=user_id, output_format="v1.1")
    results = '\n'.join([result["memory"] for result in memories["results"]])
    return str(results)

获取全部记忆

@function_tool
async def get_all_memory(
    context: RunContextWrapper[Mem0Context],
) -> str:
    """从Mem0获取所有记忆"""
    user_id = context.context.user_id or "default_user"
    memories = await client.get_all(user_id=user_id, output_format="v1.1")
    results = '\n'.join([result["memory"] for result in memories["results"]])
    return str(results)

5. 配置记忆智能体

memory_agent = Agent[Mem0Context](
    name="记忆助手",
    instructions="""你是一个具备记忆能力的智能助手，可以：
    1. 使用add_to_memory存储新信息
    2. 使用search_memory搜索已有信息
    3. 使用get_all_memory检索所有存储的信息
    
    当用户提问时：
    - 如需存储信息，使用add_to_memory
    - 如需搜索特定信息，使用search_memory
    - 如需查看所有存储内容，使用get_all_memory""",
    tools=[add_to_memory, search_memory, get_all_memory],
)

6. 实现主运行循环

async def main():
    current_agent: Agent[Mem0Context] = memory_agent
    input_items: list[TResponseInputItem] = []
    context = Mem0Context()
    while True:
        user_input = input("请输入消息(输入'quit'退出): ")
        if user_input.lower() == 'quit':
            break
        input_items.append({"content": user_input, "role": "user"})
        result = await Runner.run(current_agent, input_items, context=context)
        for new_item in result.new_items:
            agent_name = new_item.agent.name
            if isinstance(new_item, MessageOutputItem):
                print(f"{agent_name}: {ItemHelpers.text_message_output(new_item)}")
            elif isinstance(new_item, ToolCallItem):
                print(f"{agent_name}: 正在调用工具")
            elif isinstance(new_item, ToolCallOutputItem):
                print(f"{agent_name}: 工具调用输出: {new_item.output}")
            else:
                print(f"{agent_name}: 跳过项目: {new_item.__class__.__name__}")
        input_items = result.to_input_list()

if __name__ == "__main__":
    asyncio.run(main())

使用场景示例

存储信息

用户: 记住我最喜欢的颜色是蓝色
智能体: 正在调用工具
智能体: 工具调用输出: 已存储信息: 最喜欢的颜色是蓝色
智能体: 我已将您最喜欢的颜色是蓝色存入记忆，以后会记住这个信息。

搜索记忆

用户: 我最喜欢什么颜色？
智能体: 正在调用工具
智能体: 工具调用输出: 最喜欢的颜色是蓝色
智能体: 根据之前的对话，您最喜欢的颜色是蓝色。

检索所有记忆

用户: 你了解我哪些信息？
智能体: 正在调用工具
智能体: 工具调用输出: 最喜欢的颜色是蓝色
生日是3月15日
智能体: 根据我们的对话记录，我知道：
1. 您最喜欢的颜色是蓝色
2. 您的生日是3月15日

高级配置

自定义用户ID

可以为不同用户指定不同的ID，维护独立的记忆空间：

context = Mem0Context(user_id="user123")

最佳实践

1. 记忆组织：将相关信息分组存储，便于后续检索
2. 定期清理：对于不再相关的记忆可以定期清理
3. 隐私保护：敏感信息存储前应考虑加密处理
4. 记忆验证：关键信息可设计二次确认机制

总结

通过Mem0与OpenAI Agents SDK的集成，开发者可以轻松为AI智能体添加强大的记忆能力。这种能力使得智能体能够提供更加个性化、上下文相关的服务，显著提升用户体验。本文详细介绍了从环境配置到实际应用的完整流程，开发者可以根据实际需求进行调整和扩展。