تحليل تنفيذ "وكيل حزمة تطوير التطبيقات" باستخدام "المكوّن الإضافي لتحليلات وكيل BigQuery"

1. مقدمة

في هذا الدرس العملي، ستنشئ نظامًا متعدد الوكلاء باستخدام "حزمة تطوير الوكلاء" (ADK) وتفعّل ميزة "مراقبة أداء الوكيل" باستخدام BigQuery Agent Analytics Plugin.ستطرح سلسلة من الأسئلة على الوكيل، ثم ستستخدم BigQuery لتحليل عمليات تتبُّع المحادثات واستخدام أدوات الوكيل.

المهام التي ستنفذها

إنشاء مساعد بيع بالتجزئة متعدد الوكلاء باستخدام حزمة تطوير الوكلاء
ابدأ BigQuery Agent Analytics Plugin لتسجيل بيانات التتبُّع وتخزينها حول تنفيذ هذا الوكيل في BigQuery
تحليل بيانات سجلّ الوكيل في BigQuery

المتطلبات

متصفّح ويب، مثل Chrome
مشروع Google Cloud تم تفعيل الفوترة فيه
حساب Gmail سيوضّح لك القسم التالي كيفية تحصيل رصيد مجاني بقيمة 5 دولار أمريكي لهذا الدرس العملي وإعداد مشروع جديد.

هذا الدرس العملي مخصّص للمطوّرين من جميع المستويات، بما في ذلك المبتدئين. ستستخدم واجهة سطر الأوامر في Google Cloud Shell ورمز Python لتطوير حزمة تطوير التطبيقات (ADK). ليس عليك أن تكون خبيرًا في Python، ولكنّ فهمًا أساسيًا لطريقة قراءة الرموز البرمجية سيساعدك في فهم المفاهيم.

2. قبل البدء

إنشاء مشروع على Google Cloud

في Google Cloud Console، في صفحة اختيار المشروع، اختَر مشروعًا على Google Cloud أو أنشِئ مشروعًا.

تأكَّد من تفعيل الفوترة لمشروعك على Cloud. تعرَّف على كيفية التحقّق مما إذا كانت الفوترة مفعَّلة في مشروع.

بدء Cloud Shell

Cloud Shell هي بيئة سطر أوامر تعمل في Google Cloud ومحمّلة مسبقًا بالأدوات اللازمة.

انقر على تفعيل Cloud Shell في أعلى "وحدة تحكّم Google Cloud":

بعد الاتصال بـ Cloud Shell، نفِّذ الأمر التالي للتحقّق من مصادقتك في Cloud Shell:

gcloud auth list

نفِّذ الأمر التالي للتأكّد من إعداد مشروعك لاستخدامه مع gcloud:

gcloud config get project

إذا لم يتم ضبط مشروعك على النحو المتوقّع، استخدِم الأمر التالي لضبط مشروعك:

export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID

تفعيل واجهات برمجة التطبيقات

نفِّذ الأمر التالي لتفعيل جميع واجهات برمجة التطبيقات والخدمات المطلوبة:

gcloud services enable bigquery.googleapis.com \
cloudresourcemanager.googleapis.com \
aiplatform.googleapis.com

عند تنفيذ الأمر بنجاح، من المفترض أن تظهر لك رسالة مشابهة للرسالة الموضّحة أدناه:

اكتملت العملية "operations/..." بنجاح.

3- التثبيت والإعداد

ارجع إلى Cloud Shell وتأكَّد من أنّك في دليل المنزل.

نفِّذ الأمر التالي في Cloud Shell لإنشاء مجموعة بيانات جديدة باسم adk_logs في BigQuery:

bq mk --dataset --location=US adk_logs

لننشئ الآن بيئة Python افتراضية ونثبّت الحِزم المطلوبة.

افتح علامة تبويب جديدة في Cloud Shell وأدخِل الأمر التالي لإنشاء مجلد باسم adk-agent-observability والانتقال إليه:

mkdir adk-agent-observability
cd adk-agent-observability

أنشئ بيئة Python افتراضية:

python -m venv .venv

فعِّل البيئة الافتراضية:

source .venv/bin/activate

تثبيت حزمة ADK:

pip install --upgrade google-adk

4. إنشاء تطبيق ADK

الآن، لننشئ وكيل المساعدة في البيع بالتجزئة. سيتم تصميم هذا الوكيل من أجل ...

نفِّذ أمر أداة إنشاء حزمة تطوير التطبيقات (ADK) لإنشاء بنية تطبيق وكيل جديد تتضمّن المجلدات والملفات اللازمة:

adk create retail_assistant_app

اتّبِع التعليمات:

اختَر gemini-2.5-flash للنموذج.
اختَر Vertex AI للواجهة الخلفية.
أكِّد الإعدادات التلقائية لرقم تعريف مشروع Google Cloud والمنطقة.

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

