Mem0: 56K+ Stars — AI智能体内存性能调优指南 2026

# 安装 Python 客户端
pip install mem0ai

# 设置 API 密钥（从 https://app.mem0.ai 获取）
export MEM0_API_KEY="m0-your-key-here"

# mem0_quickstart.py
import os
from mem0 import MemoryClient

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

# 从对话中存储记忆
messages = [
    {"role": "user", "content": "I'm a vegetarian and allergic to nuts."},
    {"role": "assistant", "content": "Got it! I'll remember your dietary preferences."},
]
client.add(messages, user_id="user123")

# 检索相关记忆
results = client.search(
    "What are my dietary restrictions?",
    user_id="user123"
)
print(results)

# 克隆仓库
git clone https://github.com/mem0ai/mem0.git
cd mem0

# 使用 Docker 启动
make bootstrap
# 创建管理员用户，生成 API 密钥，启动服务器和管理面板

# docker-compose.yml 生产环境配置
docker run -d \
  -p 8000:8000 \
  -e MEM0_API_KEY=your-admin-key \
  -e VECTOR_STORE_PROVIDER=qdrant \
  -e VECTOR_STORE_URL=http://qdrant:6333 \
  -e LLM_PROVIDER=openai \
  -e OPENAI_API_KEY=$OPENAI_API_KEY \
  mem0/mem0-server:latest

pip install mem0ai openai chromadb

from mem0 import Memory

# 使用自定义向量存储初始化
m = Memory()

# 添加对话
messages = [
    {"role": "user", "content": "I'm planning to watch a movie tonight."},
    {"role": "assistant", "content": "How about thrillers?"},
    {"role": "user", "content": "I love sci-fi but hate horror."},
]
m.add(messages, user_id="alice", metadata={"category": "movies"})

# 使用元数据过滤搜索
results = m.search("movie recommendations", filters={"user_id": "alice"})

# mem0config.yaml — 生产级调优配置
llm:
  provider: openai
  config:
    model: "gpt-4o-mini"
    temperature: 0.1
    max_tokens: 2000

embedder:
  provider: openai
  config:
    model: "text-embedding-3-small"
    embedding_dims: 1536

vector_store:
  provider: qdrant
  config:
    host: "localhost"
    port: 6333
    collection_name: "mem0"
    on_disk: true  # 启用持久化存储

reranker:
  provider: cohere
  config:
    model: "rerank-multilingual-v3.0"

custom_instructions: |
  提取用户偏好、个人事实和上下文。
  重点关注饮食限制、过敏和技术偏好。
  忽略临时状态和一次性请求。

from mem0 import Memory

# 加载自定义配置
config_path = "mem0config.yaml"
m = Memory.from_config(config_path)

# 1. 为高吞吐量应用启用异步记忆
from mem0 import MemoryClient
import asyncio

client = MemoryClient()

async def batch_store(messages_list):
    tasks = [client.add_async(msgs, user_id=f"user_{i}")
             for i, msgs in enumerate(messages_list)]
    return await asyncio.gather(*tasks)

# 2. 使用元数据过滤限定搜索范围
results = client.search(
    "project updates",
    filters={
        "user_id": "alice",
        "metadata.category": "work"
    },
    limit=10
)

# 3. 使用 top_k 调整检索深度
results = client.search(
    "dietary preferences",
    user_id="alice",
    top_k=5,  # 减少提升速度，增加提升覆盖度
    rerank=True
)

# 引导提取和存储哪些事实
m = Memory.from_config({
    "custom_instructions": """
    提取并存储：
    - 用户的姓名、职业、地点
    - 技术偏好（语言、框架、工具）
    - 饮食限制和过敏
    - 沟通偏好

    不要存储：
    - 临时情绪或情感状态
    - 一次性请求
    - 未经同意的第三方信息
    """
})

pip install langchain langchain-openai mem0ai

