Home
Blog
Cara Membuat Agen AI Siap Produksi dengan Memori Jangka Panjang Menggunakan Google ADK dan Milvus

Cara Membuat Agen AI Siap Produksi dengan Memori Jangka Panjang Menggunakan Google ADK dan Milvus

Tutorials

February 26, 2026

Min Yin

Ketika membangun agen cerdas, salah satu masalah tersulit adalah manajemen memori: memutuskan apa yang harus diingat dan apa yang harus dilupakan oleh agen.

Tidak semua memori dimaksudkan untuk bertahan lama. Beberapa data hanya diperlukan untuk percakapan saat ini dan harus dihapus ketika percakapan berakhir. Data lain, seperti preferensi pengguna, harus tetap ada di seluruh percakapan. Ketika data ini tercampur, data sementara akan menumpuk, dan informasi penting akan hilang.

Masalah yang sebenarnya adalah arsitektur. Sebagian besar kerangka kerja tidak menerapkan pemisahan yang jelas antara memori jangka pendek dan jangka panjang, sehingga para pengembang harus menanganinya secara manual.

Agent Development Kit (ADK) sumber terbuka Google, yang dirilis pada tahun 2025, menangani hal ini di tingkat kerangka kerja dengan menjadikan manajemen memori sebagai perhatian utama. ADK memberlakukan pemisahan default antara memori sesi jangka pendek dan memori jangka panjang.

Dalam artikel ini, kita akan melihat bagaimana pemisahan ini bekerja dalam praktiknya. Dengan menggunakan Milvus sebagai basis data vektor, kita akan membangun agen siap produksi dengan memori jangka panjang yang nyata dari awal.

Prinsip Desain Inti ADK

ADK dirancang untuk mengambil alih manajemen memori dari tangan pengembang. Kerangka kerja ini secara otomatis memisahkan data sesi jangka pendek dari memori jangka panjang dan menangani masing-masing dengan tepat. Ini dilakukan melalui empat pilihan desain inti.

Antarmuka Bawaan untuk Memori Jangka Pendek dan Jangka Panjang

Setiap agen ADK dilengkapi dengan dua antarmuka bawaan untuk mengelola memori:

SessionService (data sementara)

Apa yang disimpan: konten percakapan saat ini dan hasil perantara dari panggilan alat
Kapan dihapus: secara otomatis dihapus ketika sesi berakhir
Tempat penyimpanannya: di memori (tercepat), basis data, atau layanan cloud

MemoryService (memori jangka panjang)

Apa yang disimpan: informasi yang harus diingat, seperti preferensi pengguna atau catatan sebelumnya
Kapan dihapus: tidak dihapus secara otomatis; harus dihapus secara manual
Di mana ia disimpan: ADK hanya mendefinisikan antarmuka; backend penyimpanan terserah Anda (misalnya, Milvus)

Arsitektur Tiga Lapisan

ADK membagi sistem menjadi tiga lapisan, masing-masing dengan satu tanggung jawab:

Lapisan agen: tempat logika bisnis berada, seperti "mengambil memori yang relevan sebelum menjawab pengguna."
Runtime layer: dikelola oleh framework, bertanggung jawab untuk membuat dan menghancurkan sesi dan melacak setiap langkah eksekusi.
Lapisan layanan: terintegrasi dengan sistem eksternal, seperti basis data vektor seperti Milvus atau API model besar.

Struktur ini menjaga agar masalah tetap terpisah: logika bisnis berada di dalam agen, sementara penyimpanan berada di tempat lain. Anda bisa memperbarui salah satunya tanpa merusak yang lain.

Segala Sesuatu Dicatat sebagai Peristiwa

Setiap tindakan yang dilakukan agen-memanggil alat pemanggilan memori, memanggil model, menghasilkan respons-direkam sebagai peristiwa.

Ini memiliki dua manfaat praktis. Pertama, ketika terjadi kesalahan, pengembang dapat memutar ulang seluruh interaksi langkah demi langkah untuk menemukan titik kegagalan yang tepat. Kedua, untuk audit dan kepatuhan, sistem menyediakan jejak eksekusi lengkap dari setiap interaksi pengguna.

Pelingkupan Data Berbasis Awalan

ADK mengontrol visibilitas data menggunakan prefiks kunci sederhana:

