Mem0: 56K+ Stars — Hướng Dẫn Tinh Chỉnh Hiệu Suất Bộ Nhớ AI Agent 2026

# Cài đặt Python client
pip install mem0ai

# Thiết lập API key từ 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"))

# Lưu trữ ký ức từ cuộc hội thoại
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")

# Truy xuất ký ức liên quan
results = client.search(
    "What are my dietary restrictions?",
    user_id="user123"
)
print(results)

# Clone repository
git clone https://github.com/mem0ai/mem0.git
cd mem0

# Bootstrap với Docker
make bootstrap
# Tạo admin, phát sinh API key, khởi động server + dashboard

# docker-compose.yml cho production
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

# Khởi tạo với vector store tùy chỉnh
m = Memory()

# Thêm cuộc hội thoại
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"})

# Tìm kiếm với lọc metadata
results = m.search("movie recommendations", filters={"user_id": "alice"})

# mem0config.yaml — Cấu hình tinh chỉnh production
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  # Bật lưu trữ liên tục

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

custom_instructions: |
  Trích xuất sở thích ngườ dùng, sự kiện cá nhân và ngữ cảnh.
  Tập trung vào hạn chế chế độ ăn, dị ứng và sở thích công nghệ.
  Bỏ qua trạng thái tạm thờ và yêu cầu một lần.

from mem0 import Memory

# Tải cấu hình tùy chỉnh
config_path = "mem0config.yaml"
m = Memory.from_config(config_path)

# 1. Bật async memory cho ứng dụng throughput cao
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. Sử dụng lọc metadata để giới hạn phạm vi tìm kiếm
results = client.search(
    "project updates",
    filters={
        "user_id": "alice",
        "metadata.category": "work"
    },
    limit=10
)

# 3. Điều chỉnh độ sâu truy xuất với top_k
results = client.search(
    "dietary preferences",
    user_id="alice",
    top_k=5,  # Giảm để tăng tốc, tăng để mở rộng bao phủ
    rerank=True
)

# Hớng dẫn trích xuất và lưu trữ những sự kiện nào
m = Memory.from_config({
    "custom_instructions": """
    Trích xuất và lưu trữ:
    - Tên, nghề nghiệp, vị trí của ngườ dùng
    - Sở thích công nghệ (ngôn ngữ, framework, công cụ)
    - Hạn chế chế độ ăn và dị ứng
    - Sở thích giao tiếp

    KHÔNG lưu trữ:
    - Trạng thái tâm trạng tạm thờ
    - Yêu cầu một lần
    - Thông tin bên thứ ba không có sự đồng ý
    """
})

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

# Khởi tạo
llm = ChatOpenAI(model="gpt-4o-mini")
mem0 = MemoryClient(api_key=os.getenv("MEM0_API_KEY"))

# Template prompt với memory injection
prompt = ChatPromptTemplate.from_messages([
    ("system", """Bạn là trợ lý hữu ích với bộ nhớ dài hạn.
    Ngữ cảnh quá khứ liên quan về ngườ dùng:
    {memories}

    Sử dụng ngữ cảnh này để cá nhân hóa phản hồi."""),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

def get_memories(user_id: str, query: str) -> str:
    """Truy xuất ký ức liên quan dưới dạng chuỗi."""
    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):
    """Chat với ngữ cảnh tăng cường bộ nhớ."""
    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)

    # Lưu trữ tương tác
    messages = [
        {"role": "user", "content": message},
        {"role": "assistant", "content": response.content}
    ]
    mem0.add(messages, user_id=user_id)

    return response.content

# Chạy cuộc hội thoại
user_id = "traveler_42"
response1 = chat(user_id, "I'm planning a trip to Tokyo next month.")
print(response1)

# Phiên sau — agent nhớ
response2 = chat(user_id, "What should I pack for my trip?")
# Output tham chiếu Tokyo, thờ điểm và sở thích ngườ đi du lịch

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:
    """Truy xuất ký ức về ngườ dùng để cá nhân hóa."""
    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:
    """Lưu trữ sự kiện học được trong tương tác agent."""
    messages = [{"role": "assistant", "content": content}]
    mem0.add(messages, user_id=user_id)
    return "Stored."

# Định nghĩa agent với công cụ bộ nhớ
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 sử dụng bộ nhớ
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:
    """Lưu trữ sự kiện về ngườ dùng."""
    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:
    """Tìm kiếm ký ức liên quan."""
    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:

# Kiểm soát truy cập bộ nhớ bằng metadata
def store_sensitive_memory(user_id: str, fact: str, classification: str):
    """Lưu trữ bộ nhớ với phân loại bảo mật."""
    messages = [{"role": "user", "content": fact}]
    mem0.add(
        messages,
        user_id=user_id,
        metadata={
            "classification": classification,  # "public", "internal", "confidential"
            "encrypted": True,
            "retention_days": 90
        }
    )

# Tìm kiếm với bộ lọc phân loại
results = client.search(
    "project preferences",
    filters={
        "user_id": "alice",
        "metadata.classification": ["public", "internal"]
    }
)

# Bộ nhớ phạm vi tổ chức cho ứng dụng SaaS
def add_org_scoped_memory(org_id: str, user_id: str, messages: list):
    """Lưu trữ bộ nhớ phạm vi tổ chức và ngườ dùng."""
    client.add(
        messages,
        user_id=f"{org_id}:{user_id}",
        metadata={"org_id": org_id, "isolation": "org-scoped"}
    )

# Truy vấn admin xuyên suốt mọi ngườ dùng trong tổ chức
results = client.get_all(
    filters={"metadata.org_id": "org_123"},
    limit=100
)

# Theo dõi metrics bộ nhớ
import time

def timed_search(user_id: str, query: str):
    """Tìm kiếm với ghi log độ trễ."""
    start = time.time()
    results = client.search(query, user_id=user_id)
    latency = (time.time() - start) * 1000

    print(f"Độ trễ tìm kiếm: {latency:.1f}ms | Kết quả: {len(results)}")

    # Log vào hệ thống giám sát
    # prometheus_histogram.observe(latency)
    return results

# Kiểm tra sức khỏe bộ nhớ định kỳ
def memory_health_check(user_id: str):
    """Xác minh tính toàn vẹn bộ nhớ cho ngườ dùng."""
    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
    }

# Triển khai giới hạn tốc độ phía client
from functools import wraps
import time

class Mem0RateLimiter:
    """Bộ giới hạn tốc độ đơn giản cho cuộc gọi API Mem0."""
    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():
        # Xếp hàng cho sau hoặc bỏ qua bộ nhớ không quan trọng
        print("Đạt giới hạn tốc độ, xếp hàng bộ nhớ")
        return {"status": "queued"}
    limiter.record_call()
    return client.add(messages, user_id=user_id)