Cómo crear agentes de IA listos para la producción con memoria a largo plazo utilizando Google ADK y Milvus

Cuando se construyen agentes inteligentes, uno de los problemas más difíciles es la gestión de la memoria: decidir qué debe recordar y qué debe olvidar el agente.

No toda la memoria está pensada para durar. Algunos datos sólo son necesarios para la conversación en curso y deben borrarse cuando ésta termina. Otros datos, como las preferencias del usuario, deben perdurar a lo largo de las conversaciones. Cuando se mezclan, los datos temporales se acumulan y se pierde información importante.

El verdadero problema es arquitectónico. La mayoría de los marcos de trabajo no imponen una separación clara entre la memoria a corto plazo y la memoria a largo plazo, lo que obliga a los desarrolladores a gestionarla manualmente.

El kit de desarrollo de agentes (ADK) de código abierto de Google, publicado en 2025, aborda este problema a nivel de marco de trabajo haciendo de la gestión de la memoria una preocupación de primer orden. Impone una separación por defecto entre la memoria de sesión a corto plazo y la memoria a largo plazo.

En este artículo, veremos cómo funciona esta separación en la práctica. Usando Milvus como base de datos vectorial, construiremos un agente listo para producción con memoria real a largo plazo desde cero.

Principios básicos de diseño de ADK

ADK está diseñado para eliminar la gestión de memoria del desarrollador. El framework separa automáticamente los datos de sesión a corto plazo de la memoria a largo plazo y gestiona cada uno de ellos de forma apropiada. Para ello, cuenta con cuatro opciones de diseño fundamentales.

Interfaces integradas para memoria a corto y largo plazo

Cada agente ADK viene con dos interfaces integradas para gestionar la memoria:

SessionService (datos temporales)

• Qué almacena: contenido actual de la conversación y resultados intermedios de las llamadas a herramientas.
• Cuándo se borra: se borra automáticamente cuando finaliza la sesión.
• Dónde se almacena: en la memoria (la más rápida), en una base de datos o en un servicio en la nube.

MemoryService (memoria a largo plazo)

• Qué almacena: información que debe recordarse, como las preferencias del usuario o registros anteriores
• Cuándo se borra: no se borra automáticamente; debe borrarse manualmente
• Dónde se almacena: ADK define sólo la interfaz; el backend de almacenamiento depende de usted (por ejemplo, Milvus)

Arquitectura de tres capas

ADK divide el sistema en tres capas, cada una con una responsabilidad:

• Capa de agente: donde reside la lógica de negocio, como "recuperar la memoria relevante antes de responder al usuario".
• Capa de tiempo de ejecución: gestionada por el framework, responsable de crear y destruir sesiones y de hacer un seguimiento de cada paso de la ejecución.
• Capa de servicio: se integra con sistemas externos, como bases de datos vectoriales como Milvus o grandes API de modelos.

Esta estructura mantiene las preocupaciones separadas: la lógica empresarial vive en el agente, mientras que el almacenamiento vive en otro lugar. Se puede actualizar una sin romper la otra.

Todo se registra como eventos

Cada acción que realiza un agente -llamar a una herramienta de recuperación de memoria, invocar un modelo, generar una respuesta- se registra como un evento.

Esto tiene dos ventajas prácticas. En primer lugar, cuando algo va mal, los desarrolladores pueden reproducir toda la interacción paso a paso para encontrar el punto exacto del fallo. En segundo lugar, a efectos de auditoría y cumplimiento de normativas, el sistema proporciona una traza completa de la ejecución de cada interacción de usuario.

Ámbito de datos basado en prefijos

ADK controla la visibilidad de los datos mediante prefijos de clave simples:

• temp:xxx - visible sólo dentro de la sesión actual y se elimina automáticamente cuando ésta finaliza
• user:xxx: compartido en todas las sesiones del mismo usuario, lo que permite mantener las preferencias del usuario.
• app:xxx: compartido globalmente por todos los usuarios, adecuado para el conocimiento de toda la aplicación, como la documentación del producto.

Mediante el uso de prefijos, los desarrolladores pueden controlar el alcance de los datos sin necesidad de escribir lógica de acceso adicional. El framework gestiona la visibilidad y el tiempo de vida automáticamente.

Milvus como backend de memoria para ADK

En ADK, MemoryService es sólo una interfaz. Define cómo se utiliza la memoria a largo plazo, pero no cómo se almacena. La elección de la base de datos depende del desarrollador. Entonces, ¿qué tipo de base de datos funciona bien como backend de memoria de un agente?

Qué necesita un sistema de memoria de agente y cómo lo ofrece Milvus

