إنشاء وكلاء مخصّصين ومحتفظين بالحالة باستخدام "حزمة تطوير الوكلاء"

1. مقدمة

title

مشكلة "السمكة الذهبية"

تخيّل أنّك استعنت بوكيل سفر لتخطيط عطلة أحلامك في طوكيو. استخدِم وكيل الجلسة لمشاهدة "مشكلة السمكة الذهبية" أثناء العمل.

تدخل إلى مكتبه وتقول:

"مرحبًا! أريد التخطيط لرحلة مدتها يومان إلى طوكيو. أنا مهتم بالمواقع التاريخية والسوشي".

يردّ الوكيل بحماس:

"رائع! لقد خطّطت لزيارة القصر الإمبراطوري وتناول العشاء في مطعم سوشي في Sukiyabashi Jiro".

تبتسم وتقول:

"هذا يبدو مثاليًا! هل يمكنك إرسال برنامج الرحلة إليّ؟"

ينظر إليك الموظف بتعابير فارغة ويسألك:

"مرحبًا! كيف يمكنني مساعدتك في التخطيط لرحلة اليوم؟"

هذه هي "مشكلة السمكة الذهبية". بدون ذاكرة، يكون كل تفاعل عبارة عن صفحة فارغة. تتوفّر الذكاء الاصطناعي، فالمساعد يعرف كيفية التخطيط للرحلات، ولكن لا تتوفّر الاستمرارية. لكي يكون وكيل الذكاء الاصطناعي مفيدًا حقًا، يجب أن يتذكّر.

مهمتك اليوم

في ورشة العمل هذه، ستتمكّن من حلّ مشكلة "السمكة الذهبية" من خلال إنشاء "وكيل سفر" يتذكّر ويتعلّم ويتكيّف. ستنتقل من خلال 6 مستويات من ذاكرة الوكيل، ما يؤدي إلى إنشاء نظام لا يتصرف كروبوت محادثة بل كمساعد شخصي مخصّص.

المستوى

الفكرة

"القوة الخارقة"

المستوى 1

الجلسة والحالة

إجراء محادثة بدون نسيان أي تفاصيل

المستوى 2

حالة الوكيل المتعدد

مشاركة الملاحظات بين أعضاء الفريق

المستوى 3

الثبات

تذكُّر معلوماتك حتى بعد إعادة تشغيل النظام

المستوى 4

عمليات إعادة الاستدعاء

تعديل الذكرى بشكل مستقل تمامًا

المستوى 5

الأدوات المخصّصة

قراءة وكتابة الملفات الشخصية المنظَّمة للمستخدمين

المستوى 6

الذاكرة المتعدّدة الوسائط

"رؤية" الصور والفيديوهات وتذكُّرها

حزمة الذاكرة في ADK

قبل كتابة الرمز، دعونا نتعرّف على الأدوات التي سنستخدمها. توفّر حزمة تطوير وكيل Google (ADK) طريقة منظَّمة للتعامل مع الذاكرة:

  1. الجلسة: هي الحاوية التي تضمّ المحادثة. ويحتوي على سجلّ المحادثات السابقة.
  2. الحالة: "لوحة تجارب" تتضمّن مفتاحًا وقيمةً، ويتم إرفاقها بالجلسة. تستخدم البرامج الآلية هذه السمة لتخزين حقائق معيّنة (مثل destination="Tokyo").
  3. ‫MemoryService: مساحة تخزين طويلة الأمد هذا هو المكان الذي نحتفظ فيه بالبيانات إلى الأبد، مثل الإعدادات المفضّلة للمستخدم أو المستندات التي تم تحليلها.

2. إعداد

لتشغيل وكلاء الذكاء الاصطناعي، نحتاج إلى أمرين: مشروع على Google Cloud لتوفير الأساس.

الجزء الأول: تفعيل حساب الفوترة

  • للمطالبة بحساب الفوترة الذي يتضمّن رصيدًا بقيمة 5 دولار أمريكي، ستحتاج إليه عند نشر تطبيقك. تأكَّد من تسجيل الدخول إلى حسابك على Gmail.

