Практычныя прытамулкі: Лабараторная робота: 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: не кладзіце ключы прадаўцоў у репазітарый, задаце верхнюю межу токеноў на сесію і зберагачыце транскрыпты празаўсёды пад фіксы для ацэнкі, каб пазнейшыя замены моделяў заставаліся порównаннімі.
Для прыметкі па забезпечэнню надзяймоўнасці стадіі 0 адначасова задаце вхідныя даны, власніка крока і критэрыя завершэння пры зміне коду. Аперацыйныя системы павінны магчымае перзапуск крока з вядомай точкі контролю без неабясненага вычыслення скрытых станоў. Спрацаввайце гэту стадію як кантракт межа вхіднымі данымі і перакананымі выходнымі рэзультатамі. Даце назвы артыфактам, задаце перакананні на успех і адмовіцеся ад тыхоўскага частковага завершэння.
Дзеянне паўжасткі 0/759: звярніце увагу на час выканання, класы памылак і витрату токенаў для гэтага запісу, а пасля, на аднойчынай базе пытанняў, а не на асобістых спазырах, выявіце, чы хачаце застаўіць змяну.
Калі працуеце над першым этапам запісу паўжасткі, спачатку запішыце контракт: неабходныя данні, сігнал успеху і тое, што выканаецца у разе частковай памылки. Такі список контроля дапамагае заставіць пазнейшыя змяны коду чыстымі. Зберагаеце канфігурацыю праза код аплікацыі. Файлы сераўнавання, хранільнікі секрэтных дадзенняў і флагі функцыйяй должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць без неабходнасці чытання всей структуры.
Дзеянне паўжасткі 1/759: звярніце увагу на час выканання, класы памылак і витрату токенаў для гэтага запісу, а пасля, на аднойчынай базе пытанняў, а не на асобістых спазырах, выявіце, чы хачаце застаўіць змяну.
Этап 2 прыцеленняя на зміцнэнне работае найкраща, калі яго розглядаць як вимероўваную паверхню. Запісаце адна «золатая» транскрыпцыя, адин прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштабы. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выходзіць, прычына неудачы павінна вказываць на адную адпаведальнасць, а не на заплутаны процес.
Дзеянне прыцеленняя на зміцнэнне 2/759: вимеравайце час выканання, класію памылак і колькасць выкорыстоўваных токенав для гэтага запісу, а потым вырашайце, чы рашыцца застаўляць змяну, стварываючыся на адной фіксаванай сэтцы пытанняў, а не на анекдотах.