• Recuperación semántica

La necesidad:

Los usuarios rara vez hacen la misma pregunta de la misma manera. "No se conecta" y "tiempo de espera de la conexión" significan lo mismo. El sistema de memoria debe comprender el significado, no sólo coincidir con palabras clave.

Cómo lo cumple Milvus:

Milvus admite muchos tipos de índices vectoriales, como HNSW y DiskANN, lo que permite a los desarrolladores elegir el que mejor se adapte a su carga de trabajo. Incluso con decenas de millones de vectores, la latencia de la consulta puede mantenerse por debajo de los 10 ms, lo que es suficientemente rápido para el uso de agentes.

• Consultas híbridas

La necesidad:

La recuperación de la memoria requiere algo más que una búsqueda semántica. El sistema también debe filtrar por campos estructurados como user_id para que sólo se devuelvan los datos del usuario actual.

Cómo lo satisface Milvus:

Milvus admite de forma nativa consultas híbridas que combinan la búsqueda vectorial con el filtrado escalar. Por ejemplo, puede recuperar registros semánticamente similares aplicando un filtro como user_id = 'xxx' en la misma consulta, sin perjudicar el rendimiento ni la calidad de la recuperación.

• Escalabilidad

La necesidad:

A medida que crece el número de usuarios y memorias almacenadas, el sistema debe escalar sin problemas. El rendimiento debe permanecer estable a medida que aumentan los datos, sin ralentizaciones ni fallos repentinos.

Cómo lo consigue Milvus:

Milvus utiliza una arquitectura de separación de cálculo y almacenamiento. La capacidad de consulta puede escalarse horizontalmente añadiendo nodos de consulta según sea necesario. Incluso la versión autónoma, que se ejecuta en una sola máquina, puede manejar decenas de millones de vectores, lo que la hace adecuada para las primeras fases de despliegue.

Nota: Para el desarrollo local y las pruebas, los ejemplos de este artículo utilizan Milvus Lite o Milvus Standalone.

Construir un Agente con Long-TermMemory Potenciado por Milvus

En esta sección, construimos un simple agente de soporte técnico. Cuando un usuario hace una pregunta, el agente busca tickets de soporte anteriores similares para responder, en lugar de repetir el mismo trabajo.

Este ejemplo es útil porque muestra tres problemas comunes que los sistemas de memoria de agentes reales deben manejar.

• Memoria a largo plazo entre sesiones

Una pregunta formulada hoy puede estar relacionada con un ticket creado hace semanas. El agente debe recordar la información a través de las conversaciones, no sólo dentro de la sesión actual. Por eso es necesaria la memoria a largo plazo, gestionada a través de MemoryService.

• Aislamiento del usuario

El historial de soporte de cada usuario debe permanecer privado. Los datos de un usuario nunca deben aparecer en los resultados de otro usuario. Esto requiere filtrar campos como user_id, que Milvus soporta mediante consultas híbridas.

• Correspondencia semántica

Los usuarios describen el mismo problema de diferentes maneras, como "no se puede conectar" o "tiempo de espera". La concordancia de palabras clave no es suficiente. El agente necesita una búsqueda semántica, que proporciona la recuperación vectorial.

Configuración del entorno

• Python 3.11+
• Docker y Docker Compose
• Clave API Gemini

Esta sección cubre la configuración básica para asegurarse de que el programa puede ejecutarse correctamente.

pip install google-adk pymilvus google-generativeai  

"""  
ADK + Milvus + Gemini Long-term Memory Agent  
Demonstrates how to implement a cross-session memory recall system  
"""  
import os  
import asyncio  
import time  
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility  
import google.generativeai as genai  
from google.adk.agents import Agent  
from google.adk.tools import FunctionTool  
from google.adk.runners import Runner  
from google.adk.sessions import InMemorySessionService  
from google.genai import types  

Paso 1: Desplegar Milvus Standalone (Docker)

(1) Descargue los archivos de despliegue

wget <https://github.com/Milvus-io/Milvus/releases/download/v2.5.12/Milvus-standalone-docker-compose.yml> -O docker-compose.yml  

(2) Inicie el servicio Milvus

docker-compose up -d  

docker-compose ps -a  

Paso 2 Configuración del modelo y la conexión

Configure la API de Gemini y los ajustes de conexión de Milvus.

# ==================== Configuration ====================  
# 1. Gemini API configuration  
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")  
if not GOOGLE_API_KEY:  
   raise ValueError("Please set the GOOGLE_API_KEY environment variable")  
