Build an AI Agent with Real Memory Using Mem0, LangChain & Groq

The following notes reconstruct a practical path around “Build an AI Agent with Real Memory Using Mem0, LangChain & Groq”. Emphasis stays on contracts, checks, and drop-in code placeholders rather than motivational framing.

Why LLMs Forget — And Why That Matters

When working through the Why LLMs Forget And stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Cache stable system instructions and tool schemas. Re-sending identical preamble is a common source of burn.

What We Are Building

When working through the What We Are Building stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

Tech Stack at a Glance

When working through the Tech Stack at a stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

Mem0 — the memory layer

When working through the Mem0 the memory layer stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

Groq — inference

When working through the Groq inference stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node. When working through the Groq inference stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline.

LangChain — message formatting

The LangChain message formatting stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

ChromaDB — vector store

The ChromaDB vector store stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

sentence-transformers/all-MiniLM-L6-v2 — embeddings

The sentence-transformers all-MiniLM-L6-v2 embeddings stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move. The sentence-transformers all-MiniLM-L6-v2 embeddings stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline.

Code — Walk-Through

For the Code Walk-Through stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.

1. Central Configuration

For the 1 Central Configuration stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.

@dataclass
class AgentConfig:
    groq_model: str = "llama-3.1-8b-instant"
    temperature: float = 0.7
    max_tokens: int = 1024
    chroma_collection: str = "dev_assistant_memories"
    chroma_path: str = "./mem0_db"
    embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2"
    memory_trigger_phrases: tuple = (
        "my name is", "i am building", "i am working on",
        "my stack is", "i prefer", "my team", ...
    )

2. Memory Helpers

For the 2 Memory Helpers stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness. For the 2 Memory Helpers stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline.

def store_memory(memory, text, user_id) -> None:
    try:
        memory.add(text, user_id=user_id, infer=False)
    except Exception as exc:
        logger.warning("Could not store memory: %s", exc)

def retrieve_memories(memory, query, user_id, top_k=5) -> list[str]:
    try:
        results = memory.search(query=query,
                                filters={'user_id': user_id},
                                limit=top_k)
        return [item['memory'] for item in results.get('results', [])
                if 'memory' in item]
    except Exception as exc:
        logger.warning("Memory retrieval failed: %s", exc)
        return []

3. Smarter Memory Trigger

When working through the 3 Smarter Memory Trigger stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

def should_store(text: str, triggers: tuple) -> bool:
    lowered = text.lower()
    return any(phrase in lowered for phrase in triggers)

4. Grounded Chat Prompt

When working through the 4 Grounded Chat Prompt stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Cache stable system instructions and tool schemas. Re-sending identical preamble is a common source of burn.

system = SystemMessage(content=(
    "You are a memory-aware developer assistant. "
    "Answer ONLY from the provided memory context. "
    "If memory is empty or irrelevant, say: "
    "'I don't have that in memory yet — tell me and I'll remember it.' "
    "Never fabricate projects, names, or technical choices."
))
human = HumanMessage(content=(
    f"### Memory context\n{memory_block}\n\n"
    f"### User question\n{user_query}"
))

5. Dual Modes: REPL & Demo

When working through the 5 Dual Modes REPL stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node. When working through the 5 Dual Modes REPL stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline.

# Interactive session
python memory_agent.py

# Scripted demo (no terminal needed)
python memory_agent.py demo

Setup in Under 5 Minutes

The Setup in Under 5 stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.

Install dependencies

!pip install mem0ai langchain langchain-groq chromadb sentence-transformers

Get a free Groq API key

Run

export GROQ_API_KEY=gsk_...
python memory_agent.py

What the Demo Looks Like

User: My name is Naveen and I am building a multimodal RAG pipeline using Groq and LangChain.
AI: Got it, Aditya! I'll remember your multimodal RAG project.

User: I recently switched from OpenAI embeddings to HuggingFace sentence-transformers.
AI: Noted — HuggingFace sentence-transformers for embeddings.

User: What AI project am I working on?
AI: You are building a multimodal RAG pipeline using Groq and LangChain.

User: Which embedding provider did I switch to?
AI: You switched from OpenAI embeddings to HuggingFace sentence-transformers.

Full code:

import os
import logging
from dataclasses import dataclass, field
from typing import Optional
from getpass import getpass

from mem0 import Memory
from langchain_groq import ChatGroq
from langchain_core.messages import SystemMessage, HumanMessage

# ── Logging ──────────────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)


# ── Configuration ─────────────────────────────────────────────
@dataclass
class AgentConfig:
    """Central config – tweak here, nowhere else."""
    groq_model: str = "llama-3.1-8b-instant"
    temperature: float = 0.7
    max_tokens: int = 1024
    chroma_collection: str = "dev_assistant_memories"
    chroma_path: str = "./mem0_db"
    embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2"
    # Phrases that signal the user is sharing personal context
    memory_trigger_phrases: tuple = (
        "my name is", "i am building", "i am researching",
        "i am working on", "my favorite", "i like", "i study",
        "i use", "i prefer", "my stack is", "i switched to",
        "i recently", "my team", "my company",
    )


# ── Mem0 + LLM bootstrap ──────────────────────────────────────
def build_memory(cfg: AgentConfig) -> Memory:
    mem0_config = {
        "vector_store": {
            "provider": "chroma",
            "config": {
                "collection_name": cfg.chroma_collection,
                "path": cfg.chroma_path,
            },
        },
        "embedder": {
            "provider": "huggingface",
            "config": {"model": cfg.embedding_model},
        },
    }
    memory = Memory.from_config(mem0_config)
    logger.info("Mem0 memory store ready  (collection=%s)", cfg.chroma_collection)
    return memory


