Semantische Agentenmemorie in LangGraph mit Qdrant und cross-Thread-Aufruf
Verleihen Sie einem LangGraph-Agenten mit Qdrant eine Langzeitmerkmung: Embedden Sie Benutzerdaten, rufen Sie sie über Thread-IDs ab und fügen Sie Payload-Filter hinzu, um eine Isolierung für mehrere Nutzer zu gewährleisten.
Quellen: **Lab:Langgraph: Integrating Qdrant for Semantic Agentic Memory.**
Bevor man irgendetwas entwickelt, ist es hilfreich, klarzustellen, warum ein Agent überhaupt semantisches Gedächtnis benötigt und wie es sich von episodem Gedächtnis unterscheidet.
Drei Gründe sind besonders hervorzuheben:
- Über-Sitz-Persistenz. Herkömmliche LangGraph-Checkpointer verknüpfen den Zustand mit einer bestimmten
thread_id. Wenn eine Sitzung endet oder ein neues Gespräch beginnt, wird der Kontext auf Null zurückgesetzt. - Sublineare Abrufgeschwindigkeit. Ein halbes Jahr an Rohdaten aus Gesprächsprotokollen passt nicht in ein Kontextfenster. Die Latenz steigt stark an, die Genauigkeit leidet unter dem sogenannten „lost-in-the-middle“-Effekt, und die Token-Limits werden überschritten.
- Intent-Abgleich ist effektiver als exakte Schlüsselwörter. Relationale Abfragen scheitern, wenn ein Benutzer ein Konzept umformuliert, während vektorbasierte Ähnlichkeitsbetrachtungen diese konzeptionellen Lücken überbrücken können.
Qdrant kann lokal mit Docker Desktop eingesetzt werden, als verwalteter Dienst auf Qdrant Cloud oder in der Produktion auf Kubernetes. Im Folgenden wird der lokale Einsatzweg verwendet.
Umgebungs-Einrichtung
Beginnen Sie mit einer Python-Virtuellen Umgebung. Die Abhängigkeiten werden in requirements.txt aufgeführt:
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
Die übliche Methode zur Erstellung der Umgebung unter Ubuntu oder Debian verwendet das venv-Modul:
#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
Dieselbe Aufgabe mit uv, Astrals schnellem Paketinstaller und Projektmanager, sieht so aus:
#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>
Qdrant mit Docker Compose ausführen
Zum lokalen Test startet eine docker-compose.yaml-Datei die Datenbank:
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 beschreibt eine mehrkontainerbasierte Anwendung in einer einzigen Datei. Anstatt lange docker run-Befehle mit zahlreichen Flags einzugeben, werden Container, Netzwerke, Umgebungsvariablen und Volumes alle in docker-compose.yml deklariert. Hier wird das offizielle Qdrant-Bild gestartet:
image: qdrant/qdrant:latest
Nehmen Sie es hoch und überprüfen Sie den 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
Der Vektor-Speicher
Sobald der Container läuft und der Health-Check erfolgreich ist, folgt die Datei vector_store.py. Die darin enthaltene Python-Klasse öffnet eine Client-Verbindung zu einer lokalen oder entfernten Qdrant-Instanz und stellt sicher, dass eine Sammlung mit dem Namen agent_memories und der richtigen Vektorgeometrie vorhanden ist. Vektoren haben 1536 Dimensionen, was die native Größe für OpenAI’s text-embedding-3-small und text-embedding-ada-002 ist, und die Entfernung wird mithilfe der Kosinusähnlichkeit gemessen.
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()
Die Kosinusähnlichkeit berechnet den Kosinus des Winkels zwischen zwei Vektoren. Da sie durch die Länge des Vektors geteilt wird, werden Längendifferenzen ausgeglichen, sodass nur die Richtung relevant ist. Dadurch wird eine konzeptionelle Übereinstimmung gemessen und nicht die Anzahl der Wörter oder die Länge der Sätze. Das Bild vergleicht die Ähnlichkeitsmaße sowie die Situationen, in denen jeweils das eine am besten passt:
In einer .env-Datei sind der OpenAI-Schlüssel sowie die Verbindungs-Einstellungen für Qdrant gespeichert:
# 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
Das Diagramm
graph.py definiert drei Knoten und kompiliert den Workflow.
Der recall_node holt Erinnerungen ab und injiziert sie. Er embeddet die neueste Anfrage des Benutzers mithilfe von text-embedding-3-small in einen 1536-dimensionalen Vektor, sucht in der Sammlung agent_memories nach den drei relevantesten Fakten und fügt eine Systemanweisung, die diese enthält, an Index 0 von state["messages"] ein.
Der agent_node führt Schlussfolgerungen durch und gibt eine Antwort. Er ruft gpt-4o asynchron mit der aktualisierten Historie sowie den injizierten Erinnerungen auf und fügt anschließend die Antwort des Modells zu state["messages"] hinzu.
memorize_node filtert und speichert. Es durchläuft state[„messages“], um nur die neueste rohe Benutzermitteilung zu finden, wobei Systemanweisungen und KI-Antworten übersprungen werden. Anschließend erstellt es einen Vektor für diese Eingabe und fügt eine neue PointStruct mit einer eindeutigen UUID, dem Text, dem thread_id sowie Metadaten in Qdrant ein.
Eine API zum Testen
main.py stellt einen FastAPI-Endpunkt zur Verfügung, um es auszuprobieren:
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))
Führen Sie den Server mit uvicorn main:app --reload aus und öffnen Sie anschließend die interaktiven Dokumente unter /docs auf Port 8000. Zuerst der Anfrage, die in Thread-001 eine Tatsache angibt:
{
"message": "My favorite color is Obsidian Blue.",
"thread_id": "thread-001",
"metadata": {"source": "mobile_app"}
}
Die Antwort:
{
"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!"
}
]
}
Dann eine Frage in einem anderen Thread, Thread-002:
{
"message": "What is my favorite color?",
"thread_id": "thread-002",
"metadata": {"source": "web_dashboard" }
}
Und die Antwort, die die Farbe aus dem früheren Thread erinnert:
{
"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."
}
]
}
Warum ruft recall Threads ab
Sämtliche Embeddings befinden sich in einer Sammlung, agent_memories. Jeder Eintrag speichert einen thread_id in seinem Payload, doch die Suche in recall_node filtert nicht danach:
# 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
Da query_points nur den Sammelnamen sowie den Abfragevektor erhält, vergleicht Qdrant die Abfrage mit jedem Punkt in der Datenbank – unabhängig davon, welcher Thread ihn erstellt hat. Die Abfrage des Threads 002 wird zu einem 1.536-dimensionalen Vektor, und Qdrant misst seinen Kosinuswinkel zu allen gespeicherten Vektoren, einschließlich denen des Threads 001. Die Ähnlichkeit zwischen „Was ist meine Lieblingsfarbe?“ und „Meine Lieblingsfarbe ist Obsidian Blue“ überschreitet die Übereinstimmungsschwelle, weshalb Qdrant diesen Punkt zurückgibt und recall_node den Text in die neue Anfrage einfügt, bevor GPT-4o ausgeführt wird.
Um einen Agenten auf den aktuellen Thread zu beschränken, fügen Sie einen Filter zum Feld thread_id hinzu:
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
Was das bedeutet
Der erste Vorteil ist das Langzeitgedächtnis über verschiedene Sitzungen hinweg. Das kurzfristige episodische Gedächtnis, das in PostgreSQL- oder SQLite-Checkpointern gespeichert wird, verschwindet, wenn eine thread_id endet oder sich ändert, während Qdrant die Erinnerungen global über isolierte Threads aufbewahrt.
Zweitens folgt die Suche dem Nutzungsziel. Embeddings übertragen Text in einen 1.536-dimensionalen Raum, und die Kosinus-Distanz findet Erinnerungen anhand der Bedeutung statt von exakten Zeichenketten.
Drittens ist der Abrufumfang wählbar. Das Auslassen von Payload-Filtern ermöglicht eine globale Abrufung über verschiedene Threads hinweg. Das Hinzufügen eines models.Filter für Schlüssel wie user_id oder tenant_id sorgt bei Bedarf für eine strenge Isolation.
Filterung für mehrere Mieter
Für den Einsatz mit mehreren Mietern wendet eine aktualisierte Version von graph.py den Payload-Filter auf jede Abfrage an:
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()
Die Vorteile sind praktisch. Benutzer oder Organisationen können nur ihre eigenen Speicherpunkte abrufen, sodass der Kontext eines Mieters niemals an einen anderen durchsickert. Eine einzige Sammlung dient allen, was tausende separater Sammlungen vermeidet, die RAM und CPU verbrauchen würden. Zudem kann eine einzige laufende Qdrant-Instanz Tausende von Mietern bedienen, anstatt für jeden Kunden einen dedizierten Knoten zu benötigen.
Verwandte Artikel
- Ein QuickBooks-Finanzagent, der mit Ollama und LangChain auf einem lokalen LLM läuft — Ein ReAct-Agent auf Qwen 3.5 9B und einer RTX 4070 liest Echtzeit-QuickBooks-Daten ohne jegliche Cloud-Kosten. Sie finden hier den Code, die Ausgabe sowie die bekannten Einschränkungen.