Notas prácticas: Laboratorio: Langgraph: Integración de Qdrant para memoria agente semántica.
Guía paso a paso operativa de las notas prácticas: Laboratorio: Langgraph: Integración de Qdrant para la memoria agente semántica; contratos, verificaciones y espacios de código listos para uso destinados a los equipos que implementan este patrón.
Úselo como una versión reestructurada dirigida a operadores de las ideas presentadas en “Lab:Langgraph: Integrating Qdrant for Semantic Agentic Memory.”: etapas claras, espacios ordenados para el código y notas de recuperación que perduran tras la transferencia de tareas. La etapa de Resumen funciona mejor si se considera como una superficie medible. Capture una transcripción ejemplar, un caso de fallo y la nota de reversión antes de ampliar el alcance. Registre los tiempos y el costo en tokens o consultas junto con los resultados funcionales. La visibilidad temprana del costo evita facturas inesperadas cuando el proceso pasa de la demostración a entornos compartidos.
¿Por qué es necesaria la memoria semántica?
Para que la memoria semántica de tipo “por qué” funcione como etapa, defina las entradas, el responsable de dicha etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben encontrarse en un lugar donde los operadores puedan auditarlos sin necesidad de leer todo el sistema. Coloque la aprobación humana en aquellos procesos que involucren gastos o modifiquen datos de producción. La conexión en tiempo de compilación no equivale a la completitud del proceso empresarial.
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
Conceptos clave aprendidos:
En la etapa de Conceptos Clave Aprendidos, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Incluya la aprobación humana en aquellos casos que impliquen gastos o cambios en los datos de producción. La configuración en tiempo de compilación no equivale a la completitud del producto desde el punto de vista empresarial.
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()
Lista de verificación operativa
Al trabajar en la etapa de Lista de Verificación Operativa, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista mantiene honestas las futuras modificaciones del código.
Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas.
Punto de control después de pasos costosos. La reanudación no debe volver a facturar la misma llamada al LLM cuando un operador intenta nuevamente un nodo posterior.
Fije las versiones de las dependencias y registre el resumen de la imagen que ejecutó la demostración. La reproducibilidad es mejor que el conocimiento tribal.
Registre los tiempos y el costo en tokens o consultas junto con los resultados funcionales. La visibilidad temprana de los costos evita facturas inesperadas cuando el proceso pasa de la demostración a entornos compartidos.
Punto de control después de pasos costosos. La reanudación no debe volver a facturar la misma llamada al LLM cuando un operador intenta nuevamente un nodo posterior.
Antes de promocionar la solución, congele las versiones, capture una transcripción de referencia para el camino crítico y confirme los pasos de reversión. Los entornos compartidos requieren límites de velocidad, verificaciones de tenencia y un responsable claro para la rotación de credenciales secretas. Prefiera una fiabilidad sencilla a demostraciones ingeniosas pero puntuales.
Nota por lotes para e7110284d0c4: mantenga las claves del proveedor fuera del repositorio, establezca un límite máximo para tokens por sesión y almacene las transcripciones junto a los archivos de prueba para que los cambios posteriores en el modelo sigan siendo comparables.
Para la nota de fortalecimiento de la etapa 0, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los archivos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas.
Detalle de refuerzo 0/759: mida el tiempo de ejecución, la clase de error y el consumo de tokens para esta nota, y luego decida si mantener el cambio basándose en un conjunto fijo de preguntas en lugar de en anécdotas.
Al trabajar en la primera etapa de la nota de refuerzo, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el grafo.
Detalle de refuerzo 1/759: mida el tiempo de ejecución, la clase de error y el consumo de tokens para esta nota, y luego decida si mantener el cambio basándose en un conjunto fijo de preguntas en lugar de en anécdotas.
La fase 2 de las notas de fortalecimiento funciona mejor cuando se trata como una superficie medible. Capture una transcripción ejemplar, un caso de fallo y la nota de reversión antes de ampliar el alcance. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el fallo debe apuntar a una única responsabilidad y no a un proceso complicado.
Detalle de fortalecimiento 2/759: mida el tiempo de ejecución, la clase del error y el consumo de tokens para esta nota, y luego decida si mantener el cambio basándose en un conjunto fijo de preguntas en lugar de anécdotas.