Mémoire d’agent sémantique dans LangGraph avec Qdrant et récupération entre threads

Avant de construire quoi que ce soit, il est utile de bien comprendre pourquoi un agent a besoin de mémoire sémantique et en quoi elle diffère de la mémoire épisodique.

Trois raisons se détachent :

1. Persistence entre sessions. Les outils de sauvegarde standard de LangGraph lient l’état à un thread_id. Lorsqu’une session se termine ou qu’une nouvelle conversation commence, le contexte est réinitialisé à zéro.
2. Récupération sub-linéaire. Une demi-année de journaux de conversation bruts ne peut pas tenir dans une fenêtre de contexte. La latence augmente fortement, la précision diminue en raison de l’effet « perdu au milieu », et les limites de tokens sont atteintes.
3. La correspondance d’intention est plus fiable que les mots-clés exacts. Les requêtes relationnelles échouent lorsque l’utilisateur reformule un concept, tandis que la similarité vectorielle comble cette lacune conceptuelle.

Qdrant peut être déployé localement avec Docker Desktop, en tant que service géré sur Qdrant Cloud, ou en production sur Kubernetes. La méthode locale est utilisée ci-dessous.

Configuration de l’environnement

Commencez par un environnement virtuel Python. Les dépendances sont indiquées dans 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 méthode standard pour créer l’environnement sur Ubuntu ou Debian consiste à utiliser le module 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

La même tâche effectuée avec uv, l’installateur rapide de paquets et gestionnaire de projets d’Astral, se présente comme suit :

#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>

Exécuter Qdrant avec Docker Compose

Pour un test local, un docker-compose.yaml lance la base de données :

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 décrit une application multi-conteneurs dans un seul fichier. Au lieu de taper de longues commandes docker run avec de nombreux paramètres, les conteneurs, les réseaux, les variables d’environnement et les volumes sont tous déclarés dans docker-compose.yml. Ici, il lance l’image officielle de Qdrant :

image: qdrant/qdrant:latest

Apportez-le et vérifiez le conteneur :

$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

Le stockage de vecteurs

Lorsque le conteneur est en cours d’exécution et que sa vérification de santé est réussie, le fichier suivant est vector_store.py. Sa classe Python établit une connexion client vers une instance Qdrant locale ou distante et s’assure qu’une collection nommée agent_memories existe avec la géométrie vectorielle appropriée. Les vecteurs ont 1536 dimensions, ce qui correspond à la taille native des modèles d’OpenAI text-embedding-3-small et text-embedding-ada-002, et la distance est mesurée à l’aide de la similarité cosinus.

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 similarité cosinus prend le cosinus de l’angle entre deux vecteurs. Comme elle est divisée par la longueur du vecteur, les différences de longueur disparaissent, de sorte que seule la direction compte. Cela mesure l’alignement conceptuel et non le nombre de mots ou la longueur de la phrase. L’image compare les mesures de similarité et indique dans quel cas chacune convient le mieux :

Un fichier .env contient la clé OpenAI ainsi que les paramètres de connexion à 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

Le graphique

graph.py définit trois nœuds et compile le flux de travail.

recall_node récupère les mémoires et les injecte. Il incorpore la dernière requête de l’utilisateur, traitée avec text-embedding-3-small, dans un vecteur de 1536 dimensions, cherche dans la collection agent_memories les trois faits les plus pertinents, puis insère une directive système les contenant à l’index 0 de state["messages"].

agent_node raisonne et répond. Il appelle gpt-4o de manière asynchrone en utilisant l’historique mis à jour ainsi que la mémoire injectée, puis ajoute la réponse du modèle à state["messages"].

memorize_node filtre et enregistre les données. Il parcourt à nouveau state["messages"] pour ne retenir que la dernière déclaration brute de l’utilisateur, en ignorant les instructions système et les réponses de l’IA. Il crée ensuite un vecteur pour cette entrée et insère un nouveau PointStruct dans Qdrant, accompagné d’un UUID unique, du texte, du thread_id ainsi que de métadonnées.

Une API pour les tests

main.py met à disposition une interface FastAPI permettant de l’essayer :

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

Exécutez le serveur avec uvicorn main:app --reload, puis ouvrez les documents interactifs à l’adresse /docs sur le port 8000. Tout d’abord, la requête qui indique un fait dans le thread-001 :

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

La réponse :

{
  "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!"
    }
  ]
}

Puis une question dans un autre fil de discussion, fil-002 :

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

Et la réponse, qui rappelle la couleur du fil de discussion précédent :

{
  "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."
    }
  ]
}

Pourquoi le rappel croise les threads

Tous les embeddings se trouvent dans une même collection, agent_memories. Chaque enregistrement stocke un thread_id dans son payload, mais la recherche dans recall_node ne le prend pas en compte pour filtrer :

# 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

Puisque query_points ne reçoit que le nom de la collection et le vecteur de requête, Qdrant compare cette requête avec chaque point de la base de données, quel que soit le thread qui l’a créé. La question du thread 002 est transformée en un vecteur de 1 536 dimensions, et Qdrant mesure son angle de cosinus par rapport à tous les vecteurs stockés, y compris ceux du thread 001. La similarité entre « Quelle est ma couleur préférée ? » et « Ma couleur préférée est le bleu obsidienne » dépasse le seuil de correspondance, ce qui amène Qdrant à retourner ce point, et recall_node injecte le texte dans le nouveau prompt avant que GPT-4o ne s’exécute.

Pour restreindre un agent au thread actuel, ajoutez un Filter sur le champ de charge utile 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

En résumé

Le premier avantage réside dans la mémoire à long terme entre les sessions. La mémoire épisodique à court terme, stockée dans les fichiers de vérification de PostgreSQL ou SQLite, disparaît lorsque thread_id prend fin ou change, tandis que Qdrant conserve les mémoires de manière globale au sein des threads isolés.

Deuxièmement, la recherche suit l’intention. Les embeddings transforment le texte dans un espace de 1 536 dimensions, et la distance cosinus permet de retrouver des mémoires en fonction de leur sens plutôt que de chaînes exactes.

Troisièmement, la portée de la récupération est une option. En omettant les filtres de charge utile, on obtient une récupération globale entre threads. L’ajout d’un models.Filter sur des clés telles que user_id ou tenant_id assure une isolation stricte lorsque c’est nécessaire.

Filtrage multi-locataire

Pour une utilisation multi-locataire, une version mise à jour de graph.py applique le filtre de charge utile à chaque requête :

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

Les avantages sont pratiques. Les utilisateurs ou les organisations ne peuvent récupérer que leurs propres points de mémoire, de sorte que le contexte d’un utilisateur ne s’échappe jamais vers un autre. Une seule collection sert tout le monde, ce qui évite des milliers de collections distinctes qui consommeraient de la RAM et du CPU. De plus, une seule instance Qdrant en cours d’exécution peut servir des milliers d’utilisateurs, au lieu d’un nœud dédié pour chaque client.