Memoria de agente semántico en LangGraph con Qdrant y recuperación entre hilos

Antes de construir cualquier cosa, es útil tener claro por qué un agente necesita memoria semántica en primer lugar, y cómo difiere de la memoria episódica.

Se destacan tres razones:

1. Persistencia entre sesiones. Los checkpoints estándar de LangGraph vinculan el estado a un thread_id específico. Cuando termina una sesión o comienza una nueva conversación, el contexto se reinicia a cero.
2. Recuperación sublineal. Medio año de registros brutos de conversaciones no cabe en una ventana de contexto. La latencia aumenta exponencialmente, la precisión se ve afectada por el efecto de información perdida en el proceso, y se agotan los límites de tokens.
3. La coincidencia de intenciones es más efectiva que las palabras clave exactas. Las consultas relacionales fallan cuando el usuario reformula un concepto, mientras que la similitud vectorial supera esa brecha conceptual.

Qdrant puede implementarse localmente con Docker Desktop, como servicio gestionado en Qdrant Cloud, o en entornos de producción con Kubernetes. A continuación se utiliza la opción local.

Configuración del entorno

Comience con un entorno virtual de Python. Las dependencias se encuentran en requirements.txt:

fastapi>=0.110.0
uvicorn>=0.28.0
pydantic>=2.6.0
python-dotenv>=1.0.1
langchain-core>=0.1.30
langchain-openai>=0.1.0
langgraph>=0.0.30
qdrant-client>=1.10.0

La forma estándar de crear el entorno en Ubuntu o Debian utiliza el módulo venv:

#Install venv module (Ubuntu/Debian)
$sudo apt update && sudo apt install python3-venv

#Create a virtual environment
$python3 -m venv myenv

#Activate the environment
$source myenv/bin/activate

#Install saved dependencies
$pip install -r requirements.txt

#Save dependencies
#pip freeze > requirements.txt

#Deactivate when finished:
$deactivate

#Delete the environment:
$rm -rf myenv

El mismo trabajo realizado con uv, el rápido instalador de paquetes y gestor de proyectos de Astral, se ve así:

