‫🤖 إنشاء وكيل ذكاء اصطناعي متعدد الوسائط باستخدام Graph RAG وADK وMemory Bank

1. مقدمة

أداء لأغنية تابعة لطرف ثالث

1. التحدّي

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

  1. 🗄️ قاعدة بيانات رسومية (Spanner): لتخزين العلاقات المعقّدة بين الناجين والمهارات والموارد
  2. 🔍 البحث المستند إلى الذكاء الاصطناعي: بحث مختلط دلالي ومستند إلى الكلمات الرئيسية باستخدام التضمينات
  3. 📸 المعالجة المتعددة الوسائط: استخراج البيانات المنظَّمة من الصور والنصوص والفيديوهات
  4. 🤖 تنظيم مهام الوكلاء المتعدّدين: تنسيق عمل الوكلاء المتخصّصين في مهام سير العمل المعقّدة
  5. 🧠 الذاكرة الطويلة الأمد: التخصيص باستخدام "بنك الذاكرة" في Vertex AI

محادثة

2. ما ستنشئه

قاعدة بيانات "الرسم البياني لشبكة الناجين" تتضمّن:

  • ‫🗺️ تصوّر تفاعلي ثلاثي الأبعاد للرسم البياني لعلاقات الناجين
  • 🔍 البحث الذكي (الكلمات الرئيسية والدلالي والمختلط)
  • 📸 مسار التحميل المتعدد الوسائط (استخراج الكيانات من الصور أو الفيديو)
  • 🤖 نظام متعدّد الوكلاء لتنظيم المهام المعقّدة
  • ‫🧠 دمج "سجلّ الذكريات" لتفاعلات مخصّصة

3- التقنيات الأساسية

المكوّن

تكنولوجيا

الغرض

قاعدة البيانات

Cloud Spanner Graph

تخزين العُقد (الناجون والمهارات) والحواف (العلاقات)

AI Search

‫Gemini + Embeddings

الفهم الدلالي + البحث عن التشابه

Agent Framework

ADK (Agent Development Kit)

تنظيم مهام سير عمل الذكاء الاصطناعي

الذاكرة

‫Vertex AI Memory Bank

تخزين الإعدادات المفضّلة للمستخدم على المدى الطويل

Frontend

React + Three.js

تمثيل بصري تفاعلي ثلاثي الأبعاد للرسومات البيانية

2. إعداد البيئة (يمكنك تخطّي هذه الخطوة إذا كنت في ورشة عمل)

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

  • للمطالبة بحساب الفوترة الذي يتضمّن رصيدًا بقيمة 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/google-americas/way-back-home.git

3- إعداد البيئة

1. إعداد

في نافذة محرِّر Cloud Shell، إذا لم تظهر النافذة الطرفية في أسفل الشاشة، افتحها باتّباع الخطوات التالية:

  • انقر على عرض.
  • انقر على Terminal.

فتح نافذة طرفية جديدة في "محرِّر Cloud Shell"

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

cd ~/way-back-home/level_2
./init.sh

2. ضبط إعدادات المشروع

👉💻 اضبط معرّف مشروعك:

gcloud config set project $(cat ~/project_id.txt) --quiet

👉💻 فعِّل واجهات برمجة التطبيقات المطلوبة (يستغرق ذلك من دقيقتَين إلى 3 دقائق تقريبًا):

gcloud services enable compute.googleapis.com \
                       aiplatform.googleapis.com \
                       run.googleapis.com \
                       cloudbuild.googleapis.com \
                       artifactregistry.googleapis.com \
                       spanner.googleapis.com \
                       storage.googleapis.com

3- تشغيل نص الإعداد البرمجي

👉💻 نفِّذ نص الإعداد البرمجي:

cd ~/way-back-home/level_2
./setup.sh

سيؤدي ذلك إلى إنشاء .env لك. في Cloud Shell، افتح way_back_homeproject. ضمن المجلد level_2، يمكنك الاطّلاع على الملف .env الذي تم إنشاؤه لك. إذا لم تتمكّن من العثور عليه، يمكنك النقر على View -> Toggle Hidden File للاطّلاع عليه. open_project

4. تحميل نموذج البيانات

👉💻 انتقِل إلى الخلفية وثبِّت التبعيات:

cd ~/way-back-home/level_2/backend
uv sync

👉💻 تحميل بيانات الناجين الأولية:

uv run python ~/way-back-home/level_2/backend/setup_data.py

يؤدي ذلك إلى إنشاء ما يلي:

  • مثيل Spanner (survivor-network)
  • قاعدة البيانات (graph-db)
  • جميع جداول العُقد والحواف
  • الرسومات البيانية الخاصة بالسمات لطلب البحث عن الناتج المتوقَّع:
============================================================
SUCCESS! Database setup complete.
============================================================

Instance:  survivor-network
Database:  graph-db
Graph:     SurvivorGraph

Access your database at:
https://console.cloud.google.com/spanner/instances/survivor-network/databases/graph-db?project=waybackhome

إذا نقرت على الرابط بعد Access your database at في الناتج، يمكنك فتح Google Cloud Console Spanner.

open_spanner

وسيظهر لك Spanner في Google Cloud Console.

spanner

4. عرض بيانات الرسم البياني بشكل مرئي في Spanner Studio

يساعدك هذا الدليل في عرض بيانات الرسم البياني لشبكة الناجين والتفاعل معها مباشرةً في Google Cloud Console باستخدام Spanner Studio. هذه طريقة رائعة للتحقّق من بياناتك وفهم بنية الرسم البياني قبل إنشاء وكيل الذكاء الاصطناعي.

1. الوصول إلى Spanner Studio

  1. في الخطوة الأخيرة، احرص على النقر على الرابط وفتح Spanner Studio.

spanner_studio

2. فهم بنية الرسم البياني ("الصورة الكبيرة")

يمكنك اعتبار مجموعة بيانات "شبكة الناجين" بمثابة لغز منطقي أو حالة لعبة:

الكيان

الدور في النظام

التشبيه

Survivors

الوكلاء/اللاعبون

اللاعبون

المناطق الأحيائية

موقعها الجغرافي

مناطق الخريطة

المهارات

الإجراءات التي يمكنهم اتّخاذها

الإمكانات

الاحتياجات

النقاط السلبية (الأزمات)

المهام

المراجع

العناصر التي تم العثور عليها في العالم

غنيمة

الهدف: مهمة وكيل الذكاء الاصطناعي هي ربط المهارات (الحلول) بالاحتياجات (المشاكل)، مع الأخذ في الاعتبار المناطق الأحيائية (قيود الموقع الجغرافي).

🔗 الحواف (العلاقات):

  • ‫SurvivorInBiome: تتبُّع الموقع الجغرافي
  • ‫SurvivorHasSkill: مستودع القدرات
  • SurvivorHasNeed: قائمة بالمشاكل النشطة
  • ‫SurvivorFoundResource: مستودع السلع
  • SurvivorCanHelp: علاقة مستنتَجة (يحسبها الذكاء الاصطناعي)

3- الاستعلام عن الرسم البياني

لننفّذ بعض طلبات البحث للاطّلاع على "القصة" في البيانات.

يستخدم Spanner Graph لغة طلبات الرسم البياني (GQL). لتنفيذ طلب بحث، استخدِم GRAPH SurvivorNetwork متبوعًا بنمط المطابقة.

👉 طلب البحث 1: قائمة المستخدمين العالمية (مَن أين؟) هذه هي المعلومات الأساسية، إذ إنّ فهم الموقع الجغرافي أمر بالغ الأهمية لعمليات الإنقاذ.

GRAPH SurvivorNetwork
MATCH result = (s:Survivors)-[:SurvivorInBiome]->(b:Biomes)
RETURN TO_JSON(result) AS json_result

