Como criar agentes de IA prontos para produção com memória de longo prazo usando o Google ADK e o Milvus

Ao construir agentes inteligentes, um dos problemas mais difíceis é a gestão da memória: decidir o que o agente deve lembrar e o que deve esquecer.

Nem toda a memória é para durar. Alguns dados são necessários apenas para a conversa atual e devem ser apagados quando esta termina. Outros dados, como as preferências do utilizador, devem persistir ao longo das conversações. Quando estes são misturados, os dados temporários acumulam-se e perde-se informação importante.

O verdadeiro problema é arquitetónico. A maioria das estruturas não impõe uma separação clara entre a memória de curto prazo e a memória de longo prazo, deixando os programadores a tratar disso manualmente.

O Agent Development Kit (ADK) de código aberto da Google, lançado em 2025, resolve este problema ao nível da estrutura, tornando a gestão da memória uma preocupação de primeira classe. Ele impõe uma separação padrão entre a memória de sessão de curto prazo e a memória de longo prazo.

Neste artigo, veremos como essa separação funciona na prática. Usando o Milvus como banco de dados vetorial, criaremos um agente pronto para produção com memória de longo prazo real a partir do zero.

Princípios básicos de design do ADK

O ADK foi projetado para tirar o gerenciamento de memória do prato do desenvolvedor. O framework separa automaticamente os dados de sessão de curto prazo da memória de longo prazo e trata cada um deles apropriadamente. Ele faz isso através de quatro escolhas centrais de design.

Interfaces embutidas para memória de curto e longo prazo

Todo agente ADK vem com duas interfaces integradas para gerenciar a memória:

SessionService (dados temporários)

• O que armazena: conteúdo da conversa atual e resultados intermédios de chamadas de ferramentas
• Quando é apagada: automaticamente apagada quando a sessão termina
• Onde é armazenada: na memória (mais rápida), numa base de dados ou num serviço de nuvem

MemoryService (memória de longo prazo)

• O que armazena: informações que devem ser recordadas, como as preferências do utilizador ou registos anteriores
• Quando é apagada: não é apagada automaticamente; tem de ser apagada manualmente
• Onde é armazenado: O ADK define apenas a interface; o backend de armazenamento é da responsabilidade do utilizador (por exemplo, Milvus)

Uma arquitetura de três camadas

O ADK divide o sistema em três camadas, cada uma com uma única responsabilidade:

• Camada de agente: onde reside a lógica empresarial, como "recuperar memória relevante antes de responder ao utilizador".
• Camada de tempo de execução: gerida pela estrutura, responsável pela criação e destruição de sessões e pelo controlo de cada passo da execução.
• Camada de serviço: integra-se com sistemas externos, como bases de dados vectoriais como o Milvus ou APIs de modelos de grande dimensão.

Esta estrutura mantém as preocupações separadas: a lógica empresarial reside no agente, enquanto o armazenamento reside noutro local. É possível atualizar um sem quebrar o outro.

Tudo é registado como eventos

Cada ação de um agente - chamar uma ferramenta de recuperação de memória, invocar um modelo, gerar uma resposta - é registada como um evento.

Isto tem dois benefícios práticos. Primeiro, quando algo corre mal, os programadores podem reproduzir toda a interação passo a passo para encontrar o ponto exato da falha. Em segundo lugar, para auditoria e conformidade, o sistema fornece um traço de execução completo de cada interação do utilizador.

Escopo de dados baseado em prefixo

O ADK controla a visibilidade dos dados usando prefixos de chave simples:

• temp:xxx - visível apenas na sessão atual e removido automaticamente quando esta termina
• user:xxx - partilhado em todas as sessões para o mesmo utilizador, permitindo preferências de utilizador persistentes
• app:xxx - partilhado globalmente por todos os utilizadores, adequado para conhecimento de toda a aplicação, como documentação de produtos

Ao utilizar prefixos, os programadores podem controlar o âmbito dos dados sem escrever lógica de acesso adicional. A estrutura lida com a visibilidade e o tempo de vida automaticamente.

Milvus como Backend de Memória para ADK

No ADK, o MemoryService é apenas uma interface. Ele define como a memória de longo prazo é usada, mas não como ela é armazenada. A escolha da base de dados é da responsabilidade do programador. Então, que tipo de base de dados funciona bem como backend de memória de um agente?

O que um sistema de memória de agente precisa - e como o Milvus o fornece

• Recuperação semântica

A necessidade:

Os utilizadores raramente fazem a mesma pergunta da mesma forma. "Não se conecta" e "tempo limite de conexão" significam a mesma coisa. O sistema de memória tem de compreender o significado e não apenas corresponder a palavras-chave.

Como o Milvus atende a isso:

O Milvus suporta muitos tipos de índices vetoriais, como HNSW e DiskANN, permitindo que os desenvolvedores escolham o que melhor se adapta à sua carga de trabalho. Mesmo com dezenas de milhões de vetores, a latência da consulta pode ficar abaixo de 10 ms, o que é rápido o suficiente para o uso do agente.

• Consultas híbridas

A necessidade:

A recuperação de memória requer mais do que pesquisa semântica. O sistema também deve filtrar por campos estruturados, como user_id, para que apenas os dados do utilizador atual sejam devolvidos.

Como o Milvus atende a essa necessidade:

O Milvus suporta nativamente consultas híbridas que combinam a pesquisa vetorial com a filtragem escalar. Por exemplo, pode obter registos semanticamente semelhantes enquanto aplica um filtro como user_id = 'xxx' na mesma consulta, sem prejudicar o desempenho ou a qualidade da pesquisa.