الجزء الثاني: البيئة المفتوحة

  1. 👉 انقر على هذا الرابط للانتقال مباشرةً إلى محرّر Cloud Shell
  2. 👉 إذا طُلب منك منح الإذن في أي وقت اليوم، انقر على منح الإذن للمتابعة. انقر لتفويض Cloud Shell
  3. 👉 إذا لم تظهر نافذة Terminal في أسفل الشاشة، افتحها باتّباع الخطوات التالية:
    • انقر على عرض.
    • انقر على Terminalفتح نافذة طرفية جديدة في "محرِّر Cloud Shell".
  4. 👉💻 في نافذة الأوامر، تأكَّد من أنّك قد أثبتّ هويتك وأنّ المشروع مضبوط على رقم تعريف مشروعك باستخدام الأمر التالي:
    gcloud auth list
  5. 👉💻 استنسِخ مشروع bootstrap من GitHub:
    git clone https://github.com/cuppibla/memory_agent_starter
  6. 👉💻 شغِّل نص الإعداد البرمجي من دليل المشروع.
    cd ~/memory_agent_starter
./init.sh
    سيتولّى النص البرمجي بقية عملية الإعداد تلقائيًا.
  7. 👉💻 اضبط رقم تعريف المشروع المطلوب:
    gcloud config set project $(cat ~/project_id.txt) --quiet

الجزء الثالث: إعداد الأذونات

  1. 👉💻 فعِّل واجهات برمجة التطبيقات المطلوبة باستخدام الأمر التالي. قد يستغرق هذا بضع دقائق.
    gcloud services enable \
    cloudresourcemanager.googleapis.com \
    servicenetworking.googleapis.com \
    run.googleapis.com \
    aiplatform.googleapis.com \
    compute.googleapis.com
  2. 👉💻 امنح الأذونات اللازمة من خلال تنفيذ الأوامر التالية في الوحدة الطرفية:
    . ~/memory_agent_starter/set_env.sh

لاحظ أنّه تم إنشاء ملف .env لك. تعرض هذه الصفحة معلومات مشروعك.

3- The Foundation - Session & State

جدار الحماية المتتبِّع لحالة الاتصال

المفهوم: السياق هو الأهم

أبسط أشكال الذاكرة هو ذاكرة الجلسة. هذا ما يسمح للوكيل بمعرفة أنّ الضمير "هذا" في الجملة "أريد شراء هذا" يشير إلى الحذاء الذي كنت تتحدث عنه قبل 10 ثوانٍ.

في حزمة تطوير التطبيقات على Android، ندير ذلك باستخدام العنصر Session.

  • الأسلوب غير الاحتفاظ بالحالة: إنشاء جلسة جديدة لكل رسالة
  • الأسلوب المستند إلى الحالة: إنشاء جلسة واحدة وإعادة استخدامها للمحادثة بأكملها

الخطوة 1: فحص الوكيل

👉💻 في وحدة Cloud Shell الطرفية، افتح الملف في "محرِّر Cloud Shell" من خلال تنفيذ الأمر التالي:

cloudshell edit ~/memory_agent_starter/01_session_agent/agent.py

فتح "~/memory_agent_starter/01_session_agent/agent.py"

👉 ابحث عن التعليق # TODO: Create a root agent داخل الدالة agent.py.

استبدِل هذا السطر بالكامل بالرمز التالي:

root_agent = LlmAgent(
    name="multi_day_trip_agent",
    model="gemini-2.5-flash",
    description="Agent that progressively plans a multi-day trip, remembering previous days and adapting to user feedback.",
    instruction="""
    You are the "Adaptive Trip Planner" 🗺️ - an AI assistant that builds multi-day travel itineraries step-by-step.

    Your Defining Feature:
    You have short-term memory. You MUST refer back to our conversation to understand the trip's context, what has already been planned, and the user's preferences. If the user asks for a change, you must adapt the plan while keeping the unchanged parts consistent.

    Your Mission:
    1.  **Initiate**: Start by asking for the destination, trip duration, and interests.
    2.  **Plan Progressively**: Plan ONLY ONE DAY at a time. After presenting a plan, ask for confirmation.
    3.  **Handle Feedback**: If a user dislikes a suggestion (e.g., "I don't like museums"), acknowledge their feedback, and provide a *new, alternative* suggestion for that time slot that still fits the overall theme.
    4.  **Maintain Context**: For each new day, ensure the activities are unique and build logically on the previous days. Do not suggest the same things repeatedly.
    5.  **Final Output**: Return each day's itinerary in MARKDOWN format.
    """,
    tools=[google_search]
)

يطلب الأمر من النموذج اللغوي الكبير تذكُّر المعلومات، ولكن يجب أن يوفّر الرمز إمكانية تذكُّر المعلومات.

الخطوة 2: السيناريوهان

فتح "~/memory_agent_starter/01_session_agent/main.py"

👉 في وحدة Cloud Shell الطرفية، افتح الملف في "محرّر Cloud Shell" من خلال تنفيذ الأمر التالي:

cloudshell edit ~/memory_agent_starter/01_session_agent/main.py