من المتوقّع أن تظهر النتيجة كما يلي: query1

👉 طلب البحث 2: مصفوفة المهارات (الإمكانات) بعد أن عرفت مكان كل شخص، يمكنك الآن معرفة ما يمكنه فعله.

GRAPH SurvivorNetwork
MATCH result = (s:Survivors)-[h:SurvivorHasSkill]->(k:Skills)
RETURN TO_JSON(result) AS json_result

من المتوقّع أن تظهر النتيجة كما يلي: query2

👉 طلب البحث 3: مَن هم الأشخاص الذين يواجهون أزمة؟ (لوحة المهام) يمكنك الاطّلاع على الناجين الذين يحتاجون إلى المساعدة ونوع المساعدة التي يحتاجون إليها.

GRAPH SurvivorNetwork
MATCH result = (s:Survivors)-[h:SurvivorHasNeed]->(n:Needs)
RETURN TO_JSON(result) AS json_result

من المتوقّع أن تظهر النتيجة كما يلي: query3

🔎 مستوى متقدّم: التوفيق بين الأشخاص - مَن يمكنه مساعدة مَن؟

هنا يصبح الرسم البياني قويًا. يبحث هذا الاستعلام عن الناجين الذين لديهم مهارات يمكنها تلبية احتياجات الناجين الآخرين.

GRAPH SurvivorNetwork
MATCH result = (helper:Survivors)-[:SurvivorHasSkill]->(skill:Skills)-[:SkillTreatsNeed]->(need:Needs)<-[:SurvivorHasNeed]-(helpee:Survivors)
RETURN TO_JSON(result) AS json_result

من المتوقّع أن تظهر النتيجة كما يلي: query4

aside positive الغرض من طلب البحث هذا:

بدلاً من عرض "الإسعافات الأولية تعالج الحروق" فقط (وهو أمر واضح من المخطط)، يعثر طلب البحث هذا على:

  • الدكتورة "إيلينا فروست" (التي لديها تدريب طبي) → يمكنها علاج → الكابتن "تاناكا" (الذي يعاني من حروق)
  • ديفيد تشين (الذي لديه مجموعة الإسعافات الأولية) → يمكنه علاج → الملازم بارك (الذي يعاني من التواء في الكاحل)

أهمية هذه الميزة:

المهام التي سينفّذها وكيل الذكاء الاصطناعي:

عندما يطرح مستخدم السؤال "من يمكنه معالجة الحروق؟"، سيفعل الوكيل ما يلي:

  1. تنفيذ طلب بحث مشابه للرسم البياني
  2. النتيجة: "يملك الدكتور فروست تدريبًا طبيًا ويمكنه مساعدة الكابتن تاناكا"
  3. ولا يحتاج المستخدم إلى معرفة الجداول أو العلاقات الوسيطة.

5- التضمينات المستندة إلى الذكاء الاصطناعي في Spanner

1. لماذا التضمينات؟ (بدون اتّخاذ أي إجراء، للقراءة فقط)

في سيناريو البقاء على قيد الحياة، الوقت مهم جدًا. عندما يبلغ أحد الناجين عن حالة طوارئ، مثل I need someone who can treat burns أو Looking for a medic، لا يمكنه إضاعة الوقت في تخمين أسماء المهارات الدقيقة في قاعدة البيانات.

السيناريو الحقيقي: Survivor: Captain Tanaka has burns—we need medical help NOW!

البحث التقليدي عن الكلمات الرئيسية "مسعف" → 0 نتيجة ❌

البحث الدلالي باستخدام التضمينات → يعثر على "التدريب الطبي" و"الإسعافات الأولية" ✅

وهذا هو بالضبط ما تحتاجه الوكالات: بحث ذكي يشبه البحث الذي يجريه الإنسان ويفهم النية، وليس الكلمات الرئيسية فقط.

2. إنشاء نموذج تضمين

spanner_embedding

لننشئ الآن نموذجًا يحوّل النص إلى تضمينات باستخدام text-embedding-004 من Google.

👉 في Spanner Studio، شغِّل SQL هذا (استبدِل $YOUR_PROJECT_ID برقم تعريف مشروعك الفعلي):

‼️ في محرّر Cloud Shell، افتح File -> Open Folder -> way-back-home/level_2 للاطّلاع على المشروع بأكمله.

project_id

👉 شغِّل هذا الاستعلام في Spanner Studio من خلال نسخ الاستعلام أدناه ولصقه، ثم النقر على الزر "تشغيل":

CREATE MODEL TextEmbeddings
INPUT(content STRING(MAX))
OUTPUT(embeddings STRUCT<values ARRAY<FLOAT32>>)
REMOTE OPTIONS (
    endpoint = '//aiplatform.googleapis.com/projects/$YOUR_PROJECT_ID/locations/us-central1/publishers/google/models/text-embedding-004'
);

وظيفة هذا الإجراء:

  • إنشاء نموذج افتراضي في Spanner (بدون تخزين أوزان النموذج محليًا)
  • نقاط إلى text-embedding-004 من Google على Vertex AI
  • تحديد العقد: الإدخال هو نص، والإخراج هو مصفوفة عائمة ذات 768 بُعدًا

لماذا "خيارات عن بُعد"؟

  • لا ينفّذ Spanner النموذج نفسه
  • يتم استدعاء Vertex AI من خلال واجهة برمجة التطبيقات عند استخدام ML.PREDICT
  • Zero-ETL: لا حاجة إلى تصدير البيانات إلى Python ومعالجتها وإعادة استيرادها

انقر على الزر Run، وبعد نجاح العملية، يمكنك الاطّلاع على النتيجة كما هو موضح أدناه:

spanner_result

3- إضافة عمود التضمين

👉 إضافة عمود لتخزين التضمينات:

ALTER TABLE Skills ADD COLUMN skill_embedding ARRAY<FLOAT32>;

انقر على الزر Run، وبعد نجاح العملية، يمكنك الاطّلاع على النتيجة كما هو موضح أدناه:

embedding_result

4. إنشاء تضمينات

👉 استخدِم الذكاء الاصطناعي لإنشاء تضمينات متجهة لكل مهارة:

UPDATE Skills
SET skill_embedding = (
    SELECT embeddings.values
    FROM ML.PREDICT(
        MODEL TextEmbeddings,
        (SELECT name AS content)
    )
)
WHERE skill_embedding IS NULL;

انقر على الزر Run، وبعد نجاح العملية، يمكنك الاطّلاع على النتيجة كما هو موضح أدناه:

skills_result

ما يحدث: كل اسم مهارة (مثل "الإسعافات الأولية") إلى متّجه ذي 768 سمة يمثّل معناه الدلالي.

5- التحقّق من صحة عمليات التضمين

👉 تأكَّد من إنشاء عمليات التضمين:

SELECT 
    skill_id,
    name,
    ARRAY_LENGTH(skill_embedding) AS embedding_dimensions
FROM Skills
LIMIT 5;

الناتج المتوقَّع:

spanner_result

الآن، نختبر حالة الاستخدام المحدّدة من السيناريو: العثور على مهارات طبية باستخدام المصطلح "طبيب".

👉 البحث عن مهارات مشابهة لـ "مسعف":

WITH query_embedding AS (
    SELECT embeddings.values AS val
    FROM ML.PREDICT(MODEL TextEmbeddings, (SELECT "medic" AS content))
)
SELECT
    s.name AS skill_name,
    s.category,
    COSINE_DISTANCE(s.skill_embedding, (SELECT val FROM query_embedding)) AS distance
