1. مقدمة
في هذا الدرس التطبيقي حول الترميز، ستتعرّف على كيفية استخدام حزمة تطوير البرامج (SDK) الخاصة بلغة Python في واجهة برمجة التطبيقات Conversational Analytics (CA) API مع مصدر بيانات BigQuery. ستتعرّف على كيفية إنشاء وكيل جديد، وكيفية الاستفادة من إدارة حالة المحادثة، وكيفية إرسال الردود وبثها من واجهة برمجة التطبيقات.
المتطلبات الأساسية
- فهم أساسي لـ Google Cloud وGoogle Cloud Console
- مهارات أساسية في واجهة سطر الأوامر وCloud Shell
- إتقان أساسيات البرمجة بلغة Python
ما ستتعلمه
- كيفية استخدام حزمة تطوير البرامج (SDK) المستندة إلى لغة Python الخاصة بواجهة برمجة التطبيقات Conversational Analytics مع مصدر بيانات BigQuery
- كيفية إنشاء وكيل جديد باستخدام واجهة برمجة التطبيقات الخاصة بـ CA
- كيفية الاستفادة من إدارة حالة المحادثة
- كيفية إرسال الردود وبثّها من واجهة برمجة التطبيقات
المتطلبات
- حساب Google Cloud ومشروع Google Cloud
- متصفّح ويب، مثل Chrome
2. الإعداد والمتطلبات
اختيار مشروع
- سجِّل الدخول إلى Google Cloud Console وأنشِئ مشروعًا جديدًا أو أعِد استخدام مشروع حالي. إذا لم يكن لديك حساب على Gmail أو Google Workspace، عليك إنشاء حساب.
- اسم المشروع هو الاسم المعروض للمشاركين في هذا المشروع. وهي سلسلة أحرف لا تستخدمها Google APIs. ويمكنك تعديلها في أي وقت.
- معرّف المشروع هو معرّف فريد في جميع مشاريع Google Cloud ولا يمكن تغييره (لا يمكن تغييره بعد ضبطه). تنشئ Cloud Console تلقائيًا سلسلة فريدة، ولا يهمّك عادةً ما هي. في معظم دروس البرمجة، عليك الرجوع إلى رقم تعريف مشروعك (يُشار إليه عادةً باسم
PROJECT_ID
). إذا لم يعجبك رقم التعريف الذي تم إنشاؤه، يمكنك إنشاء رقم تعريف عشوائي آخر. يمكنك بدلاً من ذلك تجربة اسم مستخدم من اختيارك ومعرفة ما إذا كان متاحًا. لا يمكن تغيير هذا الخيار بعد هذه الخطوة وسيظل ساريًا طوال مدة المشروع. - للعلم، هناك قيمة ثالثة، وهي رقم المشروع، تستخدمها بعض واجهات برمجة التطبيقات. يمكنك الاطّلاع على مزيد من المعلومات حول هذه القيم الثلاث في المستندات.
- بعد ذلك، عليك تفعيل الفوترة في Cloud Console لاستخدام موارد/واجهات برمجة تطبيقات Cloud. لن تكلفك تجربة هذا الدرس البرمجي الكثير، إن وُجدت أي تكلفة على الإطلاق. لإيقاف الموارد وتجنُّب تكبُّد رسوم فوترة تتجاوز هذا البرنامج التعليمي، يمكنك حذف الموارد التي أنشأتها أو حذف المشروع. يمكن لمستخدمي Google Cloud الجدد الاستفادة من برنامج الفترة التجريبية المجانية بقيمة 300 دولار أمريكي.
بدء Cloud Shell
على الرغم من إمكانية تشغيل Google Cloud عن بُعد من الكمبيوتر المحمول، ستستخدم في هذا الدرس العملي Google Cloud Shell، وهي بيئة سطر أوامر تعمل في السحابة الإلكترونية.
من Google Cloud Console، انقر على رمز Cloud Shell في شريط الأدوات أعلى يسار الصفحة:
لن يستغرق توفير البيئة والاتصال بها سوى بضع لحظات. عند الانتهاء، من المفترض أن يظهر لك ما يلي:
يتم تحميل هذه الآلة الافتراضية مزوّدة بكل أدوات التطوير التي ستحتاج إليها. توفّر هذه الخدمة دليلًا منزليًا ثابتًا بسعة 5 غيغابايت، وتعمل على Google Cloud، ما يؤدي إلى تحسين أداء الشبكة والمصادقة بشكل كبير. يمكن إكمال جميع المهام في هذا الدرس العملي ضمن المتصفّح. ليس عليك تثبيت أي تطبيق.
3- قبل البدء
تفعيل واجهات برمجة التطبيقات المطلوبة
لاستخدام خدمات Google Cloud، يجب أولاً تفعيل واجهات برمجة التطبيقات الخاصة بها لمشروعك. ستستخدم خدمات Google Cloud التالية في هذا الدرس العملي:
- Data Analytics API مع Gemini
- Gemini في Google Cloud
- BigQuery API
لتفعيل هذه الخدمات، نفِّذ الأوامر التالية في وحدة Cloud Shell الطرفية:
gcloud services enable geminidataanalytics.googleapis.com
gcloud services enable cloudaicompanion.googleapis.com
gcloud services enable bigquery.googleapis.com
تثبيت حِزم Python
قبل بدء أي مشروع Python، من الممارسات الجيدة إنشاء بيئة افتراضية. يؤدي ذلك إلى عزل التبعيات الخاصة بالمشروع، ما يمنع حدوث تعارضات مع المشاريع الأخرى أو حِزم Python العامة في النظام. في هذا القسم، ستثبِّت uv
من pip، لأنّ Cloud Shell يتضمّن pip.
تثبيت حزمة uv
pip install uv
التأكّد من تثبيت uv بشكل صحيح
uv --version
الناتج المتوقّع
إذا ظهر سطر إخراج يتضمّن uv، يمكنك الانتقال إلى الخطوة التالية. يُرجى العِلم أنّ رقم الإصدار قد يختلف:
إنشاء بيئة افتراضية وتثبيت حِزم
uv init ca-api-codelab
cd ca-api-codelab
uv venv --python 3.12
uv add google-cloud-geminidataanalytics pandas altair
uv pip list | grep -E 'altair|pandas|google-cloud-geminidataanalytics'
الناتج المتوقّع
إذا ظهرت أسطر الإخراج مع الحِزم الثلاث، يمكنك الانتقال إلى الخطوة التالية. يُرجى العِلم أنّ أرقام الإصدارات قد تختلف:
بدء Python
uv run python
يجب أن تبدو شاشتك على النحو التالي:
4. إنشاء وكيل
بعد إعداد بيئة التطوير وتجهيزها، حان الوقت لوضع الأساس لواجهة برمجة التطبيقات Gemini Data Analytics API. تسهّل حزمة تطوير البرامج (SDK) هذه العملية، إذ تتطلّب بضع إعدادات أساسية فقط لإنشاء برنامجك.
ضبط المتغيّرات
استورِد حزمة geminidataanalytics
واضبط متغيّرات البيئة:
import os
from google.cloud import geminidataanalytics
data_agent_client = geminidataanalytics.DataAgentServiceClient()
location = "global"
billing_project = os.environ.get('DEVSHELL_PROJECT_ID')
data_agent_id = "google_trends_analytics_agent"
ضبط تعليمات النظام للوكيل
تستند واجهة برمجة التطبيقات CA API إلى البيانات الوصفية في BigQuery للحصول على مزيد من السياق حول الجداول والأعمدة التي تتم الإشارة إليها. بما أنّ مجموعة البيانات العامة هذه لا تتضمّن أي أوصاف للأعمدة، يمكنك تقديم سياق إضافي للوكيل كسلسلة بتنسيق YAML. يمكنك الرجوع إلى المستندات للاطّلاع على أفضل الممارسات ونموذج يمكنك استخدامه:
system_instruction = """
system_instruction:
- You are a data analyst specializing in the Google Trends dataset.
- When querying, always use the 'week' column for date-based filtering. This needs to be a Sunday. If you are doing week over week comparison, make sure you specify a date that is a Sunday.
- The following columns should be ignored in all queries 'dma_id', 'refresh_date'
- The 'dma_name' column represents the city and state for about 210 metro areas in the USA.
tables:
top_terms:
description: "Represents the 25 most popular search terms by weekly search volume in a given US metro area (DMA)."
fields:
term: "The search query string."
week: "The start date of the week (Sunday) for which the ranking is valid."
rank: "The term's popularity rank from 1 (most popular) to 25."
score: "Relative search interest, where 100 is the peak popularity for the term in that week."
dma_name: "The name of the US metro area, e.g., 'New York NY'."
top_rising_terms:
description: "Represents the 25 fastest-growing ('breakout') search terms by momentum in a given US metro area (DMA)."
fields:
term: "The surging search query string."
week: "The start date of the week (Sunday) for which the ranking is valid."
rank: "The term's breakout rank from 1 (top rising) to 25."
percent_gain: "The percentage growth in search volume compared to the previous period."
dma_name: "The name of the US metro area, e.g., 'Los Angeles CA'."
score: "Relative search interest, where 100 is the peak popularity for the term in that week."
join_instructions:
goal: "Find terms that are simultaneously popular and rising in the same week and metro area."
method: "INNER JOIN the two tables on their common keys."
keys:
- "term"
- "week"
- "dma_name"
golden_queries:
- natural_language_query: "Find all terms in the 'New York NY' area that were in both the top 25 and top 25 rising lists for the week of July 6th, 2025, and show their ranks and percent gain."
sql_query: |
SELECT
top.term,
top.rank AS top_25_rank,
rising.rank AS rising_25_rank,
rising.percent_gain
FROM
`bigquery-public-data.google_trends.top_terms` AS top
INNER JOIN
`bigquery-public-data.google_trends.top_rising_terms` AS rising
ON
top.term = rising.term
AND top.week = rising.week
AND top.dma_name = rising.dma_name
WHERE
top.week = '2025-07-06'
AND top.dma_name = 'New York NY'
ORDER BY
top.rank;
"""
ضبط مصادر بيانات جداول BigQuery
يمكنك الآن ضبط مصادر بيانات جداول BigQuery. تقبل واجهة برمجة التطبيقات CA جداول BigQuery في مصفوفة:
# BigQuery table data sources
bq_top = geminidataanalytics.BigQueryTableReference(
project_id="bigquery-public-data", dataset_id="google_trends", table_id="top_terms"
)
bq_rising = geminidataanalytics.BigQueryTableReference(
project_id="bigquery-public-data", dataset_id="google_trends", table_id="top_rising_terms"
)
datasource_references = geminidataanalytics.DatasourceReferences(
bq=geminidataanalytics.BigQueryTableReferences(table_references=[bq_top, bq_rising]))
ضبط سياق المحادثة ذات الحالة
يمكنك إنشاء الوكيل الجديد باستخدام السياق المنشور الذي يجمع بين تعليمات النظام و مراجع مصادر البيانات وأي خيارات أخرى.
يُرجى العِلم أنّه يمكنك إنشاء stagingContext لاختبار التغييرات والتحقّق من صحتها قبل نشرها. يسمح ذلك للمطوّر بإضافة نظام إدارة إصدارات إلى وكيل البيانات من خلال تحديد contextVersion في طلب المحادثة. في هذا الدرس التطبيقي حول الترميز، ستنشر التطبيق مباشرةً:
# Context setup for stateful chat
published_context = geminidataanalytics.Context(
system_instruction=system_instruction,
datasource_references=datasource_references,
options=geminidataanalytics.ConversationOptions(
analysis=geminidataanalytics.AnalysisOptions(
python=geminidataanalytics.AnalysisOptions.Python(
enabled=False
)
)
),
)
data_agent = geminidataanalytics.DataAgent(
data_analytics_agent=geminidataanalytics.DataAnalyticsAgent(
published_context=published_context
),
)
# Create the agent
data_agent_client.create_data_agent(request=geminidataanalytics.CreateDataAgentRequest(
parent=f"projects/{billing_project}/locations/{location}",
data_agent_id=data_agent_id,
data_agent=data_agent,
))
من المفترض أن تظهر لك نتيجة مشابهة لما يلي بعد إنشاء الوكيل:
الحصول على الوكيل
لنختبر الوكيل للتأكّد من أنّه تم إنشاؤه:
# Test the agent
request = geminidataanalytics.GetDataAgentRequest(
name=data_agent_client.data_agent_path(
billing_project, location, data_agent_id)
)
response = data_agent_client.get_data_agent(request=request)
print(response)
يجب أن تظهر البيانات الوصفية على الوكيل الجديد. سيشمل ذلك عناصر مثل وقت الإنشاء وسياق الوكيل بشأن تعليمات النظام ومصادر البيانات.
5- إنشاء محادثة
أنت الآن جاهز لإنشاء محادثتك الأولى. في هذا الدرس العملي، ستستخدم مرجع محادثة لإجراء محادثة ذات حالة مع وكيلك.
للعلم، يوفّر CA API طرقًا مختلفة للدردشة مع خيارات مختلفة لإدارة الحالة والوكيل. في ما يلي ملخّص سريع عن الطرق الثلاث:
الولاية | سجلّ المحادثات | الوكيل | Code | الوصف | |
الدردشة باستخدام مرجع محادثة | جدار الحماية المتتبِّع لحالة الاتصال | تتم إدارتها من خلال واجهة برمجة التطبيقات | نعم | تواصل محادثة ذات حالة من خلال إرسال رسالة محادثة تشير إلى محادثة حالية وسياق الموظف المرتبط بها. بالنسبة إلى المحادثات المتعددة الأدوار، يخزّن Google Cloud سجلّ المحادثات ويديره. | |
الدردشة باستخدام مرجع وكيل بيانات | جدار الحماية غير المتتبِّع لحالة الاتصال | يديرها المستخدم | نعم | يرسل رسالة محادثة بدون حالة تشير إلى وكيل بيانات محفوظ للحصول على السياق. بالنسبة إلى المحادثات المتعددة الأدوار، يجب أن يدير تطبيقك سجلّ المحادثات ويقدّمه مع كل طلب. | |
الدردشة باستخدام السياق المضمّن | جدار الحماية غير المتتبِّع لحالة الاتصال | يديرها المستخدم | لا | يرسل رسالة محادثة بدون حالة من خلال تقديم كل السياق مباشرةً في الطلب، بدون استخدام وكيل بيانات محفوظ. بالنسبة إلى المحادثات المتعددة الأدوار، يجب أن يدير تطبيقك سجلّ المحادثات ويقدّمه مع كل طلب. |
ستنشئ دالة لإعداد محادثتك وتوفير رقم تعريف فريد لها:
def setup_conversation(conversation_id: str):
data_chat_client = geminidataanalytics.DataChatServiceClient()
conversation = geminidataanalytics.Conversation(
agents=[data_chat_client.data_agent_path(
billing_project, location, data_agent_id)],
)
request = geminidataanalytics.CreateConversationRequest(
parent=f"projects/{billing_project}/locations/{location}",
conversation_id=conversation_id,
conversation=conversation,
)
try:
data_chat_client.get_conversation(name=data_chat_client.conversation_path(
billing_project, location, conversation_id))
print(f"Conversation '{conversation_id}' already exists.")
except Exception:
response = data_chat_client.create_conversation(request=request)
print("Conversation created successfully:")
print(response)
conversation_id = "my_first_conversation"
setup_conversation(conversation_id=conversation_id)
من المفترض أن تظهر لك رسالة تفيد بأنّه تم إنشاء المحادثة بنجاح.
6. إضافة دوال مساعدة
أنت على وشك بدء الدردشة مع موظف الدعم. قبل ذلك، لنضِف بعض دوال الأدوات المساعدة لتنسيق الرسائل حتى يسهل قراءتها وعرض الرسومات البيانية. سترسل واجهة برمجة التطبيقات CA مواصفات vega التي يمكنك رسمها باستخدام حزمة altair:
# Utility functions for streaming and formatting responses
import altair as alt
import http.server
import pandas as pd
import proto
import socketserver
import threading
_server_thread = None
_httpd = None
# Prints a formatted section title
def display_section_title(text):
print(f"\n--- {text.upper()} ---")
# Handles and displays data responses
def handle_data_response(resp):
if "query" in resp:
query = resp.query
display_section_title("Retrieval query")
print(f"Query name: {query.name}")
print(f"Question: {query.question}")
print("Data sources:")
for datasource in query.datasources:
display_datasource(datasource)
elif "generated_sql" in resp:
display_section_title("SQL generated")
print(resp.generated_sql)
elif "result" in resp:
display_section_title("Data retrieved")
fields = [field.name for field in resp.result.schema.fields]
d = {field: [] for field in fields}
for el in resp.result.data:
for field in fields:
d[field].append(el[field])
print(pd.DataFrame(d))
# Starts a local web server to preview charts
def preview_in_browser(port: int = 8080):
"""Starts a web server in a background thread and waits for user to stop it."""
global _server_thread, _httpd
if _server_thread and _server_thread.is_alive():
print(
f"\n--> A new chart was generated. Refresh your browser at http://localhost:{port}")
return
Handler = http.server.SimpleHTTPRequestHandler
socketserver.TCPServer.allow_reuse_address = True
try:
_httpd = socketserver.TCPServer(("", port), Handler)
except OSError as e:
print(f"❌ Could not start server on port {port}: {e}")
return
_server_thread = threading.Thread(target=_httpd.serve_forever)
_server_thread.daemon = False
_server_thread.start()
print("\n" + "=" * 60)
print(" 📈 CHART READY - PREVIEW IN BROWSER ".center(60))
print("=" * 60)
print(
f"1. In the Cloud Shell toolbar, click 'Web Preview' and select port {port}.")
print(f"2. Or, open your local browser to http://localhost:{port}")
print("=" * 60)
try:
input(
"\n--> Press Enter here after viewing all charts to shut down the server...\n\n")
finally:
print("Shutting down server...")
_httpd.shutdown()
_server_thread.join()
_httpd, _server_thread = None, None
print("Server stopped.")
# Handles chart responses
def handle_chart_response(resp, chart_generated_flag: list):
def _value_to_dict(v):
if isinstance(v, proto.marshal.collections.maps.MapComposite):
return {k: _value_to_dict(v[k]) for k in v}
elif isinstance(v, proto.marshal.collections.RepeatedComposite):
return [_value_to_dict(el) for el in v]
return v
if "query" in resp:
print(resp.query.instructions)
elif "result" in resp:
vega_config_dict = _value_to_dict(resp.result.vega_config)
chart = alt.Chart.from_dict(vega_config_dict)
chart_filename = "index.html"
chart.save(chart_filename)
if chart_generated_flag:
chart_generated_flag[0] = True
# Displays the schema of a data source
def display_schema(data):
fields = getattr(data, "fields")
df = pd.DataFrame({
"Column": [f.name for f in fields],
"Type": [f.type for f in fields],
"Description": [getattr(f, "description", "-") for f in fields],
"Mode": [f.mode for f in fields],
})
print(df)
# Displays information about a BigQuery data source
def display_datasource(datasource):
table_ref = datasource.bigquery_table_reference
source_name = f"{table_ref.project_id}.{table_ref.dataset_id}.{table_ref.table_id}"
print(source_name)
display_schema(datasource.schema)
# Handles and displays schema resolution responses
def handle_schema_response(resp):
if "query" in resp:
print(resp.query.question)
elif "result" in resp:
display_section_title("Schema resolved")
print("Data sources:")
for datasource in resp.result.datasources:
display_datasource(datasource)
# Handles and prints simple text responses
def handle_text_response(resp):
parts = resp.parts
print("".join(parts))
# Processes and displays different types of system messages
def show_message(msg, chart_generated_flag: list):
m = msg.system_message
if "text" in m:
handle_text_response(getattr(m, "text"))
elif "schema" in m:
handle_schema_response(getattr(m, "schema"))
elif "data" in m:
handle_data_response(getattr(m, "data"))
elif "chart" in m:
handle_chart_response(getattr(m, "chart"), chart_generated_flag)
print("\n")
7. إنشاء دالة دردشة
الخطوة الأخيرة هي إنشاء دالة محادثة يمكنك إعادة استخدامها واستدعاء الدالة show_message لكل جزء في بث الرد:
def stream_chat_response(question: str):
"""
Sends a chat request, processes the streaming response, and if a chart
was generated, starts the preview server and waits for it to be closed.
"""
data_chat_client = geminidataanalytics.DataChatServiceClient()
chart_generated_flag = [False]
messages = [
geminidataanalytics.Message(
user_message=geminidataanalytics.UserMessage(text=question)
)
]
conversation_reference = geminidataanalytics.ConversationReference(
conversation=data_chat_client.conversation_path(
billing_project, location, conversation_id
),
data_agent_context=geminidataanalytics.DataAgentContext(
data_agent=data_chat_client.data_agent_path(
billing_project, location, data_agent_id
),
),
)
request = geminidataanalytics.ChatRequest(
parent=f"projects/{billing_project}/locations/{location}",
messages=messages,
conversation_reference=conversation_reference,
)
stream = data_chat_client.chat(request=request)
for response in stream:
show_message(response, chart_generated_flag)
if chart_generated_flag[0]:
preview_in_browser()
تم الآن تحديد الدالة stream_chat_response
وأصبحت جاهزة للاستخدام مع طلباتك.
8. Start Chattin'
السؤال 1
أنت الآن جاهز لبدء طرح الأسئلة. في ما يلي بعض المهام التي يمكن لهذا الوكيل تنفيذها:
question = "Hey what data do you have access to?"
stream_chat_response(question=question)
يجب أن يردّ الوكيل بشيء مشابه لما يلي:
السؤال 2
حسنًا، لنحاول العثور على مزيد من المعلومات حول أحدث عبارات البحث الرائجة:
question = "What are the top 20 most popular search terms last week in NYC based on rank? Display each term and score as a column chart"
stream_chat_response(question=question)
سيستغرق تنفيذ هذا الإجراء بعض الوقت. سيظهر لك الوكيل وهو ينفّذ خطوات مختلفة ويقدّم تحديثات مباشرة، بدءًا من استرداد المخطط والبيانات الوصفية، وكتابة طلب بحث SQL، والحصول على النتائج، وتحديد تعليمات العرض المرئي، وتلخيص النتائج.
لعرض الرسم البياني، انتقِل إلى شريط أدوات Cloud Shell، وانقر على "معاينة الويب" (Web Preview) واختَر المنفذ 8080:
من المفترض أن يظهر عرض مرئي للرسم البياني على النحو التالي:
اضغط على Enter لإيقاف الخادم والمتابعة.
السؤال 3
لنجرِّب طرح سؤال متابعة والتعمّق في هذه النتائج:
question = "What was the percent gain in growth for these search terms from the week before?"
stream_chat_response(question=question)
من المفترض أن يظهر لك شيء مشابه لما يلي. في هذه الحالة، أنشأ الوكيل طلب بحث لدمج الجدولَين للعثور على النسبة المئوية للزيادة. يُرجى العِلم أنّ استعلامك قد يبدو مختلفًا قليلاً:
وسيبدو العرض المرئي على النحو التالي:
9- تنظيف
بما أنّ هذا الدرس العملي لا يتضمّن أي منتجات تعمل لفترة طويلة، يكفي إيقاف جلسة Python النشطة ببساطة عن طريق إدخال exit()
في الوحدة الطرفية.
حذف مجلدات وملفات المشروع
إذا كنت تريد إزالة الرمز من بيئة Cloud Shell، استخدِم الأوامر التالية:
cd ~
rm -rf ca-api-codelab
إيقاف واجهات برمجة التطبيقات
لإيقاف واجهات برمجة التطبيقات التي تم تفعيلها سابقًا، نفِّذ هذا الأمر
gcloud services disable geminidataanalytics.googleapis.com
gcloud services disable cloudaicompanion.googleapis.com
gcloud services disable bigquery.googleapis.com
10. الخاتمة
تهانينا، لقد نجحت في إنشاء "وكيل إحصاءات المحادثات" بسيط باستخدام حزمة تطوير البرامج (SDK) الخاصة بـ "إحصاءات المحادثات". اطّلِع على المواد المرجعية لمعرفة المزيد.