def build_llm(cfg: AgentConfig) -> ChatGroq:
    llm = ChatGroq(model=cfg.groq_model, temperature=cfg.temperature, max_tokens=cfg.max_tokens)
    logger.info("Groq LLM ready  (model=%s)", cfg.groq_model)
    return llm


# ── Memory helpers ────────────────────────────────────────────
def store_memory(memory: Memory, text: str, user_id: str) -> None:
    """Persist a memory fragment for the given user."""
    try:
        memory.add(text, user_id=user_id, infer=False)
        logger.debug("Stored memory for %s: %s", user_id, text[:80])
    except Exception as exc:
        logger.warning("Could not store memory: %s", exc)


def retrieve_memories(memory: Memory, query: str, user_id: str, top_k: int = 5) -> list[str]:
    """Return the top-k relevant memory strings for a query."""
    try:
        results = memory.search(query=query, filters={"user_id": user_id}, limit=top_k)
        memories = [
            item["memory"]
            for item in results.get("results", [])
            if "memory" in item
        ]
        logger.debug("Retrieved %d memories for query: %s", len(memories), query[:60])
        return memories
    except Exception as exc:
        logger.warning("Memory retrieval failed: %s", exc)
        return []


def should_store(text: str, triggers: tuple) -> bool:
    """Return True if the utterance carries information worth storing."""
    lowered = text.lower()
    return any(phrase in lowered for phrase in triggers)


# ── Core chat function ────────────────────────────────────────
def chat(
    llm: ChatGroq,
    memory: Memory,
    user_query: str,
    user_id: str,
    cfg: AgentConfig,
) -> str:
    """
    Full turn:
      1. Retrieve relevant memories
      2. Build a grounded prompt
      3. Call LLM
      4. Conditionally store new memories
      5. Return response text
    """
    # 1. Retrieve
    memories = retrieve_memories(memory, user_query, user_id)
    memory_block = "\n".join(f"- {m}" for m in memories) if memories else "(none yet)"

    # 2. Build prompt
    system = SystemMessage(content=(
        "You are a memory-aware developer assistant. "
        "Answer ONLY from the provided memory context. "
        "If memory is empty or irrelevant, say: "
        "'I don't have that in memory yet — tell me and I'll remember it.' "
        "Never fabricate projects, names, or technical choices."
    ))
    human = HumanMessage(content=(
        f"### Memory context\n{memory_block}\n\n"
        f"### User question\n{user_query}"
    ))

    # 3. Call LLM
    response = llm.invoke([system, human])
    answer = response.content

    # 4. Store if the query carries new personal context
    if should_store(user_query, cfg.memory_trigger_phrases):
        store_memory(memory, user_query, user_id)

    return answer


# ── Interactive REPL ──────────────────────────────────────────
def run_repl(user_id: str = "dev_user") -> None:
    """
    A simple command-line loop – type 'quit' or 'exit' to stop.
    Special commands:
      /memories   – dump all stored memories
      /clear      – (future) clear memories for this user
    """
    cfg = AgentConfig()
    memory = build_memory(cfg)
    llm = build_llm(cfg)

    print("\nDeveloper Assistant  (type 'exit' to quit)\n")

    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nGoodbye!")
            break

        if not user_input:
            continue

        if user_input.lower() in {"exit", "quit"}:
            print("Goodbye!")
            break

        if user_input == "/memories":
            mems = retrieve_memories(memory, "", user_id, top_k=20)
            if mems:
                print("\nStored memories:")
                for i, m in enumerate(mems, 1):
                    print(f"  {i}. {m}")
            else:
                print("  (no memories stored yet)")
            print()
            continue

        answer = chat(llm, memory, user_input, user_id, cfg)
        print(f"\nAssistant: {answer}\n")


# ── Demo (non-interactive) ────────────────────────────────────
def run_demo() -> None:
    """
    Scripted demo that shows the memory round-trip without
    needing a terminal.  Useful for notebooks or CI checks.
    """
    cfg = AgentConfig()
    memory = build_memory(cfg)
    llm = build_llm(cfg)

    USER = "demo_user"

    turns = [
        "My name is Naveen and I am building a multimodal RAG pipeline using Groq and LangChain.",
        "I recently switched from OpenAI embeddings to HuggingFace sentence-transformers.",
        "My team prefers async FastAPI endpoints for all our inference services.",
        # Recall queries
        "What AI project am I working on?",
        "Which embedding provider did I switch to and why?",
        "What's my team's preferred API framework?",
    ]

    for query in turns:
        print(f"\n{query}")
        response = chat(llm, memory, query, USER, cfg)
        print(f"{response}")


# ── Entry point ───────────────────────────────────────────────
if __name__ == "__main__":
    import sys

    # Set keys
    if not os.getenv("GROQ_API_KEY"):
        os.environ["GROQ_API_KEY"] = getpass("Enter GROQ API Key: ")
    # Mem0 needs OPENAI_API_KEY present even when using local embeddings
    os.environ.setdefault("OPENAI_API_KEY", "sk-dummy")

    mode = sys.argv[1] if len(sys.argv) > 1 else "repl"

    if mode == "demo":
        run_demo()
    else:
        run_repl()

Conclusion

Operational checklist