انقر على الزر "فتح المحرّر" في Cloud Shell لفتح Cloud Shell Editor وعرض المجلدات والملفات التي تم إنشاؤها حديثًا:

دوِّن الملفات التي تم إنشاؤها:

retail_assistant_app/
├── .venv/
└── retail_assistant_app/
    ├── __init__.py
    ├── agent.py
    └── .env

init.py: يضع علامة على المجلد باعتباره وحدة Python.
agent.py: يحتوي على تعريف الوكيل الأولي.
‫.env: قد تحتاج إلى النقر على "عرض" > "تبديل الملفات المخفية" لعرض هذا الملف

يحتوي ملف ‎ .env على متغيرات البيئة لمشروعك، لذا عدِّل أي متغيرات لم يتم ضبطها بشكل صحيح من الطلبات:

GOOGLE_GENAI_USE_VERTEXAI=1
GOOGLE_CLOUD_PROJECT=<YOUR_GOOGLE_PROJECT_ID>
GOOGLE_CLOUD_LOCATION=<YOUR_GOOGLE_CLOUD_REGION>

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

لنعرّف الآن نظامًا هرميًا متعدّد الوكلاء.

Real Time Trend Agent: تستخدم هذه الأداة "بحث Google" للعثور على أحدث صيحات الموضة.
Inventory Data Agent: يستخدم مجموعة أدوات BigQuery لطلب البحث في مجموعة بيانات thelook_ecommerce العامة عن المنتجات المتاحة.
وكيل المساعدة في البيع بالتجزئة (الوكيل الرئيسي): ينظّم سير العمل من خلال طلب المشورة من "وكيل رصد المؤشرات" والمنتجات المطابقة من "وكيل المستودع".

استبدِل محتوى retail_assistant_app/agent.py بالكامل بالرمز التالي.

import os
import uuid
import asyncio
import google.auth
import dotenv
from google.genai import types
from google.adk.agents import Agent
from google.adk.apps import App
from google.adk.runners import InMemoryRunner
from google.adk.tools import AgentTool, google_search
from google.adk.tools.bigquery import BigQueryCredentialsConfig, BigQueryToolset
from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryAgentAnalyticsPlugin

dotenv.load_dotenv()

# --- Configuration ---

PROJECT_ID = os.getenv('GOOGLE_CLOUD_PROJECT', 'project_not_set')
DATASET_ID = "adk_logs"
TABLE_ID = "retail_assistant_agent_logs"
APP_NAME = "retail_assistant_agent"
USER_ID = "test_user"

# --- Toolsets ---

credentials, _ = google.auth.default()
credentials_config = BigQueryCredentialsConfig(credentials=credentials)
bigquery_toolset = BigQueryToolset(
 credentials_config=credentials_config
)

# --- Agents ---

# 1. Trend Spotter
real_time_agent = Agent(
   name="real_time_agent",
   model="gemini-2.5-flash",
    description="Researches external factors like weather, local events, and current fashion trends.",
   instruction="""
       You are a real-time research agent.
       Use Google Search to find real-time information relevant to the user's request,
       such as the current weather in their location or trending styles.
   """,
   tools=[google_search]
)

# 2. Inventory Manager
inventory_data_agent = Agent(
   name="inventory_data_agent",
   model="gemini-2.5-flash",
   description="Oversees product inventory in the BigQuery `thelook_ecommerce` dataset to find available items and prices.",
   instruction=f"""
   You manage the inventory. You have access to the `bigquery-public-data.thelook_ecommerce` dataset via the BigQuery toolset.
   Run all BigQuery queries the project id of: '{PROJECT_ID}'
  
   Your workflow:
   1. Look at the products table.
   2. Find items that match the requirements, factor in the results from the trend_setter agent if there are any.
   3. Return with a user friendly response, including the list of specific products and prices.
   """,
   tools=[bigquery_toolset]
)

# 3. Root Orchestrator
root_agent = Agent(
   name="retail_assistant",
   model="gemini-2.5-flash",   
   description="The primary orchestrator, responsible for handling user input, delegating to sub-agents, and synthesizing the final product recommendation.",
   instruction="""
   You are a Retail Assistant.   
   You can ask the 'real_time_agent' agent for any realtime information needed, or style advice, include any information provided by the user.
   You should ask the 'inventory_data_agent' agent to find a maximum of 3 available items matching that style.
   Combine the results into a recommendation.
   """,
   tools=[AgentTool(agent=real_time_agent)],
   sub_agents=[inventory_data_agent]
)

6. إنشاء سجلّات باستخدام "إضافة إحصاءات وكيل BigQuery"

