פיתוח סוכן נתונים באמצעות Python SDK

בדף הזה מוסבר איך משתמשים ב-Python SDK כדי לשלוח בקשות אל Conversational Analytics API. בדוגמה לקוד Python אפשר לראות איך לבצע את הפעולות הבאות:

אימות והגדרת הסביבה

כדי להשתמש ב-Python SDK ל-Conversational Analytics API, צריך לפעול לפי ההוראות שב-Conversational Analytics API SDK Colaboratory notebook כדי להוריד ולהתקין את ה-SDK. שימו לב: שיטת ההורדה והתוכן של SDK Colab עשויים להשתנות.

אחרי שמבצעים את הוראות ההגדרה במחברת, אפשר להשתמש בקוד הבא כדי לייבא את ספריות ה-SDK הנדרשות, לאמת את חשבון Google בסביבת Colaboratory ולהפעיל לקוח לשליחת בקשות API:

from google.colab import auth
auth.authenticate_user()

from google.cloud import geminidataanalytics

data_agent_client = geminidataanalytics.DataAgentServiceClient()
data_chat_client = geminidataanalytics.DataChatServiceClient()

ציון פרויקט החיוב והוראות המערכת

בדוגמת הקוד הבאה ב-Python מוגדרים פרויקט החיוב, המיקום וההוראות למערכת שבהם נעשה שימוש לאורך הסקריפט:

# Billing project
billing_project = "my_project_name"
location = "global"

# System instructions
system_instruction = "Help the user analyze their data."

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫my_project_name: המזהה של פרויקט החיוב שבו מופעלים ממשקי ה-API הנדרשים.
  • ‫Help the user analyze their data.: הוראות למערכת שמנחות את התנהגות הסוכן ומאפשרות להתאים אותה לצורכי הנתונים שלכם. לדוגמה, אפשר להשתמש בהוראות מערכת כדי להגדיר מונחים עסקיים, לשלוט באורך התשובה או להגדיר את עיצוב הנתונים. מומלץ להגדיר את ההוראות למערכת באמצעות פורמט ה-YAML המומלץ במאמר איך כותבים הוראות למערכת בצורה יעילה כדי לספק הנחיות מפורטות ומובְנות.

התחברות למקור נתונים

בקטעים הבאים מוסבר איך מגדירים את פרטי החיבור למקורות הנתונים של הסוכן. הסוכן יכול להתחבר לנתונים ב-Looker, ב-BigQuery או ב-Looker Studio.

חיבור לנתונים ב-Looker

בדוגמאות הקוד הבאות אפשר לראות איך מגדירים את הפרטים של חיבור ל-Looker Explore באמצעות מפתחות API או אסימון גישה. אפשר לקשר עד חמישה Looker Explores בכל פעם באמצעות Conversational Analytics API.

כשמתחברים למקור נתונים של Looker, חשוב לשים לב לנקודות הבאות:

  • אתם יכולים לשאול שאלות על כל דוח Explore שכלול בשיחה.
  • סוכן יכול לשלוח שאילתה רק למסמך Explore אחד בכל פעם. אי אפשר להריץ שאילתות בכמה דוחות Explore בו-זמנית.
  • נציג יכול לשאול כמה שאלות ב-Explore באותה שיחה.

  • סוכן יכול לשלוח שאילתות לכמה ניתוחים בשיחה שכוללת שאלות עם כמה חלקים, או בשיחות שכוללות שאלות המשך.

    לדוגמה: משתמש מחבר שני דוחות Explore, אחד בשם cat-explore ואחד בשם dog-explore. המשתמש מזין את השאלה 'מה גדול יותר: מספר החתולים או מספר הכלבים?' כך יישאלו שתי שאילתות: אחת לספירת מספר החתולים ב-cat-explore ואחת לספירת מספר הכלבים ב-dog-explore. הנציג משווה את המספר משתי השאילתות אחרי שהוא משלים את שתיהן.

בדוגמת הקוד הבאה מוגדר חיבור לכמה אפשרויות נוספות ב-Looker. כדי לשפר את הביצועים של הסוכן, אתם יכולים לספק שאילתות לדוגמה כהקשר מובנה לניתוחים שלכם. מידע נוסף זמין במאמר בנושא הגדרת הקשר של סוכן נתונים למקורות נתונים של Looker.

מפתחות API

אתם יכולים ליצור קישור למופע Looker באמצעות מפתחות Looker API שנוצרו, כמו שמתואר במאמר אימות וקישור למקור נתונים באמצעות Conversational Analytics API.

looker_client_id = "my_looker_client_id"
looker_client_secret = "my_looker_client_secret"
looker_instance_uri = "https://my_company.looker.com"
lookml_model_1 = "my_model"
explore_1 = "my_explore"
lookml_model_2 = "my_model_2"
explore_2 = "my_explore_2"

looker_explore_reference = geminidataanalytics.LookerExploreReference()
looker_explore_reference.looker_instance_uri = looker_instance_uri
looker_explore_reference.lookml_model = "my_model"
looker_explore_reference.explore = "my_explore"

looker_explore_reference2 = geminidataanalytics.LookerExploreReference()
looker_explore_reference2.looker_instance_uri = looker_instance_uri
looker_explore_reference2.lookml_model = "my_model_2"
looker_explore_reference2.explore = "my_explore_2"

