Erstellen Sie einen KI-Agenten mit echter Speicherfunktion mithilfe von Mem0, LangChain und Groq

Die folgenden Notizen skizzieren einen praktischen Weg zu „Build an AI Agent with Real Memory Using Mem0, LangChain & Groq“. Der Schwerpunkt liegt auf Verträgen, Überprüfungen und Code-Platzhaltern statt auf motivierenden Formulierungen.

Warum LLMs vergessen – und warum das wichtig ist

Während der Phase „Warum LLMs vergessen“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikatoren sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Fehlerbehebungsprozess gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Cachen Sie stabile Systemanweisungen sowie Tool-Schemata. Das erneute Senden identischer Vorlagen ist eine häufige Ursache für Ressourcenverschwendung.

Was wir bauen

Während der Phase „Was wir entwickeln“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Legen Sie nach teuren Schritten Kontrollpunkte an. Das System sollte bei einer Neuprobe eines späteren Knotens nicht denselben LLM-Aufruf erneut berechnen.

Tech-Stack im Überblick

Wenn Sie in einer bestimmten Phase den Technologiestack durchgehen, notieren Sie zunächst den Vertrag: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgsprüfungen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Erstellen Sie Kontrollpunkte nach kostspieligen Schritten. Das Resume sollte keine doppelten Abrechnungen für denselben LLM-Aufruf tätigen, wenn ein Operator einen späteren Knoten erneut ausführt.

Mem0 – die Speicherschicht

Beim Arbeiten an der Mem0-Memory-Layer-Ebene sollte man zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo-Umgebung in gemeinsam genutzte Umgebungen wechselt. Erstellen Sie nach aufwändigen Schritten einen Checkpoint. Das Wiederaufnehmen des Vorgangs sollte keine erneuten Gebühren für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt.

Groq – Inferenz

Beim Arbeiten an der Groq-Inferenzphase sollte man zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Operator ohne das Lesen des gesamten Graphen überprüfen können. Erstellen Sie Checkpoints nach aufwändigen Schritten. Das Wiederaufnehmen des Vorgangs sollte keine doppelte Abrechnung für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt. Beim Arbeiten an der Groq-Inferenzphase sollte man zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortungsbereich verweisen und nicht auf einen verworrenen Ablauf.

LangChain – Formatierung von Nachrichten

Die Nachrichtenformatierungsphase von LangChain funktioniert am besten, wenn sie als messbarer Prozess betrachtet wird. Erfassen Sie ein perfektes Beispiel, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Ergebnisse ab. Halten Sie den Graphenzustand strukturiert und typisiert. Verschachtelte Datenstrukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Problemen bei der Wiederaufnahme nach Unterbrechungen.

ChromaDB – Vektordatenspeicher

Die Vektorlagerebene von ChromaDB funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie ein optimales Ergebnis, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Erfassen Sie außerdem die Laufzeiten sowie die Kosten für Token oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Einsatzbereich von einer Demo in gemeinsame Umgebungen wechselt. Halten Sie den Graphenzustand flach und typisiert – eingebettete Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.

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

Die Embedding-Ebene von sentence-transformers all-MiniLM-L6-v2 funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zum Rollback, bevor Sie den Umfang erweitern. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreuer ohne Durchsicht des gesamten Systems prüfen können. Trennen Sie die Chunking-Strategie von der Abrufstrategie. Ein Änderung in einer sollte nicht zwangsläufig zu einem Neuschreiben der anderen führen, wenn sich die Qualitätsmetriken ändern. Die Embedding-Ebene von sentence-transformers all-MiniLM-L6-v2 funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zum Rollback, bevor Sie den Umfang erweitern. Ziehen Sie kleine, testbare Einheiten vor großen, komplexen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortungsbereich verweisen und nicht auf einen verworrenen Ablauf.

Code – Schritt für Schritt

Zur Phase des Code-Walk-Throughs sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Setzen Sie menschliche Freigabe bei Schritten ein, die Geld kosten oder Produktionsdaten ändern. Kompilierzeitliche Verbindungen entsprechen nicht der Geschäftskomplettheit.

1. Zentrale Konfiguration

Zur Phase der zentralen Konfiguration sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Erfassen Sie die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Weg von einer Demo-Umgebung in gemeinsam genutzte Umgebungen wechselt. Setzen Sie menschliche Freigabe für Schritte ein, die Geld kosten oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung.

@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. Hilfsfunktionen für das Speichern

Für die Phase „2 Memory Helpers“ sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor der Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckten Zuständen schließen zu müssen. Die Konfiguration sollte außerhalb des Anwendungscode gespeichert werden. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort zusammengefasst sein, den die Operator überprüfen können, ohne den gesamten Ablauf durchzulesen. Menschliche Freigabe sollte für Schritte erforderlich sein, die Geld ausgeben oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung. Für die Phase „2 Memory Helpers“ sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor der Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckten Zuständen schließen zu müssen. Man sollte kleine, testbare Einheiten vor umfangreichen Skripten bevorzugen. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf.

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. Intelligentere Speichtrigger

Während der Arbeit an der Stufe „Intelligenterere Speichtrigger“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Stufe als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Erstellen Sie Kontrollpunkte nach aufwändigen Schritten. Das Wiederaufnehmen des Vorgangs sollte keine erneute Abrechnung für denselben LLM-Aufruf veranlassen, wenn ein Operator einen späteren Knoten erneut versucht.

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

4. Grundierter Chat-Aufruf

Beim Bearbeiten der 4 Phasen des Grounded Chat Prompt schreiben Sie zunächst den Vertrag auf: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Einsatzbereich von einer Demo in gemeinsame Umgebungen wechselt. Speichern Sie stabile Systemanweisungen sowie Tool-Schemata im Cache – das erneute Senden identischer Vorlagen ist eine häufige Ursache für Ressourcenverbrauch.

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. Doppelte Modi: REPL & Demo

Beim Durchlaufen der 5 Phasen des Dual Modes REPL sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Operator ohne das Durchlesen des gesamten Systems überprüfen können. Erstellen Sie nach aufwändigen Schritten einen Checkpoint. Das Wiederaufnehmen des Vorgangs sollte keine doppelte Abrechnung für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt. Beim Durchlaufen der 5 Phasen des Dual Modes REPL sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortungsbereich verweisen und nicht auf ein verworrenes Ablaufschema.

# Interactive session
python memory_agent.py

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

Einrichtung in unter 5 Minuten

Die Phase „Setup in Under 5“ funktioniert am besten, wenn sie als messbare Ebene betrachtet wird. Erfassen Sie vor Erweiterung des Umfangs ein erfolgreiches Beispiel, einen Fehlerfall sowie die Notizen zur Rücksetzung. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Ergebnisdokumente, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab.

Abhängigkeiten installieren

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

Kostenlose Groq-API-Schlüssel erhalten

Ausführen

export GROQ_API_KEY=gsk_...
python memory_agent.py

Wie die Demo aussieht

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.

Vollständiger 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()

Fazit

Operative Kontrollliste