لنضبط الآن مكوّن BigQuery الإضافي لتحليلات الوكيل من أجل تسجيل بيانات التنفيذ.

لإجراء ذلك، عليك إنشاء مثيل للفئة App. يعمل هذا الصف كحاوية وقت التشغيل للوكيل، ويدير حلقة المحادثة ويتعامل مع حالة المستخدم وينسّق أي مكوّنات إضافية مرفقة (مثل أداة تسجيل إحصاءات الوكيل).

الرمز أدناه:

تهيئة إضافة تسجيل الدخول: يتم إنشاء BigQueryAgentAnalyticsPlugin باستخدام تفاصيل الربط المطلوبة.
دمج المكوّن الإضافي: يتم تمرير المكوّن الإضافي BigQuery الذي تمّت تهيئته إلى الدالة الإنشائية App، ما يضمن تسجيل أحداث تنفيذ الوكيل تلقائيًا.
تنفيذ الوكيل للسجلّات وعمليات التشغيل: ينفّذ مسار المحادثة من خلال runner.run_async، ويجمع المكوّن الإضافي في الوقت نفسه تسلسل الأحداث بالكامل ويرسله إلى BigQuery قبل إغلاق موارده.

انسخ الرمز التالي وألصقه أسفل تعريفات الوكيل في ملف agent.py:

async def main(prompt: str):
    """Runs a conversation with the BigQuery agent using the ADK Runner."""
    bq_logger_plugin = BigQueryAgentAnalyticsPlugin(
        project_id=PROJECT_ID, dataset_id=DATASET_ID, table_id=TABLE_ID
    )
    app = App(name=APP_NAME, root_agent=root_agent, plugins=[bq_logger_plugin])
    runner = InMemoryRunner(app=app)

    try:
        session_id = f"{USER_ID}_{uuid.uuid4().hex[:8]}"

        my_session = await runner.session_service.create_session(
            app_name=APP_NAME, user_id=USER_ID, session_id=session_id
        )

        async for event in runner.run_async(
            user_id=USER_ID,
            new_message=types.Content(
                role="user", parts=[types.Part.from_text(text=prompt)]
            ),
            session_id=my_session.id,
        ):
            if event.content.parts and event.content.parts[0].text:
                print(f"** {event.author}: {event.content.parts[0].text}")

    except Exception as e:
        print(f"Error in main: {e}")
    finally:
        print("Closing BQ Plugin...")
        await bq_logger_plugin.close()
        print("BQ Plugin closed.")

async def run_all_prompts():
    """Runs all prompts in a single event loop."""
    prompts = [
        "what outfits do you have available that are suitable for the weather in london this week?",
        "You are such a cool agent! I need a gift idea for my friend who likes yoga.",
        "I'd like to complain - the products sold here are not very good quality!"
    ]
    for prompt in prompts:
        await main(prompt)

if __name__ == "__main__":
    asyncio.run(run_all_prompts())

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

python retail_assistant_app/agent.py

من المفترض أن ترى مساعد البيع بالتجزئة وهو ينظّم سير العمل:

يطلب من وكيل مؤشرات البحث في الوقت الفعلي (real_time_agent) تحديد حالة الطقس في لندن والبحث عن مؤشرات الموضة المناسبة.
بعد ذلك، يتم تفويض وكيل بيانات المستودع (inventory_data_agent) لطلب البحث في مجموعة بيانات thelook_ecommerce BigQuery عن منتجات معيّنة تتطابق مع هذه المؤشرات.
أخيرًا، يجمع منسّق الجذر النتائج في توصية نهائية.

في الوقت نفسه، يرسل المكوّن الإضافي بشكل متواصل تتبُّع تنفيذ الوكيل إلى BigQuery.

7. تحليل سجلّات الوكيل

استخدام الأداة

أصبح بإمكاننا الآن معرفة ما كان يفعله موظّف الدعم في الكواليس. تمّ نقل البيانات إلى BigQuery وهي جاهزة للتحليل:

في Google Cloud Console، ابحث عن BigQuery.
في جزء المستكشف، ابحث عن مشروعك.
وسِّع مجموعة بيانات adk_logs.
افتح جدول retail_assitant_agent_logs وانقر على طلب بحث.

للاطّلاع على طلبات الأدوات التي أرسلها وكيلك ورصد أي أخطاء في الأدوات، شغِّل طلب البحث التالي في "محرّر BigQuery":

SELECT
  -- Extract the tool name directly from the JSON key "tool"
  JSON_VALUE(content, '$.tool') AS tool_name,
  -- Count every time a tool finished (successfully or with an error)
  COUNT(*) AS total_finished_runs,
  -- Count as a failure if  event_type is ERROR, result object contains a status of 'ERROR', or error_details exist
  COUNTIF(
    event_type = 'TOOL_ERROR' OR
    JSON_VALUE(content, '$.result.status') = 'ERROR' OR
    JSON_VALUE(content, '$.result.error_details') IS NOT NULL
  ) AS failure_count
