1. مقدمه
در این لبه کد یاد خواهید گرفت که چگونه از API Python SDK از Conversational Analytics (CA) با منبع داده BigQuery استفاده کنید. شما یاد خواهید گرفت که چگونه یک عامل جدید ایجاد کنید، چگونه از مدیریت وضعیت مکالمه استفاده کنید، و چگونه پاسخ ها را از API ارسال و پخش کنید.
پیش نیازها
- درک اولیه از Google Cloud و کنسول Google Cloud
- مهارت های اولیه در رابط خط فرمان و Cloud Shell
- مهارت اولیه در برنامه نویسی پایتون
چیزی که یاد خواهید گرفت
- نحوه استفاده از Conversational Analytics API Python SDK با منبع داده BigQuery
- نحوه ایجاد یک عامل جدید با استفاده از CA API
- نحوه استفاده از مدیریت حالت مکالمه
- نحوه ارسال و پخش پاسخ ها از API
آنچه شما نیاز دارید
- یک حساب Google Cloud و پروژه Google Cloud
- یک مرورگر وب مانند کروم
2. راه اندازی و الزامات
یک پروژه را انتخاب کنید
- به Google Cloud Console وارد شوید و یک پروژه جدید ایجاد کنید یا از یک موجود استفاده مجدد کنید. اگر قبلاً یک حساب Gmail یا Google Workspace ندارید، باید یک حساب ایجاد کنید .
- نام پروژه نام نمایشی برای شرکت کنندگان این پروژه است. این یک رشته کاراکتری است که توسط API های Google استفاده نمی شود. همیشه می توانید آن را به روز کنید.
- شناسه پروژه در تمام پروژههای Google Cloud منحصربهفرد است و تغییرناپذیر است (پس از تنظیم نمیتوان آن را تغییر داد). Cloud Console به طور خودکار یک رشته منحصر به فرد تولید می کند. معمولاً برای شما مهم نیست که چیست. در اکثر کدها، باید شناسه پروژه خود را ارجاع دهید (معمولاً با نام
PROJECT_ID
شناخته می شود). اگر شناسه تولید شده را دوست ندارید، ممکن است یک شناسه تصادفی دیگر ایجاد کنید. از طرف دیگر، میتوانید خودتان را امتحان کنید، و ببینید آیا در دسترس است یا خیر. پس از این مرحله نمی توان آن را تغییر داد و در طول مدت پروژه باقی می ماند. - برای اطلاع شما، یک مقدار سوم وجود دارد، یک شماره پروژه ، که برخی از API ها از آن استفاده می کنند. در مورد هر سه این مقادیر در مستندات بیشتر بیاموزید.
- در مرحله بعد، برای استفاده از منابع Cloud/APIها باید صورتحساب را در کنسول Cloud فعال کنید . اجرا کردن از طریق این کد لبه هزینه زیادی نخواهد داشت. برای خاموش کردن منابع برای جلوگیری از تحمیل صورتحساب فراتر از این آموزش، میتوانید منابعی را که ایجاد کردهاید حذف کنید یا پروژه را حذف کنید. کاربران جدید Google Cloud واجد شرایط برنامه آزمایشی رایگان 300 دلاری هستند.
Cloud Shell را راه اندازی کنید
در حالی که Google Cloud را می توان از راه دور از لپ تاپ شما کار کرد، در این کد لبه از Google Cloud Shell استفاده خواهید کرد، یک محیط خط فرمان که در Cloud اجرا می شود.
از Google Cloud Console ، روی نماد Cloud Shell در نوار ابزار بالا سمت راست کلیک کنید:
تهیه و اتصال به محیط فقط چند لحظه طول می کشد. وقتی تمام شد، باید چیزی شبیه به این را ببینید:
این ماشین مجازی با تمام ابزارهای توسعه که شما نیاز دارید بارگذاری شده است. این یک فهرست اصلی 5 گیگابایتی دائمی را ارائه می دهد و در Google Cloud اجرا می شود و عملکرد و احراز هویت شبکه را تا حد زیادی افزایش می دهد. تمام کارهای شما در این کد لبه را می توان در یک مرورگر انجام داد. شما نیازی به نصب چیزی ندارید.
3. قبل از شروع
API های مورد نیاز را فعال کنید
برای استفاده از سرویسهای Google Cloud، ابتدا باید API مربوط به پروژه خود را فعال کنید. شما از خدمات 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
بسته های پایتون را نصب کنید
قبل از شروع هر پروژه پایتون، تمرین خوبی است که یک محیط مجازی ایجاد کنید. این امر وابستگی های پروژه را جدا می کند و از درگیری با پروژه های دیگر یا بسته های جهانی پایتون سیستم جلوگیری می کند. در این بخش، uv
از pip نصب خواهید کرد، زیرا Cloud Shell از قبل پیپ در دسترس دارد.
بسته 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'
خروجی مورد انتظار
اگر خطوط خروجی را با سه بسته مشاهده کردید، خوب است که به مرحله بعدی بروید. توجه داشته باشید که شماره نسخه ممکن است متفاوت باشد:
پایتون را راه اندازی کنید
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 API جداول 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 راه های مختلفی برای چت با گزینه های مختلف مدیریت ایالت و عامل ارائه می دهد. در اینجا خلاصه ای سریع از 3 رویکرد آورده شده است:
ایالت | تاریخچه مکالمه | عامل | کد | توضیحات | |
با استفاده از مرجع مکالمه چت کنید | دولتی | API مدیریت شد | بله | با ارسال یک پیام چت که به یک مکالمه موجود و زمینه عامل مرتبط با آن اشاره می کند، یک مکالمه حالت دار را ادامه می دهد. برای مکالمات چند نوبتی، 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 API مشخصات 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. چت را شروع کنید
سوال 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 بروید، روی «پیشنمایش وب» کلیک کنید و پورت 8080 را انتخاب کنید:
شما باید رندر تصویرسازی را مانند این ببینید:
Enter را فشار دهید تا سرور خاموش شود و ادامه یابد.
سوال 3
بیایید یک سوال بعدی را امتحان کنیم و این نتایج را عمیق تر کنیم:
question = "What was the percent gain in growth for these search terms from the week before?"
stream_chat_response(question=question)
شما باید چیزی شبیه به زیر ببینید. در این حالت عامل یک پرس و جو برای پیوستن به 2 جدول برای یافتن درصد سود ایجاد کرده است. توجه داشته باشید که درخواست شما ممکن است کمی متفاوت به نظر برسد:
و تجسم آن به این صورت خواهد بود:
9. پاکسازی
از آنجایی که این کد لبه هیچ محصول طولانی مدتی را شامل نمی شود، صرفاً متوقف کردن جلسه فعال پایتون با وارد کردن exit()
در ترمینال کافی است.
پوشه ها و فایل های پروژه را حذف کنید
اگر می خواهید کد را از محیط Cloud Shell خود حذف کنید، از دستورات زیر استفاده کنید:
cd ~
rm -rf ca-api-codelab
API ها را غیرفعال کنید
برای غیرفعال کردن API هایی که قبلاً فعال شده بودند، این دستور را اجرا کنید
gcloud services disable geminidataanalytics.googleapis.com
gcloud services disable cloudaicompanion.googleapis.com
gcloud services disable bigquery.googleapis.com
10. نتیجه گیری
تبریک می گوییم، شما با موفقیت یک عامل ساده Conversational Analytics با استفاده از CA SDK ساخته اید. برای کسب اطلاعات بیشتر، مواد مرجع را بررسی کنید!