credentials = geminidataanalytics.Credentials()
credentials.oauth.secret.client_id = looker_client_id
credentials.oauth.secret.client_secret = looker_client_secret

datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.looker.explore_references = [looker_explore_reference, looker_explore_reference2]

# Do not include the following line during agent creation
datasource_references.credentials = credentials

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫my_looker_client_id: מזהה הלקוח של מפתח Looker API שנוצר.
  • ‫my_looker_client_secret: ה-client secret של מפתח Looker API שנוצר.
  • ‫https://my_company.looker.com: כתובת ה-URL המלאה של מכונת Looker.
  • ‫my_model: השם של מודל LookML שכולל את כלי הניתוח שרוצים להתחבר אליו.
  • ‫my_explore: השם של Looker Explore שרוצים שהסוכן לגישה לנתונים ישלח אליו שאילתה.
  • ‫my_model_2: השם של מודל LookML השני שכולל את התכונה Explore שאליה רוצים להתחבר. אפשר לחזור על התהליך הזה כדי להוסיף משתנים למודלים נוספים, עד חמישה ניתוחים.
  • ‫my_explore_2: השם של Looker Explore נוסף שרוצים שהסוכן לגישה לנתונים יריץ עליו שאילתה. אפשר לחזור על המשתנה הזה כדי לכלול עד חמישה ניתוחים.

טוקן גישה

אתם יכולים ליצור קישור למופע Looker באמצעות טוקן גישה, כמו שמתואר במאמר אימות וקישור למקור נתונים באמצעות Conversational Analytics API.

looker_access_token = "my_access_token"
looker_instance_uri = "https://my_company.looker.com"
lookml_model = "my_model"
explore = "my_explore"
lookml_model_2 = "my_model_2"
explore_2 = "my_explore_2"

looker_explore_reference = geminidataanalytics.LookerExploreReference()
looker_explore_reference.looker_instance_uri = looker_instance_uri
looker_explore_reference.lookml_model = "my_model"
looker_explore_reference.explore = "my_explore"

looker_explore_reference2 = geminidataanalytics.LookerExploreReference()
looker_explore_reference2.looker_instance_uri = looker_instance_uri
looker_explore_reference2.lookml_model = "my_model_2"
looker_explore_reference2.explore = "my_explore_2"

credentials = geminidataanalytics.Credentials()
credentials.oauth.secret.client_id = looker_client_id
credentials.oauth.secret.client_secret = looker_client_secret

datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.looker.explore_references = [looker_explore_reference, looker_explore_reference2]

# Do not include the following line during agent creation
datasource_references.credentials = credentials

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫my_access_token: הערך access_token שיוצרים כדי לבצע אימות ל-Looker.
  • ‫https://my_company.looker.com: כתובת ה-URL המלאה של מכונת Looker.
  • ‫my_model: השם של מודל LookML שכולל את כלי הניתוח שרוצים להתחבר אליו.
  • ‫my_explore: השם של Looker Explore שרוצים שהסוכן לגישה לנתונים ישלח אליו שאילתה.
  • ‫my_model_2: השם של מודל LookML השני שכולל את התכונה Explore שאליה רוצים להתחבר. אפשר לחזור על התהליך הזה כדי להוסיף משתנים למודלים נוספים, עד חמישה ניתוחים.
  • ‫my_explore_2: השם של Looker Explore נוסף שרוצים שהסוכן לגישה לנתונים יריץ עליו שאילתה. אפשר לחזור על המשתנה הזה כדי לכלול עד חמישה ניתוחים.

התחברות לנתוני BigQuery

ב-Conversational Analytics API, אין מגבלות קשיחות על מספר הטבלאות ב-BigQuery שאפשר להתחבר אליהן. עם זאת, חיבור למספר גדול של טבלאות עלול להפחית את רמת הדיוק או לגרום לחריגה ממגבלת הטוקנים של הקלט במודל.

קוד לדוגמה הבא מגדיר חיבור לכמה טבלאות ב-BigQuery, והוא כולל דוגמאות לשדות אופציונליים של הקשר מובנה. כדי לשפר את הביצועים של הסוכן, אתם יכולים לספק הקשר מובנה לטבלאות BigQuery, כמו תיאורים של טבלאות ועמודות, מילים נרדפות, תגים ושאילתות לדוגמה. מידע נוסף זמין במאמר בנושא הגדרת הקשר של סוכן הנתונים למקורות נתונים של BigQuery.

bigquery_table_reference_1 = geminidataanalytics.BigQueryTableReference()
bigquery_table_reference_1.project_id = "my_project_id_1"
bigquery_table_reference_1.dataset_id = "my_dataset_id_1"
bigquery_table_reference_1.table_id = "my_table_id_1"
bigquery_table_reference_1.schema = geminidataanalytics.Schema()
bigquery_table_reference_1.schema.description = "my_table_description"
bigquery_table_reference_1.schema.fields = [
  geminidataanalytics.Field(
    name="my_column_name",
    description="my_column_description"
  )
]

bigquery_table_reference_2 = geminidataanalytics.BigQueryTableReference()
bigquery_table_reference_2.project_id = "my_project_id_2"
bigquery_table_reference_2.dataset_id = "my_dataset_id_2"
bigquery_table_reference_2.table_id = "my_table_id_2"