#Install uv
$curl -LsSf [https://astral.sh/uv/install.sh](https://astral.sh/uv/install.sh) | sh

#Create a virtual environment
$uv venv myenv

#Activate the environment
$source myenv/bin/activate

#Install saved dependencies:
$uv pip install -r requirements.txt

#Save dependencies:
$uv pip freeze > requirements.txt

#Sync dependencies from lockfile
$uv sync

#Add a new package and auto-update file
$uv add <package_name>

Ejecutar Qdrant con Docker Compose

Para pruebas locales, un docker-compose.yaml inicia la base de datos:

version: '3.8'

services:
  qdrant:
    image: qdrant/qdrant:latest    #image name download from docker.io
    container_name: sbi_semantic_memory  #container name
    ports:
      - "6333:6333"   # REST HTTP API & Web Dashboard
      - "6334:6334"   # High-speed gRPC API
    volumes:
      - ./qdrant_storage:/qdrant/storage #Binds local directory to container for data persistence across restarts
    networks:
      - agent-network  #Attaches container to isolated network for communication
    healthcheck:   #Executes internal HTTP ping against health check endpoint
      test: ["CMD", "curl", "-f", "http://localhost:6333/healthz"]
      interval: 10s  #Run health check every 10 seconds
      timeout: 5s    #Fail if check takes longer than 5 seconds
      retries: 5     #Startup grace period before recording failures

networks:    #Private bridged network for secure container-to-container communication
  agent-network:
    driver: bridge   #Standard single-host bridge driver

Docker Compose describe una aplicación multicontenedor en un único archivo. En lugar de escribir comandos largos de docker run con numerosas opciones, los contenedores, redes, variables de entorno y volúmenes se declaran todos en docker-compose.yml. Aquí se inicia la imagen oficial de Qdrant:

image: qdrant/qdrant:latest

Sáquelo y revise el contenedor:

$docker compose up -d

$  docker ps
CONTAINER ID   IMAGE                  COMMAND             CREATED        STATUS                    PORTS                                                             NAMES
8922a12f1029   qdrant/qdrant:latest   "./entrypoint.sh"   45 hours ago   Up 45 hours (unhealthy)   0.0.0.0:6333-6334->6333-6334/tcp, [::]:6333-6334->6333-6334/tcp   sbi_semantic_memory

$ curl http://localhost:6333/healthz
healthz check passed

#View recent logs
$docker logs sbi_semantic_memory

#Tail live logs in real time
$docker logs -f sbi_semantic_memory

#View the last 50 lines of logs
$docker logs --tail 50 sbi_semantic_memory

#Inspect health check failure reasons
$docker inspect --format='{{json .State.Health}}' sbi_semantic_memory

#Check container resource consumption (CPU/RAM):
$docker stats sbi_semantic_memory

#Inspect container runtime details and exit codes:
$docker inspect sbi_semantic_memory

#Execute an interactive shell inside the container
$docker exec -it sbi_semantic_memory sh

#Restart the container:
$docker restart sbi_semantic_memory

#Stop, remove, and recreate the container:
$docker compose down && docker compose up -d

#Force-kill a stuck container:
$docker kill sbi_semantic_memory

El almacén de vectores

Una vez que el contenedor está en ejecución y su verificación de estado es exitosa, el siguiente archivo es vector_store.py. Su clase en Python establece una conexión de cliente con una instancia local o remota de Qdrant y se asegura de que exista una colección llamada agent_memories con la geometría de vectores adecuada. Los vectores tienen 1536 dimensiones, que es el tamaño nativo de los modelos text-embedding-3-small y text-embedding-ada-002 de OpenAI, y la distancia se mide mediante la similitud coseno.

import os
from dotenv import load_dotenv
from qdrant_client import QdrantClient
from qdrant_client.http import models

# Load environment variables
load_dotenv()

class SemanticMemory:
    def __init__(self):
        host = os.getenv("QDRANT_HOST", "localhost")
        port = int(os.getenv("QDRANT_PORT", 6333))

        self.client = QdrantClient(host=host, port=port)
        self.collection_name = "agent_memories"
        self._ensure_collection()

    def _ensure_collection(self):
        collections = self.client.get_collections().collections
        exists = any(c.name == self.collection_name for c in collections)

        if not exists:
            self.client.create_collection(
                collection_name=self.collection_name,
                vectors_config=models.VectorParams(
                    size=1536,
                    distance=models.Distance.COSINE
                ),
            )

memory_vault = SemanticMemory()

La similitud coseno utiliza el coseno del ángulo entre dos vectores. Dado que se divide entre la longitud del vector, las diferencias en la longitud desaparecen, por lo que solo cuenta la dirección. Esto mide la alineación conceptual y no el número de palabras ni la longitud de las oraciones. La imagen compara las medidas de similitud y en qué caso cada una se ajusta mejor:

Un archivo .env contiene la clave de OpenAI y la configuración de conexión de Qdrant:

# OpenAI API Key
OPENAI_API_KEY=

# Qdrant Database Configuration
QDRANT_HOST=localhost
QDRANT_PORT=6333
QDRANT_COLLECTION_NAME=agent_memories

# FastAPI Server Setup
HOST=0.0.0.0
PORT=8000

El gráfico

graph.py define tres nodos y compila el flujo de trabajo.

recall_node recupera los recuerdos e inyecta ellos. Incorpora la última consulta del usuario mediante text-embedding-3-small en un vector de 1536 dimensiones, busca en la colección agent_memories los tres hechos más relevantes e inserta una directiva del sistema que los contiene en la posición 0 de state["messages"].

agent_node razona y responde. Llama de forma asíncrona a gpt-4o con el historial actualizado y el recuerdo inyectado, y luego agrega la respuesta del modelo a state["messages"].

memorize_node filtra y guarda los datos. Recorre state["messages"] para encontrar únicamente la declaración más reciente del usuario en formato raw, omitiendo las instrucciones del sistema y las respuestas de la IA. Luego crea un vector para esa entrada e inserta una nueva PointStruct en Qdrant con un UUID único, el texto, el thread_id y metadatos.

Una API para pruebas

main.py ofrece un endpoint de FastAPI para probarlo:

from typing import Dict, Any, List
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

# Ensure environment variables are loaded at application start
load_dotenv(override=True)

from graph import app_graph

app = FastAPI(title="Qdrant Semantic Memory Service")

class ChatRequest(BaseModel):
    message: str
    thread_id: str
    metadata: Dict[str, Any] = {}

class ChatResponse(BaseModel):
    thread_id: str
    messages: List[Dict[str, Any]]

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(req: ChatRequest):
    try:
        initial_state = {
            "messages": [{"role": "user", "content": req.message}],
            "thread_id": req.thread_id,
            "metadata": req.metadata
        }

        final_state = await app_graph.ainvoke(initial_state)

        return ChatResponse(
            thread_id=req.thread_id,
            messages=final_state["messages"]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Ejecuta el servidor con uvicorn main:app --reload, luego abre las documentaciones interactivas en /docs en el puerto 8000. Primero, la solicitud que indica un hecho en el hilo-001:

{
  "message": "My favorite color is Obsidian Blue.",
  "thread_id": "thread-001",
  "metadata": {"source": "mobile_app"}
}

La respuesta:

{
  "thread_id": "thread-001",
  "messages": [
    {
      "role": "system",
      "content": "You have access to the following long-term memory facts about the user:\n- My favorite color is Obsidian Blue.\n- My favorite color is Obsidian Blue.\n- Obsidian Blue is a deep, rich shade of blue that often resembles the color of the volcanic glass obsidian. It's a striking and elegant color choice! If you have any questions or need assistance related to colors or anything else, feel free to ask!\n\nUse the facts above to directly answer the user's question."
    },
    {
      "role": "user",
      "content": "My favorite color is Obsidian Blue."
    },
    {
      "role": "assistant",
      "content": "That's a great choice! Obsidian Blue is a deep, rich shade of blue that resembles the color of volcanic glass. It's both striking and elegant. If you have any questions or need assistance related to colors or anything else, feel free to ask!"
    }
  ]
}

Luego, una pregunta en otro hilo, hilo-002:

{
  "message": "What is my favorite color?",
  "thread_id": "thread-002",
  "metadata": {"source": "web_dashboard" }
}

Y la respuesta, que recuerda el color del hilo anterior:

{
  "thread_id": "thread-002",
  "messages": [
    {
      "role": "system",
      "content": "You have access to the following long-term memory facts about the user:\n- What is my favorite color?\n- My favorite color is Obsidian Blue.\n- My favorite color is Obsidian Blue.\n\nUse the facts above to directly answer the user's question."
    },
    {
      "role": "user",
      "content": "What is my favorite color?"
    },
    {
      "role": "assistant",
      "content": "Your favorite color is Obsidian Blue."
    }
  ]
}

Por qué el recall cruza hilos

Todos los embeddings se encuentran en una misma colección, agent_memories. Cada registro almacena un thread_id en su carga útil, pero la búsqueda en recall_node no lo filtra:

# 1. Generate query vector from user's message
query_vector = embeddings.embed_query(user_query)

# 2. Search Qdrant WITHOUT a payload filter
relevant_docs = memory_vault.client.query_points(
    collection_name="agent_memories",
    query=query_vector,
    limit=2  # Grabs top 2 matches globally across all threads
).points

Dado que query_points recibe únicamente el nombre de la colección y el vector de consulta, Qdrant compara la consulta con cada punto de la base de datos, independientemente del hilo que lo haya creado. La consulta del hilo 002 se convierte en un vector de 1,536 dimensiones, y Qdrant mide su ángulo de coseno con respecto a todos los vectores almacenados, incluidos los del hilo 001. La similitud entre “¿Cuál es mi color favorito?” y “Mi color favorito es azul óbsidiano” supera el umbral de coincidencia, por lo que Qdrant devuelve ese punto y recall_node inserta el texto en el nuevo prompt antes de que ejecute GPT-4o.

Para restringir a un agente al hilo actual, agregue un Filter al campo de carga thread_id:

from qdrant_client.http import models

# Returns matches ONLY if thread_id matches current thread
relevant_docs = memory_vault.client.query_points(
    collection_name="agent_memories",
    query=query_vector,
    query_filter=models.Filter(
        must=[
            models.FieldCondition(
                key="thread_id",
                match=models.MatchValue(value=state["thread_id"])
            )
        ]
    ),
    limit=2
).points

Qué implica esto

La primera ventaja es la memoria a largo plazo entre sesiones. La memoria episódica a corto plazo, almacenada en los puntos de control de PostgreSQL o SQLite, desaparece cuando finaliza o cambia un thread_id, mientras que Qdrant mantiene las memorias de forma global entre hilos aislados.

En segundo lugar, la búsqueda sigue la intención. Los embeddings mapean el texto a un espacio de 1,536 dimensiones, y la distancia de coseno permite encontrar memorias por su significado en lugar de por cadenas exactas.

En tercer lugar, el alcance de la recuperación es una opción. Excluir los filtros de carga brinda una recuperación global entre hilos. Al agregar un models.Filter en claves como user_id o tenant_id se logra una aislación estricta cuando es necesario.

Filtrado multi-tenant

Para el uso multi-tenant, una versión actualizada de graph.py aplica el filtro de carga en cada consulta:

import os
import uuid
from typing import TypedDict, List, Dict, Any
from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from qdrant_client.http import models
from langgraph.graph import StateGraph, START, END
from app.memory.vector_store import memory_vault

load_dotenv()

class AgentState(TypedDict):
    messages: List[Dict[str, Any]]
    thread_id: str
    tenant_id: str  # Enforces multi-tenancy boundaries
    user_id: str    # Identifies the end-user
    metadata: Dict[str, Any]

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
llm = ChatOpenAI(model="gpt-4o", temperature=0)

async def recall_node(state: AgentState) -> dict:
    """Retrieves semantic memories STRICTLY bounded by tenant_id and user_id."""
    user_query = state["messages"][-1]["content"]
    query_vector = embeddings.embed_query(user_query)

    # Multi-Tenant Payload Filter Enforcement
    tenant_filter = models.Filter(
        must=[
            models.FieldCondition(
                key="tenant_id",
                match=models.MatchValue(value=state["tenant_id"])
            ),
            models.FieldCondition(
                key="user_id",
                match=models.MatchValue(value=state["user_id"])
            )
        ]
    )

    # Scoped Query Execution
    relevant_docs = memory_vault.client.query_points(
        collection_name="agent_memories",
        query=query_vector,
        query_filter=tenant_filter,  # Prevents cross-tenant data leaks
        limit=3
    ).points

    if relevant_docs:
        memory_context = "\n".join([f"- {hit.payload['text']}" for hit in relevant_docs])

        system_instruction = {
            "role": "system",
            "content": (
                "You have access to the following long-term memory facts about the user:\n"
                f"{memory_context}\n\n"
                "Use the facts above to directly answer the user's question."
            )
        }
        state["messages"].insert(0, system_instruction)

    return {"messages": state["messages"]}

async def agent_node(state: AgentState) -> dict:
    """LLM reasoning node."""
    response = await llm.ainvoke(state["messages"])
    state["messages"].append({"role": "assistant", "content": response.content})
    return {"messages": state["messages"]}

async def memorize_node(state: AgentState) -> dict:
    """Embeds user facts along with mandatory multi-tenant metadata payloads."""
    user_message = next(
        (m["content"] for m in reversed(state["messages"]) if m.get("role") == "user"),
        None
    )

    if user_message:
        vector = embeddings.embed_query(user_message)

        memory_vault.client.upsert(
            collection_name="agent_memories",
            points=[
                models.PointStruct(
                    id=str(uuid.uuid4()),
                    vector=vector,
                    payload={
                        "text": user_message,
                        "tenant_id": state["tenant_id"],  # Tenant payload tag
                        "user_id": state["user_id"],      # User payload tag
                        "thread_id": state["thread_id"],
                        "metadata": state.get("metadata", {})
                    }
                )
            ]
        )
    return state

# Graph Workflow Wiring
workflow = StateGraph(AgentState)

workflow.add_node("recall", recall_node)
workflow.add_node("agent", agent_node)
workflow.add_node("memorize", memorize_node)

workflow.add_edge(START, "recall")
workflow.add_edge("recall", "agent")
workflow.add_edge("agent", "memorize")
workflow.add_edge("memorize", END)

app_graph = workflow.compile()

Las ventajas son prácticas. Los usuarios u organizaciones solo pueden recuperar sus propios puntos de memoria, por lo que el contexto de un inquilino nunca se filtra al de otro. Una sola colección sirve a todos, lo que evita miles de colecciones separadas que consumirían RAM y CPU. Además, una única instancia de Qdrant puede servir a miles de inquilinos, en lugar de un nodo dedicado para cada cliente.