temp:xxx - hanya dapat dilihat dalam sesi saat ini dan secara otomatis dihapus ketika sesi berakhir
user:xxx - dibagikan di semua sesi untuk pengguna yang sama, sehingga memungkinkan preferensi pengguna yang persisten
app:xxx - dibagikan secara global ke semua pengguna, cocok untuk pengetahuan di seluruh aplikasi seperti dokumentasi produk

Dengan menggunakan prefix, pengembang dapat mengontrol cakupan data tanpa menulis logika akses tambahan. Kerangka kerja ini menangani visibilitas dan masa pakai secara otomatis.

Milvus sebagai Backend Memori untuk ADK

Dalam ADK, MemoryService hanyalah sebuah antarmuka. Ini mendefinisikan bagaimana memori jangka panjang digunakan, tetapi bukan bagaimana memori disimpan. Memilih basis data tergantung pada pengembang. Jadi, database seperti apa yang bekerja dengan baik sebagai backend memori agen?

Apa yang Dibutuhkan Sistem Memori Agen - dan Bagaimana Milvus Memberikannya

Pengambilan Semantik

Kebutuhan:

Pengguna jarang mengajukan pertanyaan yang sama dengan cara yang sama. "Tidak tersambung" dan "batas waktu koneksi" memiliki arti yang sama. Sistem memori harus memahami makna, bukan hanya mencocokkan kata kunci.

Bagaimana Milvus memenuhinya:

Milvus mendukung banyak jenis indeks vektor, seperti HNSW dan DiskANN, yang memungkinkan para pengembang untuk memilih yang sesuai dengan beban kerja mereka. Bahkan dengan puluhan juta vektor, latensi kueri dapat tetap di bawah 10 ms, yang cukup cepat untuk penggunaan agen.

Kueri Hibrida

Kebutuhan:

Pemanggilan kembali memori membutuhkan lebih dari sekadar pencarian semantik. Sistem juga harus memfilter berdasarkan bidang terstruktur seperti user_id sehingga hanya data pengguna saat ini yang dikembalikan.

Bagaimana Milvus memenuhinya:

Milvus secara native mendukung kueri hibrida yang menggabungkan pencarian vektor dengan pemfilteran skalar. Sebagai contoh, ia dapat mengambil catatan yang secara semantik mirip sambil menerapkan filter seperti user_id = 'xxx' dalam kueri yang sama, tanpa merusak kinerja atau kualitas penarikan.

Skalabilitas

Kebutuhan:

Seiring dengan bertambahnya jumlah pengguna dan memori yang tersimpan, sistem harus dapat berkembang dengan lancar. Performa harus tetap stabil seiring bertambahnya data, tanpa perlambatan atau kegagalan yang tiba-tiba.

Bagaimana Milvus memenuhinya:

Milvus menggunakan arsitektur pemisahan komputasi-penyimpanan. Kapasitas kueri dapat ditingkatkan secara horizontal dengan menambahkan Query Node sesuai kebutuhan. Bahkan versi mandiri, yang berjalan pada satu mesin, dapat menangani puluhan juta vektor, sehingga cocok untuk penerapan tahap awal.

Catatan: Untuk pengembangan dan pengujian lokal, contoh-contoh dalam artikel ini menggunakan Milvus Lite atau Milvus Standalone.

Membangun Agen dengan Memori Jangka Panjang yang Didukung oleh Milvus

Pada bagian ini, kita akan membangun agen dukungan teknis sederhana. Ketika pengguna mengajukan pertanyaan, agen akan mencari tiket dukungan yang serupa di masa lalu untuk dijawab, daripada mengulangi pekerjaan yang sama.

Contoh ini berguna karena menunjukkan tiga masalah umum yang harus ditangani oleh sistem memori agen yang sebenarnya.

Memori jangka panjang di seluruh sesi

Pertanyaan yang diajukan hari ini mungkin berhubungan dengan tiket yang dibuat beberapa minggu yang lalu. Agen harus mengingat informasi di seluruh percakapan, tidak hanya dalam sesi saat ini. Inilah sebabnya mengapa memori jangka panjang, yang dikelola melalui MemoryService, dibutuhkan.

Isolasi pengguna

Riwayat dukungan setiap pengguna harus tetap bersifat pribadi. Data dari satu pengguna tidak boleh muncul di hasil pengguna lain. Hal ini membutuhkan penyaringan pada bidang seperti user_id, yang didukung oleh Milvus melalui kueri hibrida.

Pencocokan semantik

