Crea un agente de IA con memoria real utilizando Mem0, LangChain y Groq

Las notas siguientes reconstruyen un camino práctico para abordar “Crear un agente de IA con memoria real utilizando Mem0, LangChain y Groq”. Se da énfasis en los contratos, las verificaciones y los marcadores de posición para código listo para usar, en lugar de en un enfoque motivacional.

Por qué los LLM olvidan — y por qué eso es importante

Al trabajar en la etapa de “Por qué los LLM olvidan”, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones de código. Documente tanto el camino óptimo como el camino de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no de mejoras posteriores. Almacene en caché las instrucciones estables del sistema y los esquemas de las herramientas. Reenviar un preámbulo idéntico es una causa común de agotamiento.

Qué estamos construyendo

Al trabajar en la etapa de “Qué estamos construyendo”, anote primero el contrato: los datos necesarios, la señal de éxito y qué ocurre en caso de un fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Prefiera unidades pequeñas y probables sobre scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad y no a un proceso complicado. Haga una verificación después de los pasos costosos. El sistema de reanudación no debe volver a facturar la misma llamada al LLM cuando un operador intenta nuevamente un nodo posterior.

Resumen del stack tecnológico

Cuando se trabaja en la pila tecnológica en una etapa determinada, primero escribe el contrato: los datos de entrada requeridos, la señal de éxito y qué ocurre en caso de un fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Considere esta etapa como un contrato entre los datos de entrada y las salidas validadas. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas. Haga una verificación después de los pasos costosos. El sistema de reanudación no debe volver a facturar la misma llamada al LLM cuando un operador intenta nuevamente un nodo posterior.

Mem0 — la capa de memoria

Al trabajar en la etapa de capa de memoria de Mem0, anote primero el contrato: entradas requeridas, señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Registre los tiempos y el costo en tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la versión de demostración a entornos compartidos. Haga un punto de control después de los pasos costosos. La reanudación no debe volver a facturar la misma llamada al LLM cuando un operador intenta nuevamente un nodo posterior.

Groq — inferencia

Al trabajar en la fase de inferencia de Groq, primero escribe el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Mantén la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Haz puntos de control después de los pasos costosos. La reanudación no debe volver a facturar la misma llamada al LLM cuando un operador vuelve a intentar un nodo posterior. Al trabajar en la fase de inferencia de Groq, primero escribe el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Prefiere unidades pequeñas y probables a scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad en lugar de a un proceso complicado.

LangChain — formato de mensajes

La etapa de formato de mensajes de LangChain funciona mejor cuando se trata como una superficie medible. Capture una transcripción ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Trate esta etapa como un contrato entre las entradas y las salidas validadas. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Mantenga el estado del grafo plano y tipado. Los bloques anidados ocultan qué nodo escribió qué campo y provocan interrupciones en la continuación posterior.

ChromaDB — almacén vectorial

La etapa de almacenamiento vectorial de ChromaDB funciona mejor cuando se trata como una superficie medible. Capture una transcripción ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando se pasa de entornos de demostración a entornos compartidos. Mantenga el estado del grafo plano y tipado; los blobs anidados ocultan qué nodo escribió qué campo y provocan interrupciones en la continuación del proceso.

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

La etapa de incrustaciones de sentence-transformers all-MiniLM-L6-v2 funciona mejor cuando se trata como una superficie medible. Capture un transcripto ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Separe la política de particionamiento de la política de recuperación. Cambiar una no debe obligar a reescribir la otra cuando cambian las métricas de calidad. La etapa de incrustaciones de sentence-transformers all-MiniLM-L6-v2 funciona mejor cuando se trata como una superficie medible. Capture un transcripto ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Prefiera unidades pequeñas y probables sobre scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad en lugar de a un proceso complicado.

Código — Desglose paso a paso

En la fase de análisis del código, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Considere esta fase como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Incluya la aprobación humana en aquellos casos que impliquen gastos o cambios en datos de producción. La configuración en tiempo de compilación no equivale a la completitud del proceso empresarial.

1. Configuración central

En la etapa de Configuración Central 1, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo de tokens o consultas junto con los resultados funcionales. La visibilidad temprana del costo evita facturas inesperadas cuando el proceso pasa de entornos de demostración a entornos compartidos. Implemente la aprobación humana en aquellos casos en que se gasten fondos o se modifiquen datos de producción. La conexión en tiempo de compilación no equivale a la completitud del proceso empresarial.

@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. Ayudas de memoria

En la etapa de 2 Memory Helpers, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin necesidad de leer todo el sistema. Coloque la aprobación humana en aquellos procesos que generan gastos o modifican datos de producción. La conexión en tiempo de compilación no equivale a la completitud del proceso empresarial. En la etapa de 2 Memory Helpers, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el error debe indicar una única responsabilidad y no un proceso complicado.

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. Disparador de memoria más inteligente

Al trabajar en la etapa del Disparador de memoria más inteligente, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Trate esta etapa como un contrato entre las entradas y las salidas validadas. Asigne nombres a los elementos generados, defina comprobaciones de éxito y evite completaciones parciales silenciosas. Haga una verificación después de los pasos costosos. La función de reanudación no debe volver a facturar la misma llamada al LLM cuando un operador intenta nuevamente un nodo posterior.

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

4. Prompt de chat fundamentado

Al trabajar en la etapa 4 de los prompts para chats basados en datos reales, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de un fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Registre los tiempos y el costo en tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando se pasa de entornos de demostración a entornos compartidos. Almacene en caché las instrucciones del sistema estables y los esquemas de las herramientas. Reenviar un preámbulo idéntico es una causa común de consumo excesivo.

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. Modos duales: REPL y Demo

Al trabajar en las 5 etapas del REPL de Modos Duales, anote primero el contrato: entradas requeridas, señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Haga un punto de control después de los pasos costosos. La reanudación no debe volver a facturar la misma llamada al LLM cuando un operador intente nuevamente un nodo posterior. Al trabajar en las 5 etapas del REPL de Modos Duales, anote primero el contrato: entradas requeridas, señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Prefiera unidades pequeñas y probables sobre scripts extensos. Cuando un paso falla, el fallo debe apuntar a una sola responsabilidad y no a un proceso complicado.

# Interactive session
python memory_agent.py

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

Configuración en menos de 5 minutos

La etapa “Configuración en menos de 5 minutos” funciona mejor cuando se trata como una superficie medible. Capture un registro exitoso, un caso de fallo y la nota de reversión antes de ampliar el alcance. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace cualquier completación parcial silenciosa.

Instalar dependencias

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

Obtener una clave API gratuita de Groq

Ejecutar

export GROQ_API_KEY=gsk_...
python memory_agent.py

Así es la demostración

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.

Código completo:

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()

Conclusión

Lista de verificación operativa

Lecturas relacionadas

• Notas prácticas: Construir un enrutador multi-agente con LangGraph en 30 minutos — Guía paso a paso de las Notas prácticas: Construir un enrutador multi-agente con LangGraph en 30 minutos: contratos, verificaciones y espacios para código listo para usar para los equipos que implementan este patrón.
• Notas prácticas: Mejorar nuestro agente LangGraph para el comercio electrónico en el mundo real — Guía paso a paso de las Notas prácticas: Mejorar nuestro agente LangGraph para el comercio electrónico en el mundo real: contratos, verificaciones y espacios para código listo para usar para los equipos que implementan este patrón.