genai.configure(api_key=GOOGLE_API_KEY)  
# 2. Milvus connection configuration  
MILVUS_HOST = os.getenv("MILVUS_HOST", "localhost")  
MILVUS_PORT = os.getenv("MILVUS_PORT", "19530")  
# 3. Model selection (best combination within the free tier limits)  
LLM_MODEL = "gemini-2.5-flash-lite"  # LLM model: 1000 RPD  
EMBEDDING_MODEL = "models/text-embedding-004"  # Embedding model: 1000 RPD  
EMBEDDING_DIM = 768  # Vector dimension  
# 4. Application configuration  
APP_NAME = "tech_support"  
USER_ID = "user_123"  
print(f"✓ Using model configuration:")  
print(f"  LLM: {LLM_MODEL}")  
print(f"  Embedding: {EMBEDDING_MODEL} (dimension: {EMBEDDING_DIM})")  

Paso 3 Inicialización de la base de datos Milvus

Crear una colección de base de datos vectorial (similar a una tabla en una base de datos relacional)

# ==================== Initialize Milvus ====================  
def init_milvus():  
   """Initialize Milvus connection and collection"""  
   # Step 1: Establish connection  
   Try:  
       connections.connect(  
           alias="default",  
           host=MILVUS_HOST,  
           port=MILVUS_PORT  
       )  
       print(f"✓ Connected to Milvus: {MILVUS_HOST}:{MILVUS_PORT}")  
   except Exception as e:  
       print(f"✗ Failed to connect to Milvus: {e}")  
       print("Hint: make sure Milvus is running")  
       Raise  
   # Step 2: Define data schema  
   fields = [  
       FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),  
       FieldSchema(name="user_id", dtype=DataType.VARCHAR, max_length=100),  
       FieldSchema(name="session_id", dtype=DataType.VARCHAR, max_length=100),  
       FieldSchema(name="question", dtype=DataType.VARCHAR, max_length=2000),  
       FieldSchema(name="solution", dtype=DataType.VARCHAR, max_length=5000),  
       FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=EMBEDDING_DIM),  
       FieldSchema(name="timestamp", dtype=DataType.INT64)  
   ]  
   schema = CollectionSchema(fields, description="Tech support memory")  
   collection_name = "support_memory"  
   # Step 3: Create or load the collection  
   if utility.has_collection(collection_name):  
       memory_collection = Collection(name=collection_name)  
       print(f"✓ Collection '{collection_name}' already exists")  
   Else:  
       memory_collection = Collection(name=collection_name, schema=schema)  
   # Step 4: Create vector index  
   index_params = {  
       "index_type": "IVF_FLAT",  
       "metric_type": "COSINE",  
       "params": {"nlist": 128}  
   }  
   memory_collection.create_index(field_name="embedding", index_params=index_params)  
   print(f"✓ Created collection '{collection_name}' and index")  
   return memory_collection  
# Run initialization  
memory_collection = init_milvus()  

Paso 4 Funciones de operación de memoria

Encapsule la lógica de almacenamiento y recuperación como herramientas para el agente.

(1) Función de almacenamiento de memoria

# ==================== Memory Operation Functions ====================  
def store_memory(question: str, solution: str) -> str:  
   """  
   Store a solution record into the memory store  
   Args:  
       question: the user's question  
       solution: the solution  
   Returns:  
       str: result message  
   """  
   Try:  
       print(f"\\n[Tool Call] store_memory")  
       print(f" - question: {question[:50]}...")  
       print(f" - solution: {solution[:50]}...")  
       # Use global USER_ID (in production, this should come from ToolContext)  
       user_id = USER_ID  
       session_id = f"session_{int(time.time())}"  
       # Key step 1: convert the question into a 768-dimensional vector  
       embedding_result = genai.embed_content(  
           model=EMBEDDING_MODEL,  
           content=question,  
           task_type="retrieval_document",  # specify document indexing task  
           output_dimensionality=EMBEDDING_DIM  
       )  
       embedding = embedding_result["embedding"]  
       # Key step 2: insert into Milvus  
       memory_collection.insert([{  
           "user_id": user_id,  
           "session_id": session_id,  
           "question": question,  
           "solution": solution,  
           "embedding": embedding,  
           "timestamp": int(time.time())  
       }])  
       # Key step 3: flush to disk (ensure data persistence)  
       memory_collection.flush()  
       result = "✓ Successfully stored in memory"  
       print(f"[Tool Result] {result}")  
       return result  
   except Exception as e:  
       error_msg = f"✗ Storage failed: {str(e)}"  
       print(f"[Tool Error] {error_msg}")  
       return error_msg  

(2) Función de recuperación de memoria

