Память семантических агентов в 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. Сходство между фразами «Какой у меня любимый цвет?» и «Мой любимый цвет — обсидианово-синий» превышает порог совпадения, поэтому 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 сохраняет данные в глобальном масштабе между изолированными потоками.
Во-вторых, поиск осуществляется с учётом намерения пользователя. Эмбеддинги преобразуют текст в 1536-мерное пространство, а косинусовое расстояние позволяет находить соответствия по смыслу, а не по точным строкам.
В-третьих, объём восстановления данных можно настраивать по желанию. Игнорирование фильтров позволяет осуществлять поиск во всех потоках, тогда как добавление 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 может обслуживать тысячи пользователей вместо того, чтобы для каждого клиента требовался отдельный узел.
Связанные статьи
- Агент для QuickBooks, работающий на локальной нейросети с помощью Ollama и LangChain — агент ReAct на базе Qwen 3.5 9B и RTX 4070 считывает данные QuickBooks в реальном времени без каких-либо затрат на облачные сервисы. Представлен код, результаты работы и известные ограничения.