bigquery_table_reference_3 = geminidataanalytics.BigQueryTableReference()
bigquery_table_reference_3.project_id = "my_project_id_3"
bigquery_table_reference_3.dataset_id = "my_dataset_id_3"
bigquery_table_reference_3.table_id = "my_table_id_3"

# Connect to your data source
datasource_references = geminidataanalytics.DatasourceReferences()
datasource_references.bq.table_references = [bigquery_table_reference_1, bigquery_table_reference_2, bigquery_table_reference_3]

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫my_project_id_1: המזהה של הפרויקט Google Cloud שמכיל את מערך הנתונים והטבלה ב-BigQuery שאליהם רוצים להתחבר. כדי להתחבר למערך נתונים ציבורי, מציינים bigquery-public-data.
  • ‫my_dataset_id_1: המזהה של מערך הנתונים ב-BigQuery. לדוגמה, san_francisco.
  • ‫my_table_id_1: המזהה של הטבלה ב-BigQuery. לדוגמה, street_trees.
  • ‫my_table_description: תיאור אופציונלי של תוכן הטבלה והמטרה שלה.
  • ‫my_column_name: שם העמודה בטבלה שרוצים לספק לה תיאור אופציונלי.
  • ‫my_column_description: תיאור אופציונלי של התוכן והמטרה של העמודה.

חיבור לנתונים ב-Looker Studio

בדוגמת הקוד הבאה מוגדר חיבור למקור נתונים ב-Looker Studio.

studio_datasource_id = "my_datasource_id"

studio_references = geminidataanalytics.StudioDatasourceReference()
studio_references.datasource_id = studio_datasource_id

## Connect to your data source
datasource_references.studio.studio_references = [studio_references]

בדוגמה הקודמת, מחליפים את my_datasource_id במזהה של מקור הנתונים.

הגדרת הקשר לשיחה עם שמירת מצב או לשיחה ללא שמירת מצב

ממשק ה-API של ניתוח נתונים בממשק שיחה תומך בשיחות מרובות, שמאפשרות למשתמשים לשאול שאלות המשך שמבוססות על הקשר קודם. בדוגמה הבאה של קוד Python אפשר לראות איך מגדירים הקשר לשיחה עם שמירת מצב או לשיחה בלי שמירת מצב:

  • צ'אט עם שמירת מצב: Google Cloud שומר ומנהל את היסטוריית השיחות. צ'אט עם שמירת מצב הוא מטבעו רב-שלבי, כי ה-API שומר את ההקשר מההודעות הקודמות. צריך לשלוח רק את ההודעה הנוכחית בכל תור.
  • צ'אט ללא שמירת מצב: האפליקציה שלכם מנהלת את היסטוריית השיחות. צריך לכלול את כל היסטוריית השיחות בכל הודעה חדשה. דוגמאות מפורטות לניהול שיחות מרובות שלבים במצב חסר מצב זמינות במאמר יצירת שיחה מרובת שלבים חסרת מצב.

צ'אט עם שמירת מצב

בדוגמת הקוד הבאה מוגדר הקשר לשיחה עם שמירת מצב, שבה Google Cloud מאחסן את היסטוריית השיחות ומנהל אותה. אפשר גם להפעיל ניתוח מתקדם באמצעות Python. כדי לעשות זאת, צריך לכלול את השורה published_context.options.analysis.python.enabled = True בקוד לדוגמה הבא.

# Set up context for stateful chat
published_context = geminidataanalytics.Context()
published_context.system_instruction = system_instruction
published_context.datasource_references = datasource_references
# Optional: To enable advanced analysis with Python, include the following line:
published_context.options.analysis.python.enabled = True

צ'אט ללא שמירת מצב

קוד לדוגמה שבו מוגדר הקשר לצ'אט בלי שמירת מצב, שבו צריך לשלוח את כל היסטוריית השיחות עם כל הודעה. אפשר גם להפעיל ניתוח מתקדם באמצעות Python. כדי לעשות זאת, צריך לכלול את השורה inline_context.options.analysis.python.enabled = True בקוד לדוגמה הבא.

# Set up context for stateless chat
# datasource_references.looker.credentials = credentials
inline_context = geminidataanalytics.Context()
inline_context.system_instruction = system_instruction
inline_context.datasource_references = datasource_references
# Optional: To enable advanced analysis with Python, include the following line:
inline_context.options.analysis.python.enabled = True

יצירת סוכן נתונים

דוגמת קוד Python הבאה יוצרת בקשת API ליצירת סוכן נתונים, שאחר כך תוכלו להשתמש בו כדי לנהל שיחה על הנתונים שלכם. אפשר ליצור סוכן באופן סינכרוני או אסינכרוני. סוכן הנתונים מוגדר עם מקור הנתונים, הוראות המערכת וההקשר שצוינו.

סינכרוני

  data_agent_id = "data_agent_1"

  data_agent = geminidataanalytics.DataAgent()
  data_agent.data_analytics_agent.published_context = published_context
  data_agent.name = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}" # Optional

  request = geminidataanalytics.CreateDataAgentRequest(
    parent=f"projects/{billing_project}/locations/{location}",
    data_agent_id=data_agent_id, # Optional
    data_agent=data_agent,
  )

  try:
    response = data_agent_client.create_data_agent_sync(request=request)
    print("Data Agent created")
    print(response)
  except Exception as e:
    print(f"Error creating Data Agent: {e}")

