Семантычная памяць агента ў LangGraph з Qdrant і ўзгадваннем паміж трэдамі
Даём агенту LangGraph доўгатэрміновую памяць на Qdrant: эмбедынгі фактаў пра карыстальніка, узгадванне паміж ідэнтыфікатарамі трэдаў і фільтры payload для ізаляцыі кліентаў.
Крыніцы: **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"], ўпынкам шукаючы толькі найсвежэйшы необработаны запіс ад карыстальніка, пры чым ігнаруе інструкцыі сістэмы і адпаведзі AI. Пасля гэтага ён стварае вектар для гэтаго вхіднага данага і дадзеўляе новы элемент 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. Першым будзе запіт у гілцы-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."
}
]
}
Чаму праблема асоціяцыі нітак
Усе эмбеддынгі знаходзяцца ў адной колекцыі, 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()
Прыямыя перавагі — у практычнасці. Карыстальнікі або арганізацыі можаць атрымліваць толькі свае сабе пункты памяці, таму кантэкст аднаго карыстальніка ніколі не прасачыцца да іншага. Адна колекцыя служыць усім, чым утрымліваецца патрэба ў тыячах адзінаковых колекцыях, якія бы займалі RAM і CPU. Адна працюючая інстанцыя Qdrant можа служыць тыячам карыстальнікаў, замест таго, каб для кожнага кліента быў адзін спецыяльны вузел.
Спаднёе чытанне
- Агент QuickBooks Finance, які працуе на локальным LLM з адной Ollama і LangChain — Агент ReAct на Qwen 3.5 9B і RTX 4070 чытае даны QuickBooks у рэальны час без жадных вітатаў у хмару. Можна пазначыць код, рэзультаты і вядомыя меры.