كيفية بناء وكلاء ذكاء اصطناعي جاهزين للإنتاج مع ذاكرة طويلة المدى باستخدام Google ADK و Milvus
عند بناء وكلاء أذكياء، فإن إحدى أصعب المشاكل هي إدارة الذاكرة: تحديد ما يجب أن يتذكره الوكيل وما يجب أن ينساه.
ليس من المفترض أن تدوم كل الذاكرة. بعض البيانات مطلوبة فقط للمحادثة الحالية ويجب مسحها عند انتهائها. أما البيانات الأخرى، مثل تفضيلات المستخدم، فيجب أن تستمر عبر المحادثات. عندما تختلط هذه البيانات، تتراكم البيانات المؤقتة وتضيع المعلومات المهمة.
المشكلة الحقيقية هي مشكلة معمارية. لا تفرض معظم أطر العمل فصلًا واضحًا بين الذاكرة قصيرة الأجل والذاكرة طويلة الأجل، مما يترك المطورين للتعامل معها يدويًا.
تعالج مجموعة أدوات تطوير العوامل مفتوحة المصدر من Google (ADK)، التي تم إصدارها في عام 2025، هذه المشكلة على مستوى إطار العمل من خلال جعل إدارة الذاكرة من اهتمامات الدرجة الأولى. فهي تفرض فصلًا افتراضيًا بين ذاكرة الجلسة قصيرة المدى وذاكرة المدى الطويل.
في هذه المقالة، سنلقي نظرة على كيفية عمل هذا الفصل عمليًا. باستخدام Milvus كقاعدة بيانات المتجه، سنقوم ببناء وكيل جاهز للإنتاج مع ذاكرة حقيقية طويلة الأجل من الصفر.
مبادئ تصميم ADK الأساسية
تم تصميم ADK لإزالة إدارة الذاكرة من على عاتق المطور. يفصل الإطار تلقائيًا بيانات الجلسة قصيرة الأجل عن الذاكرة طويلة الأجل ويعالج كل منها بشكل مناسب. يقوم بذلك من خلال أربعة خيارات تصميم أساسية.
واجهات مدمجة للذاكرة قصيرة وطويلة الأمد
يأتي كل وكيل ADK مع واجهتين مدمجتين لإدارة الذاكرة:
خدمة الجلسة (البيانات المؤقتة)
- ما تقوم بتخزينه: محتوى المحادثة الحالي والنتائج الوسيطة من استدعاءات الأداة
- متى يتم مسحها: يتم مسحها تلقائيًا عند انتهاء الجلسة
- مكان تخزينها: في الذاكرة (الأسرع) أو قاعدة بيانات أو خدمة سحابية
خدمة الذاكرة (الذاكرة طويلة المدى)
- ماذا تخزن: المعلومات التي يجب تذكرها، مثل تفضيلات المستخدم أو السجلات السابقة
- متى يتم مسحها: لا يتم مسحها تلقائيًا؛ يجب حذفها يدويًا
- مكان تخزينها: تحدد ADK الواجهة فقط؛ أما الواجهة الخلفية للتخزين فهي متروكة لك (على سبيل المثال، ميلفوس)
بنية ثلاثية الطبقات
تقسم ADK النظام إلى ثلاث طبقات، لكل منها مسؤولية واحدة:
- طبقة الوكيل: حيث يوجد منطق العمل، مثل "استرجاع الذاكرة ذات الصلة قبل الرد على المستخدم".
- طبقة وقت التشغيل: يديرها إطار العمل، وهي مسؤولة عن إنشاء الجلسات وتدميرها وتتبع كل خطوة من خطوات التنفيذ.
- طبقة الخدمة: تتكامل مع أنظمة خارجية، مثل قواعد البيانات المتجهة مثل ميلفوس أو واجهات برمجة التطبيقات النموذجية الكبيرة.
تحافظ هذه البنية على الاهتمامات منفصلة: يعيش منطق العمل في الوكيل، بينما يعيش التخزين في مكان آخر. يمكنك تحديث أحدهما دون كسر الآخر.
يتم تسجيل كل شيء كأحداث
يتم تسجيل كل إجراء يتخذه الوكيل - استدعاء أداة استدعاء الذاكرة، استدعاء نموذج، توليد استجابة - كحدث.
هذا له فائدتان عمليتان. أولاً، عندما يحدث خطأ ما، يمكن للمطورين إعادة تشغيل التفاعل بأكمله خطوة بخطوة للعثور على نقطة الفشل بالضبط. ثانياً، من أجل التدقيق والامتثال، يوفر النظام تتبعاً كاملاً لتنفيذ كل تفاعل للمستخدم.
تحديد نطاق البيانات المستند إلى البادئة
يتحكم ADK في رؤية البيانات باستخدام بادئات مفاتيح بسيطة:
- temp:xxx - مرئية فقط داخل الجلسة الحالية ويتم إزالتها تلقائيًا عند انتهائها
- المستخدم:xxx - مشتركة عبر جميع الجلسات لنفس المستخدم، مما يتيح تفضيلات المستخدم المستمرة
- التطبيق:xxx - مشترك عالميًا عبر جميع المستخدمين، ومناسب للمعرفة على مستوى التطبيق مثل وثائق المنتج
باستخدام البادئات، يمكن للمطورين التحكم في نطاق البيانات دون كتابة منطق وصول إضافي. يتعامل إطار العمل مع الرؤية والعمر الافتراضي تلقائيًا.
ميلفوس كواجهة خلفية للذاكرة في ADK
في ADK، تعد MemoryService في ADK مجرد واجهة. وهي تحدد كيفية استخدام الذاكرة طويلة المدى، ولكن ليس كيفية تخزينها. اختيار قاعدة البيانات متروك للمطور. فما نوع قاعدة البيانات التي تعمل بشكل جيد كواجهة خلفية لذاكرة الوكيل؟
ما يحتاجه نظام ذاكرة الوكيل - وكيف يقدمه ميلفوس
- الاسترجاع الدلالي
الحاجة:
نادرًا ما يطرح المستخدمون نفس السؤال بنفس الطريقة. "لا يتصل" و"مهلة الاتصال" تعنيان نفس الشيء. يجب أن يفهم نظام الذاكرة المعنى، وليس فقط مطابقة الكلمات الرئيسية.
كيف يلبي ميلفوس ذلك:
يدعم Milvus العديد من أنواع فهارس المتجهات، مثل HNSW و DiskANN، مما يسمح للمطورين باختيار ما يناسب عبء العمل الخاص بهم. حتى مع وجود عشرات الملايين من المتجهات، يمكن أن يظل زمن انتقال الاستعلام أقل من 10 مللي ثانية، وهو سريع بما يكفي لاستخدام الوكيل.
- الاستعلامات الهجينة
الحاجة
يتطلب استدعاء الذاكرة أكثر من البحث الدلالي. يجب على النظام أيضًا التصفية حسب الحقول المهيكلة مثل user_id بحيث يتم إرجاع بيانات المستخدم الحالي فقط.
كيف يلبي ميلفوس ذلك:
يدعم ميلفوس أصلاً الاستعلامات المختلطة التي تجمع بين البحث المتجه والتصفية العددية. على سبيل المثال، يمكنه استرداد سجلات متشابهة دلاليًا أثناء تطبيق عامل تصفية مثل user_id = 'xxx' في نفس الاستعلام، دون الإضرار بالأداء أو جودة الاستدعاء.
- قابلية التوسع
الحاجة:
مع تزايد عدد المستخدمين والذكريات المخزنة، يجب أن يتوسع النظام بسلاسة. يجب أن يظل الأداء مستقرًا مع زيادة البيانات، دون حدوث تباطؤ أو أعطال مفاجئة.
كيف يلبي ميلفوس هذه الحاجة:
يستخدم ميلفوس بنية فصل بين الحوسبة والتخزين. يمكن توسيع سعة الاستعلام أفقياً عن طريق إضافة عقد الاستعلام حسب الحاجة. حتى النسخة المستقلة، التي تعمل على جهاز واحد، يمكنها التعامل مع عشرات الملايين من المتجهات، مما يجعلها مناسبة لعمليات النشر في المراحل المبكرة.
ملاحظة: بالنسبة للتطوير والاختبار المحليين، تستخدم الأمثلة في هذه المقالة Milvus Lite أو Milvus Standalone.
بناء وكيل مع الذاكرة طويلة الأمد مدعوم من ميلفوس
في هذا القسم، نقوم ببناء وكيل دعم فني بسيط. عندما يطرح أحد المستخدمين سؤالاً، يبحث الوكيل عن تذاكر دعم سابقة مشابهة للإجابة عليها، بدلاً من تكرار نفس العمل.
هذا المثال مفيد لأنه يوضح ثلاث مشاكل شائعة يجب على أنظمة ذاكرة الوكيل الحقيقية التعامل معها.
- الذاكرة طويلة المدى عبر الجلسات
قد يتعلق السؤال المطروح اليوم بتذكرة تم إنشاؤها منذ أسابيع. يجب على الوكيل أن يتذكر المعلومات عبر المحادثات، وليس فقط داخل الجلسة الحالية. هذا هو السبب في الحاجة إلى ذاكرة طويلة الأمد، تتم إدارتها من خلال MemoryService.
- عزل المستخدم
يجب أن يبقى سجل دعم كل مستخدم خاصاً. يجب ألا تظهر البيانات من مستخدم واحد في نتائج مستخدم آخر. يتطلب هذا تصفية حقول مثل user_id، وهو ما يدعمه ميلفوس عبر الاستعلامات المختلطة.
- المطابقة الدلالية
يصف المستخدمون نفس المشكلة بطرق مختلفة، مثل "لا يمكن الاتصال" أو "مهلة". مطابقة الكلمات الرئيسية ليست كافية. يحتاج الوكيل إلى بحث دلالي، وهو ما يوفره استرجاع المتجهات.
إعداد البيئة
- بايثون 3.11+
- Docker و Docker Compose
- مفتاح واجهة برمجة تطبيقات الجوزاء
يغطي هذا القسم الإعداد الأساسي للتأكد من إمكانية تشغيل البرنامج بشكل صحيح.
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
الخطوة 1: نشر ميلفوس مستقل (Docker)
(1) قم بتنزيل ملفات النشر
wget <https://github.com/Milvus-io/Milvus/releases/download/v2.5.12/Milvus-standalone-docker-compose.yml> -O docker-compose.yml
(2) ابدأ تشغيل خدمة Milvus
docker-compose up -d
docker-compose ps -a
الخطوة 2: تكوين النموذج والاتصال
تكوين واجهة برمجة تطبيقات الجوزاء وإعدادات اتصال 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})")
الخطوة 3 تهيئة قاعدة بيانات Milvus
إنشاء مجموعة قاعدة بيانات متجهة (مشابهة لجدول في قاعدة بيانات علائقية)
# ==================== 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()
الخطوة 4 وظائف تشغيل الذاكرة
قم بتغليف منطق التخزين والاسترجاع كأدوات للوكيل.
(1) دالة تخزين الذاكرة
# ==================== 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) دالة استرجاع الذاكرة
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) التسجيل كأداة 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]
الخطوة 5 تعريف الوكيل
الفكرة الأساسية: تحديد منطق سلوك الوكيل.
# ==================== 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
)
الخطوة 6 البرنامج الرئيسي وتدفق التنفيذ
يوضح العملية الكاملة لاسترجاع الذاكرة عبر الجلسات.
# ==================== 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
الخطوة 7 التشغيل والاختبار
(1) تعيين متغيرات البيئة
export GOOGLE_API_KEY="your-gemini-api-key"
python milvus_agent.py
المخرجات المتوقعة
يوضح الناتج كيفية عمل نظام الذاكرة في الاستخدام الحقيقي.
في المحادثة الأولى، يسأل المستخدم عن كيفية التعامل مع مهلة اتصال ميلفوس. يقدم الوكيل حلاً. بعد أن يؤكد المستخدم حل المشكلة، يقوم الوكيل بحفظ هذا السؤال والإجابة في الذاكرة.
في المحادثة الثانية، تبدأ جلسة جديدة. يسأل المستخدم نفس السؤال باستخدام كلمات مختلفة: "ميلفوس لا يمكنه الاتصال". يسترجع الوكيل تلقائياً حالة مشابهة من الذاكرة ويعطي نفس الحل.
لا حاجة لخطوات يدوية. يقرّر الوكيل متى يسترجع الحالات السابقة ومتى يخزّن الحالات الجديدة، مما يُظهر ثلاث قدرات رئيسية: الذاكرة العابرة للجلسات، والمطابقة الدلالية، وعزل المستخدم.
الخاتمة
تفصل ADK بين السياق قصير المدى والذاكرة طويلة المدى على مستوى إطار العمل باستخدام SessionService و MemoryService. يتعامل Milvus مع البحث الدلالي والتصفية على مستوى المستخدم من خلال الاسترجاع القائم على المتجهات.
عند اختيار إطار عمل، فإن الهدف مهم. إذا كنت بحاجة إلى عزل قوي للحالة، وقابلية التدقيق، واستقرار الإنتاج، فإن ADK هو الأنسب. أما إذا كنت تعمل على وضع نماذج أولية أو تجري تجارب، فإن LangChain (إطار عمل بايثون شائع لبناء التطبيقات والوكلاء المستند إلى LLM بسرعة) يوفر مرونة أكبر.
بالنسبة لذاكرة الوكيل، القطعة الأساسية هي قاعدة البيانات. تعتمد الذاكرة الدلالية على قواعد البيانات المتجهة، بغض النظر عن إطار العمل الذي تستخدمه. يعمل برنامج Milvus بشكل جيد لأنه مفتوح المصدر، ويعمل محليًا، ويدعم التعامل مع مليارات المتجهات على جهاز واحد، ويدعم البحث الهجين المتجه والقياسي والبحث في النص الكامل. تغطي هذه الميزات كلاً من الاختبار المبكر واستخدام الإنتاج.
نأمل أن تساعدك هذه المقالة على فهم تصميم ذاكرة الوكيل بشكل أفضل واختيار الأدوات المناسبة لمشاريعك.
إذا كنت تقوم ببناء وكلاء ذكاء اصطناعي يحتاجون إلى ذاكرة حقيقية - وليس فقط نوافذ سياق أكبر - نود أن نسمع كيف تتعامل مع الأمر.
هل لديك أسئلة حول ADK، أو تصميم ذاكرة الوكيل، أو استخدام Milvus كواجهة خلفية للذاكرة؟ انضم إلى قناة Slack الخاصة بنا أو احجز جلسة ساعات عمل Milvus المكتبية لمدة 20 دقيقة للتحدث عن حالة استخدامك.
Try Managed Milvus for Free
Zilliz Cloud is hassle-free, powered by Milvus and 10x faster.
Get StartedLike the article? Spread the word