אסינכרוני

  data_agent_id = "data_agent_1"

  data_agent = geminidataanalytics.DataAgent()
  data_agent.data_analytics_agent.published_context = published_context
  data_agent.name = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}" # Optional

  request = geminidataanalytics.CreateDataAgentRequest(
    parent=f"projects/{billing_project}/locations/{location}",
    data_agent_id=data_agent_id, # Optional
    data_agent=data_agent,
  )

  try:
    data_agent_client.create_data_agent(request=request)
    print("Data Agent created")
  except Exception as e:
    print(f"Error creating Data Agent: {e}")

בדוגמה הקודמת, מחליפים את הערך data_agent_1 במזהה ייחודי של סוכן הנתונים.

יצירת שיחה

בדוגמה הבאה של קוד Python מבוצעת בקשת API ליצירת שיחה.

# Initialize request arguments
data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

conversation = geminidataanalytics.Conversation()
conversation.agents = [f'projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}']
conversation.name = f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"

request = geminidataanalytics.CreateConversationRequest(
  parent=f"projects/{billing_project}/locations/{location}",
  conversation_id=conversation_id,
  conversation=conversation,
)

# Make the request
response = data_chat_client.create_conversation(request=request)

# Handle the response
print(response)

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫data_agent_1: המזהה של סוכן הנתונים, כפי שמוגדר בבלוק הקוד לדוגמה במאמר יצירת סוכן נתונים.
  • ‫conversation_1: מזהה ייחודי של השיחה.

ניהול סוכני נתונים ושיחות

בדוגמאות הקוד הבאות אפשר לראות איך לנהל את סוכני הנתונים והשיחות באמצעות Conversational Analytics API. אפשר לבצע את המשימות הבאות:

קבלת סוכן נתונים

בדוגמת הקוד הבאה ב-Python אפשר לראות איך שולחים בקשת API לאחזור סוכן נתונים שיצרתם בעבר.

# Initialize request arguments
data_agent_id = "data_agent_1"
request = geminidataanalytics.GetDataAgentRequest(
  name=f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
)

# Make the request
response = data_agent_client.get_data_agent(request=request)

# Handle the response
print(response)

בדוגמה הקודמת, מחליפים את הערך data_agent_1 במזהה הייחודי של סוכן הנתונים שרוצים לאחזר.

הצגת רשימה של סוכני נתונים

בדוגמה הבאה מוצג קוד שמראה איך אפשר להציג רשימה של כל סוכני הנתונים של פרויקט מסוים באמצעות קריאה לשיטה list_data_agents. כדי לראות את רשימת כל הסוכנים, צריכה להיות לכם הרשאה geminidataanalytics.dataAgents.list בפרויקט. למידע נוסף על תפקידי IAM שכוללים את ההרשאה הזו, אפשר לעיין ברשימת התפקידים המוגדרים מראש.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
request = geminidataanalytics.ListDataAgentsRequest(
  parent=f"projects/{billing_project}/locations/{location}",
)

# Make the request
page_result = data_agent_client.list_data_agents(request=request)

# Handle the response
for response in page_result:
  print(response)

מחליפים את YOUR-BILLING-PROJECT במזהה פרויקט החיוב.

הצגת רשימה של סוכני נתונים שאפשר לגשת אליהם

בדוגמה הבאה אפשר לראות איך מציגים רשימה של כל סוכני הנתונים שאפשר לגשת אליהם בפרויקט מסוים באמצעות קריאה לשיטה list_accessible_data_agents.

billing_project = "YOUR-BILLING-PROJECT"
creator_filter = "YOUR-CREATOR-FILTER"
location = "global"
request = geminidataanalytics.ListAccessibleDataAgentsRequest(
  parent=f"projects/{billing_project}/locations/{location}",
  creator_filter=creator_filter
)

# Make the request
page_result = data_agent_client.list_accessible_data_agents(request=request)

# Handle the response
for response in page_result:
  print(response)

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫YOUR-BILLING-PROJECT: מזהה הפרויקט לחיוב.
  • ‫YOUR-CREATOR-FILTER: המסנן להחלה על סמך יוצר סוכן הנתונים. הערכים האפשריים כוללים NONE (ברירת מחדל), CREATOR_ONLY ו-NOT_CREATOR_ONLY.

עדכון סוכן נתונים

בדוגמת הקוד הבאה אפשר לראות איך מעדכנים סוכן נתונים באופן סינכרוני או אסינכרוני על ידי הפעלת השיטה update_data_agent_sync או update_data_agent במשאב של סוכן הנתונים. הבקשה צריכה לכלול אובייקט DataAgent עם הערכים החדשים של השדות שרוצים לשנות, ופרמטר update_mask שמקבל אובייקט FieldMask כדי לציין אילו שדות לעדכן.

כדי לעדכן סוכן נתונים, צריך את הרשאת ה-IAM‏ geminidataanalytics.dataAgents.update בסוכן. למידע נוסף על תפקידי IAM שכוללים את ההרשאה הזו, אפשר לעיין ברשימת התפקידים המוגדרים מראש.

