Семантична пам’ять агента в LangGraph з Qdrant та функція відтворення даних між потоками
Надайте агенту LangGraph довгострокову пам’ять за допомогою Qdrant: вбудуйте дані користувачів, відтворюйте їх за ідентифікаторами потоків та додайте фільтри навантаження для ізоляції кількох орендарів.
Джерела: **Lab:Langgraph: Integrating Qdrant for Semantic Agentic Memory.**
Перш ніж щось будувати, корисно чітко зрозуміти, чому агенту взагалі потрібна семантична пам’ять та в чому її відмінність від епізодичної пам’яті.
Виділяються три причини:
- Стабільність між сеансами. Стандартні інструменти зберігання стану LangGraph прив’язують його до одного
thread_id. Коли сеанс закінчується або починається нова розмова, контекст скидається до нуля. - Нелінійне пошукове функціонування. Півроку необроблених записів розмов не вміщується у вікно контексту. Затримки значно зростають, точність погіршується через ефект „втрати посередині“, а обмеження на кількість токенів досягаються.
- Відповідність наміру краща за точне використання ключових слів. Відносні запити не функціонують, коли користувач перефразовує поняття, тоді як векторна схожість допомагає подолати цю концептуальну прогалину.
Qdrant можна розгорнути локально за допомогою Docker Desktop, як керовану послугу на Qdrant Cloud або у продакшені на Kubernetes. Нижче описано локальний варіант розгортання.
Налаштування середовища
Почніть з віртуального середовища Python. Залежності знаходяться у файлі 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
Стандартний спосіб створення середовища в Ubuntu або Debian передбачає використання модуля 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
Те саме завдання, виконане за допомогою uv — швидкого інсталятора пакетів та менеджера проєктів Astral, виглядає так:
#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 за допомогою Docker Compose
Для локальних тестів файл docker-compose.yaml запускає базу даних:
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 описує багатоконтейнерний додаток у одному файлі. Замість того, щоб вводити довгі команди docker run із великою кількістю параметрів, контейнери, мережі, змінні середовища та томи оголошуються у файлі docker-compose.yml. Тут запускається офіційний образ Qdrant:
image: qdrant/qdrant:latest
Підніміть його та перевірте контейнер:
$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
Магазин векторів
Коли контейнер працює та проходить перевірку стану, наступним файлом є vector_store.py. Його клас на Python відкриває клієнтське з’єднання з локальною або віддаленою інстанцією Qdrant та переконується, що існує колекція під назвою agent_memories із відповідною геометрією векторів. Вектори мають 1536 вимірів, що є стандартним розміром для моделей OpenAI text-embedding-3-small та text-embedding-ada-002, а відстань вимірюється за допомогою косинусної схожості.
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()
Схожість косинуса обчислюється за допомогою косинуса кута між двома векторами. Оскільки при цьому ділиться на довжину вектора, відмінності у довжині не враховуються, тож має значення лише напрямок. Це вимірює концептуальну відповідність, а не кількість слів чи довжину речення. На зображенні порівнюються показники схожості та показано, коли кожен з них найкраще підходить:
Файл .env містить ключ OpenAI та налаштування підключення до 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
Граф
graph.py визначає три вузли та скомпілює робочий процес.
recall_node отримує спогади та вставляє їх. Він включає останнє запитання користувача за допомогою text-embedding-3-small у 1536-вимірний вектор, шукає у колекції agent_memories три найбільш релевантні факти та вставляє системну інструкцію, яка їх містить, у індекс 0 масива state["messages"].
agent_node аналізує ситуацію та дає відповідь. Він асинхронно викликає gpt-4o із оновленою історією та вставленими спогадами, а потім додає відповідь моделі до масива state["messages"].
memorize_node фільтрує та зберігає дані. Він пройшовся по state["messages"], щоб знайти лише найновіше сире повідомлення користувача, ігноруючи системні інструкції та відповіді ШІ. Потім він створює вектор для цього вхідного даних та додає новий об’єкт PointStruct у Qdrant із унікальним UUID, текстом, thread_id та метаданими.
API для тестування
main.py надає кінцеву точку FastAPI для тестування:
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))
Запустіть сервер за допомогою uvicorn main:app --reload, потім відкрийте інтерактивну документацію за адресою /docs на порту 8000. Спочатку — запит, який вказує факт у потоці thread-001:
{
"message": "My favorite color is Obsidian Blue.",
"thread_id": "thread-001",
"metadata": {"source": "mobile_app"}
}
Відповідь:
{
"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!"
}
]
}
Потім запитання в іншій темі, тема-002:
{
"message": "What is my favorite color?",
"thread_id": "thread-002",
"metadata": {"source": "web_dashboard" }
}
І відповідь, яка нагадує колір з попередньої теми:
{
"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."
}
]
}
Чому функція recall об’єднує потоки
Усі ембеддинги знаходяться в одній колекції agent_memories. Кожен запис зберігає thread_id у своєму вмісті, але пошук у recall_node не фільтрує їх за цим параметром:
# 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
Оскільки query_points отримує лише назву колекції та вектор запиту, Qdrant порівнює цей запит із кожною точкою в базі даних, незалежно від того, яка потік її створила. Запит потоку 002 перетворюється на 1 536-вимірний вектор, і Qdrant вимірює кут косинуса між ним та всіма збереженими векторами, включаючи ті, що належать потоку 001. Схожість між фразами „Який мій улюблений колір?“ та „Мій улюблений колір – Obsidian Blue“ перевищує поріг збігу, тому Qdrant повертає цю точку, а recall_node вставляє текст у новий запит перед запуском GPT-4o.
Щоб обмежити дію агента лише поточним потоком, додайте Filter до поля навантаження 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
Що це дає
Першою перевагою є довгострокова пам’ять між сеансами. Короткострокова епізодична пам’ять, яка зберігається у механізмах контролю стану PostgreSQL чи SQLite, зникає, коли закінчується або змінюється thread_id, тоді як Qdrant зберігає пам’ять у глобальному масштабі між ізольованими потоками.
По-друге, пошук відбувається відповідно до наміру. Ембеддинги перетворюють текст у 1 536-вимірний простір, а косинусова відстань дозволяє знаходити записи за значенням змісту, а не за точними рядками.
По-третє, обсяг пошуку є обираємим. Відсутність фільтрів вантажу забезпечує глобальне пошукове відтворення між потоками. Додавання models.Filter для ключів на кшталт user_id чи tenant_id забезпечує сувору ізоляцію за потреби.
Фільтрація для кількох орендарів
Для використання кількома орендарями оновлений graph.py застосовує фільтр вантажу до кожного запиту:
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()
Переваги є практичними. Користувачі чи організації можуть отримувати доступ лише до власних пунктів пам’яті, тому контекст одного користувача ніколи не потрапляє до іншого. Одна колекція обслуговує всіх, що дозволяє уникнути тисяч окремих колекцій, які б споживали оперативну пам’ять та процесор. Крім того, один запущений екземпляр Qdrant може обслуговувати тисячі користувачів замість окремого вузла для кожного клієнта.