def recall_memory(query: str, top_k: int = 3) -> str:  
   """  
   Retrieve relevant historical cases from the memory store  
   Args:  
       query: query question  
       top_k: number of most similar results to return  
   Returns:  
       str: retrieval result  
   """  
   Try:  
       print(f"\\n[Tool Call] recall_memory")  
       print(f" - query: {query}")  
       print(f" - top_k: {top_k}")  
       user_id = USER_ID  
       # Key step 1: convert the query into a vector  
       embedding_result = genai.embed_content(  
           model=EMBEDDING_MODEL,  
           content=query,  
           task_type="retrieval_query",  # specify query task (different from indexing)  
           output_dimensionality=EMBEDDING_DIM  
       )  
       query_embedding = embedding_result["embedding"]  
       # Key step 2: load the collection into memory (required for the first query)  
       memory_collection.load()  
       # Key step 3: hybrid search (vector similarity + scalar filtering)  
       results = memory_collection.search(  
           data=[query_embedding],  
           anns_field="embedding",  
           param={"metric_type": "COSINE", "params": {"nprobe": 10}},  
           limit=top_k,  
           expr=f'user_id == "{user_id}"',  # 🔑 key to user isolation  
           output_fields=["question", "solution", "timestamp"]  
       )  
       # Key step 4: format results  
       if not results[0]:  
           result = "No relevant historical cases found"  
           print(f"[Tool Result] {result}")  
           return result  
       result_text = f"Found {len(results[0])} relevant cases:\\n\\n"  
       for i, hit in enumerate(results[0]):  
           result_text += f"Case {i+1} (similarity: {hit.score:.2f}):\\n"  
           result_text += f"Question: {hit.entity.get('question')}\\n"  
           result_text += f"Solution: {hit.entity.get('solution')}\\n\\n"  
       print(f"[Tool Result] Found {len(results[0])} cases")  
       return result_text  
   except Exception as e:  
       error_msg = f"Retrieval failed: {str(e)}"  
       print(f"[Tool Error] {error_msg}")  
       return error_msg  

(3) Registro como herramienta ADK

# Usage  
# Wrap functions with FunctionTool  
store_memory_tool = FunctionTool(func=store_memory)  
recall_memory_tool = FunctionTool(func=recall_memory)  
memory_tools = [store_memory_tool, recall_memory_tool]  

Paso 5 Definición del Agente

Idea central: definir la lógica de comportamiento del agente.

# ==================== Create Agent ====================  
support_agent = Agent(  
   model=LLM_MODEL,  
   name="support_agent",  
   description="Technical support expert agent that can remember and recall historical cases",  
   # Key: the instruction defines the agent’s behavior  
   instruction="""  
You are a technical support expert. Strictly follow the process below:  
<b>When the user asks a technical question:</b>  
1. Immediately call the recall_memory tool to search for historical cases  
  - Parameter query: use the user’s question text directly  
  - Do not ask for any additional information; call the tool directly  
2. Answer based on the retrieval result:  
  - If relevant cases are found: explain that similar historical cases were found and answer by referencing their solutions  
  - If no cases are found: explain that this is a new issue and answer based on your own knowledge  
3. After answering, ask: “Did this solution resolve your issue?”  
<b>When the user confirms the issue is resolved:</b>  
- Immediately call the store_memory tool to save this Q&A  
- Parameter question: the user’s original question  
- Parameter solution: the complete solution you provided  
<b>Important rules:</b>  
- You must call a tool before answering  
- Do not ask for user_id or any other parameters  
- Only store memory when you see confirmation phrases such as “resolved”, “it works”, or “thanks”  
""",  
   tools=memory_tools  
)  

Paso 6 Programa Principal y Flujo de Ejecución

Demuestra el proceso completo de recuperación de memoria entre sesiones.