افتح ~/memory_agent_starter/01_session_agent/main.py، وابحث عن التعليق # TODO: Create a runner with in memorysession service داخل الدالة main.py.

استبدِل هذا السطر بالكامل بالرمز التالي:

    runner = Runner(
        agent=agent,
        session_service=session_service,
        app_name=agent.name
    )

👉 ابحث عن التعليق # TODO: create a different session to test داخل الدالة main.py.

استبدِل هذا السطر بالكامل بالرمز التالي:

    tokyo_session_2 = await session_service.create_session(
        app_name=multi_day_agent.name,
        user_id=user_id
    )

اختبار الميزة

لدينا وظيفتان توضّحان الفرق بين الذاكرة "القصيرة" و "الطويلة".

السيناريو 1: الاحتفاظ بالحالة (الجلسة المشترَكة)

async def run_trip_same_session_scenario(session_service, user_id):
    # 1. Create ONE session
    trip_session = await session_service.create_session(...)

    # 2. Turn 1
    await run_agent_query(..., trip_session, ...)

    # 3. Turn 2 - REUSING the same session!
    # The agent can "see" Turn 1 because it's in the session history.
    await run_agent_query(..., trip_session, ...)

السيناريو 2: بلا حالة (جلسة جديدة في كل مرة)

async def run_trip_different_session_scenario(session_service, user_id):
    # Turn 1
    tokyo_session = await session_service.create_session(...)
    await run_agent_query(..., tokyo_session, ...)

    # Turn 2 - Creating a FREASH session
    # The agent has NO IDEA what happened in Turn 1.
    tokyo_session_2 = await session_service.create_session(...)
    await run_agent_query(..., tokyo_session_2, ...)

الخطوة 3: تشغيل "الوكيل"

لنطّلِع على الفرق عمليًا. شغِّل النص البرمجي:

👉💻 في سطر الأوامر، نفِّذ سطر الأوامر أدناه:

cd ~/memory_agent_starter
uv run python ~/memory_agent_starter/01_session_agent/main.py

المشهد 1: يتذكّر الوكيل إعداداتك المفضّلة من الرسالة الأولى ويعدّل الخطة في الرسالة الثانية.

السيناريو 2: في الجولة الثانية ("هل تتذكر ما أعجبني في الطعام؟")، يفشل الوكيل تمامًا لأنّها جلسة جديدة. وهي تعني فعليًا "لا أعرف ما تتحدث عنه".

نصيحة رئيسية

القاعدة رقم 1 في "الذاكرة": أعِد استخدام session.id دائمًا للحفاظ على سياق المحادثة. الكائن Session هو مخزن مؤقت للذاكرة القصيرة الأمد للوكيل.

4. الفريق - حالة الوكلاء المتعدّدين

الطعام

الفكرة: "لعبة الهاتف"

عندما يعمل عدّة وكلاء معًا، يكونون مثل زملاء يتبادلون مجلد ملفات. إذا كتب أحد الموظفين ملاحظة في المجلد، يجب أن يتمكّن الموظف التالي من قراءتها.

في "حزمة تطوير التطبيقات"، هذا "المجلد" هو الحالة.

  • الحالة هي قاموس ({"key": "value"}) مضمّن في الجلسة.
  • يمكن لأي وكيل في الجلسة القراءة من هذه السمة أو الكتابة إليها.

الخطوة 1: فحص سير العمل

👉💻 في وحدة Cloud Shell الطرفية، افتح الملف في "محرِّر Cloud Shell" من خلال تنفيذ الأمر التالي:

cloudshell edit ~/memory_agent_starter/02_multi_agent/agent.py

👉في الملف ~/memory_agent_starter/02_multi_agent/agent.py، ابحث عن التعليق # TODO: foodie agent.

استبدِل هذا السطر بالكامل بالرمز التالي:

foodie_agent = LlmAgent(
    name="foodie_agent",
    model="gemini-2.5-flash",
    tools=[google_search],
    instruction="""You are an expert food critic. Your goal is to find the best restaurant based on a user's request.

    When you recommend a place, you must output *only* the name of the establishment and nothing else.
    For example, if the best sushi is at 'Jin Sho', you should output only: Jin Sho
    """,
    output_key="destination"  # ADK will save the agent's final response to state['destination']
)

👉 ابحث عن التعليق # TODO: transportation agent داخل الدالة agent.py.

استبدِل هذا السطر بالكامل بالرمز التالي:

transportation_agent = LlmAgent(
    name="transportation_agent",
    model="gemini-2.5-flash",
    tools=[google_search],
    instruction="""You are a navigation assistant. Given a destination, provide clear directions.
    The user wants to go to: {destination}.

    Analyze the user's full original query to find their starting point.
    Then, provide clear directions from that starting point to {destination}.
    """,
)

👉 ابحث عن التعليق # TODO: root_agent داخل الدالة agent.py.

استبدِل هذا السطر بالكامل بالرمز التالي:

root_agent = SequentialAgent(
    name="find_and_navigate_agent",
    sub_agents=[foodie_agent, transportation_agent],
    description="A workflow that first finds a location and then provides directions to it."
)

لدينا الآن وكيلان يعملان بالتسلسل:

  1. Foodie Agent: للعثور على مطعم
  2. وكيل النقل: يقدّم اتجاهات للوصول إلى هذا المطعم.

ميزة "التسليم السحري": لاحظ كيف يسلّم foodie_agent العصا إلى transportation_agent.

foodie_agent = LlmAgent(
    # ...
    # CRITICAL: This tells ADK to save the agent's output to state['destination']
    output_key="destination"
)

transportation_agent = LlmAgent(
    # ...
    # CRITICAL: This injects state['destination'] into the prompt
    instruction="""
    The user wants to go to: {destination}.
    Provide clear directions...
    """,
)
  1. output_key="destination": يتم حفظ إجابة "وكيل الطعام" بكفاءة.
  2. {destination}: يقرأ "وكيل النقل" هذه الإجابة تلقائيًا.

(لا يلزم اتّخاذ أي إجراء) الخطوة 2: أداة التنسيق

فتح "02_multi_agent/main.py"

نستخدم SequentialAgent لتشغيلها بالترتيب.

# 1. Create a single session for the sequential agent
session = await session_service.create_session(...)

# 2. Run the query
# The SequentialAgent manages the state flow:
# Query -> Foodie -> state['destination'] -> Transportation -> Final Answer
await run_agent_query(root_agent, query, ...)

يرسل المستخدم طلبًا واحدًا:

"Find best sushi in Palo Alto and then tell me how to get there."

تعمل الوكلاء معًا للإجابة عن السؤال.

الخطوة 3: تشغيل الفريق

👉💻 في "وحدة Cloud Shell الطرفية"، نفِّذ سير العمل الذي يتضمّن عدة وكلاء:

cd ~/memory_agent_starter
uv run python ~/memory_agent_starter/02_multi_agent/main.py

ماذا يحدث؟

  1. Foodie Agent: يعثر على "جين شو" (أو ما شابه ذلك).
  2. ADK: يحفظ "جين شو" في state['destination'].
  3. وكيل النقل: يتلقّى "جين شو" في تعليماته.
  4. النتيجة: "للوصول إلى Jin Sho من محطة Caltrain، عليك السير في شارع University Ave..."

نصيحة رئيسية

القاعدة رقم 2 للذاكرة: استخدِم الحالة لنقل المعلومات المنظَّمة بين الوكلاء. استخدِم output_key للكتابة و{placeholders} للقراءة.

5- The Reboot - Persistence

جدار الحماية المتتبِّع لحالة الاتصال

المفهوم: "مشكلة إعادة التشغيل"

حتى الآن، كانت الذاكرة InMemory. إذا أوقفت النص البرمجي وأعدت تشغيله، سينسى الوكيل كل شيء. يشبه ذلك جهاز كمبيوتر يمحو محرك الأقراص الثابتة في كل مرة يتم إيقاف تشغيله.

لحلّ هذه المشكلة، نحتاج إلى الثبات. نستبدل InMemorySessionService بـ DatabaseSessionService.

الخطوة 1: تبديل قاعدة البيانات

👉💻 في وحدة Cloud Shell الطرفية، افتح الملف في "محرِّر Cloud Shell" من خلال تنفيذ الأمر التالي:

cloudshell edit ~/memory_agent_starter/03_persistent_agent/main.py

👉 في الملف ~/memory_agent_starter/03_persistent_agent/main.py، ابحث عن التعليق # TODO: Configuration for Persistent Sessions.

استبدِل هذا السطر بالكامل بالرمز التالي:

SESSIONS_DIR = Path(os.path.expanduser("~")) / ".adk_codelab" / "sessions"
os.makedirs(SESSIONS_DIR, exist_ok=True)
SESSION_DB_FILE = SESSIONS_DIR / "trip_planner.db"
SESSION_URL = f"sqlite:///{SESSION_DB_FILE}"

الآن، يتم حفظ كل جلسة وكل حدث في ملف SQLite.

الخطوة 2: استرجاع البيانات بين الجلسات