FROM Skills AS s
WHERE s.skill_embedding IS NOT NULL
ORDER BY distance ASC
LIMIT 10;
  • تحويل عبارة البحث "طبيب" التي أدخلها المستخدم إلى تضمين
  • يخزّنها في الجدول المؤقت query_embedding

النتائج المتوقّعة (المسافة الأقصر تعني التشابه الأكبر):

spanner_result

7. إنشاء نموذج Gemini للتحليل

spanner_gemini

👉 أنشئ مرجعًا لنموذج الذكاء الاصطناعي التوليدي (استبدِل $YOUR_PROJECT_ID برقم تعريف مشروعك الفعلي):

CREATE MODEL GeminiPro
INPUT(prompt STRING(MAX))
OUTPUT(content STRING(MAX))
REMOTE OPTIONS (
    endpoint = '//aiplatform.googleapis.com/projects/$YOUR_PROJECT_ID/locations/us-central1/publishers/google/models/gemini-2.5-pro',
    default_batch_size = 1
);

الاختلاف عن "نموذج التضمينات":

  • التضمينات: نص → متّجه (للبحث عن التشابه)
  • Gemini: نص → نص تم إنشاؤه (للاستدلال/التحليل)

spanner_result

8. استخدام Gemini لتحليل التوافق

👉 تحليل أزواج الناجين من أجل التوافق مع المهمة:

WITH PairData AS (
    SELECT
        s1.name AS Name_A,
        s2.name AS Name_B,
        CONCAT(
            "Assess compatibility of these two survivors for a resource-gathering mission. ",
            "Survivor 1: ", s1.name, ". ",
            "Survivor 2: ", s2.name, ". ",
            "Give a score from 1-10 and a 1-sentence reason."
        ) AS prompt
    FROM Survivors s1
    JOIN Survivors s2 ON s1.survivor_id < s2.survivor_id
    LIMIT 1
)
SELECT
    Name_A,
    Name_B,
    content AS ai_assessment
FROM ML.PREDICT(
    MODEL GeminiPro,
    (SELECT Name_A, Name_B, prompt FROM PairData)
);

الناتج المتوقَّع:

Name_A          | Name_B            | ai_assessment
----------------|-------------------|----------------
"David Chen"    | "Dr. Elena Frost" | "**Score: 9/10** Their compatibility is extremely high as David's practical, hands-on scavenging skills are perfectly complemented by Dr. Frost's specialized knowledge to identify critical medical supplies and avoid biological hazards."

6. إنشاء وكيل Graph RAG باستخدام البحث المختلط

1. نظرة عامة على بنية النظام

يبني هذا القسم نظام بحث متعدد الطرق يمنح وكيلك المرونة اللازمة للتعامل مع أنواع مختلفة من طلبات البحث. يتضمّن النظام ثلاث طبقات: طبقة الوكيل وطبقة الأدوات وطبقة الخدمة.

architecture_hybrid_search

لماذا ثلاث طبقات؟

  • فصل الاهتمامات: يركّز الوكيل على الغرض، والأدوات على الواجهة، والخدمة على التنفيذ
  • المرونة: يمكن للوكيل فرض طرق معيّنة أو السماح للذكاء الاصطناعي بتحديد المسار تلقائيًا.
  • التحسين: يمكن تخطّي تحليل الذكاء الاصطناعي المكلف عندما تكون الطريقة معروفة

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

2. تنفيذ خدمة التوليد المعزّز بالاسترجاع (RAG)

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

cloudshell edit ~/way-back-home/level_2/backend/services/hybrid_search_service.py

تحديد مكان التعليق # TODO: REPLACE_SQL

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

        # This is your working query from the successful run!
        sql = """
            WITH query_embedding AS (
                SELECT embeddings.values AS val
                FROM ML.PREDICT(
                    MODEL TextEmbeddings,
                    (SELECT @query AS content)
                )
            )
            SELECT
                s.survivor_id,
                s.name AS survivor_name,
                s.biome,
                sk.skill_id,
                sk.name AS skill_name,
                sk.category,
                COSINE_DISTANCE(
                    sk.skill_embedding, 
                    (SELECT val FROM query_embedding)
                ) AS distance
            FROM Survivors s
            JOIN SurvivorHasSkill shs ON s.survivor_id = shs.survivor_id
            JOIN Skills sk ON shs.skill_id = sk.skill_id
            WHERE sk.skill_embedding IS NOT NULL
            ORDER BY distance ASC
            LIMIT @limit
        """

3- تعريف أداة البحث الدلالي

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

cloudshell edit ~/way-back-home/level_2/backend/agent/tools/hybrid_search_tools.py

في hybrid_search_tools.py، ابحث عن التعليق # TODO: REPLACE_SEMANTIC_SEARCH_TOOL

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

async def semantic_search(query: str, limit: int = 10) -> str:
    """
    Force semantic (RAG) search using embeddings.
    
    Use this when you specifically want to find things by MEANING,
    not just matching keywords. Great for:
    - Finding conceptually similar items
    - Handling vague or abstract queries
    - When exact terms are unknown
    
    Example: "healing abilities" will find "first aid", "surgery", 
    "herbalism" even though no keywords match exactly.
    
    Args:
        query: What you're looking for (describe the concept)
        limit: Maximum results
        
    Returns:
        Semantically similar results ranked by relevance
    """
    try:
        service = _get_service()
        result = service.smart_search(
            query, 
            force_method=SearchMethod.RAG,
            limit=limit
        )
        
        return _format_results(
            result["results"],
            result["analysis"],
            show_analysis=True
        )
        
    except Exception as e:
        return f"Error in semantic search: {str(e)}"

حالات استخدام الوكيل:

  • طلبات البحث التي تسأل عن التشابه ("البحث عن محتوى مشابه لـ X")
  • طلبات البحث المفاهيمية ("قدرات الشفاء")
  • عندما يكون فهم المعنى أمرًا بالغ الأهمية

4. دليل اتخاذ القرارات من الوكيل (التعليمات)

في تعريف الوكيل، انسخ الجزء المتعلّق بالبحث الدلالي والصقه في التعليمات.

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

cloudshell edit ~/way-back-home/level_2/backend/agent/agent.py

يستخدم الوكيل هذه التعليمات لاختيار الأداة المناسبة:

👉في ملف agent.py، ابحث عن التعليق # TODO: REPLACE_SEARCH_LOGIC، استبدِل هذا السطر بالكامل بالرمز التالي:

- `semantic_search`: Force RAG/embedding search
  Use for: "Find similar to X", conceptual queries, unknown terminology
  Example: "Find skills related to healing"

👉ابحث عن التعليق # TODO: ADD_SEARCH_TOOLReplace this whole line واستبدله بالرمز التالي:

    semantic_search,         # Force RAG

5- فهم طريقة عمل البحث المختلط (للقراءة فقط، لا يلزم اتّخاذ أي إجراء)

في الخطوات من 2 إلى 4، نفّذت البحث الدلالي (التوليد المعزّز بالاسترجاع)، وهو طريقة البحث الأساسية التي تعثر على النتائج حسب المعنى. ولكن ربما لاحظت أنّ اسم النظام هو "البحث المختلط". إليك كيفية عمل كل ذلك معًا:

طريقة عمل ميزة "الدمج المختلط":

في الملف way-back-home/level_2/backend/services/hybrid_search_service.py، عند طلب hybrid_search()، تُجري الخدمة عمليتَي البحث وتدمج النتائج:

# Location: backend/services/hybrid_search_service.py

    rank_kw = keyword_ranks.get(surv_id, float('inf'))
    rank_rag = rag_ranks.get(surv_id, float('inf'))

    rrf_score = 0.0
    if rank_kw != float('inf'):
        rrf_score += 1.0 / (K + rank_kw)
    if rank_rag != float('inf'):
        rrf_score += 1.0 / (K + rank_rag)

    combined_score = rrf_score

