Cómo darle memoria semántica a un agente de IA sin base de datos vectorial

Puedes darle a un agente de IA serverless una memoria durable y basada en significado usando búsqueda vectorial nativa de DynamoDB y embeddings de Amazon Bedrock, sin una base de datos vectorial dedicada y sin pipeline de sincronización. En esta demo el embedding vive junto al ítem, un solo PutItem escribe ambos, y buscar por significado es una única llamada a SearchVectors . Medido cara a cara en us-east-1 , esta ruta nativa fue ~2.5x más rápida en escrituras y ~1.4x más rápida en búsquedas (p50) que S3 Vectors para la memoria en la ruta caliente de un agente. 📦 Clona y ⭐ strands-dynamo-vectors Este es el fallo del que trata todo el post. A un agente recién creado le preguntas por un dato que le contaron a un agente anterior. Nunca vio esa conversación, así que no tiene con qué responder - echa mano de una tool, no encuentra nada, y termina pidiéndote que se lo repitas: [session 2] user: What's my production cluster called and its region? Tool #1: retrieve_context Could you please provide the reference or context key that holds the production cluster details? Alternatively, I can try to retrieve this information from other available sources if you direct me. Esa es una ejecución real de demo.py , no un experimento mental. Esto golpea a cualquiera que corra agentes en AWS Lambda, donde el contenedor muere entre invocaciones y cada turno es, en la práctica, un agente nuevo. Lo que vas a aprender: - Las tres capas de memoria: por qué state, session y memoria a largo plazo no son lo mismo. - Cómo construir memoria semántica: enchufar la búsqueda vectorial nativa de DynamoDB al Strands Harness como backend de memoria. - El trade-off: latencia y costo medidos de DynamoDB Vectors vs un store dedicado (S3 Vectors). ¿Por qué un agente "olvida" entre turnos? Piensa en tu agente como un asistente personal al que reemplazan por otra persona distinta cada vez que le hablas. No es una metáfora de Lambda: es literalmente lo que pasa. El contenedor que tenía la última conversación desapareció, así que entra un asistente nuevo sin cuaderno y sin memoria. Para arreglar el "olvido" primero tienes que darte cuenta de que son tres problemas distintos, cada uno con su propio ciclo de vida. Mezclarlos es el bug que hace que un agente "recuerde lo que no debe y olvide lo que sí". | Capa | Qué es | Duración | ¿Va al modelo? | |---|---|---|---| | State | key/value que usan tu app y tus tools | entre requests | No - inyéctalo tú si hace falta | | Session | el historial de una conversación | hasta que dejes de reanudar ese id | Sí - es el contexto | | Memoria a largo plazo | datos durables recuperados por significado | entre ejecuciones sin relación, para siempre | Se inyecta antes de cada turno | En términos del asistente: state es la nota adhesiva sobre el escritorio, session es el cuaderno de la reunión de hoy, y la memoria a largo plazo es que el asistente de verdad recuerde que prefieres desplegar por las mañanas, incluso semanas después y aunque lo digas con otras palabras. El Strands Harness (su constructor de agentes ya ensamblado, create_harness ) expone exactamente dos interruptores para esto: create_harness(session={"id": "..."}) # reanudar ESTA conversación create_harness(memory={"stores": [store]}) # llevar datos entre TODAS las ejecuciones Fíjate en que son ortogonales. Session sobrevive al teardown de Lambda; la memoria sobrevive a todo, incluidas conversaciones que no comparten ninguna palabra. Todo el punto de la búsqueda vectorial de DynamoDB es que el vector vive junto al ítem. Una sola tabla es a la vez tu store operacional y tu índice vectorial, sin un segundo servicio. Contrasta las dos formas. RAG significa Retrieval-Augmented Generation: darle al modelo datos relevantes antes de que responda. | Enfoque | Piezas | Pipeline de sincronización | Lecturas | |---|---|---|---| | Vector DB dedicada | DynamoDB + Streams + Lambda worker + OpenSearch/Pinecone | Sí - Streams → Lambda → índice externo | latencia por consistencia eventual | | DynamoDB Vectors (esta demo) | Solo DynamoDB (ítem + vector) | Ninguno | milisegundos de un dígito en la región | Paso 1: Crear la tabla con un índice vectorial nativo Declara el índice vectorial al crear la tabla, no después. Agregarlo a una tabla que ya tiene ítems dispara un backfill que pagas en tiempo de reloj. Esta es la declaración del índice de setup_table.py : ddb.create_table( TableName=TABLE_NAME, AttributeDefinitions=[ {"AttributeName": "pk", "AttributeType": "S"}, {"AttributeName": "sk", "AttributeType": "S"}, ], KeySchema=[ {"AttributeName": "pk", "KeyType": "HASH"}, {"AttributeName": "sk", "KeyType": "RANGE"}, ], BillingMode="PAY_PER_REQUEST", # on-demand es OBLIGATORIO para índices vectoriales VectorIndexes=[ { "IndexName": INDEX_NAME, "VectorAttribute": {"AttributeName": VECTOR_ATTRIBUTE}, "SearchSchema": [ {"AttributeName": "pk", "SearchSchemaElementType": "HASH"} ], "Projection": {"ProjectionType": "ALL"}, "Dimensions": EMBED_DIMS, # DEBE coincidir con el modelo de embeddings (1024) "DistanceFunction": DISTANCE_FUNCTION, # COSINE } ], ) Por qué importa: dos campos deben coincidir o la búsqueda devuelve silenciosamente nada útil - Dimensions debe igualar la salida de tu modelo de embeddings (Titan V2 = 1024), y DistanceFunction debe coincidir con cómo generaste los vectores (COSINE para embeddings de longitud unitaria). Ejecútalo: python setup_table.py [setup] region=us-east-1 table=strands-agent-memory [setup] create-table sent for 'strands-agent-memory' with vector index 'memory-vector-idx' (1024 dims, COSINE) [setup] waiting for table to become ACTIVE ... [setup] table=ACTIVE index=ACTIVE backfilling=None [setup] confirming the search endpoint serves the index ... [setup] searchable on attempt 1 [setup] DONE. The table is ready for writes and semantic search. Por qué importa: que DescribeTable reporte ACTIVE es necesario pero no suficiente - el endpoint de búsqueda puede ir con retraso. El script demuestra que está listo lanzando una llamada real a SearchVectors en un bucle de reintentos, así que cuando imprime searchable on attempt N , el índice de verdad sirve consultas. Paso 2: Convertir texto en vector con Titan DynamoDB almacena y busca vectores, pero no los genera. Tu código llama a Bedrock. Este es el helper completo de embeddings de embeddings.py : def embed(text: str) -> list[float]: """Devuelve el embedding de Titan V2 para text como lista de floats.""" response = _bedrock.invoke_model( modelId=EMBED_MODEL_ID, # amazon.titan-embed-text-v2:0 contentType="application/json", accept="application/json", body=json.dumps( {"inputText": text, "dimensions": EMBED_DIMS, "normalize": True} ), ) return json.loads(response["body"].read())["embedding"] Por qué importa: normalize=True produce vectores de longitud unitaria, que es lo que espera la búsqueda COSINE . Y como es tu proceso el que llama a Bedrock, cada escritura semántica y cada búsqueda son dos llamadas facturables - una a Titan, una a DynamoDB - que es también por qué el rol de IAM necesita bedrock:InvokeModel , no solo permisos de DynamoDB. Paso 3: Implementar el store de memoria que el Harness entiende El Harness trae un file store cuya búsqueda es solapamiento de tokens por palabra clave: matchea palabras. Nosotros queremos que matchee significado. El store implementa el protocolo strands.memory.MemoryStore ; los dos métodos que importan son add() y search() , de ddb_memory_store.py : async def add(self, content: str, metadata: dict | None = None) -> str: """Genera el embedding del dato y lo guarda como vector buscable en DynamoDB.""" key = f"memories/{uuid.uuid4().hex[:8]}" await self._storage.write( key, content.encode("utf-8"), vector=embed(content), # primero el embedding, luego una escritura guarda ambos metadata=metadata, ) return key async def search(self, query: str, options: dict | None = None) -> list[MemoryEntry]: """Genera el embedding de la query y devuelve las memorias más similares por significado.""" results = await self._storage.search( SearchQuery( vector=embed(query), top_k=self.max_search_results, pk=self._partition, # acota la búsqueda a un solo tenant include_values=False, ) ) # ... mapea los resultados a MemoryEntry(content=..., metadata={"score": r.score}) Por qué importa: son async , tienes que hacerles await . Un write sin await no hace nada silenciosamente y tu tabla queda vacía. Ese await que falta es la trampa que más tiempo me costó (más abajo). Paso 4: Demostrar el recall por significado y luego conectarlo al agente Guarda un dato en una "conversación" y luego haz una pregunta sin relación que no comparta ninguna palabra de contenido. De demo.py : fact = "The user prefers deployments to happen on Tuesday mornings." await store.add(fact, metadata={"kind": "preference"}) await asyncio.sleep(2) # deja que el índice vectorial se asiente (consistencia eventual) query = "when does this user like to ship releases?" results = await store.search(query) [write] stored in some conversation: 'The user prefers deployments to happen on Tuesday mornings.' [search] asked in ANOTHER conversation, by meaning: 'when does this user like to ship releases?' score=0.6665 The user prefers deployments to happen on Tuesday mornings. Por qué importa: la query dice "ship releases", el dato dice "deployments"; la query no dice nada de "Tuesday". Un Query por prefijo nunca encuentra esto. El índice vectorial sí, porque busca significado, no texto - el score de coseno quedó en 0.6665. Ahora enchufa ese mismísimo store a un agente Harness completo como su backend de memoria - esta es la juntura documentada (el seam): store = DynamoDBVectorMemoryStore( TABLE_NAME, region_name=REGION, partition=DEMO_USER, index_name=INDEX_NAME, max_search_results=3, ) agent = create_harness( model=MODEL, session=False, memory={"stores": [store]}, # <-- el seam: el Harness posee la inyección + la tool search_memory ) reply = await agent.invoke_async( "I'm planning next week's release. Is there any timing preference on file for