סינכרוני

data_agent_id = "data_agent_1"
billing_project = "YOUR-BILLING-PROJECT"
data_agent = geminidataanalytics.DataAgent()
data_agent.data_analytics_agent.published_context = published_context
data_agent.name = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
data_agent.description = "Updated description of the data agent."

update_mask = field_mask_pb2.FieldMask(paths=['description', 'data_analytics_agent.published_context'])

request = geminidataanalytics.UpdateDataAgentRequest(
  data_agent=data_agent,
  update_mask=update_mask,
)

try:
  # Make the request
  response = data_agent_client.update_data_agent_sync(request=request)
  print("Data Agent Updated")
  print(response)
except Exception as e:
  print(f"Error updating Data Agent: {e}")

אסינכרוני

data_agent_id = "data_agent_1"
billing_project = "YOUR-BILLING-PROJECT"
data_agent = geminidataanalytics.DataAgent()
data_agent.data_analytics_agent.published_context = published_context
data_agent.name = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
data_agent.description = "Updated description of the data agent."

update_mask = field_mask_pb2.FieldMask(paths=['description', 'data_analytics_agent.published_context'])

request = geminidataanalytics.UpdateDataAgentRequest(
  data_agent=data_agent,
  update_mask=update_mask,
)

try:
  # Make the request
  data_agent_client.update_data_agent(request=request)
  print("Data Agent Updated")
except Exception as e:
  print(f"Error updating Data Agent: {e}")

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫data_agent_1: המזהה של סוכן הנתונים שרוצים לעדכן.
  • ‫YOUR-BILLING-PROJECT: מזהה הפרויקט לחיוב.
  • Updated description of the data agent.: תיאור של סוכן הנתונים המעודכן.

קבלת מדיניות IAM עבור סוכן נתונים

בדוגמת הקוד הבאה אפשר לראות איך משתמשים בשיטה get_iam_policy כדי לאחזר את מדיניות IAM של סוכן נתונים. בבקשה מצוין נתיב המשאב של סוכן הנתונים.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_id = "data_agent_1"

resource = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
request = iam_policy_pb2.GetIamPolicyRequest(
      resource=resource,
    )
try:
  response = data_agent_client.get_iam_policy(request=request)
  print("IAM Policy fetched successfully!")
  print(f"Response: {response}")
except Exception as e:
  print(f"Error setting IAM policy: {e}")

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫YOUR-BILLING-PROJECT: מזהה הפרויקט לחיוב.
  • ‫data_agent_1: המזהה של סוכן הנתונים שעבורו רוצים לקבל את מדיניות ה-IAM.

הגדרת מדיניות IAM לסוכן נתונים

כדי לשתף סוכן, אפשר להשתמש בשיטה set_iam_policy כדי להקצות תפקידי IAM למשתמשים בסוכן ספציפי. הבקשה כוללת קישורים שמציינים אילו תפקידים צריך להקצות לאילו משתמשים.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_id = "data_agent_1"
role = "roles/geminidataanalytics.dataAgentEditor"
users = "222larabrown@gmail.com, cloudysanfrancisco@gmail.com"

resource = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

# Construct the IAM policy
binding = policy_pb2.Binding(
  role=role,
  members= [f"user:{i.strip()}" for i in users.split(",")]
)

policy = policy_pb2.Policy(bindings=[binding])

# Create the request
request = iam_policy_pb2.SetIamPolicyRequest(
  resource=resource,
  policy=policy
)

# Send the request
try:
  response = data_agent_client.set_iam_policy(request=request)
  print("IAM Policy set successfully!")
  print(f"Response: {response}")
except Exception as e:
  print(f"Error setting IAM policy: {e}")

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫YOUR-BILLING-PROJECT: מזהה הפרויקט לחיוב.
  • ‫data_agent_1: המזהה של סוכן הנתונים שעבורו רוצים להגדיר את מדיניות ה-IAM.
  • ‫222larabrown@gmail.com, cloudysanfrancisco@gmail.com: רשימה מופרדת בפסיקים של כתובות אימייל של משתמשים שרוצים להקצות להם את התפקיד שצוין.

מחיקה של סוכן נתונים

בדוגמת הקוד הבאה אפשר לראות איך משתמשים בשיטה delete_data_agent_sync או בשיטה delete_data_agent כדי לבצע מחיקה עם יכולת שחזור של סוכן נתונים באופן סינכרוני או אסינכרוני. כשמבצעים מחיקה עם יכולת שחזור של סוכן, הוא נמחק אבל עדיין אפשר לשחזר אותו תוך 30 יום. בבקשה מצוינת כתובת ה-URL של משאב סוכן הנתונים.

סינכרוני

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_id = "data_agent_1"

request = geminidataanalytics.DeleteDataAgentRequest(
  name=f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
)

try:
  # Make the request
  data_agent_client.delete_data_agent_sync(request=request)
  print("Data Agent Deleted")
except Exception as e:
  print(f"Error deleting Data Agent: {e}")

אסינכרוני

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_id = "data_agent_1"

request = geminidataanalytics.DeleteDataAgentRequest(
  name=f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
)