في هذا الدرس التطبيقي حول الترميز، نفّذت مكوّن البحث الدلالي (RAG)، وهو الأساس. تمّت إضافة طريقتَي الكلمات الرئيسية والطريقة المختلطة إلى الخدمة، ويمكن للموظف استخدام الطرق الثلاث.

تهانينا! لقد أتممت بنجاح إنشاء Graph RAG Agent باستخدام البحث المختلط.

7. اختبار الوكيل باستخدام ADK Web

أسهل طريقة لاختبار الوكيل هي استخدام الأمر adk web، الذي يشغّل الوكيل مع واجهة محادثة مدمجة.

1. تشغيل الوكيل

👉💻 انتقِل إلى دليل الخلفية (الذي تم فيه تحديد الوكيل) وشغِّل واجهة الويب::

cd ~/way-back-home/level_2/backend
uv run adk web

يبدأ هذا الأمر الوكيل المحدّد في

agent/agent.py

ويفتح واجهة ويب لإجراء الاختبار.

👉 فتح عنوان URL:

سيُخرج الأمر عنوان URL محليًا (عادةً http://127.0.0.1:8000 أو ما شابه ذلك). افتح هذا الرابط في المتصفّح.

adk web

بعد النقر على عنوان URL، ستظهر لك واجهة مستخدم ADK على الويب. تأكَّد من اختيار "المشرف" من أعلى يمين الصفحة.

adk_ui

2. اختبار إمكانات البحث

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

العثور على العناصر استنادًا إلى المعنى والمفهوم، حتى إذا لم تتطابق الكلمات الرئيسية

طلبات الاختبار: (اختَر أيًا مما يلي)

Who can help with injuries?

What abilities are related to survival?

ما يجب البحث عنه:

  • يجب أن يذكر الاستدلال البحث الدلالي أو RAG.
  • من المفترض أن تظهر لك نتائج ذات صلة من الناحية المفاهيمية (مثل "جراحة" عند طلب "إسعافات أولية").
  • ستتضمّن النتائج الرمز 🧬.

تجمع هذه الميزة بين فلاتر الكلمات الرئيسية والفهم الدلالي لطلبات البحث المعقّدة.

طلبات الاختبار:(اختَر أيًا مما يلي)

Find someone who can ply a plane in the volcanic area

Who has healing abilities in the FOSSILIZED?

Who has healing abilities in the mountains?

ما يجب البحث عنه:

  • يجب أن يذكر السبب البحث الهجين.
  • يجب أن تتطابق النتائج مع المعيارَين (المفهوم والموقع الجغرافي/الفئة).
  • ستظهر النتائج التي تم العثور عليها باستخدام كلتا الطريقتَين مع الرمز 🔀، وسيتم ترتيبها في أعلى القائمة.

👉💻 عند الانتهاء من الاختبار، أنهِ العملية بالضغط على Ctrl+C في سطر الأوامر.

8. تشغيل التطبيق الكامل

نظرة عامة على بنية Full Stack

architecture_fullstack

إضافة SessionService وRunner

👉💻 في المحطة الطرفية، افتح الملف chat.py في "محرّر Cloud Shell" من خلال تنفيذ ما يلي (تأكَّد من الضغط على ctrl+C لإنهاء العملية السابقة قبل المتابعة):

cloudshell edit ~/way-back-home/level_2/backend/api/routes/chat.py

👉في ملف chat.py، ابحث عن التعليق # TODO: REPLACE_INMEMORY_SERVICES، ثم استبدِل هذا السطر بالكامل بالرمز التالي:

    session_service = InMemorySessionService()
    memory_service = InMemoryMemoryService()

👉في ملف chat.py، ابحث عن التعليق # TODO: REPLACE_RUNNER، ثم استبدِل هذا السطر بالكامل بالرمز التالي:

runner = Runner(
    agent=root_agent, 
    session_service=session_service,
    memory_service=memory_service,
    app_name="survivor-network"
)

1. بدء الطلب

إذا كانت الوحدة الطرفية السابقة لا تزال قيد التشغيل، يمكنك إنهاءها بالضغط على Ctrl+C.

👉💻 بدء التطبيق:

cd ~/way-back-home/level_2/
./start_app.sh

عند بدء الخلفية بنجاح، سيظهر Local: http://localhost:5173/" كما يلي: fronted

👉 انقر على Local: http://localhost:5173/‎ من نافذة Terminal.

محادثة

طلب البحث:

Find skills similar to healing

محادثة

ما يحدث:

  • يتعرّف الموظّف على طلب التشابه
  • إنشاء تضمين لكلمة "شفاء"
  • يستخدم مسافة جيب التمام للعثور على مهارات متشابهة دلاليًا
  • العودة: الإسعافات الأولية (على الرغم من أنّ الأسماء لا تتطابق مع "الشفاء")

طلب البحث:

Find medical skills in the mountains

ما يحدث:

  1. مكوّن الكلمات الرئيسية: فلترة حسب category='medical'
  2. المكوّن الدلالي: تضمين "طبي" والترتيب حسب التشابه
  3. الدمج: يتم دمج النتائج مع إعطاء الأولوية لتلك التي تم العثور عليها باستخدام الطريقتين 🔀

طلب البحث(اختياري):

Who is good at survival and in the forest?

ما يحدث:

  • نتائج البحث عن الكلمات الرئيسية: biome='forest'
  • نتائج البحث الدلالي: مهارات مشابهة لـ "البقاء على قيد الحياة"
  • تجمع الطريقة المختلطة بين الطريقتين لتحقيق أفضل النتائج

👉💻 عند الانتهاء من الاختبار، اضغط على Ctrl+C في الوحدة الطرفية.

9- مسار متعدد الوسائط — طبقة الأدوات

لماذا نحتاج إلى مسار متعدد الوسائط؟

شبكة النجاة ليست مجرد نص. يرسل الناجون في الميدان بيانات غير منظَّمة مباشرةً من خلال المحادثة:

  • 📸 الصور: صور للموارد أو المخاطر أو المعدّات
  • ‫🎥 الفيديوهات: تقارير الحالة أو رسائل البث في حالات الطوارئ
  • 📄 النص: ملاحظات أو سجلّات الحقل

ما هي الملفات التي نعالجها؟

على عكس الخطوة السابقة التي بحثنا فيها عن البيانات الحالية، نعالج هنا الملفات التي حمّلها المستخدم. تتعامل واجهة chat.py مع مرفقات الملفات بشكل ديناميكي:

المصدر

المحتوى

الهدف

مرفق المستخدم

صورة/فيديو/نص

المعلومات التي يجب إضافتها إلى الرسم البياني

سياق المحادثة

"إليك صورة للّوازم"

النية والتفاصيل الإضافية

الأسلوب المخطَّط له: مسار الإجراءات التسلسلي للوكيل

نستخدم وكيلًا تسلسليًا (multimedia_agent.py) يربط الوكلاء المتخصّصين معًا:

architecture_uploading

يتم تحديد ذلك في backend/agent/multimedia_agent.py كـ SequentialAgent.

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

1. فتح ملف الأدوات

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

cloudshell edit ~/way-back-home/level_2/backend/agent/tools/extraction_tools.py

2. تنفيذ أداة upload_media

تحمّل هذه الأداة ملفًا محليًا إلى Google Cloud Storage.

👉 في extraction_tools.py، ابحث عن التعليق pass # TODO: REPLACE_UPLOAD_MEDIA_FUNCTION.

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

    """
    Upload media file to GCS and detect its type.
    
    Args:
        file_path: Path to the local file
        survivor_id: Optional survivor ID to associate with upload
        
    Returns:
        Dict with gcs_uri, media_type, and status
    """
    try:
        if not file_path:
            return {"status": "error", "error": "No file path provided"}
        
        # Strip quotes if present
        file_path = file_path.strip().strip("'").strip('"')
        
        if not os.path.exists(file_path):
            return {"status": "error", "error": f"File not found: {file_path}"}
        
        gcs_uri, media_type, signed_url = gcs_service.upload_file(file_path, survivor_id)
        
        return {
            "status": "success",
            "gcs_uri": gcs_uri,
            "signed_url": signed_url,
            "media_type": media_type.value,
            "file_name": os.path.basename(file_path),
            "survivor_id": survivor_id
        }
    except Exception as e:
        logger.error(f"Upload failed: {e}")
        return {"status": "error", "error": str(e)}

3- تنفيذ أداة extract_from_media

هذه الأداة هي جهاز توجيه، فهي تتحقّق من media_type وترسل البيانات إلى أداة الاستخراج الصحيحة (نص أو صورة أو فيديو).

👉في extraction_tools.py، ابحث عن التعليق pass # TODO: REPLACE_EXTRACT_FROM_MEDIA.

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

    """
    Extract entities and relationships from uploaded media.
    
    Args:
        gcs_uri: GCS URI of the uploaded file
        media_type: Type of media (text/image/video)
        signed_url: Optional signed URL for public/temporary access
        
    Returns:
        Dict with extraction results
    """
    try:
        if not gcs_uri:
             return {"status": "error", "error": "No GCS URI provided"}

        # Select appropriate extractor
        if media_type == MediaType.TEXT.value or media_type == "text":
            result = await text_extractor.extract(gcs_uri)
        elif media_type == MediaType.IMAGE.value or media_type == "image":
            result = await image_extractor.extract(gcs_uri)
        elif media_type == MediaType.VIDEO.value or media_type == "video":
            result = await video_extractor.extract(gcs_uri)
        else:
            return {"status": "error", "error": f"Unsupported media type: {media_type}"}
            
        # Inject signed URL into broadcast info if present
        if signed_url:
            if not result.broadcast_info:
                result.broadcast_info = {}
            result.broadcast_info['thumbnail_url'] = signed_url
        
        return {
            "status": "success",
            "extraction_result": result.to_dict(), # Return valid JSON dict instead of object
            "summary": result.summary,
            "entities_count": len(result.entities),
            "relationships_count": len(result.relationships),
            "entities": [e.to_dict() for e in result.entities],
            "relationships": [r.to_dict() for r in result.relationships]
        }
    except Exception as e:
        logger.error(f"Extraction failed: {e}")
        return {"status": "error", "error": str(e)}

تفاصيل التنفيذ الأساسية:

  • الإدخال المتعدّد الوسائط: ننقل الطلب النصي (_get_extraction_prompt()) وعنصر الصورة إلى generate_content.
  • الإخراج المنظَّم: تضمن response_mime_type="application/json" أن يعرض النموذج اللغوي الكبير ملف JSON صالحًا، وهو أمر بالغ الأهمية في مسار العمل.
  • ربط الكيانات المرئية: يتضمّن الطلب كيانات معروفة ليتمكّن Gemini من التعرّف على شخصيات معيّنة.

4. تنفيذ أداة save_to_spanner

تحتفظ هذه الأداة بالكيانات والعلاقات المستخرَجة في قاعدة بيانات Spanner Graph.

👉في extraction_tools.py، ابحث عن التعليق pass # TODO: REPLACE_SPANNER_AGENT.

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

    """
    Save extracted entities and relationships to Spanner Graph DB.
    
    Args:
        extraction_result: ExtractionResult object (or dict from previous step if passed as dict)
        survivor_id: Optional survivor ID to associate with the broadcast
        
    Returns:
        Dict with save statistics
    """
    try:
        # Handle if extraction_result is passed as the wrapper dict from extract_from_media
        result_obj = extraction_result
        if isinstance(extraction_result, dict) and 'extraction_result' in extraction_result:
             result_obj = extraction_result['extraction_result']
        
        # If result_obj is a dict (from to_dict()), reconstruct it
        if isinstance(result_obj, dict):
            from extractors.base_extractor import ExtractionResult
            result_obj = ExtractionResult.from_dict(result_obj)
        
        if not result_obj:
            return {"status": "error", "error": "No extraction result provided"}
            
        stats = spanner_service.save_extraction_result(result_obj, survivor_id)
        
        return {
            "status": "success",
            "entities_created": stats['entities_created'],
            "entities_existing": stats['entities_found_existing'],
            "relationships_created": stats['relationships_created'],
            "broadcast_id": stats['broadcast_id'],
            "errors": stats['errors'] if stats['errors'] else None
        }
    except Exception as e:
        logger.error(f"Spanner save failed: {e}")
        return {"status": "error", "error": str(e)}

من خلال تزويد الوكلاء بأدوات رفيعة المستوى، نضمن سلامة البيانات مع الاستفادة من قدرات الوكيل على الاستدلال.

5- تعديل خدمة GCS

يتولّى GCSService عملية تحميل الملفات إلى Google Cloud Storage.

👉💻 في المحطة الطرفية، افتح الملف في "محرِّر Cloud Shell":

cloudshell edit ~/way-back-home/level_2/backend/services/gcs_service.py

👉 في الملف gcs_service.py، ابحث عن التعليق # TODO: REPLACE_SAVE_TO_GCS داخل الدالة upload_file.

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

        blob = self.bucket.blob(blob_name)
        blob.upload_from_filename(file_path)

من خلال تجريد هذه العملية إلى خدمة، لا يحتاج "الوكيل" إلى معرفة تفاصيل حول حِزم GCS أو أسماء الكائنات الثنائية الكبيرة أو إنشاء عناوين URL موقّعة. يطلب منك فقط "تحميل" الملف.

6. (للقراءة فقط)لماذا سير العمل المستند إلى الذكاء الاصطناعي التوليدي أفضل من النُهج التقليدية؟

ميزة الحلول المستندة إلى الذكاء الاصطناعي الوكيل:

الميزة

مسار الدفعات

مستنِد إلى الأحداث

سير العمل المستند إلى الذكاء الاصطناعي الوكيل

التعقيد

منخفضة (نص برمجي واحد)

مرتفع (5 خدمات أو أكثر)

منخفضة (ملف Python واحد: multimedia_agent.py)

إدارة الحالة

المتغيّرات العمومية

صعب (غير مرتبط)

موحَّدة (حالة الوكيل)

التعامل مع الأخطاء

الأعطال

السجلات الصامتة

تفاعلية ("تعذّر عليّ قراءة هذا الملف")

ملاحظات المستخدمين

لوحات مطبوعة

يجب إجراء استطلاع

فوري (جزء من المحادثة)

القدرة على التكيّف

المنطق الثابت

الدوال الثابتة

ذكي (يقرّر النموذج اللغوي الكبير الخطوة التالية)

الوعي بالسياق

بدون

بدون

كاملة (تعرف نية المستخدم)

أهمية ذلك: باستخدام multimedia_agent.py (SequentialAgent مع 4 وكلاء فرعيين: تحميل → استخراج → حفظ → ملخّص)، نستبدل البنية الأساسية المعقّدة والنصوص البرمجية الهشة بمنطق تطبيق ذكي قائم على المحادثة.

10. مسار الإجراءات المتعدّدة الوسائط — طبقة الوكيل

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

agent_diagram

1. فتح ملف الوكيل

👉💻 في المحطة الطرفية، افتح الملف في "محرِّر Cloud Shell":

cloudshell edit ~/way-back-home/level_2/backend/agent/multimedia_agent.py

2. تحديد "وكيل التحميل"

يستخرج هذا الوكيل مسار ملف من رسالة المستخدم ويحمّله إلى GCS.

👉في الملف multimedia_agent.py، ابحث عن التعليق # TODO: REPLACE_UPLOAD_AGENT.

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

upload_agent = LlmAgent(
    name="UploadAgent",
    model="gemini-2.5-flash",
    instruction="""Extract the file path from the user's message and upload it.

Use `upload_media(file_path, survivor_id)` to upload the file.
The survivor_id is optional - include it if the user mentions a specific survivor (e.g., "survivor Sarah" -> "Sarah").
If the user provides a path like "/path/to/file", use that.

Return the upload result with gcs_uri and media_type.""",
    tools=[upload_media],
    output_key="upload_result"
)

3- تحديد "عامل الاستخراج"

يستطيع هذا الوكيل "رؤية" الوسائط التي تم تحميلها واستخراج البيانات المنظَّمة باستخدام Gemini Vision.

👉في الملف multimedia_agent.py، ابحث عن التعليق # TODO: REPLACE_EXTRACT_AGENT.

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

extraction_agent = LlmAgent(
    name="ExtractionAgent", 
    model="gemini-2.5-flash",
    instruction="""Extract information from the uploaded media.

Previous step result: {upload_result}

Use `extract_from_media(gcs_uri, media_type, signed_url)` with the values from the upload result.
The gcs_uri is in upload_result['gcs_uri'], media_type in upload_result['media_type'], and signed_url in upload_result['signed_url'].

Return the extraction results including entities and relationships found.""",
    tools=[extract_from_media],
    output_key="extraction_result"
)

لاحظ كيف تشير instruction إلى {upload_result}، فهذه هي الطريقة التي يتم بها نقل الحالة بين الوكلاء في ADK.

4. تحديد Spanner Agent

يحفظ هذا الوكيل الكيانات والعلاقات المستخرَجة في قاعدة بيانات الرسومات البيانية.

👉في الملف multimedia_agent.py، ابحث عن التعليق # TODO: REPLACE_SPANNER_AGENT.

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

spanner_agent = LlmAgent(
    name="SpannerAgent",
    model="gemini-2.5-flash", 
    instruction="""Save the extracted information to the database.

Upload result: {upload_result}
Extraction result: {extraction_result}

Use `save_to_spanner(extraction_result, survivor_id)` to save to Spanner.
Pass the WHOLE `extraction_result` object/dict from the previous step.
Include survivor_id if it was provided in the upload step.

Return the save statistics.""",
    tools=[save_to_spanner],
    output_key="spanner_result"
)

يتلقّى هذا الوكيل سياقًا من كلتا الخطوتَين السابقتَين (upload_result وextraction_result).

5- تحديد وكيل التلخيص

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

👉في الملف multimedia_agent.py، ابحث عن التعليق summary_instruction="" # TODO: REPLACE_SUMMARY_AGENT_PROMPT.

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

USE_MEMORY_BANK = os.getenv("USE_MEMORY_BANK", "false").lower() == "true"
save_msg = "6. Mention that the data is also being synced to the memory bank." if USE_MEMORY_BANK else ""

summary_instruction = f"""Provide a user-friendly summary of the media processing.

Upload: {{upload_result}}
Extraction: {{extraction_result}}
Database: {{spanner_result}}

Summarize:
1. What file was processed (name and type)
2. Key information extracted (survivors, skills, needs, resources found) - list names and counts
3. Relationships identified
4. What was saved to the database (broadcast ID, number of entities)
5. Any issues encountered
{save_msg}

Be concise but informative."""

لا يحتاج هذا الوكيل إلى أدوات، بل يقرأ السياق المشترك وينشئ ملخّصًا واضحًا للمستخدم.

‫🧠 ملخّص البنية

طبقة

ملف

المسؤولية

الأدوات

extraction_tools.py + gcs_service.py

كيفية — التحميل والاستخراج والحفظ

الوكيل

multimedia_agent.py

ماذا: تنظيم مسار العمل

11. مسار البيانات المتعدّدة الوسائط — التنسيق

إنّ أساس نظامنا الجديد هو MultimediaExtractionPipeline المحدّد في backend/agent/multimedia_agent.py. تستخدِم هذه الأداة نمط الوكيل التسلسلي من حزمة تطوير الوكلاء (ADK).

1. لماذا يجب استخدام الترتيب التسلسلي؟

تتضمّن معالجة ملف تم تحميله سلسلة تبعية خطية:

  1. لا يمكنك استخراج البيانات إلا بعد الحصول على الملف (تحميل).
  2. لا يمكنك حفظ البيانات إلى أن تستخرجها (الاستخراج).
  3. لا يمكنك تلخيص النتائج إلا بعد الحصول عليها (حفظ).

SequentialAgent هي الأنسب لهذا الغرض. وينقل ناتج أحد الوكلاء كسياق/مدخل إلى الوكيل التالي.

2. تعريف الوكيل

لنلقِ نظرة على كيفية تجميع خط الأنابيب في أسفل multimedia_agent.py: 👉💻 في الوحدة الطرفية، افتح الملف في "محرِّر Cloud Shell" من خلال تنفيذ الأمر التالي:

cloudshell edit ~/way-back-home/level_2/backend/agent/multimedia_agent.py

يتلقّى هذا الإجراء مدخلات من كلتا الخطوتَين السابقتَين. ابحث عن التعليق # TODO: REPLACE_ORCHESTRATION. استبدِل هذا السطر بالكامل بالرمز التالي:

    sub_agents=[upload_agent, extraction_agent, spanner_agent, summary_agent]

3- التواصل مع موظّف الدعم الأساسي

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

cloudshell edit ~/way-back-home/level_2/backend/agent/agent.py

ابحث عن التعليق # TODO: REPLACE_ADD_SUBAGENT. استبدِل هذا السطر بالكامل بالرمز التالي:

    sub_agents=[multimedia_agent],

يجمع هذا العنصر الفردي أربعة "خبراء" في كيان واحد يمكن استدعاؤه.

4. تدفّق البيانات بين الوكلاء

يخزّن كل وكيل ناتجه في سياق مشترك يمكن للوكلاء اللاحقين الوصول إليه:

architecture_uploading

5- افتح التطبيق (تخطَّ هذه الخطوة إذا كان التطبيق لا يزال قيد التشغيل)

👉💻 بدء التطبيق:

cd ~/way-back-home/level_2/
./start_app.sh

👉 انقر على Local: http://localhost:5173/‎ من نافذة Terminal.

6. اختبار تحميل الصورة

👉 في واجهة المحادثة، اختَر أيًا من الصور هنا وحمِّلها إلى واجهة المستخدم:

في واجهة المحادثة، أخبر الوكيل عن سياقك المحدّد:

Here is the survivor note

بعد ذلك، أرفِق الصورة هنا.

upload_input

upload_result

👉💻 في الوحدة الطرفية، بعد الانتهاء من الاختبار، اضغط على "Ctrl+C" لإنهاء العملية.

6. التحقّق من إمكانية تحميل الوسائط المتعدّدة في حزمة GCS

gcs

  • اختَر الحزمة وانقر على media.

وسائط

  • يمكنك الاطّلاع على الصورة التي حمّلتها هنا. uploaded_img

7. التحقّق من ميزة "التحميل المتعدّد الوسائط" في Spanner (اختياري)

في ما يلي مثال على الناتج في واجهة المستخدم لـ test_photo1.

  • افتح Google Cloud Console Spanner.
  • اختَر مثيلك: Survivor Network
  • اختَر قاعدة البيانات: graph-db
  • في الشريط الجانبي الأيمن، انقر على Spanner Studio.

👉 في Spanner Studio، استخدِم طلب بحث عن البيانات الجديدة:

SELECT 
  s.name AS Survivor,
  s.role AS Role,
  b.name AS Biome,
  r.name AS FoundResource,
  s.created_at
FROM Survivors s
LEFT JOIN SurvivorInBiome sib ON s.survivor_id = sib.survivor_id
LEFT JOIN Biomes b ON sib.biome_id = b.biome_id
LEFT JOIN SurvivorFoundResource sfr ON s.survivor_id = sfr.survivor_id
LEFT JOIN Resources r ON sfr.resource_id = r.resource_id
ORDER BY s.created_at DESC;

يمكننا التحقّق من ذلك من خلال الاطّلاع على النتيجة أدناه:

spanner_verify

12. "بنك الذاكرة" مع "محرك الوكيل"

1. طريقة عمل ميزة "هذه السنة"

يستخدم النظام نهج الذاكرة المزدوجة للتعامل مع السياق الفوري والتعلم على المدى الطويل.

memory_bank

2. ما هي مواضيع الذكريات؟

تحدّد مواضيع الذاكرة فئات المعلومات التي يجب أن يتذكّرها الوكيل في المحادثات. يمكنك اعتبارها خزائن لحفظ أنواع مختلفة من إعدادات المستخدم المفضّلة.

الموضوعان اللذان اخترناهما:

  1. search_preferences: طريقة البحث المفضّلة لدى المستخدم
    • هل يفضّلون البحث عن الكلمات الرئيسية أو البحث الدلالي؟
    • ما هي المهارات أو البيئات التي يبحثون عنها غالبًا؟
    • مثال على الذاكرة: "يفضّل المستخدم البحث الدلالي عن المهارات الطبية"
  2. urgent_needs_context: الأزمات التي يتم تتبّعها
    • ما هي الموارد التي تتم مراقبتها؟
    • ما هي الفئات التي تثير قلقهم من بين الناجين؟
    • مثال على الذاكرة: "يتتبّع المستخدم نقص الأدوية في معسكر الشمال"

3- إعداد مواضيع الذكريات

تحدّد مواضيع الذاكرة المخصّصة ما يجب أن يتذكّره الوكيل. يتم ضبط هذه الإعدادات عند نشر "محرك الوكيل".

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

cloudshell edit ~/way-back-home/level_2/backend/deploy_agent.py

سيؤدي ذلك إلى فتح ~/way-back-home/level_2/backend/deploy_agent.py في المحرِّر.

نحدّد عناصر بنية MemoryTopic لتوجيه النموذج اللغوي الكبير بشأن المعلومات التي يجب استخراجها وحفظها.

👉في الملف deploy_agent.py، استبدِل # TODO: SET_UP_TOPIC بما يلي:

# backend/deploy_agent.py

    custom_topics = [
        # Topic 1: Survivor Search Preferences
        MemoryTopic(
            custom_memory_topic=CustomMemoryTopic(
                label="search_preferences",
                description="""Extract the user's preferences for how they search for survivors. Include:
                - Preferred search methods (keyword, semantic, direct lookup)
                - Common filters used (biome, role, status)
                - Specific skills they value or frequently look for
                - Geographic areas of interest (e.g., "forest biome", "mountain outpost")
                
                Example: "User prefers semantic search for finding similar skills."
                Example: "User frequently checks for survivors in the Swamp Biome."
                """,
            )
        ),
        # Topic 2: Urgent Needs Context
        MemoryTopic(
            custom_memory_topic=CustomMemoryTopic(
                label="urgent_needs_context",
                description="""Track the user's focus on urgent needs and resource shortages. Include:
                - Specific resources they are monitoring (food, medicine, ammo)
                - Critical situations they are tracking
                - Survivors they are particularly concerned about
                
                Example: "User is monitoring the medicine shortage in the Northern Camp."
                Example: "User is looking for a doctor for the injured survivors."
                """,
            )
        )
    ]

4. دمج الوكيل

يجب أن يكون رمز الوكيل على دراية بـ "بنك الذاكرة" لحفظ المعلومات واسترجاعها.

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

cloudshell edit ~/way-back-home/level_2/backend/agent/agent.py

سيؤدي ذلك إلى فتح ~/way-back-home/level_2/backend/agent/agent.py في المحرِّر.

إنشاء وكيل

عند إنشاء الوكيل، نمرّر after_agent_callback لضمان حفظ الجلسات في الذاكرة بعد التفاعلات. يتم تشغيل الدالة add_session_to_memory بشكل غير متزامن لتجنُّب إبطاء ردّ المحادثة.

👉في الملف agent.py، ابحث عن التعليق # TODO: REPLACE_ADD_SESSION_MEMORY، استبدِل هذا السطر بالكامل بالرمز التالي:

async def add_session_to_memory(
        callback_context: CallbackContext
) -> Optional[types.Content]:
    """Automatically save completed sessions to memory bank in the background"""
    if hasattr(callback_context, "_invocation_context"):
        invocation_context = callback_context._invocation_context
        if invocation_context.memory_service:
            # Use create_task to run this in the background without blocking the response
            asyncio.create_task(
                invocation_context.memory_service.add_session_to_memory(
                    invocation_context.session
                )
            )
            logger.info("Scheduled session save to memory bank in background")

الحفظ في الخلفية

👉في الملف agent.py، ابحث عن التعليق # TODO: REPLACE_ADD_MEMORY_BANK_TOOL، استبدِل هذا السطر بالكامل بالرمز التالي:

if USE_MEMORY_BANK:
    agent_tools.append(PreloadMemoryTool())

👉في الملف agent.py، ابحث عن التعليق # TODO: REPLACE_ADD_CALLBACK، استبدِل هذا السطر بالكامل بالرمز التالي:

    after_agent_callback=add_session_to_memory if USE_MEMORY_BANK else None

إعداد خدمة جلسات Vertex AI

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

cloudshell edit ~/way-back-home/level_2/backend/api/routes/chat.py

👉في ملف chat.py، ابحث عن التعليق # TODO: REPLACE_VERTEXAI_SERVICES، ثم استبدِل هذا السطر بالكامل بالرمز التالي:

    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
    )