• Escalabilidade

A necessidade:

À medida que o número de utilizadores e de memórias armazenadas aumenta, o sistema deve ser escalável sem problemas. O desempenho deve manter-se estável à medida que os dados aumentam, sem abrandamentos ou falhas súbitas.

Como é que o Milvus satisfaz essa necessidade:

O Milvus utiliza uma arquitetura de separação entre computação e armazenamento. A capacidade de consulta pode ser escalada horizontalmente, adicionando nós de consulta conforme necessário. Mesmo a versão autónoma, executada numa única máquina, pode lidar com dezenas de milhões de vectores, tornando-a adequada para implementações em fase inicial.

Nota: Para desenvolvimento e testes locais, os exemplos neste artigo usam o Milvus Lite ou o Milvus Standalone.

Construindo um agente com Long-TermMemory Powered by Milvus

Nesta secção, construímos um agente de suporte técnico simples. Quando um usuário faz uma pergunta, o agente procura tickets de suporte anteriores semelhantes para responder, em vez de repetir o mesmo trabalho.

Este exemplo é útil porque mostra três problemas comuns que os sistemas de memória de agentes reais devem resolver.

• Memória de longo prazo entre sessões

Uma pergunta feita hoje pode estar relacionada a um ticket criado semanas atrás. O agente deve lembrar-se das informações em todas as conversas, não apenas na sessão atual. É por isso que a memória de longo prazo, gerenciada pelo MemoryService, é necessária.

• Isolamento do utilizador

O histórico de suporte de cada utilizador deve permanecer privado. Os dados de um utilizador nunca devem aparecer nos resultados de outro utilizador. Isto requer a filtragem de campos como user_id, que o Milvus suporta através de consultas híbridas.

• Correspondência semântica

Os utilizadores descrevem o mesmo problema de formas diferentes, como "não é possível ligar" ou "tempo limite". A correspondência de palavras-chave não é suficiente. O agente precisa de uma pesquisa semântica, que é fornecida pela recuperação de vectores.

Configuração do ambiente

• Python 3.11+
• Docker e Docker Compose
• Chave da API Gemini

Esta secção abrange a configuração básica para garantir que o programa pode ser executado corretamente.

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  

Etapa 1: implantar o Milvus Standalone (Docker)

(1) Descarregar os ficheiros de implementação

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

(2) Iniciar o serviço Milvus

docker-compose up -d  

docker-compose ps -a  

Etapa 2: Configuração de modelo e conexão

Configure as definições de ligação da Gemini API e do 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})")  

Etapa 3 Inicialização do banco de dados do Milvus

Criar uma coleção de base de dados vetorial (semelhante a uma tabela numa base de dados 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()  

Etapa 4 Funções de operação de memória

Encapsular a lógica de armazenamento e recuperação como ferramentas para o agente.

(1) Função de memória de armazenamento

# ==================== 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) Função de recuperação de memória

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) Registar como uma ferramenta 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]  

Etapa 5 Definição do agente

Ideia central: definir a lógica de comportamento do 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  
)  

Etapa 6 Programa principal e fluxo de execução

Demonstra o processo completo de recuperação de memória entre sessões.

# ==================== 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

Etapa 7 Executar e testar

(1) Definir variáveis de ambiente

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

python milvus_agent.py  

Saída esperada

A saída mostra como o sistema de memória funciona em uso real.

Na primeira conversa, o utilizador pergunta como lidar com o tempo limite de uma ligação Milvus. O agente apresenta uma solução. Depois de o utilizador confirmar que o problema está resolvido, o agente guarda esta pergunta e resposta na memória.

Na segunda conversa, é iniciada uma nova sessão. O utilizador faz a mesma pergunta utilizando palavras diferentes: "O Milvus não consegue ligar-se." O agente recupera automaticamente um caso semelhante da memória e dá-lhe a mesma solução.

Não são necessários passos manuais. O agente decide quando recuperar casos passados e quando armazenar novos casos, mostrando três capacidades-chave: memória entre sessões, correspondência semântica e isolamento do utilizador.

Conclusão

O ADK separa o contexto de curto prazo e a memória de longo prazo ao nível da estrutura, utilizando o SessionService e o MemoryService. O Milvus trata da pesquisa semântica e da filtragem ao nível do utilizador através da recuperação baseada em vectores.

Ao escolher uma estrutura, o objetivo é importante. Se precisar de um forte isolamento de estado, auditabilidade e estabilidade de produção, o ADK é mais adequado. Se estiver a criar protótipos ou a experimentar, o LangChain (uma estrutura Python popular para criar rapidamente aplicações e agentes baseados em LLM) oferece mais flexibilidade.

Para a memória do agente, a peça chave é o banco de dados. A memória semântica depende de bases de dados vectoriais, independentemente da estrutura utilizada. O Milvus funciona bem porque é de código aberto, corre localmente, suporta a manipulação de milhares de milhões de vectores numa única máquina e suporta pesquisa híbrida vetorial, escalar e de texto integral. Esses recursos abrangem tanto os testes iniciais quanto o uso em produção.

Esperamos que este artigo o ajude a compreender melhor o design da memória do agente e a escolher as ferramentas certas para os seus projectos.

Se você estiver criando agentes de IA que precisam de memória real, e não apenas de janelas de contexto maiores, gostaríamos de saber como você está lidando com isso.

Tem dúvidas sobre o ADK, o design de memória do agente ou o uso do Milvus como back-end de memória? Junte-se ao nosso canal Slack ou marque uma sessão de 20 minutos do Milvus Office Hours para falar sobre o seu caso de utilização.