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)