تتيح لك ميزة "استمرار المحادثة" ليس فقط استئناف محادثة، بل الاستفادة من المحادثات السابقة.

في الملف نفسه ~/memory_agent_starter/03_persistent_agent/main.py، اطّلِع على حالة الاختبار 3: استرجاع البيانات على مستوى الجلسات.

👉 حدِّد مكان التعليق # TODO: retrieve the previous session manually

استبدِل هذا السطر بالكامل بالرمز التالي:

    old_session = await session_service.get_session(
        app_name=root_agent.name, user_id="user_01", session_id=session_id
    )

👉 ابحث عن التعليق # TODO: Extract content from the OLD session داخل الدالة main.py.

استبدِل هذا السطر بالكامل بالرمز التالي:

                    previous_context += f"- {role}: {text}\n"

👉 ابحث عن التعليق # TODO: Manually inject the context to the query داخل الدالة main.py.

استبدِل هذا السطر بالكامل بالرمز التالي:

    query_3 = f"""
    {previous_context}

    I'm planning a new trip to Osaka this time. 
    Based on my previous preferences (above), what should I eat?
    """

يحاكي ذلك عودة المستخدم بعد أشهر. ولا يمكنك استرداد هذا السجلّ القديم إلا باستخدام قاعدة بيانات.

الخطوة 3: تجاوز إعادة التشغيل

👉💻 في الوحدة الطرفية، شغِّل النص البرمجي:

cd ~/memory_agent_starter
uv run python ~/memory_agent_starter/03_persistent_agent/main.py

ينشئ ملفًا باسم ~/memory_agent_starter/trip_planner.db. جرِّب ما يلي: شغِّل النص البرمجي مرتين.

  • في عملية التشغيل الثانية، ابحث عن "استئناف جلسة حالية".
  • سيتذكّر البرنامج السياق من عملية التشغيل الأولى لأنّه يتم تحميله من ملف قاعدة البيانات.

نصيحة رئيسية

القاعدة رقم 3 للذاكرة: استخدِم DatabaseSessionService في مرحلة الإنتاج. ويضمن ذلك استمرار محادثات المستخدمين عند إعادة تشغيل الخادم، كما يتيح تحليل السجلّ على المدى الطويل.

6. The Spy - Callbacks

جدار الحماية المتتبِّع لحالة الاتصال

في بعض الأحيان، عليك تعديل الذاكرة تلقائيًا استنادًا إلى ما يفعله الوكيل، وليس ما يقوله فقط. أنت بحاجة إلى "جاسوس" يراقب الوكيل ويدوّن الملاحظات.

في حزمة تطوير التطبيقات، يكون هذا العنصر هو دالة معاودة الاتصال. adk_callback

  • ‫after_tool_callback: دالة يتم تشغيلها في كل مرة يعمل فيها الوكيل.
  • ‫ToolContext: طريقة للكتابة إلى الحالة من داخل هذه الدالة

الخطوة 1: المنطق

👉💻 في وحدة Cloud Shell الطرفية، افتح الملف في "محرِّر Cloud Shell" من خلال تنفيذ الأمر التالي:

cloudshell edit ~/memory_agent_starter/04_stateful_agent/agent.py

👉 في الملف ~/memory_agent_starter/04_stateful_agent/agent.py، ابحث عن التعليق # TODO: Implement call back logic

استبدِل هذا السطر بالكامل بالرمز التالي:

def save_activity_type_callback(
    tool,
    args: Dict[str, Any],
    tool_context: ToolContext,
    tool_response: Dict[str, Any],
) -> Optional[Dict[str, Any]]:
    """
    Callback to save the TYPE of activity just planned into the session state.
    """
    # 1. Get the actual agent name.
    if tool.name == "transfer_to_agent":
         agent_name = args.get("agent_name")
    else:
         agent_name = tool.name

    activity_type = "unknown"

    # 2. Determine the type based on which agent was actually used
    if agent_name == "museum_expert":
        activity_type = "CULTURAL"
    elif agent_name == "restaurant_expert":
        activity_type = "FOOD"
    elif agent_name == "outdoor_expert":
        activity_type = "OUTDOOR"

    print(f"\n🔔 [CALLBACK] The planner transferred to '{agent_name}'.")

    # 3. Update the state directly
    tool_context.state["last_activity_type"] = activity_type
    print(f"💾 [STATE UPDATE] 'last_activity_type' is now set to: {activity_type}\n")

    return tool_response

👉 في الملف نفسه، ابحث عن التعليق # TODO: add callback to root agent داخل الدالة 04_stateful_agent/agent.py.