4. الإعداد والنشر

قبل اختبار ميزات الذاكرة، عليك نشر الوكيل مع مواضيع الذاكرة الجديدة والتأكّد من إعداد بيئتك بشكل صحيح.

لقد وفّرنا نصًا برمجيًا مناسبًا للتعامل مع هذه العملية.

تشغيل نص برمجي للنشر

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

cd ~/way-back-home/level_2
./deploy_and_update_env.sh

ينفِّذ هذا النص البرمجي الإجراءات التالية:

  • يُنفِّذ backend/deploy_agent.py لتسجيل الوكيل ومواضيع الذاكرة في Vertex AI.
  • تعرض هذه السمة معرّف محرّك بحث الوكيل الجديد.
  • يتم تعديل ملف .env تلقائيًا باستخدام AGENT_ENGINE_ID.
  • تأكَّد من ضبط USE_MEMORY_BANK=TRUE في ملف .env.

[!IMPORTANT] إذا أجريت تغييرات على custom_topics في deploy_agent.py، يجب إعادة تشغيل هذا النص البرمجي لتعديل Agent Engine.

13. التحقّق من صحة "بنك الذاكرة" باستخدام بيانات متعددة الوسائط

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

1. افتح التطبيق (يمكنك تخطّي هذه الخطوة إذا كان تطبيقك قيد التشغيل)

