The Integration Pattern

Mem0 integrates at two points in any agent interaction: before the agent runs (retrieve relevant memories and inject into context), and after the agent runs (add the exchange to memory). The exact hook varies by framework, but the pattern is always the same.

# The universal Mem0 integration pattern
# 1. Before agent call:
memories = mem0.search(query=user_input, user_id=user_id)
context = format_memories(memories)
 
# 2. Run agent with context injected
response = agent.run(system=base_system + context, input=user_input)
 
# 3. After agent call:
mem0.add([{"role": "user", "content": user_input},
          {"role": "assistant", "content": response}],
         user_id=user_id)

LangGraph Integration

In LangGraph, the cleanest integration point is a dedicated memory node at the start and end of your graph.

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from mem0 import Memory
import operator
 
mem0 = Memory()
 
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    user_id: str
    memory_context: str
 
def retrieve_memories(state: AgentState) -> dict:
    """Node: fetch relevant memories before the agent runs."""
    last_message = state["messages"][-1]["content"]
    memories = mem0.search(query=last_message, user_id=state["user_id"], limit=5)
    context = "\n".join([f"- {m['memory']}" for m in memories])
    return {"memory_context": context}
 
def agent_node(state: AgentState) -> dict:
    """Node: run the agent with memory context injected."""
    system = "You are a helpful assistant.\n\nRelevant facts about this user:\n" + state["memory_context"]
    # ... call your LLM with system prompt ...
    response_content = call_llm(system, state["messages"])
    return {"messages": [{"role": "assistant", "content": response_content}]}
 
def store_memories(state: AgentState) -> dict:
    """Node: store the completed exchange after agent responds."""
    last_user = next(m for m in reversed(state["messages"]) if m["role"] == "user")
    last_asst = next(m for m in reversed(state["messages"]) if m["role"] == "assistant")
    mem0.add([last_user, last_asst], user_id=state["user_id"])
    return {}
 
# Build the graph
graph = StateGraph(AgentState)
graph.add_node("retrieve_memories", retrieve_memories)
graph.add_node("agent", agent_node)
graph.add_node("store_memories", store_memories)
graph.set_entry_point("retrieve_memories")
graph.add_edge("retrieve_memories", "agent")
graph.add_edge("agent", "store_memories")
graph.add_edge("store_memories", END)
 
app = graph.compile()
 
Add memory_context to your AgentState TypedDict so it flows through the graph cleanly. Avoid calling mem0.search() inside your main agent node — keep retrieval and storage as separate, testable nodes.

CrewAI Integration

CrewAI does not have a built-in memory hook that fires at the agent level per-user. The cleanest integration is to inject memories via the agent's backstory or via a dedicated memory-retrieval task at the start of the crew.

Option A: Memory in Agent Backstory (Simple)

from crewai import Agent, Task, Crew
from mem0 import Memory
 
mem0 = Memory()
 
def build_crew_with_memory(user_input: str, user_id: str) -> str:
    # Retrieve relevant memories
    memories = mem0.search(query=user_input, user_id=user_id, limit=5)
    memory_facts = "\n".join([f"- {m['memory']}" for m in memories])
 
    analyst = Agent(
        role="Research Analyst",
        goal="Provide accurate, personalised research",
        backstory=(
            "You are an expert research analyst. "
            f"Known facts about the person you are helping:\n{memory_facts}"
        ),
        verbose=False
    )
 
    task = Task(
        description=user_input,
        agent=analyst,
        expected_output="A thorough research response"
    )
 
    crew = Crew(agents=[analyst], tasks=[task])
    result = crew.kickoff()
 
    # Store the exchange
    mem0.add(
        [{"role": "user", "content": user_input},
         {"role": "assistant", "content": str(result)}],
        user_id=user_id
    )
 
    return str(result)

Option B: Memory Retrieval Task (Structured)

For multi-agent crews, add a dedicated memory retrieval agent as the first task in your crew. Its output becomes context for all subsequent agents.

from crewai import Agent, Task, Crew
from mem0 import Memory
 
mem0 = Memory()
 
def memory_retriever_tool(query: str, user_id: str) -> str:
    memories = mem0.search(query=query, user_id=user_id, limit=8)
    if not memories:
        return "No prior context found for this user."
    return "\n".join([f"- {m['memory']}" for m in memories])
 
Do not use CrewAI's built-in memory=True alongside Mem0 simultaneously — they will both attempt to store conversation history and you will get duplicate, conflicting memory entries. Pick one.

PydanticAI Integration

PydanticAI's dependency injection system (deps_type) makes Mem0 integration clean — inject the memory instance and user_id as dependencies, then use them inside tool functions and the system prompt.

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
from mem0 import Memory
from dataclasses import dataclass
 
mem0_instance = Memory()
 
@dataclass
class AgentDeps:
    user_id: str
    mem0: Memory
 
def build_system_prompt(ctx: RunContext[AgentDeps], user_input: str) -> str:
    memories = ctx.deps.mem0.search(query=user_input, user_id=ctx.deps.user_id, limit=5)
    facts = "\n".join([f"- {m['memory']}" for m in memories])
    return f"You are a helpful assistant.\n\nWhat you know about this user:\n{facts}"
 
agent = Agent(
    model="claude-sonnet-4-6",
    deps_type=AgentDeps,
    system_prompt=build_system_prompt
)
 
async def run_with_memory(user_input: str, user_id: str) -> str:
    deps = AgentDeps(user_id=user_id, mem0=mem0_instance)
 
    result = await agent.run(user_input, deps=deps)
    response = result.data
 
    # Store the exchange
    mem0_instance.add(
        [{"role": "user", "content": user_input},
         {"role": "assistant", "content": response}],
        user_id=user_id
    )
 
    return response

Common Mistakes Across All Integrations

  • Storing every message, not just meaningful exchanges — Mem0 extraction is not free (one LLM call per add). Only add exchanges that likely contain extractable facts, not short back-and-forth confirmations.
  • Not scoping memories by user_id — if you omit user_id, memories are global across all users. This is rarely what you want and is a privacy issue.
  • Retrieving memories after the agent has already generated its response — memory must be retrieved before the LLM call to influence the response.
  • Forgetting to handle empty memory results — if a user is new, mem0.search() returns an empty list. Your prompt formatting must handle this gracefully rather than injecting an empty 'What you know about this user:' section.
  • Running mem0.add() synchronously in a high-traffic API — memory extraction involves an LLM call. For latency-sensitive applications, run add() in the background after returning the response to the user.

Background Memory Storage Pattern

import asyncio
from mem0 import AsyncMemory
 
mem0 = AsyncMemory()
 
async def run_agent_with_background_memory(user_input: str, user_id: str) -> str:
    # Retrieve synchronously (needed before agent runs)
    memories = await mem0.search(query=user_input, user_id=user_id, limit=5)
    context = "\n".join([f"- {m['memory']}" for m in memories])
 
    # Run agent
    response = await call_agent(user_input, context)
 
    # Store in background — don't block the response
    asyncio.create_task(
        mem0.add(
            [{"role": "user", "content": user_input},
             {"role": "assistant", "content": response}],
            user_id=user_id
        )
    )
 
    return response  # returns immediately, memory stored async