Pengguna menggambarkan masalah yang sama dengan cara yang berbeda, seperti "tidak dapat terhubung" atau "waktu habis." Pencocokan kata kunci saja tidak cukup. Agen membutuhkan pencarian semantik, yang disediakan oleh pengambilan vektor.

Penyiapan lingkungan

Python 3.11+
Docker dan Docker Compose
Kunci API Gemini

Bagian ini mencakup penyiapan dasar untuk memastikan program dapat berjalan dengan benar.

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

Langkah 1: Menerapkan Milvus Standalone (Docker)

(1) Unduh berkas penyebaran

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

(2) Mulai layanan Milvus

docker-compose up -d

docker-compose ps -a

Langkah 2 Model dan Konfigurasi Koneksi

Konfigurasikan API Gemini dan pengaturan koneksi 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})")

Langkah 3 Inisialisasi Basis Data Milvus

Membuat koleksi basis data vektor (mirip dengan tabel dalam basis data relasional)

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

Langkah 4 Fungsi Operasi Memori

Enkapsulasi logika penyimpanan dan pengambilan sebagai alat bantu untuk agen.

(1) Menyimpan fungsi memori

# ==================== 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) Mengambil fungsi memori

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) Mendaftarkan sebagai Alat 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]

Langkah 5 Definisi Agen

Ide inti: mendefinisikan logika perilaku agen.

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

Langkah 6 Program Utama dan Alur Eksekusi

Mendemonstrasikan proses lengkap pengambilan memori lintas sesi.

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

Langkah 7 Jalankan dan Uji

(1) Tetapkan variabel lingkungan

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

python milvus_agent.py

Keluaran yang diharapkan

Keluaran menunjukkan bagaimana sistem memori bekerja dalam penggunaan nyata.

Pada percakapan pertama, pengguna bertanya bagaimana cara menangani batas waktu koneksi Milvus. Agen memberikan solusi. Setelah pengguna mengonfirmasi bahwa masalahnya telah terpecahkan, agen akan menyimpan pertanyaan dan jawaban ke dalam memori.

Pada percakapan kedua, sesi baru dimulai. Pengguna mengajukan pertanyaan yang sama dengan kata-kata yang berbeda: "Milvus tidak bisa terhubung." Agen secara otomatis mengambil kasus serupa dari memori dan memberikan solusi yang sama.

Tidak ada langkah manual yang diperlukan. Agen memutuskan kapan harus mengambil kasus-kasus sebelumnya dan kapan harus menyimpan kasus yang baru, dengan menunjukkan tiga kemampuan utama: memori lintas sesi, pencocokan semantik, dan isolasi pengguna.

Kesimpulan

ADK memisahkan konteks jangka pendek dan memori jangka panjang pada tingkat kerangka kerja menggunakan SessionService dan MemoryService. Milvus menangani pencarian semantik dan pemfilteran tingkat pengguna melalui pengambilan berbasis vektor.

Ketika memilih kerangka kerja, tujuan itu penting. Jika Anda membutuhkan isolasi state yang kuat, kemampuan audit, dan stabilitas produksi, ADK lebih cocok. Jika Anda membuat prototipe atau bereksperimen, LangChain (kerangka kerja Python yang populer untuk membangun aplikasi dan agen berbasis LLM dengan cepat) menawarkan lebih banyak fleksibilitas.

Untuk memori agen, bagian kuncinya adalah basis data. Memori semantik bergantung pada basis data vektor, apa pun kerangka kerja yang Anda gunakan. Milvus bekerja dengan baik karena Milvus bersifat open source, berjalan secara lokal, mendukung penanganan miliaran vektor dalam satu mesin, dan mendukung vektor hibrida, skalar, dan pencarian teks lengkap. Fitur-fitur ini mencakup pengujian awal dan penggunaan produksi.

Kami harap artikel ini membantu Anda lebih memahami desain memori agen dan memilih alat yang tepat untuk proyek Anda.

Jika Anda sedang membangun agen AI yang membutuhkan memori nyata-bukan hanya jendela konteks yang lebih besar-kami ingin mendengar bagaimana Anda mendekatinya.

Ada pertanyaan tentang ADK, desain memori agen, atau menggunakan Milvus sebagai backend memori? Bergabunglah dengan saluran Slack kami atau pesan sesi Jam Kantor Milvus selama 20 menit untuk membicarakan kasus penggunaan Anda.