استبدِل هذا السطر بالكامل بالرمز التالي:

    after_tool_callback=save_activity_type_callback,

التعليمات الديناميكية: أصبحت تعليمات الوكيل الآن دالة، وليست سلسلة. ويتغيّر بناءً على الحالة.

def get_planner_instruction(context):
    last_activity = context.state.get("last_activity_type", "None")
    
    return f"""
    The last activity was: {last_activity}
    
    If last_activity is 'CULTURAL' -> `museum_expert` is BANNED.
    """

الخطوة 3: اختبار Spy

👉💻 في الوحدة الطرفية، شغِّل النص البرمجي عن طريق نسخ الأمر أدناه ولصقه:

cd ~/memory_agent_starter
uv run python ~/memory_agent_starter/04_stateful_agent/main.py

عند تشغيل هذا الوكيل، ستظهر لك حلقة.

  1. الجولة 1: تطلب من "خرائط Google" البحث عن متحف. يضبط الجاسوس last_activity="CULTURAL".
  2. الجولة 2: تطلب متحفًا آخر.
  3. تعديلات على تعليمات الوكيل: "تم حظر CULTURAL".
  4. يقول الوكيل: "لا يمكنني البحث عن متحف آخر. ما رأيك في الذهاب إلى حديقة؟"

اطّلِع على سجلّات وحدة التحكّم بحثًا عن [CALLBACK] و[STATE UPDATE]. يمكنك الاطّلاع على الذاكرة التي تتغيّر في الوقت الفعلي أثناء عمل الوكيل.

نصيحة رئيسية

القاعدة رقم 4 من الذاكرة: استخدِم عمليات رد الاتصال لأتمتة إدارة الحالة. ينشئ الوكيل سياقه الخاص ببساطة من خلال تنفيذ مهمته.

7. خزانة الملفات - أدوات مخصّصة

المفهوم: "الذاكرة المنظَّمة"

جدار الحماية المتتبِّع لحالة الاتصال

حتى الآن، كانت "الذاكرة" عبارة عن سجلّ محادثات أو زوج بسيط من المفتاح والقيمة. ولكن ماذا لو كنت بحاجة إلى تذكُّر ملف مستخدم معقّد؟ على سبيل المثال: diet: vegan, budget: high, pets: [cat, dog].

لهذا السبب، نتعامل مع الذاكرة على أنّها أداة. يقرّر الوكيل بشكل صريح متى يفتح خزانة الملفات (قراءة) ومتى يقدّم تقريرًا (كتابة). مخطّط الأدوات المخصّصة

الخطوة 1: الأدوات

👉💻 في وحدة Cloud Shell الطرفية، افتح الملف في "محرِّر Cloud Shell" من خلال تنفيذ الأمر التالي:

cloudshell edit ~/memory_agent_starter/05_profile_agent/tools.py

👉 في هذا الملف: ~/memory_agent_starter/05_profile_agent/tools.py.

علينا تنفيذ الأداتَين المحدّدتين التاليتَين:

  1. save_user_preferences: تكتب في قاعدة بيانات.
  2. ‫recall_user_preferences: قراءة البيانات من قاعدة بيانات

ابحث عن التعليق # TODO: implement save_user_preferences tools داخل الدالة ~/memory_agent_starter/05_profile_agent/tools.py.

استبدِل هذا السطر بالكامل بالرمز التالي:

def save_user_preferences(tool_context: ToolContext, new_preferences: Dict[str, Any]) -> str:
    user_id = tool_context.session.user_id
    with sqlite3.connect(USER_DB_FILE) as conn:
        for key, value in new_preferences.items():
            conn.execute("INSERT INTO user_preferences (user_id, pref_key, pref_value) VALUES (?, ?, ?) ON CONFLICT(user_id, pref_key) DO UPDATE SET pref_value = excluded.pref_value;",
                         (user_id, key, json.dumps(value)))
    return f"Preferences updated: {list(new_preferences.keys())}"

👉 ابحث عن التعليق # TODO: implement recall_user_preferences tools داخل الدالة 05/tools.py.

استبدِل هذا السطر بالكامل بالرمز التالي:

def recall_user_preferences(tool_context: ToolContext) -> Dict[str, Any]:
    user_id = tool_context.session.user_id
    preferences = {}
    with sqlite3.connect(USER_DB_FILE) as conn:
        rows = conn.execute("SELECT pref_key, pref_value FROM user_preferences WHERE user_id = ?", (user_id,)).fetchall()
        if not rows: return {"message": "No preferences found."}
        for key, value_str in rows: preferences[key] = json.loads(value_str)
    return preferences

