Semantic Agent Memory in LangGraph with Qdrant and Cross-Thread Recall

Before building anything, it helps to be clear about why an agent needs semantic memory at all, and how it differs from episodic memory.

Three reasons stand out:

1. Cross-session persistence. Standard LangGraph checkpointers tie state to one thread_id. When a session ends or a new conversation begins, the context resets to zero.
2. Sub-linear retrieval. Half a year of raw conversation logs can't go into a context window. Latency explodes, accuracy suffers from the lost-in-the-middle effect, and token limits run out.
3. Intent matching beats exact keywords. Relational queries fail when a user rephrases a concept, whereas vector similarity bridges that conceptual gap.

Qdrant can be deployed locally with Docker Desktop, as a managed service on Qdrant Cloud, or in production on Kubernetes. The local route is used below.

Environment setup

Start with a Python virtual environment. The dependencies go in 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

The standard way to create the environment on Ubuntu or Debian uses the venv module:

#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

The same job done with uv, Astral's fast package installer and project manager, looks like this:

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

Running Qdrant with Docker Compose

For a local test, a docker-compose.yaml starts the database:

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 describes a multi-container application in a single file. Rather than typing long docker run commands with piles of flags, containers, networks, environment variables and volumes are all declared in docker-compose.yml. Here it launches the official Qdrant image:

image: qdrant/qdrant:latest

Bring it up and check the container:

$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

The vector store

With the container running and its health check passing, the next file is vector_store.py. Its Python class opens a client connection to a local or remote Qdrant instance and makes sure a collection named agent_memories exists with the right vector geometry. Vectors have 1536 dimensions, the native size for OpenAI's text-embedding-3-small and text-embedding-ada-002, and distance is measured with cosine similarity.

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

Cosine similarity takes the cosine of the angle between two vectors. Because it divides by vector length, differences in length drop out, so only direction counts. That measures conceptual alignment and not word count or sentence length. The image compares the similarity measures and when each fits best:

An .env file holds the OpenAI key and the Qdrant connection settings:

# 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

The graph

graph.py defines three nodes and compiles the workflow.

The recall_node retrieves memories and injects them. It embeds the user's latest query with text-embedding-3-small into a 1536-dimensional vector, searches the agent_memories collection for the three most relevant facts, and inserts a system directive containing them at index 0 of state["messages"].

The agent_node reasons and answers. It calls gpt-4o asynchronously with the updated history and the injected memory, then appends the model's reply to state["messages"].

The memorize_node filters and saves. It walks back through state["messages"] to find only the newest raw user statement, skipping system instructions and AI replies. It then creates a vector for that input and upserts a new PointStruct into Qdrant with a unique UUID, the text, the thread_id and metadata.

An API for testing

main.py exposes a FastAPI endpoint for trying it out:

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

Run the server with uvicorn main:app --reload, then open the interactive docs at /docs on port 8000. First, the request that states a fact in thread-001:

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

The response:

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

Then a question in a different thread, thread-002:

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

And the answer, which recalls the color from the earlier thread:

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

Why recall crosses threads

All embeddings live in one collection, agent_memories. Every record stores a thread_id in its payload, but the search in recall_node doesn't filter on it:

# 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

Since query_points receives only the collection name and the query vector, Qdrant compares the query against every point in the database, whichever thread created it. The thread-002 question becomes a 1,536-dimensional vector, and Qdrant measures its cosine angle against all stored vectors, including the ones from thread-001. The similarity between "What is my favorite color?" and "My favorite color is Obsidian Blue" clears the match threshold, so Qdrant returns that point and recall_node injects the text into the new prompt before GPT-4o runs.

To restrict an agent to the current thread, add a Filter on the thread_id payload field:

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

What this adds up to

Long-term memory across sessions is the first gain. Short-term episodic memory, held in PostgreSQL or SQLite checkpointers, disappears when a thread_id ends or changes, while Qdrant keeps memories globally across isolated threads.

Second, search follows intent. Embeddings map text into a 1,536-dimension space, and cosine distance finds memories by meaning instead of exact strings.

Third, retrieval scope is a choice. Leaving out payload filters gives global cross-thread recall. Adding a models.Filter on keys such as user_id or tenant_id enforces strict isolation when needed.

Multi-tenant filtering

For multi-tenant use, an updated graph.py applies the payload filter on every query:

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

The benefits are practical. Users or organizations can only retrieve their own memory points, so one tenant's context never leaks to another. A single collection serves everyone, which avoids thousands of separate collections that would eat RAM and CPU. And one running Qdrant instance can serve thousands of tenants, instead of a dedicated node for each client.