افتح التطبيق مرة أخرى باتّباع التعليمات أدناه: إذا كانت النافذة السابقة لا تزال قيد التشغيل، أنهِها بالضغط على Ctrls+C.

👉💻 بدء التطبيق:

cd ~/way-back-home/level_2/
./start_app.sh

👉 انقر على Local: http://localhost:5173/‎ من نافذة Terminal.

2. اختبار "بنك الذاكرة" باستخدام النص

في واجهة المحادثة، أخبر الوكيل عن سياقك المحدّد:

"I'm planning a medical rescue mission in the mountains. I need survivors with first aid and climbing skills."

👉 انتظِر لمدة 30 ثانية تقريبًا حتى تتم معالجة الذاكرة في الخلفية.

2. بدء جلسة جديدة

أعِد تحميل الصفحة لمحو سجلّ المحادثات الحالي (الذاكرة القصيرة المدى).

اطرح سؤالاً يعتمد على السياق الذي قدّمته سابقًا:

"What kind of missions am I interested in?"

الردّ المتوقّع:

"استنادًا إلى محادثاتك السابقة، أنت مهتم بما يلي:

  • مهام الإنقاذ الطبي
  • عمليات في الجبال أو على ارتفاعات عالية
  • المهارات المطلوبة: الإسعافات الأولية، التسلق