يفرض التعليمات سير عمل على النحو التالي:

instruction="""
1. RECALL FIRST: First action MUST be `recall_user_preferences`.
3. LEARN: If a user states a new preference, use `save_user_preferences`.
"""

الخطوة 2: التنفيذ

👉💻 في وحدة Cloud Shell الطرفية، افتح الملف في "محرِّر Cloud Shell" من خلال تنفيذ الأمر التالي:

cloudshell edit ~/memory_agent_starter/05_profile_agent/main.py

فتح "~/memory_agent_starter/05_profile_agent/main.py"

على عكس الوحدات السابقة التي كانت فيها حزمة تطوير التطبيقات (ADK) تتعامل مع الحالة تلقائيًا، يتحكّم الوكيل هنا في الحالة.

  • يتم اختيار الاتصال بـ recall_user_preferences في البداية.
  • يختار الاتصال بـ save_user_preferences عندما تقول "أنا نباتي".

الخطوة 3: إنشاء الملف الشخصي

👉💻 شغِّل النص البرمجي:

cd ~/memory_agent_starter
uv run python ~/memory_agent_starter/05_profile_agent/main.py

جرِّب مسار المحادثة التالي:

  1. "مرحبًا، أريد التخطيط لعشاء". -> يتحقّق الوكيل من قاعدة البيانات، ولا يعثر على أي نتائج. يطلب تحديد الإعدادات المفضّلة.
  2. "أنا نباتي صرف". -> يحفظ الوكيل "نباتي" في قاعدة البيانات.
  3. أعِد تشغيل النص البرمجي.
  4. "مرحبًا، أريد التخطيط لعشاء". -> يتحقّق الموظّف من قاعدة البيانات، ويرى كلمة "نباتي"، ويقترح مطعمًا نباتيًا على الفور.

نصيحة رئيسية

القاعدة رقم 5 من الذاكرة: بالنسبة إلى البيانات المعقّدة والمنظَّمة، امنح وكيلك أدوات القراءة والكتابة. السماح للنموذج اللغوي الكبير بإدارة مساحة التخزين الطويل الأمد الخاصة به

8. الدماغ - الذاكرة المتعدّدة الوسائط

جدار الحماية المتتبِّع لحالة الاتصال

الفكرة: "التجربة الإنسانية"

يتذكّر البشر أكثر من النص. نتذكّر الأجواء في صورة، والصوت في تسجيل صوتي، والشعور في فيديو.

تتيح Vertex AI Memory Bank للوكيل التعامل مع الذاكرة المتعدّدة الوسائط. يمكنه استيعاب الصور والفيديوهات والمحتوى الصوتي و"فهمها" واسترجاعها لاحقًا.

الخطوة 1: الإعداد

👉💻 في وحدة Cloud Shell الطرفية، افتح الملف في "محرِّر Cloud Shell" من خلال تنفيذ الأمر التالي:

cloudshell edit ~/memory_agent_starter/06_multimodal_agent/main.py

👉 افتح 06_multimodal_agent/main.py. ابحث عن التعليق # TODO: Configure Memory Bank Topic.

استبدِل هذا السطر بالكامل بالرمز التالي:

travel_topics = [
    MemoryTopic(
        managed_memory_topic=ManagedMemoryTopic(
            managed_topic_enum=ManagedTopicEnum.USER_PREFERENCES
        )
    ),
    MemoryTopic(
        managed_memory_topic=ManagedMemoryTopic(
            managed_topic_enum=ManagedTopicEnum.USER_PERSONAL_INFO
        )
    ),
    MemoryTopic(
        custom_memory_topic=CustomMemoryTopic(
            label="travel_experiences",
            description="""Memorable travel experiences including:
                - Places visited and impressions
                - Favorite restaurants, cafes, and food experiences
                - Preferred accommodation types and locations
                - Activities enjoyed (museums, hiking, beaches, etc.)
                - Travel companions and social preferences
                - Photos and videos from trips with location context""",
        )
    ),
    MemoryTopic(
        custom_memory_topic=CustomMemoryTopic(
            label="travel_preferences",
            description="""Travel style and preferences:
                - Budget preferences (luxury, mid-range, budget)
                - Transportation preferences (flying, trains, driving)
                - Trip duration preferences
                - Season and weather preferences
                - Cultural interests and language abilities
                - Dietary restrictions and food preferences""",
        )
    ),
    MemoryTopic(
        custom_memory_topic=CustomMemoryTopic(
            label="travel_logistics",
            description="""Practical travel information:
                - Passport and visa information
                - Frequent flyer numbers and hotel loyalty programs
                - Emergency contacts
                - Medical considerations and insurance
                - Packing preferences and essentials
                - Time zone preferences and jet lag strategies""",
        )
    ),
]

