Практичні зауваження: Лабораторна робота: Langgraph – інтеграція Qdrant для семантичної агентної пам’яті.
Покрокова інструкція з практичних нотаток: Лабораторія: Langgraph – інтеграція Qdrant для семантичної агентської пам’яті: контракти, перевірки та готові блоки коду для команд, які впроваджують цю схему.
Використовуйте цей документ як оновлену версію ідей з “Lab:Langgraph: Integrating Qdrant for Semantic Agentic Memory.” для співробітників-операторів: чіткі етапи, впорядковані блоки коду та примітки щодо відновлення, які зберігаються під час передачі обов’язків. Етап Огляду найкраще функціонує, якщо його розглядати як вимірювану поверхню. Запишіть один ідеальний запис, один випадок збою та примітки щодо скасування змін перед розширенням обсягу роботи. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-версії до спільних середовищ.
Чому необхідна семантична пам’ять?
Щоб семантична пам’ять була етапом процесу, необхідно визначити вхідні дані, виконавця кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись визначити прихований стан. Конфігурацію слід тримати окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функціоналу мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь алгоритм. Необхідно передбачити людське схвалення для дій, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення на етапі компіляції не є гарантією повноти бізнес-функціоналу.
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
#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
#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>
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
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
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()
# 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
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))
{
"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!"
}
]
}
{
"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."
}
]
}
# 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
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
Ключові концепції, які були опановані:
На етапі вивчення ключових концепцій необхідно спочатку визначити вхідні дані, відповідальну особу за крок та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Необхідно документувати як шлях успішного виконання, так і шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої доробки. Встановлюйте людське схвалення для тих кроків, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення на етапі компіляції не є гарантією повності функціоналу продукту.
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()
Чек-лист для експлуатації
Під час роботи за чек-листом для експлуатації спочатку запишіть умови виконання: необхідні вхідні дані, сигнал про успіх та наслідки часткової невдачі. Цей чек-лист допомагає зберігати чесність у подальших змінах коду.
Розглядайте цю стадію як контракт між вхідними даними та перевіреними результатами. Призначте назви елементам, визначте критерії успіху та не допускайте мовчазного часткового виконання.
Створюйте контрольні точки після дорогих кроків. Функція відновлення не повинна знову стягувати плату за той самий виклик LLM, коли оператор намагається знову виконати пізнішу операцію.
Фіксуйте версії залежностей та записуйте хеш-значення зображення, яке використовувалося під час демонстрації. Відтворюваність краща за індивідуальні знання.
Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Чітке бачення витрат заздалегідь запобігає несподіваним рахункам під час переходу від демонстрації до спільних середовищ.
Створюйте контрольні точки після дорогих кроків. Функція відновлення не повинна знову стягувати плату за той самий виклик LLM, коли оператор намагається знову виконати пізнішу операцію.
Перш ніж запускати стек у продакшн, заморозьте версії, створіть «золотий» запис для критичного шляху та підтвердьте кроки відкату. У спільних середовищах необхідні обмеження на частоту використання, перевірки прав доступу та чіткий власник для зміни секретів. Віддавайте перевагу надійності перед креативними одноразовими демонстраціями.
Примітка до пакету e7110284d0c4: не включайте ключі постачальника до репозиторію, встановіть ліміт токенів на сеанс та зберігайте записи поруч із фікстурами для оцінки, щоб подальша заміна моделей залишалася порівнянною.
Для етапу 0 щодо посилення безпеки визначте вхідні дані, власника кроку та критерії завершення ще до змін у коді. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не здогадуючись про прихований стан. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте артефакти, визначте критерії успіху та не допускайте беззвучного часткового завершення.
Деталь посилення безпеки 0/759: виміряйте час виконання, клас помилки та кількість витрачених токенів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих випадках.
Під час роботи над першим етапом запису про посилення безпеки спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається при частковій невдачі. Цей перелік допомагає зберігати чесність у подальших змінах коду. Зберігайте конфігурацію окремо від коду додатку. Файли середовища, сховища секретів та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код.
Деталь посилення безпеки 1/759: виміряйте час виконання, клас помилки та кількість витрачених токенів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих випадках.
Етап 2 процедури зміцнення найкраще функціонує, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний зразок виконання, один випадок збою та запис про скасування змін перед розширенням обсягу роботи. Віддавайте перевагу невеликим, тестованим одиницям перед складними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність дій.
Деталь 2/759 процедури зміцнення: виміряйте час виконання, клас помилки та кількість витрачених ресурсів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих прикладах.