هل تريد أن أعثر لك على ناجين يستوفون هذه المعايير؟"

3- الاختبار باستخدام ميزة "تحميل صورة"

حمِّل صورة واطرح السؤال التالي:

remember this

يمكنك اختيار أي من الصور هنا أو صورتك الخاصة وتحميلها إلى واجهة المستخدم:

4. التأكّد من صحة المعلومات في "محرك وكلاء Vertex AI"

الانتقال إلى Google Cloud Console Agent Engine

  1. تأكَّد من اختيار المشروع من أداة اختيار المشاريع في أعلى يمين الصفحة:أداة اختيار المشاريع
  2. تحقَّق من محرك الوكيل الذي نشرته للتو من الأمر السابق use_memory_bank.sh:محرّك الحواريانقر على محرك الوكيل الذي أنشأته للتو.
  3. انقر على علامة التبويب Memories في هذا الوكيل الذي تم نشره، ويمكنك الاطّلاع على كل الذكريات هنا.عرض الذكرى

👉💻 عند الانتهاء من الاختبار، انقر على "Ctrl + C" في الوحدة الطرفية لإنهاء العملية.

🎉 تهانينا! لقد أضفت للتوّ قاعدة بيانات الذاكرة إلى الوكيل.

14. النشر على Cloud Run