try:
  # Make the request
  data_agent_client.delete_data_agent(request=request)
  print("Data Agent Deleted")
except Exception as e:
  print(f"Error deleting Data Agent: {e}")

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫YOUR-BILLING-PROJECT: מזהה הפרויקט לחיוב.
  • ‫data_agent_1: המזהה של סוכן הנתונים שרוצים למחוק.

קבלת שיחה

בדוגמת הקוד הבאה אפשר לראות איך משתמשים בשיטה get_conversation כדי לאחזר מידע על שיחה קיימת. הבקשה מציינת את נתיב המשאב של השיחה.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
conversation_id = "conversation_1"

request = geminidataanalytics.GetConversationRequest(
  name = f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"
)

# Make the request
response = data_chat_client.get_conversation(request=request)

# Handle the response
print(response)

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫YOUR-BILLING-PROJECT: מזהה הפרויקט לחיוב.
  • ‫conversation_1: המזהה של השיחה שרוצים לאחזר.

רשימת השיחות

בדוגמת הקוד הבאה מוצג איך אפשר להציג רשימה של שיחות בפרויקט מסוים באמצעות קריאה לשיטה list_conversations. בבקשה מציינים את כתובת ה-URL של משאב האב, כלומר הפרויקט והמיקום (לדוגמה, projects/my-project/locations/global).

כברירת מחדל, השיטה הזו מחזירה את השיחות שיצרתם. אדמינים (משתמשים עם תפקיד IAM‏ cloudaicompanion.topicAdmin) יכולים לראות את כל השיחות בפרויקט.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
request = geminidataanalytics.ListConversationsRequest(
  parent=f"projects/{billing_project}/locations/{location}",
)

# Make the request
response = data_chat_client.list_conversations(request=request)

# Handle the response
print(response)

מחליפים את YOUR-BILLING-PROJECT במזהה של פרויקט החיוב שבו הפעלתם את ממשקי ה-API הנדרשים.

הצגת רשימת ההודעות בשיחה

בדוגמת הקוד הבאה אפשר לראות איך משתמשים במתודה list_messages כדי לאחזר את כל ההודעות בשיחה. הבקשה מציינת את נתיב המשאב של השיחה.

כדי להציג את רשימת ההודעות, צריכה להיות לכם הרשאה מסוג cloudaicompanion.topics.get בשיחה.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"

conversation_id = "conversation_1"

request = geminidataanalytics.ListMessagesRequest(
  parent=f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}",
)

# Make the request
response = data_chat_client.list_messages(request=request)

# Handle the response
print(response)

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫YOUR-BILLING-PROJECT: מזהה הפרויקט לחיוב.
  • ‫conversation_1: המזהה של השיחה שרוצים להציג את ההודעות שלה.

איך מוחקים שיחות

בדוגמה הבאה של קוד לדוגמה אפשר לראות איך מוחקים שיחה. אדמינים (משתמשים עם התפקיד cloudaicompanion.topicAdmin בניהול הזהויות והרשאות הגישה) או משתמשים עם cloudaicompanion.topics.delete ההרשאה בניהול הזהויות והרשאות הגישה יכולים למחוק שיחות בפרויקט.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
conversation_id = "conversation_1"

request = geminidataanalytics.DeleteConversationRequest(
  name = f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"
)

# Make the request
response = data_chat_client.delete_conversation(request=request)

# Handle the response
print(response)

מחליפים את הערכים לדוגמה באופן הבא:

  • ‫YOUR-BILLING-PROJECT: מזהה הפרויקט לחיוב.
  • ‫conversation_1: המזהה של השיחה שרוצים להציג את ההודעות שלה.

איך לשאול שאלות באמצעות ה-API

