Pamięć semantycznych agentów w LangGraph z Qdrant i funkcją przywoływania między wątkami
Przysposób agenta LangGraph do długoterminowej pamięci za pomocą Qdrant: umieść dane użytkowników, odwołuj się do nich przy różnych identyfikatorach wątków oraz dodaj filtry ładunku w celu izolacji wielu użytkowników.
Źródła: **Lab:Langgraph: Integrating Qdrant for Semantic Agentic Memory.**
Zanim cokolwiek zbudujemy, przydatne jest jasne zrozumienie, dlaczego agent w ogóle potrzebuje pamięci semantycznej oraz w jaki sposób różni się ona od pamięci epizodycznej.
Wyróżniają się trzy główne powody:
- Trwałość między sesjami. Standardowe narzędzia do zapisywania stanu w LangGraph powiązują stan z jednym
thread_id. Gdy sesja się kończy lub rozpoczyna nowa rozmowa, kontekst jest resetowany do zera. - Pobieranie danych poniżej linii liniowej. Pół roku surowych zapisów rozmów nie mieści się w oknie kontekstowym. Opóźnienia rosną, dokładność spada z powodu efektu „zagubienia się pośrodku”, a limity tokenów są przekraczane.
- Porównywanie intencji jest skuteczniejsze niż dokładne słowa kluczowe. Zapytania relacyjne zawodzą, gdy użytkownik przeformułuje koncepcję, natomiast podobieństwo wektorowe pomaga przezwyciężyć tę luki koncepcyjną.
Qdrant można zainstalować lokalnie za pomocą Docker Desktop, jako usługę zarządzaną w Qdrant Cloud lub w środowisku produkcyjnym na Kubernetes. Poniżej użyto wariantu lokalnego.
Ustawienie środowiska
Zacznij od wirtualnego środowiska Pythona. Zależności znajdują się w pliku 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
Standardowy sposób tworzenia środowiska na Ubuntu lub Debian wykorzystuje moduł 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
To samo zadanie wykonywane za pomocą uv, szybkiego instalatora pakietów i menedżera projektów Astral, wygląda w ten sposób:
#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>
Uruchamianie Qdrant z Docker Compose
Dla testów lokalnych plik docker-compose.yaml uruchamia bazę danych:
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 opisuje aplikację wielokontenerową w jednym pliku. Zamiast wpisywać długie polecenia docker run z mnóstwem flag, kontenery, sieci, zmiennych środowiskowych i woluminy są wszystkie deklarowane w pliku docker-compose.yml. W tym przykładzie uruchamia oficjalny obraz Qdrant:
image: qdrant/qdrant:latest
Przynieś to i sprawdź pojemnik:
$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
Magazyn wektorów
Gdy kontener jest uruchomiony i jego sprawdzenie stanu przebiega pomyślnie, następnym plikiem jest vector_store.py. Znajdująca się w nim klasa w języku Python nawiązuje połączenie klienta z lokalną lub zdalną instancją Qdrant i upewnia się, że istnieje kolekcja o nazwie agent_memories z odpowiednią geometrią wektorową. Wektory mają 1536 wymiarów, co jest standardową wielkością dla modeli OpenAI text-embedding-3-small i text-embedding-ada-002, a odległość mierzona jest za pomocą podobieństwa kosinowego.
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()
Podobieństwo kosinowe opiera się na wartości kosinusa kąta pomiędzy dwoma wektorami. Ponieważ dzieli się przez długość wektora, różnice w długości znoszą się, więc liczy się tylko kierunek. Mierzy to zgodność koncepcyjną, a nie liczbę słów czy długość zdania. Obrazek porównuje różne wskaźniki podobieństwa oraz określa, w którym przypadku dany wskaźnik najlepiej pasuje:
Plik .env zawiera klucz OpenAI oraz ustawienia połączenia z 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
Graf
graph.py definiuje trzy węzły i kompiluje proces pracy.
recall_node pobiera wspomnienia i wstrzykuje je do systemu. Włącza najnowsze zapytanie użytkownika za pomocą modelu text-embedding-3-small do wektora o wymiarze 1536, wyszukuje w kolekcji agent_memories trzy najbardziej istotne fakty i wstawia instrukcję systemową zawierającą je pod indeksem 0 w state["messages"].
agent_node analizuje sytuację i udziela odpowiedzi. Wywołuje model gpt-4o asynchronicznie, przekazując zaktualizowaną historię oraz wstrzyknięte wspomnienie, a następnie dodaje odpowiedź modelu do state["messages"].
memorize_node filtruje i zapisuje dane. Przechodzi ponownie przez state["messages"], aby znaleźć tylko najnowsze surowe komunikaty użytkownika, pomijając instrukcje systemowe oraz odpowiedzi AI. Następnie tworzy wektor dla tego wprowadzenia i dodaje nowy obiekt PointStruct do Qdrant z unikalnym UUID, tekstem, thread_id oraz metadanymi.
API do testowania
main.py udostępnia punkt końcowy FastAPI do przetestowania tego rozwiązania:
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))
Zainstaluj serwer za pomocą uvicorn main:app --reload, a następnie otwórz interaktywne dokumentacje pod adresem /docs na porcie 8000. Najpierw przeglądamy żądanie przedstawiające fakt w wątku thread-001:
{
"message": "My favorite color is Obsidian Blue.",
"thread_id": "thread-001",
"metadata": {"source": "mobile_app"}
}
Odpowiedź:
{
"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!"
}
]
}
Następnie pytanie w innym wątku, wątek-002:
{
"message": "What is my favorite color?",
"thread_id": "thread-002",
"metadata": {"source": "web_dashboard" }
}
A odpowiedź, która przypomina kolor z poprzedniego wątku:
{
"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."
}
]
}
Dlaczego funkcja recall łączy wątki
Wszystkie embeddingi znajdują się w jednej kolekcji, agent_memories. Każdy rekord przechowuje thread_id w swojej treści, ale wyszukiwanie w recall_node nie filtrowuje na jego podstawie:
# 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
Ponieważ query_points otrzymuje jedynie nazwę kolekcji oraz wektor zapytania, Qdrant porównuje to zapytanie z każdym punktem w bazie danych, niezależnie od tego, który wątek go stworzył. Zapytanie z wątku 002 przekształca się w wektor o wymiarze 1 536, a Qdrant mierzy kąt kosinusowy tego wektora w stosunku do wszystkich przechowywanych wektorów, w tym tych z wątku 001. Podobieństwo między „Jaki jest mój ulubiony kolor?” a „Mój ulubiony kolor to niebieski obsydian” przekracza próg dopasowania, więc Qdrant zwraca ten punkt, a recall_node wstrzykuje tekst do nowego promptu przed uruchomieniem GPT-4o.
Aby ograniczyć agenta do bieżącego wątku, należy dodać Filter do pola zawartości 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
Co to oznacza
Pierwszą zaletą jest pamięć długoterminowa między sesjami. Krótka pamięć epizodyczna, przechowywana w mechanizmach checkpointingu PostgreSQL lub SQLite, znika, gdy thread_id się kończy lub zmienia, podczas gdy Qdrant przechowuje informacje globalnie wśród izolowanych wątków.
Po drugie, wyszukiwanie odbywa się zgodnie z intencją. Embeddingi mapują tekst do przestrzeni o 1 536 wymiarach, a odległość kosinusowa umożliwia odnajdywanie informacji na podstawie znaczenia, a nie dokładnych ciągów znaków.
Po trzecie, zakres odzyskiwania danych jest do wyboru. Pominięcie filtrów treści zapewnia możliwość odzyskiwania danych na poziomie globalnym pomiędzy wątkami. Dodanie models.Filter do kluczy takich jak user_id lub tenant_id zapewnia ścisłą izolację w razie potrzeby.
Filtrowanie w modelu multi-tenant
W przypadku zastosowań multi-tenant zaktualizowany plik graph.py stosuje filtr treści do każdej zapytania:
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()
Korzyści są praktyczne. Użytkownicy lub organizacje mogą uzyskać dostęp tylko do swoich własnych punktów pamięci, dzięki czemu kontekst jednego użytkownika nigdy nie wycieka do innego. Jedna kolekcja służy wszystkim, co pozwala uniknąć tysięcy oddzielnych kolekcji, które zajmowałyby pamięć RAM i zasoby CPU. Ponadto jedna działająca instancja Qdrant może obsługiwać tysiące użytkowników, zamiast dedykowanego węzła dla każdego klienta.
Literatura pokrewna
- Agent finansowy QuickBooks działający na lokalnym LLM z Ollama i LangChain — Agent ReAct na bazie Qwen 3.5 9B i karty RTX 4070 odczytuje dane z QuickBooks w czasie rzeczywistym bez żadnych kosztów chmurowych. Zobacz kod, wyniki oraz znane ograniczenia.
- Koncepcje LangGraph w prostym języku: stan, pamięć i orkiestracja — Przed pisaniem kodu: co oznaczają orkiestracja, czas wykonywania, agenci stanowi, pamięć krótko- i długoterminowa, systemy wieloagentowe oraz udział człowieka w procesie w LangGraph.
- Notatki praktyczne: Rozszyfrowanie Agentic Mesh: skalowanie agentów AI i modeli LLM — Szczegółowy przewodnik po notatkach praktycznych: Rozszyfrowanie Agentic Mesh: skalowanie agentów AI i modeli LLM, w tym informacje o kontraktach, sprawdzaniach oraz gotowych fragmentach kodu dostępnych dla zespołów wdrażających ten model.
- Kompletny lokalny pipeline GraphRAG z użyciem LangExtract, Neo4j, Qdrant i Ollama — Budowa GraphRAG na lokalnych modelach: wyodrębnianie entytetów za pomocą LangExtract, przechowywanie ich w Neo4j, indeksowanie w Qdrant, a następnie rozwijanie podgrafów w celu uzyskania odpowiedzi opartych na grafach.
- Czterowarstwowy projekt pamięci dla agentów AI z informacjami o pochodzeniu danych — Podział pamięci krótkoterminowej i długoterminowej, cztery warstwy (buforowa, epizodyczna, semantyczna, proceduralna), opcje przechowywania, zapisy asynchroniczne oraz metody ochrony przed zatruwaniem pamięci.
- Przepływy bankowe oparte na wielu agentach na Kubernetes: LangGraph, rejestr oraz AgentGateway — Tworzenie specjalistycznych agentów FastAPI, orkiestratora LangGraph z planerem LLM, ich wdrażanie do klastra z rejestratem, a następnie dodawanie routingu i ograniczeń szybkości przy użyciu AgentGateway.
- Ocena agentów AI: zbiory referencyjne, sędziowie LLM oraz bramy regresji CI — Ocena offline i online agentów za pomocą AgentCore, LangFuse, Ragas i DeepEval, wraz z zbiorkami danych referencyjnymi, bramami CI, testami A/B oraz informacjami zwrotnymi z pamięci.
- Praktyczne notatki: Stek rozwojowca agencyjnego, warstwa po warstwie — Szczegółowy przewodnik po Praktycznych notatkach: Stek rozwojowca agencyjnego, warstwa po warstwie: kontrakty, sprawdzania oraz gotowe elementy kodu dostępne dla zespołów wdrażających ten wzorzec.