FROM
  `adk_logs.retail_assistant_agent_logs`
WHERE
  event_type IN ('TOOL_COMPLETED', 'TOOL_ERROR')
GROUP BY
  1;

انقر على "تصوّر البيانات" لعرض النتائج في رسم بياني (قد تختلف نتائجك):

استخدام الرمز المميز

لاستنتاج تكلفة الوكلاء، يمكنك تجميع الرموز المميزة للطلبات والرموز المميزة للمرشحين التي يستهلكها كل وكيل مميز:

SELECT
 t.agent,
 SUM(LAX_INT64(t.content.usage.prompt)) AS prompt_tokens,
 SUM(LAX_INT64(t.content.usage.completion)) AS completion_tokens
FROM
 `adk_logs.retail_assistant_agent_logs` AS t
WHERE
 t.event_type = 'LLM_RESPONSE'
 -- Filter for records that actually contain usage metadata
 AND t.content.usage IS NOT NULL
GROUP BY 1;

انقر على "تصوّر البيانات" لعرض النتائج في رسم بياني (قد تختلف نتائجك):

8. [مكافأة] تحليل آراء المستخدمين

لنحلّل الآن المشاعر التي عبّر عنها المستخدم في طلبه المقدَّم إلى الوكيل.

في Cloud Shell، أنشئ عملية ربط بمورد على السحابة الإلكترونية لتمكين BigQuery من التفاعل مع خدمات Vertex AI:

bq mk --connection --location=us \
    --connection_type=CLOUD_RESOURCE test_connection

من المفترض أن يظهر لك ردّ على النحو التالي:

تم إنشاء عملية الربط 517325854360.us.test_connection بنجاح

إنشاء عملية ربط بمورد على السحابة الإلكترونية:

export SERVICE_ACCOUNT_EMAIL=$(bq show --format=prettyjson --connection us.test_connection | grep "serviceAccountId" | cut -d '"' -f 4)

نفِّذ هذا الأمر للتأكّد من إنشاء حساب الخدمة بنجاح:

echo $SERVICE_ACCOUNT_EMAIL

من المفترض أن يظهر حساب الخدمة على النحو التالي:

امنح حساب خدمة ربط الموارد أذونات على مستوى المشروع مطلوبة للتفاعل مع Vertex AI:

gcloud projects add-iam-policy-binding $(gcloud config get-value project) \
    --member="serviceAccount:$SERVICE_ACCOUNT_EMAIL" \
    --role='roles/bigquery.connectionUser' && \
gcloud projects add-iam-policy-binding $(gcloud config get-value project) \
    --member="serviceAccount:$SERVICE_ACCOUNT_EMAIL" \
    --role='roles/aiplatform.user'

انتظِر بضع دقائق حتى يتم نشر الأذونات. بعد ذلك، ارجع إلى BigQuery، ونفِّذ الاستعلام التالي الذي يحتوي على الدالة AI.SCORE لتحليل آراء المستخدمين:

SELECT
 timestamp,
 user_id,
 content,
 AI.SCORE((
   'What is the sentiment of the user in this text:', JSON_VALUE(content.text_summary),
   'Use a scale from 1 to 5.'),
 connection_id => 'us.test_connection') AS user_sentiment
FROM
 `adk_logs.retail_assistant_agent_logs`
WHERE
  event_type = 'USER_MESSAGE_RECEIVED'
ORDER BY
 user_sentiment DESC;

ستعيّن دالة AI.SCORE قيمة شعور تتراوح بين 1 و5 لكل إدخال من المستخدم. من المفترض أن تظهر لك نتائج مشابهة لما يلي:

9- تنظيف

لتجنُّب فرض رسوم مستمرة على حسابك على Google Cloud، احذف الموارد التي تم إنشاؤها خلال ورشة العمل هذه.

احذف مجموعة بيانات التسجيل التي أنشأتها السمة البرمجية:

bq rm -r -f -d $PROJECT_ID:adk_logs

احذف ربط مورد السحابة الإلكترونية باتّباع الخطوات التالية:

bq rm --connection --project_id=$PROJECT_ID --location=us test_connection

لإزالة الدليل bigquery-adk-codelab ومحتواه، اتّبِع الخطوات التالية:

cd .. 
rm -rf adk-agent-observability

10. تهانينا

تهانينا! لقد أنشأت نظامًا متعدد الوكلاء باستخدام "حزمة تطوير الوكلاء" (ADK) ودمجت بنجاح "المكوّن الإضافي لتحليلات الوكلاء في BigQuery" لتتبُّع سلوك وكيلك وتدقيقه.