אחרי שיוצרים סוכן נתונים ושיחה (בצ'אט עם שמירת מצב), אפשר לשלוח שאילתות לסוכן. בדוגמאות הקוד הבאות אפשר לראות איך מפעילים את השיטה data_chat_client.chat. בדוגמאות האלה נעשה שימוש במשתני ההקשר (כמו מקור נתונים והוראות מערכת) שהגדרתם במאמר הגדרת הקשר לצ'אט עם שמירת מצב או בלי שמירת מצב.

כששולחים שאילתה, ה-API מחזיר זרם של אובייקטים מסוג Message. הזרם הזה יכול להכיל סוגים שונים של הודעות, כולל טקסט, טבלאות נתונים ותרשימים. הודעות טקסט יכולות לספק תובנות לגבי ההיגיון של הסוכן, לדווח על ההתקדמות שלו או לתת את התשובה הסופית. המטרה של כל הודעת טקסט מצוינת על ידי הערך TextType שלה:

  • ‫THOUGHT: מציג את תהליך החשיבה הפנימי של הסוכן בזמן שהוא מתכנן איך לענות על השאילתה. הודעות THOUGHT מספקות תובנות שלב אחר שלב לגבי תהליך החשיבה הרציונלית וקבלת ההחלטות של ה-Agent, והן כוללות שני חלקים: parts[0] הוא סיכום החשיבה, שכולל סיכום קצר של טקסט החשיבה המלא, וparts[1] הוא טקסט החשיבה המלא.
  • ‫PROGRESS: דיווח על התקדמות הסוכן בפעולה, כמו אחזור נתונים או הפעלה של כלי. הערך הזה מוחזר רק למקורות נתונים של Looker, והוא מכיל שני חלקים: parts[0] הוא הסיכום, ו-parts[1] הוא הטקסט המלא של ההתקדמות.
  • ‫FINAL_RESPONSE: מספקת את התשובה הסופית לשאילתה.

בדוגמאות הקוד הבאות נעשה שימוש בפונקציית העזר show_message, שמוגדרת במאמר הגדרת פונקציות עזר, כדי לעבד ולהציג כל הודעה בזרם. הוראות להצגת ההודעות האלה בממשק משתמש כשמשתמשים במקורות נתונים של Looker מופיעות במאמר הצגת תשובות של סוכנים למקורות נתונים של Looker.

צ'אט עם שמירת מצב

שליחת בקשת צ'אט עם שמירת מצב עם הפניה Conversation

אתם יכולים לשלוח בקשה לצ'אט עם סוכן הנתונים תוך הפניה למשאב Conversation שיצרתם בעבר.

# Create a request that contains a single user message (your question)
question = "Which species of tree is most prevalent?"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

# Create a conversation_reference
conversation_reference = geminidataanalytics.ConversationReference()
conversation_reference.conversation = f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"
conversation_reference.data_agent_context.data_agent = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
# conversation_reference.data_agent_context.credentials = credentials

# Form the request
request = geminidataanalytics.ChatRequest(
    parent = f"projects/{billing_project}/locations/{location}",
    messages = messages,
    conversation_reference = conversation_reference
)

# Make the request
stream = data_chat_client.chat(request=request, timeout=300 #custom timeout up to 600s)

# Handle the response
for response in stream:
    show_message(response)

מחליפים את הערכים לדוגמה באופן הבא:

  • Which species of tree is most prevalent?: שאלה בשפה טבעית לשליחה לסוכן הנתונים.
  • ‫data_agent_1: המזהה הייחודי של סוכן הנתונים, כפי שמוגדר במאמר יצירת סוכן נתונים.
  • ‫conversation_1: המזהה הייחודי של השיחה, כפי שמוגדר במאמר יצירת שיחה.

צ'אט ללא שמירת מצב

דוגמאות הקוד הבאות מדגימות איך לשלוח שאילתה לסוכן הנתונים כשהגדרתם הקשר לצ'אט בלי שמירת מצב. אפשר לשלוח שאילתות ללא מצב (stateless) על ידי הפניה למשאב DataAgent שהוגדר קודם לכן או על ידי שימוש בהקשר מוטבע בבקשה.

שליחת בקשה לצ'אט בלי שמירת מצב עם הפניה DataAgent

אפשר לשלוח שאילתה לסוכן הנתונים על ידי הפניה לDataAgent משאב שיצרתם בעבר.

# Create a request that contains a single user message (your question)
question = "Which species of tree is most prevalent?"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

data_agent_id = "data_agent_1"

data_agent_context = geminidataanalytics.DataAgentContext()
data_agent_context.data_agent = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
# data_agent_context.credentials = credentials

# Form the request
request = geminidataanalytics.ChatRequest(
    parent=f"projects/{billing_project}/locations/{location}",
    messages=messages,
    data_agent_context = data_agent_context
)

# Make the request
stream = data_chat_client.chat(request=request, timeout=300 #custom timeout up to 600s)

# Handle the response
for response in stream:
    show_message(response)

מחליפים את הערכים לדוגמה באופן הבא:

  • Which species of tree is most prevalent?: שאלה בשפה טבעית לשליחה לסוכן הנתונים.
  • ‫data_agent_1: המזהה הייחודי של סוכן הנתונים, כפי שמוגדר במאמר יצירת סוכן נתונים.

שליחת בקשה לצ'אט בלי שמירת מצב עם הקשר מוטבע

בדוגמת הקוד הבאה אפשר לראות איך משתמשים בפרמטר inline_context כדי לספק הקשר ישירות בבקשת הצ'אט חסרת הסטטוס.

# Create a request that contains a single user message (your question)
question = "Which species of tree is most prevalent?"
messages = [geminidataanalytics.Message()]
messages[0].user_message.text = question

request = geminidataanalytics.ChatRequest(
    inline_context=inline_context,
    parent=f"projects/{billing_project}/locations/{location}",
    messages=messages,
)

# Make the request
stream = data_chat_client.chat(request=request, timeout=300 #custom timeout up to 600s)

# Handle the response
for response in stream:
    show_message(response)

בדוגמה הקודמת, מחליפים את Which species of tree is most prevalent? בשאלה בשפה טבעית כדי לשלוח אותה לסוכן הנתונים.

יצירת שיחה מרובת תפניות ללא שמירת מצב

כדי לשאול שאלות להמשך בשיחה בלי שמירת מצב, האפליקציה צריכה לנהל את ההקשר של השיחה על ידי שליחת היסטוריית ההודעות המלאה עם כל בקשה חדשה. בדוגמה הבאה מוצגות שתי דרכים ליצור שיחה מרובת תפניות: הפניה לסוכן נתונים או שימוש בהקשר מוטבע כדי לספק את מקור הנתונים ישירות.

# List that is used to track previous turns and is reused across requests
conversation_messages = []

data_agent_id = "data_agent_1"

# Use data agent context
data_agent_context = geminidataanalytics.DataAgentContext()
data_agent_context.data_agent = f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
# data_agent_context.credentials = credentials

# Helper function for calling the API
def multi_turn_Conversation(msg):

  message = geminidataanalytics.Message()
  message.user_message.text = msg

  # Send a multi-turn request by including previous turns and the new message
  conversation_messages.append(message)

  request = geminidataanalytics.ChatRequest(
    parent=f"projects/{billing_project}/locations/{location}",
    messages=conversation_messages,
    # Use data agent context
    data_agent_context=data_agent_context,
    # Use inline context
    # inline_context=inline_context,
  )

  # Make the request
  stream = data_chat_client.chat(request=request, timeout=300 #custom timeout up to 600s)

  # Handle the response
  for response in stream:
    show_message(response)
    conversation_messages.append(response)

# Send the first turn request
multi_turn_Conversation("Which species of tree is most prevalent?")

# Send follow-up turn request
multi_turn_Conversation("Can you show me the results as a bar chart?")

בדוגמה הקודמת, מחליפים את הערכים לדוגמה באופן הבא:

  • ‫data_agent_1: המזהה הייחודי של סוכן הנתונים, כפי שמוגדר בדוגמה של בלוק הקוד במאמר יצירת סוכן נתונים.
  • Which species of tree is most prevalent?: שאלה בשפה טבעית לשליחה לסוכן הנתונים.
  • Can you show me the results as a bar chart?: שאלה המשך שמבוססת על השאלה הקודמת או משפרת אותה.

הגדרת פונקציות עזר

קוד לדוגמה הבא מכיל הגדרות של פונקציות עזר שמשמשות בדוגמאות הקוד הקודמות. הפונקציות האלה עוזרות לנתח את התגובה מה-API ולהציג את התוצאות.

import json as json_lib
import textwrap
import time

import altair as alt
import IPython
import pandas as pd
import proto
import requests

from google.protobuf.json_format import MessageToDict, MessageToJson
from IPython.display import display, HTML
from pygments import highlight, lexers, formatters
from google.protobuf import field_mask_pb2
from google.iam.v1 import policy_pb2
from google.iam.v1 import iam_policy_pb2

def handle_text_response(resp):
  parts = resp.parts
  full_text = "".join(parts)
  if "\n" not in full_text and len(full_text) > 80:
    wrapped_text = textwrap.fill(full_text, width=80)
    print(wrapped_text)
  else:
    print(full_text)

def display_schema(data):
  fields = getattr(data, "fields")
  df = pd.DataFrame({
    "Column": map(lambda field: getattr(field, 'name'), fields),
    "Type": map(lambda field: getattr(field, 'type'), fields),
    "Description": map(lambda field: getattr(field, 'description', '-'), fields),
    "Mode": map(lambda field: getattr(field, 'mode'), fields)
  })
  display(df)

def display_section_title(text):
  display(HTML('<h2>{}</h2>'.format(text)))

def format_looker_table_ref(table_ref):
  return 'lookmlModel: {}, explore: {}, lookerInstanceUri: {}'.format(table_ref.lookml_model, table_ref.explore, table_ref.looker_instance_uri)

def format_bq_table_ref(table_ref):
  return '{}.{}.{}'.format(table_ref.project_id, table_ref.dataset_id, table_ref.table_id)

def display_datasource(datasource):
  source_name = ''
  if 'studio_datasource_id' in datasource:
    source_name = getattr(datasource, 'studio_datasource_id')
  elif 'looker_explore_reference' in datasource:
    source_name = format_looker_table_ref(getattr(datasource, 'looker_explore_reference'))
  else:
    source_name = format_bq_table_ref(getattr(datasource, 'bigquery_table_reference'))

  print(source_name)
  display_schema(datasource.schema)

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)

def handle_data_response(resp):
  if "query" in resp:
    query = resp.query
    display_section_title("Retrieval query")
    print(f"Query name: {query.name}")
    if "question" in query:
      print(f"Question: {query.question}")
    if "datasources" in query:
      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 = {}
    for el in resp.result.data:
      for field in fields:
        if field in d:
          d[field].append(el[field])
        else:
          d[field] = [el[field]]

    display(pd.DataFrame(d))

def handle_chart_response(resp):
  def _value_to_dict(v):
    if isinstance(v, proto.marshal.collections.maps.MapComposite):
      return _map_to_dict(v)
    elif isinstance(v, proto.marshal.collections.RepeatedComposite):
      return [_value_to_dict(el) for el in v]
    elif isinstance(v, (int, float, str, bool)):
      return v
    else:
      return MessageToDict(v)

  def _map_to_dict(d):
    out = {}
    for k in d:
      if isinstance(d[k], proto.marshal.collections.maps.MapComposite):
        out[k] = _map_to_dict(d[k])
      else:
        out[k] = _value_to_dict(d[k])
    return out

  if 'query' in resp:
    print(resp.query.instructions)
  elif 'result' in resp:
    vegaConfig = resp.result.vega_config
    vegaConfig_dict = _map_to_dict(vegaConfig)
    alt.Chart.from_json(json_lib.dumps(vegaConfig_dict)).display();

def show_message(msg):
  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'))
  print('\n')