# ==================== Main Program ====================  
async def main():  
   """Demonstrate cross-session memory recall"""  
   # Create Session service and Runner  
   session_service = InMemorySessionService()  
   runner = Runner(  
       agent=support_agent,  
       app_name=APP_NAME,  
       session_service=session_service  
   )  
   # ========== First round: build memory ==========  
   print("\\n" + "=" \* 60)  
   print("First conversation: user asks a question and the solution is stored")  
   print("=" \* 60)  
   session1 = await session_service.create_session(  
       app_name=APP_NAME,  
       user_id=USER_ID,  
       session_id="session_001"  
   )  
   # User asks the first question  
   print("\\n[User]: What should I do if Milvus connection times out?")  
   content1 = types.Content(  
       role='user',  
       parts=[types.Part(text="What should I do if Milvus connection times out?")]  
   )  
   async for event in runner.run_async(  
       user_id=USER_ID,  
       session_id=[session1.id](http://session1.id),  
       new_message=content1  
   ):  
       if event.content and event.content.parts:  
           for part in event.content.parts:  
               if hasattr(part, 'text') and part.text:  
                   print(f"[Agent]: {part.text}")  
   # User confirms the issue is resolved  
   print("\\n[User]: The issue is resolved, thanks!")  
   content2 = types.Content(  
       role='user',  
       parts=[types.Part(text="The issue is resolved, thanks!")]  
   )  
   async for event in runner.run_async(  
       user_id=USER_ID,  
       session_id=[session1.id](http://session1.id),  
       new_message=content2  
   ):  
       if event.content and event.content.parts:  
           for part in event.content.parts:  
               if hasattr(part, 'text') and part.text:  
                   print(f"[Agent]: {part.text}")  
   # ========== Second round: recall memory ==========  
   print("\\n" + "=" \* 60)  
   print("Second conversation: new session with memory recall")  
   print("=" \* 60)  
   session2 = await session_service.create_session(  
       app_name=APP_NAME,  
       user_id=USER_ID,  
       session_id="session_002"  
   )  
   # User asks a similar question in a new session  
   print("\\n[User]: Milvus can't connect")  
   content3 = types.Content(  
       role='user',  
       parts=[types.Part(text="Milvus can't connect")]  
   )  
   async for event in runner.run_async(  
       user_id=USER_ID,  
       session_id=[session2.id](http://session2.id),  
       new_message=content3  
   ):  
       if event.content and event.content.parts:  
           for part in event.content.parts:  
               if hasattr(part, 'text') and part.text:  
                   print(f"[Agent]: {part.text}")
# Program entry point  
if name == "main":

Try:

asyncio.run(main())

except KeyboardInterrupt:

print(“\n\nProgram exited”)

except Exception as e:

print(f"\n\nProgram error: {e}")

import traceback

traceback.print_exc()

Finally:

Try:

connections.disconnect(alias=“default”)

print(“\n✓ Disconnected from Milvus”)

Except:

pass

Paso 7 Ejecutar y Probar

(1) Establecer variables de entorno

export GOOGLE_API_KEY="your-gemini-api-key"  

python milvus_agent.py  

Salida Esperada

La salida muestra cómo funciona el sistema de memoria en uso real.

En la primera conversación, el usuario pregunta cómo manejar el tiempo de espera de una conexión Milvus. El agente da una solución. Después de que el usuario confirme que el problema está resuelto, el agente guarda la pregunta y la respuesta en la memoria.

En la segunda conversación, se inicia una nueva sesión. El usuario formula la misma pregunta utilizando palabras diferentes: "Milvus no puede conectarse". El agente recupera automáticamente un caso similar de la memoria y da la misma solución.

No es necesario ningún paso manual. El agente decide cuándo recuperar los casos anteriores y cuándo almacenar los nuevos, lo que demuestra tres capacidades clave: memoria entre sesiones, correspondencia semántica y aislamiento del usuario.

Conclusión

ADK separa el contexto a corto plazo y la memoria a largo plazo a nivel de marco utilizando SessionService y MemoryService. Milvus gestiona la búsqueda semántica y el filtrado a nivel de usuario mediante la recuperación basada en vectores.

A la hora de elegir un framework, el objetivo es importante. Si necesita un fuerte aislamiento del estado, auditabilidad y estabilidad de producción, ADK es la mejor opción. Si está creando prototipos o experimentando, LangChain (un popular marco de Python para crear rápidamente aplicaciones y agentes basados en LLM) ofrece más flexibilidad.

Para la memoria de agentes, la pieza clave es la base de datos. La memoria semántica depende de bases de datos vectoriales, independientemente del framework que utilices. Milvus funciona bien porque es de código abierto, se ejecuta localmente, permite manejar miles de millones de vectores en una sola máquina y admite búsquedas híbridas vectoriales, escalares y de texto completo. Estas características cubren tanto las pruebas iniciales como el uso en producción.

Esperamos que este artículo te ayude a comprender mejor el diseño de la memoria de los agentes y a elegir las herramientas adecuadas para tus proyectos.

Si está creando agentes de IA que necesitan memoria real, no sólo ventanas de contexto más grandes, nos encantaría saber cómo lo está abordando.

¿Tiene preguntas sobre ADK, el diseño de la memoria del agente o el uso de Milvus como backend de memoria? Únase a nuestro canal de Slack o reserve una sesión de 20 minutos de Milvus Office Hours para hablar sobre su caso de uso.