1. تشغيل نص النشر البرمجي

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

cd ~/way-back-home/level_2
./deploy_cloud_run.sh

بعد نشرها بنجاح، سيتوفّر لديك عنوان URL، وهو عنوان URL الذي تم نشره. تم نشرها

👉💻 قبل الحصول على عنوان URL، امنح الإذن من خلال تنفيذ ما يلي:

source .env && gcloud run services add-iam-policy-binding survivor-frontend --region $REGION --member=allUsers --role=roles/run.invoker && gcloud run services add-iam-policy-binding survivor-backend --region $REGION --member=allUsers --role=roles/run.invoker

انتقِل إلى عنوان URL الذي تم نشره، وسيظهر تطبيقك مباشرةً هناك.

2. التعرّف على مسار الإنشاء

يحدّد ملف cloudbuild.yaml الخطوات التسلسلية التالية:

  1. إنشاء الخلفية: ينشئ صورة Docker من backend/Dockerfile.
  2. نشر الخلفية: ينشر حاوية الخلفية إلى Cloud Run.
  3. عنوان URL لعملية الالتقاط: يحصل على عنوان URL الجديد للخادم الخلفي.
  4. إنشاء الواجهة الأمامية:
    • تثبيت التبعيات
    • ينشئ تطبيق React، ويضيف VITE_API_URL=.
  5. صورة الواجهة الأمامية: تنشئ صورة Docker من frontend/Dockerfile (حزم مواد العرض الثابتة).
  6. نشر الواجهة الأمامية: ينشر حاوية الواجهة الأمامية.

3- التحقّق من عملية النشر

بعد اكتمال عملية الإنشاء (راجِع رابط السجلات الذي يوفّره النص البرمجي)، يمكنك التأكّد مما يلي:

  1. انتقِل إلى وحدة تحكّم Cloud Run.
  2. ابحث عن خدمة survivor-frontend.
  3. انقر على عنوان URL لفتح التطبيق.
  4. أدخِل طلب بحث للتأكّد من أنّ الواجهة الأمامية يمكنها التواصل مع الواجهة الخلفية.

4. (!ONLY FOR WORKSHOP ATTENDEE) تعديل موقعك الجغرافي

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

cd ~/way-back-home/level_2
./set_level_2.sh

افتح الآن waybackhome.dev، وسيظهر لك أنّ موقعك الجغرافي قد تم تعديله. تهانينا على إكمال المستوى 2.

النتيجة النهائية

(اختياري) 5. النشر اليدوي

إذا كنت تفضّل تنفيذ الأوامر يدويًا أو فهم العملية بشكل أفضل، إليك كيفية استخدام cloudbuild.yaml مباشرةً.

كتابة cloudbuild.yaml

يخبر ملف cloudbuild.yaml خدمة Google Cloud Build بالخطوات التي يجب تنفيذها.

  • الخطوات: قائمة بالإجراءات المتسلسلة. يتم تنفيذ كل خطوة في حاوية (مثل docker، gcloud، node، bash).
  • عمليات الاستبدال: المتغيرات التي يمكن تمريرها في وقت الإنشاء (مثل $_REGION).
  • مساحة العمل: دليل مشترك يمكن للخطوات مشاركة الملفات فيه (مثل طريقة مشاركة backend_url.txt).

تشغيل عملية النشر

للنشر يدويًا بدون النص البرمجي، استخدِم الأمر gcloud builds submit. يجب تمرير متغيّرات الاستبدال المطلوبة.

# Load your env vars first or replace these values manually
export PROJECT_ID=your-project-id
export REGION=us-central1

gcloud builds submit --config cloudbuild.yaml \
    --project "$PROJECT_ID" \
    --substitutions _REGION="us-central1",_GOOGLE_API_KEY="",_AGENT_ENGINE_ID="your-agent-id",_USE_MEMORY_BANK="TRUE",_GOOGLE_GENAI_USE_VERTEXAI="TRUE"

15. الخاتمة

1. المشاريع التي أنشأتها

‫✅ قاعدة بيانات الرسومات البيانية: Spanner مع عُقد (الناجون والمهارات) وحواف (العلاقات)
‫✅ بحث مستند إلى الذكاء الاصطناعي: بحث مستند إلى الكلمات الرئيسية والبحث الدلالي والبحث المختلط باستخدام عمليات التضمين
‫✅ مسار متعدد الوسائط: استخراج الكيانات من الصور/الفيديوهات باستخدام Gemini
‫✅ نظام متعدد الوكلاء: سير عمل منسَّق باستخدام "حزمة تطوير التطبيقات"
‫✅ بنك الذاكرة: تخصيص طويل الأمد باستخدام Vertex AI
‫✅ عملية نشر الإنتاج: Cloud Run + Agent Engine

2. ملخّص البنية

architecture_fullstack

3- النقاط الرئيسية التي تم تعلّمها

  1. ‫Graph RAG: يجمع بين بنية قاعدة بيانات الرسوم البيانية والتضمينات الدلالية لإجراء بحث ذكي
  2. أنماط الوكلاء المتعدّدين: مسارات متسلسلة لسير العمل المعقّد والمتعدّد الخطوات
  3. الذكاء الاصطناعي المتعدد الوسائط: استخراج بيانات منظَّمة من وسائط غير منظَّمة (صور/فيديوهات)
  4. الوكلاء ذوو الحالة: يتيح "بنك الذاكرة" التخصيص على مستوى الجلسات

4. محتوى ورشة العمل

5- الموارد