# langchain_mem0_agent.py
import os
from typing import List, Dict
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from mem0 import MemoryClient

# 初始化
llm = ChatOpenAI(model="gpt-4o-mini")
mem0 = MemoryClient(api_key=os.getenv("MEM0_API_KEY"))

# 带记忆注入的提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是拥有长期记忆的助手。
    关于用户的相关历史上下文：
    {memories}

    使用这些上下文来个性化你的回答。"""),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

def get_memories(user_id: str, query: str) -> str:
    """检索相关记忆并格式化为字符串。"""
    results = mem0.search(query, user_id=user_id, limit=5)
    return "\n".join([r["memory"] for r in results])

def chat(user_id: str, message: str, history: List = None):
    """带记忆增强上下文的对话。"""
    if history is None:
        history = []

    memories = get_memories(user_id, message)
    formatted_prompt = prompt.format_messages(
        memories=memories,
        history=history,
        input=message
    )

    response = llm.invoke(formatted_prompt)

    # 存储交互
    messages = [
        {"role": "user", "content": message},
        {"role": "assistant", "content": response.content}
    ]
    mem0.add(messages, user_id=user_id)

    return response.content

# 运行对话
user_id = "traveler_42"
response1 = chat(user_id, "I'm planning a trip to Tokyo next month.")
print(response1)

# 后续会话 — 智能体记住了
response2 = chat(user_id, "What should I pack for my trip?")
# 输出引用东京、时间以及旅行者的偏好

pip install crewai mem0ai

# crewai_mem0_crew.py
import os
from crewai import Agent, Task, Crew
from crewai_tools import tool
from mem0 import MemoryClient

mem0 = MemoryClient(api_key=os.getenv("MEM0_API_KEY"))

@tool
def retrieve_user_context(user_id: str, query: str) -> str:
    """检索关于用户的记忆用于个性化。"""
    results = mem0.search(query, user_id=user_id, limit=5)
    return "\n".join([f"- {r['memory']}" for r in results])

@tool
def store_interaction(user_id: str, content: str) -> str:
    """存储智能体交互中学到的事实。"""
    messages = [{"role": "assistant", "content": content}]
    mem0.add(messages, user_id=user_id)
    return "Stored."

# 定义带有记忆工具的智能体
researcher = Agent(
    role="Personal Researcher",
    goal="Provide personalized research based on user history",
    backstory="You remember user preferences and tailor research accordingly.",
    tools=[retrieve_user_context, store_interaction],
    verbose=True,
    memory=True
)

# 使用记忆的任务
task = Task(
    description="""Research travel options for user {{user_id}}.
    First retrieve their preferences, then provide personalized recommendations.
    Query: travel preferences""",
    expected_output="Personalized travel recommendations",
    agent=researcher
)

crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff(inputs={"user_id": "user123"})
print(result)

pip install openai-agents mem0ai

# openai_agents_mem0.py
import os
from dataclasses import dataclass
from agents import Agent, Runner, function_tool
from mem0 import MemoryClient

mem0 = MemoryClient(api_key=os.getenv("MEM0_API_KEY"))

@dataclass
class UserContext:
    user_id: str

@function_tool
def add_to_memory(ctx, messages: str) -> str:
    """存储关于用户的事实。"""
    parsed = [{"role": "user", "content": m} for m in messages.split("\n")]
    mem0.add(parsed, user_id=ctx.context.user_id)
    return "Memory stored."

@function_tool
def search_memory(ctx, query: str) -> str:
    """搜索相关记忆。"""
    results = mem0.search(query, user_id=ctx.context.user_id, limit=5)
    return "\n".join([r["memory"] for r in results])

memory_agent = Agent(
    name="MemoryAgent",
    instructions="You are a helpful assistant that remembers user preferences.",
    tools=[add_to_memory, search_memory],
    model="gpt-4o-mini"
)

async def run_agent():
    context = UserContext(user_id="user_42")
    result = await Runner.run(
        memory_agent,
        "I'm a vegetarian who loves Italian food.",
        context=context
    )
    print(result.final_output)

# asyncio.run(run_agent())

# mem0-production-stack.yml
version: "3.8"

services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_storage:/qdrant/storage
    environment:
      - QDRANT__SERVICE__GRPC_PORT=6334

  mem0-server:
    image: mem0/mem0-server:latest
    ports:
      - "8000:8000"
    environment:
      - MEM0_API_KEY=${MEM0_API_KEY}
      - VECTOR_STORE_PROVIDER=qdrant
      - VECTOR_STORE_URL=http://qdrant:6333
      - LLM_PROVIDER=openai
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - EMBEDDER_PROVIDER=openai
      - OPENAI_EMBEDDING_MODEL=text-embedding-3-small
    depends_on:
      - qdrant

  mem0-dashboard:
    image: mem0/mem0-dashboard:latest
    ports:
      - "3000:3000"
    environment:
      - MEM0_API_URL=http://mem0-server:8000
      - MEM0_API_KEY=${MEM0_API_KEY}

volumes:
  qdrant_storage:

# 使用元数据进行记忆访问控制
def store_sensitive_memory(user_id: str, fact: str, classification: str):
    """存储带安全等级的记忆。"""
    messages = [{"role": "user", "content": fact}]
    mem0.add(
        messages,
        user_id=user_id,
        metadata={
            "classification": classification,  # "public", "internal", "confidential"
            "encrypted": True,
            "retention_days": 90
        }
    )

# 带分类过滤的搜索
results = client.search(
    "project preferences",
    filters={
        "user_id": "alice",
        "metadata.classification": ["public", "internal"]
    }
)

# 面向 SaaS 应用的组织级记忆隔离
def add_org_scoped_memory(org_id: str, user_id: str, messages: list):
    """存储同时按组织和用户范围隔离的记忆。"""
    client.add(
        messages,
        user_id=f"{org_id}:{user_id}",
        metadata={"org_id": org_id, "isolation": "org-scoped"}
    )

# 管理员跨用户查询组织内所有记忆
results = client.get_all(
    filters={"metadata.org_id": "org_123"},
    limit=100
)

# 追踪记忆指标
import time

def timed_search(user_id: str, query: str):
    """带延迟日志的搜索。"""
    start = time.time()
    results = client.search(query, user_id=user_id)
    latency = (time.time() - start) * 1000

    print(f"搜索延迟: {latency:.1f}ms | 结果数: {len(results)}")

    # 记录到你的监控系统
    # prometheus_histogram.observe(latency)
    return results

# 定期记忆健康检查
def memory_health_check(user_id: str):
    """验证用户记忆的完整性。"""
    all_memories = client.get_all(filters={"user_id": user_id})

    return {
        "total_memories": len(all_memories),
        "oldest_memory": min(m["created_at"] for m in all_memories) if all_memories else None,
        "categories": len(set(m.get("metadata", {}).get("category", "") for m in all_memories)),
        "avg_score": sum(m.get("score", 0) for m in all_memories) / len(all_memories) if all_memories else 0
    }

# 客户端速率限制实现
from functools import wraps
import time

class Mem0RateLimiter:
    """Mem0 API 调用的简单速率限制器。"""
    def __init__(self, max_calls_per_minute=100):
        self.max_calls = max_calls_per_minute
        self.calls = []

    def can_call(self) -> bool:
        now = time.time()
        self.calls = [c for c in self.calls if now - c < 60]
        return len(self.calls) < self.max_calls

    def record_call(self):
        self.calls.append(time.time())

limiter = Mem0RateLimiter(max_calls_per_minute=60)

def rate_limited_add(messages, user_id):
    if not limiter.can_call():
        # 稍后排队或跳过非关键记忆
        print("触发速率限制，记忆入队")
        return {"status": "queued"}
    limiter.record_call()
    return client.add(messages, user_id=user_id)