تحديد مكان التعليق # TODO: Configure Memory Bank Customization

استبدِل هذا السطر بالكامل بالرمز التالي:

memory_bank_config = {
    "customization_configs": [
        {
            "memory_topics": travel_topics,
        }
    ],
    "similarity_search_config": {
        "embedding_model": f"projects/{PROJECT_ID}/locations/{LOCATION}/publishers/google/models/gemini-embedding-001"
    },
    "generation_config": {
        "model": f"projects/{PROJECT_ID}/locations/{LOCATION}/publishers/google/models/gemini-2.5-flash"
    },
}

الخطوة 2: استيعاب العالم

في test_trip_planner، نرسل:

  1. رسالة نصية ("مرحبًا")
  2. صورة (مَعلَم)
  3. فيديو (البحر الأبيض المتوسط)
  4. مقطع صوتي (ملاحظة صوتية حول مدينة غايتا)

ابحث عن التعليق # TODO create session service and memory service داخل الدالة 6_multimodal_agent/main.py.

استبدِل هذا السطر بالكامل بالرمز التالي:

session_service = VertexAiSessionService(
    project=PROJECT_ID, location=LOCATION, agent_engine_id=agent_engine_id
)
memory_service = VertexAiMemoryBankService(
    project=PROJECT_ID, location=LOCATION, agent_engine_id=agent_engine_id
)

👉 في الملف نفسه 06_multimodal_agent/main.py، ابحث عن التعليق # TODO: create memory from session

استبدِل هذا السطر بالكامل بالرمز التالي:

    await memory_service.add_session_to_memory(final_session_state)

هذا هو الخط السحري. ترسل هذه الأداة جميع الوسائط الغنية إلى Vertex AI، التي تعالجها وتفهرسها.

الخطوة 3: الاسترداد

👉💻 في وحدة Cloud Shell الطرفية، افتح الملف في "محرِّر Cloud Shell" من خلال تنفيذ الأمر التالي:

cloudshell edit ~/memory_agent_starter/06_multimodal_agent/agent.py

يحتوي الوكيل على PreloadMemoryTool.

tools=[PreloadMemoryTool(), budget_tool]

عند بدء جلسة جديدة، تبحث هذه الأداة تلقائيًا في "بنك الذاكرة" عن التجارب السابقة ذات الصلة وتدرجها في السياق.

الخطوة 4: تشغيل "الدماغ"

👉💻 في وحدة Cloud Shell الطرفية، شغِّل النص البرمجي (ملاحظة: يتطلّب ذلك مشروع Google Cloud تم تفعيل Vertex AI فيه):

cd ~/memory_agent_starter
uv run python ~/memory_agent_starter/06_multimodal_agent/main.py

شاهِد خطوة إثبات الملكية النهائية:

"استنادًا إلى الصورة والفيديو والصوت الذين شاركتُهم معك من قبل..."

سيردّ الوكيل بما يلي:

"أنصحك بزيارة غايتا. لقد عرضتَ لي فيديو للبحر الأبيض المتوسط ومقطعًا صوتيًا قلتَ فيه إنّك تحبّ غايتا".

وقد ربطت هذه الميزة بين أنواع مختلفة من الوسائط من الماضي.

نصيحة رئيسية

القاعدة رقم 6 من Memory: استخدِم Vertex AI Memory Bank للحصول على أفضل تجربة ذاكرة. فهو يوحّد النصوص والصور والفيديوهات في قاعدة بيانات واحدة قابلة للبحث.

9- الخاتمة

لقد تطوّرت من سمكة ذهبية تنسى بسرعة إلى فيل متعدد الوسائط.

You Built

الإمكانية

وكيل الجلسة

ذاكرة المحادثات القصيرة الأجل

Multi-Agent

الذكريات المشتركة بين الفريق

الوكيل الدائم

السجلّ الطويل الأمد

الوكيل ذو الحالة

ذاكرة ديناميكية يتم تعديلها تلقائيًا

Profile Agent

ذاكرة البيانات المنظَّمة

الوكيل المتعدّد الوسائط

الذاكرة الحسية الشبيهة بالذاكرة البشرية

الثقة مبنية على الذاكرة. من خلال تنفيذ هذه الأنماط، يمكنك إنشاء وكلاء يحترمون وقت المستخدم وسجلّه، ما يؤدي إلى تفاعلات أعمق وأكثر فعالية.

ابدأ بإنشاء وكلاء مخصّصين اليوم.