HTTP 및 Python을 사용하여 데이터 에이전트 빌드

이 페이지에서는 Python을 사용하여 Conversational Analytics API(geminidataanalytics.googleapis.com를 통해 액세스)에 HTTP 요청을 보내는 방법을 안내합니다.

이 페이지의 샘플 Python 코드는 다음 작업을 완료하는 방법을 보여줍니다.

Conversational Analytics API HTTP Colaboratory 노트북에서 이 페이지의 코드 샘플을 실행할 수도 있습니다.

샘플 코드의 전체 버전은 페이지 끝에 포함되어 있으며 API 응답을 스트리밍하는 데 사용되는 도우미 함수도 함께 제공됩니다.

초기 설정 및 인증 구성

다음 샘플 Python 코드는 다음 작업을 수행합니다.

  • 필수 Python 라이브러리를 가져옵니다.
  • 결제 프로젝트, 시스템 안내, 데이터 에이전트를 위한 질문의 변수를 정의합니다.
  • Google Cloud CLI를 사용하여 HTTP 인증을 위한 액세스 토큰을 가져옵니다.
from pygments import highlight, lexers, formatters
import pandas as pd
import json as json_lib
import requests
import json
import altair as alt
import IPython
from IPython.display import display, HTML
import google.auth
from google.auth.transport.requests import Request

from google.colab import auth
auth.authenticate_user()

billing_project = 'YOUR-BILLING-PROJECT'
system_description = 'YOUR-SYSTEM-INSTRUCTIONS'
question = 'YOUR-QUESTION-HERE'

access_token = !gcloud auth application-default print-access-token
headers = {
    "Authorization": f"Bearer {access_token[0]}",
    "Content-Type": "application/json",
}

샘플 값을 다음과 같이 바꿉니다.

  • YOUR-BILLING-PROJECT: 필요한 API를 사용 설정한 결제 프로젝트의 ID입니다.
  • YOUR-SYSTEM-INSTRUCTIONS: 에이전트의 동작을 안내하고 필요에 따라 에이전트를 맞춤설정할 수 있는 시스템 안내입니다. 예를 들어 시스템 안내를 사용하여 비즈니스 용어(예: '충성도 높은 고객'의 정의)를 정의하거나, 응답 길이('20단어 미만으로 요약')를 제어하거나, 데이터 형식을 설정('회사 표준에 부합')할 수 있습니다. 자리표시자 텍스트를 데이터 및 사용 사례와 관련된 안내로 바꾸세요.
  • YOUR-QUESTION-HERE: 데이터 에이전트에게 보낼 자연어 질문입니다.

Looker에 인증

Looker 데이터 소스에 연결하려면 Looker 인스턴스에 인증해야 합니다.

API 키 사용

다음 Python 코드 샘플은 API 키를 사용하여 Looker 인스턴스에 에이전트를 인증하는 방법을 보여줍니다.

looker_credentials = {
    "oauth": {
        "secret": {
            "client_id": "YOUR-LOOKER-CLIENT-ID",
            "client_secret": "YOUR-LOOKER-CLIENT-SECRET",
        }
    }
}

샘플 값을 다음과 같이 바꿉니다.

  • YOUR-LOOKER-CLIENT-ID: 생성된 Looker API 키의 클라이언트 ID입니다.
  • YOUR-LOOKER-CLIENT-SECRET: 생성된 Looker API 키의 클라이언트 보안 비밀번호입니다.

액세스 토큰 사용

다음 Python 코드 샘플은 액세스 토큰을 사용하여 Looker 인스턴스에 에이전트를 인증하는 방법을 보여줍니다.

looker_credentials = {
    "oauth": {
        "token": {
            "access_token": "YOUR-TOKEN",
        }
    }
}

샘플 값을 다음과 같이 바꿉니다.

  • YOUR-TOKEN: Looker에 인증하기 위해 생성하는 access_token 값입니다.

데이터 소스에 연결

다음 Python 코드 샘플은 에이전트가 사용할 Looker, BigQuery 또는 Looker Studio 데이터 소스를 정의하는 방법을 보여줍니다.

Looker 데이터에 연결

다음 샘플 코드는 Looker Explore에 대한 연결을 정의합니다. Looker 인스턴스와의 연결을 설정하려면 Conversational Analytics API로 데이터 소스를 인증하고 연결에 설명된 대로 Looker API 키를 생성해야 합니다.

looker_data_source = {
    "looker": {
        "explore_references": {
            "looker_instance_uri": "https://your_company.looker.com",
            "lookml_model": "your_model",
            "explore": "your_explore",
       },
    }
}

샘플 값을 다음과 같이 바꿉니다.

  • https://your_company.looker.com: Looker 인스턴스의 전체 URL입니다.
  • your_model: 연결할 Explore가 포함된 LookML 모델의 이름입니다.
  • your_explore: 데이터 에이전트가 쿼리할 Looker Explore의 이름입니다.

BigQuery 데이터에 연결

Conversational Analytics API를 사용하면 한 번에 최대 10개의 BigQuery 테이블에 연결하고 쿼리할 수 있습니다.

다음 샘플 코드는 BigQuery 테이블에 대한 연결을 정의합니다.

bigquery_data_sources = {
    "bq": {
        "tableReferences": [
            {
                "projectId": "bigquery-public-data",
                "datasetId": "san_francisco",
                "tableId": "street_trees",
            }
        ]
    }
}

샘플 값을 다음과 같이 바꿉니다.

  • bigquery-public-data: 연결하려는 BigQuery 데이터 세트 및 테이블이 포함된 Google Cloud 프로젝트의 ID입니다. 공개 데이터 세트에 연결하려면 bigquery-public-data를 지정합니다.
  • san_francisco: BigQuery 데이터 세트의 ID입니다.
  • street_trees: BigQuery 테이블의 ID입니다.

Looker Studio 데이터에 연결

다음 샘플 코드는 Looker Studio 데이터 소스에 대한 연결을 정의합니다.

looker_studio_data_source = {
    "studio":{
        "studio_references": [
            {
              "studio_datasource_id": "studio_datasource_id"
            }
        ]
    }
}

studio_datasource_id를 데이터 소스 ID로 바꿉니다.

데이터 에이전트 만들기

다음 샘플 코드는 데이터 에이전트 생성 엔드포인트에 HTTP POST 요청을 전송하여 데이터 에이전트를 만드는 방법을 보여줍니다. 요청 페이로드에는 다음 세부정보가 포함됩니다.

  • 에이전트의 전체 리소스 이름. 이 값에는 프로젝트 ID, 위치, 에이전트의 고유 식별자가 포함됩니다.
  • 데이터 에이전트 설명
  • 시스템 설명(초기 설정 및 인증 구성에 정의됨) 및 에이전트가 사용하는 데이터 소스(데이터 소스에 연결에 정의됨)를 포함한 데이터 에이전트의 컨텍스트

요청 페이로드에 options 파라미터를 포함하여 Python으로 고급 분석을 선택적으로 사용 설정할 수도 있습니다.

data_agent_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}/dataAgents"

data_agent_id = "data_agent_1"

data_agent_payload = {
      "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
      "description": "This is the description of data_agent_1.", # Optional

      "data_analytics_agent": {
          "published_context": {
              "datasource_references": bigquery_data_sources,
              "system_instruction": system_instruction,
              # Optional: To enable advanced analysis with Python, include the following options block:
              "options": {
                  "analysis": {
                      "python": {
                          "enabled": True
                      }
                  }
              }
          }
      }
  }

params = {"data_agent_id": data_agent_id} # Optional

data_agent_response = requests.post(
    data_agent_url, params=params, json=data_agent_payload, headers=headers
)

if data_agent_response.status_code == 200:
    print("Data Agent created successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error creating Data Agent: {data_agent_response.status_code}")
    print(data_agent_response.text)

샘플 값을 다음과 같이 바꿉니다.

  • data_agent_1: 데이터 에이전트의 고유 식별자입니다. 이 값은 에이전트의 리소스 이름에 사용되며 data_agent_id URL 쿼리 파라미터로 사용됩니다.
  • This is the description of data_agent_1.: 데이터 에이전트에 대한 설명입니다.

대화 만들기

다음 샘플 코드는 데이터 에이전트와 대화를 만드는 방법을 보여줍니다.

conversation_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}/conversations"

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

conversation_payload = {
    "agents": [
        f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
    ],
    "name": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"
}
params = {
    "conversation_id": conversation_id
}

conversation_response = requests.post(conversation_url, headers=headers, params=params, json=conversation_payload)

if conversation_response.status_code == 200:
    print("Conversation created successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error creating Conversation: {conversation_response.status_code}")
    print(conversation_response.text)

샘플 값을 다음과 같이 바꿉니다.

  • data_agent_1: 데이터 에이전트의 ID로, 데이터 에이전트 만들기의 샘플 코드 블록에 정의되어 있습니다.
  • conversation_1: 대화의 고유 식별자입니다.

API를 사용하여 질문하기

데이터 에이전트와 대화를 만든 후 데이터에 대해 질문할 수 있습니다.

Conversational Analytics API는 사용자가 이전 컨텍스트를 기반으로 후속 질문을 할 수 있는 멀티턴 대화를 지원합니다. 이 API는 대화 기록을 관리하기 위한 다음 메서드를 제공합니다.

  • 스테이트풀(Stateful) 채팅: Google Cloud 에서 대화 기록을 저장하고 관리합니다. 스테이트풀(Stateful) 채팅은 API가 이전 메시지의 컨텍스트를 유지하므로 본질적으로 멀티턴입니다. 각 대화 차례에 현재 메시지만 전송하면 됩니다.

  • 스테이트리스(Stateless) 채팅: 애플리케이션에서 대화 기록을 관리합니다. 각 새 메시지와 함께 관련된 이전 메시지를 포함해야 합니다. 스테이트리스(Stateless) 모드에서 멀티턴 대화를 관리하는 방법에 대한 자세한 예는 스테이트리스(Stateless) 멀티턴 대화 만들기를 참조하세요.

스테이트풀(Stateful) 채팅

대화 참조를 사용하여 스테이트풀(Stateful) 채팅 요청 보내기

다음 샘플 코드는 이전 단계에서 정의한 대화를 사용하여 API에 질문하는 방법을 보여줍니다. 이 샘플에서는 get_stream 도우미 함수를 사용하여 대답을 스트리밍합니다.

chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "data_agent_1"
conversation_id = "conversation_1"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "conversation_reference": {
        "conversation": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}",
        "data_agent_context": {
            "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
            # "credentials": looker_credentials
        }
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

샘플 값을 다음과 같이 바꿉니다.

  • data_agent_1: 데이터 에이전트의 ID로, 데이터 에이전트 만들기의 샘플 코드 블록에 정의되어 있습니다.
  • conversation_1: 대화의 고유 식별자입니다.
  • Make a bar graph for the top 5 states by the total number of airports가 샘플 프롬프트로 사용되었습니다.

스테이트리스(Stateless) 채팅

데이터 에이전트 참조를 사용하여 스테이트리스(Stateless) 채팅 요청 보내기

다음 샘플 코드는 이전 단계에서 정의한 데이터 에이전트를 사용하여 API에 스테이트리스(Stateless) 질문을 하는 방법을 보여줍니다. 이 샘플에서는 get_stream 도우미 함수를 사용하여 대답을 스트리밍합니다.

chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}:chat"

data_agent_id = "data_agent_1"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "data_agent_context": {
        "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
        # "credentials": looker_credentials
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

샘플 값을 다음과 같이 바꿉니다.

  • data_agent_1: 데이터 에이전트의 ID로, 데이터 에이전트 만들기의 샘플 코드 블록에 정의되어 있습니다.
  • Make a bar graph for the top 5 states by the total number of airports가 샘플 프롬프트로 사용되었습니다.

인라인 컨텍스트를 사용하여 스테이트리스(Stateless) 채팅 요청 보내기

다음 샘플 코드는 인라인 컨텍스트를 사용하여 API에 스테이트리스(Stateless) 질문을 하는 방법을 보여줍니다. 이 샘플에서는 get_stream 도우미 함수를 사용하여 대답을 스트리밍하고 BigQuery 데이터 소스를 예로 사용합니다.

요청 페이로드에 options 파라미터를 포함하여 Python으로 고급 분석을 선택적으로 사용 설정할 수도 있습니다.

chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/global:chat"

# Construct the payload
chat_payload = {
    "parent": f"projects/{billing_project}/locations/global",
    "messages": [
        {
            "userMessage": {
                "text": "Make a bar graph for the top 5 states by the total number of airports"
            }
        }
    ],
    "inline_context": {
        "datasource_references": bigquery_data_sources,
          # Optional: To enable advanced analysis with Python, include the following options block:
          "options": {
              "analysis": {
                  "python": {
                      "enabled": True
                  }
              }
          }
    }
}

# Call the get_stream function to stream the response
get_stream(chat_url, chat_payload)

스테이트리스(Stateless) 멀티턴 대화 만들기

스테이트리스(Stateless) 대화에서 후속 질문을 하려면 애플리케이션이 각 새 요청과 함께 전체 메시지 기록을 전송하여 대화의 컨텍스트를 관리해야 합니다. 다음 섹션에서는 멀티턴 대화를 만들기 위해 도우미 함수를 정의하고 호출하는 방법을 보여줍니다.

멀티턴 요청 보내기

다음 multi_turn_Conversation 도우미 함수는 목록에 메시지를 저장하여 대화 컨텍스트를 관리합니다. 이렇게 하면 이전 턴을 기반으로 하는 후속 질문을 보낼 수 있습니다. 함수의 페이로드에서 데이터 에이전트를 참조하거나 인라인 컨텍스트를 사용하여 데이터 소스를 직접 제공할 수 있습니다.

chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/global:chat"

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

data_agent_id = "data_agent_1"

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

  userMessage = {
      "userMessage": {
          "text": msg
      }
  }

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

  # Construct the payload
  chat_payload = {
      "parent": f"projects/{billing_project}/locations/global",
      "messages": conversation_messages,
      # Use a data agent reference
      "data_agent_context": {
          "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
          # "credentials": looker_credentials
      },
      # Use inline context
      # "inline_context": {
      #     "datasource_references": bigquery_data_sources,
      # }
  }

  # Call the get_stream_multi_turn helper function to stream the response
  get_stream_multi_turn(chat_url, chat_payload, conversation_messages)

이전 예시에서 data_agent_1을 데이터 에이전트 만들기의 샘플 코드 블록에 정의된 데이터 에이전트의 ID로 바꿉니다.

대화의 각 턴에서 multi_turn_Conversation 도우미 함수를 호출할 수 있습니다. 다음 샘플 코드는 초기 요청을 전송한 후 이전 대답을 기반으로 하는 후속 요청을 전송하는 방법을 보여줍니다.

# Send 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?")

이전 예시에서 샘플 값을 다음과 같이 바꿉니다.

  • Which species of tree is most prevalent?: 데이터 에이전트에게 보낼 자연어 질문입니다.
  • Can you show me the results as a bar chart?: 이전 질문을 기반으로 하거나 이전 질문을 구체화하는 후속 질문입니다.

응답 처리

다음 get_stream_multi_turn 함수는 스트리밍 API 응답을 처리합니다. 이 함수는 get_stream 도우미 함수와 비슷하지만 다음 턴의 대화 컨텍스트를 저장하기 위해 conversation_messages 목록에 응답을 저장합니다.

def get_stream_multi_turn(url, json, conversation_messages):
    s = requests.Session()

    acc = ''

    with s.post(url, json=json, headers=headers, stream=True) as resp:
        for line in resp.iter_lines():
            if not line:
                continue

            decoded_line = str(line, encoding='utf-8')

            if decoded_line == '[{':
                acc = '{'
            elif decoded_line == '}]':
                acc += '}'
            elif decoded_line == ',':
                continue
            else:
                acc += decoded_line

            if not is_json(acc):
                continue

            data_json = json_lib.loads(acc)
            # Store the response that will be used in the next iteration
            conversation_messages.append(data_json)

            if not 'systemMessage' in data_json:
                if 'error' in data_json:
                    handle_error(data_json['error'])
                continue

            if 'text' in data_json['systemMessage']:
                handle_text_response(data_json['systemMessage']['text'])
            elif 'schema' in data_json['systemMessage']:
                handle_schema_response(data_json['systemMessage']['schema'])
            elif 'data' in data_json['systemMessage']:
                handle_data_response(data_json['systemMessage']['data'])
            elif 'chart' in data_json['systemMessage']:
                handle_chart_response(data_json['systemMessage']['chart'])
            else:
                colored_json = highlight(acc, lexers.JsonLexer(), formatters.TerminalFormatter())
                print(colored_json)
            print('\n')
            acc = ''

엔드 투 엔드 코드 샘플

다음의 확장 가능한 코드 샘플에는 이 가이드에서 다루는 모든 작업이 포함되어 있습니다.

HTTP 및 Python을 사용하여 데이터 에이전트 빌드

    from pygments import highlight, lexers, formatters
    import pandas as pd
    import json as json_lib
    import requests
    import json
    import altair as alt
    import IPython
    from IPython.display import display, HTML
    import requests
    import google.auth
    from google.auth.transport.requests import Request

    from google.colab import auth
    auth.authenticate_user()

    access_token = !gcloud auth application-default print-access-token
    headers = {
        "Authorization": f"Bearer {access_token[0]}",
        "Content-Type": "application/json",
    }

    ################### Data source details ###################

    billing_project = "your_billing_project"
    location = "global"
    system_instruction = "Help the user in analyzing their data"


    # BigQuery data source
    bigquery_data_sources = {
        "bq": {
        "tableReferences": [
            {
            "projectId": "bigquery-public-data",
            "datasetId": "san_francisco",
            "tableId": "street_trees"
            }
        ]
        }
    }

    # Looker data source
    looker_credentials = {
        "oauth": {
            "secret": {
            "client_id": "your_looker_client_id",
            "client_secret": "your_looker_client_secret",
            }
        }
    }
    # # To use access_token for authentication, uncomment the following looker_credentials code block and comment out the previous looker_credentials code block.
    # looker_credentials = {
    #     "oauth": {
    #         "token": {
    #           "access_token": "your_looker_access_token",
    #         }
    #     }
    # }
    looker_data_source = {
        "looker": {
        "explore_references": {
            "looker_instance_uri": "https://my_company.looker.com",
            "lookml_model": "my_model",
            "explore": "my_explore",
        },
        # "credentials": looker_credentials
    }

    # Looker Studio data source
    looker_studio_data_source = {
        "studio":{
            "studio_references":
            [
                {
                "datasource_id": "your_studio_datasource_id"
                }
            ]
        }
    }

    ################### Create data agent ###################
    data_agent_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}/dataAgents"

    data_agent_id = "data_agent_1"

    data_agent_payload = {
        "name": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}", # Optional
        "description": "This is the description of data_agent.", # Optional

        "data_analytics_agent": {
            "published_context": {
                "datasource_references": bigquery_data_sources,
                "system_instruction": system_instruction,
                # Optional: To enable advanced analysis with Python, include the following options block:
                "options": {
                    "analysis": {
                        "python": {
                            "enabled": True
                        }
                    }
                }
            }
        }
    }

    params = {"data_agent_id": data_agent_id} # Optional

    data_agent_response = requests.post(
        data_agent_url, params=params, json=data_agent_payload, headers=headers
    )

    if data_agent_response.status_code == 200:
        print("Data Agent created successfully!")
        print(json.dumps(data_agent_response.json(), indent=2))
    else:
        print(f"Error creating Data Agent: {data_agent_response.status_code}")
        print(data_agent_response.text)


    ################### Create conversation ###################

    conversation_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}/conversations"

    data_agent_id = "data_agent_1"
    conversation_id = "conversation _1"

    conversation_payload = {
        "agents": [
            f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
        ],
        "name": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}"
    }
    params = {
        "conversation_id": conversation_id
    }

    conversation_response = requests.post(conversation_url, headers=headers, params=params, json=conversation_payload)

    if conversation_response.status_code == 200:
        print("Conversation created successfully!")
        print(json.dumps(conversation_response.json(), indent=2))
    else:
        print(f"Error creating Conversation: {conversation_response.status_code}")
        print(conversation_response.text)


    ################### Chat with the API by using conversation (stateful) ####################

    chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}:chat"

    data_agent_id = "data_agent_1"
    conversation_id = "conversation _1"

    # Construct the payload
    chat_payload = {
        "parent": f"projects/{billing_project}/locations/global",
        "messages": [
            {
                "userMessage": {
                    "text": "Make a bar graph for the top 5 states by the total number of airports"
                }
            }
        ],
        "conversation_reference": {
            "conversation": f"projects/{billing_project}/locations/{location}/conversations/{conversation_id}",
            "data_agent_context": {
                "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
                # "credentials": looker_credentials
            }
        }
    }

    # Call the get_stream function to stream the response
    get_stream(chat_url, chat_payload)

    ################### Chat with the API by using dataAgents (stateless) ####################

    chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/{location}:chat"

    data_agent_id = "data_agent_1"

    # Construct the payload
    chat_payload = {
        "parent": f"projects/{billing_project}/locations/global",
        "messages": [
            {
                "userMessage": {
                    "text": "Make a bar graph for the top 5 states by the total number of airports"
                }
            }
        ],
        "data_agent_context": {
            "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
            # "credentials": looker_credentials
        }
    }

    # Call the get_stream function to stream the response
    get_stream(chat_url, chat_payload)

    ################### Chat with the API by using inline context (stateless) ####################

    chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/global:chat"

    # Construct the payload
    chat_payload = {
        "parent": f"projects/{billing_project}/locations/global",
        "messages": [
            {
                "userMessage": {
                    "text": "Make a bar graph for the top 5 states by the total number of airports"
                }
            }
        ],
        "inline_context": {
            "datasource_references": bigquery_data_sources,
            # Optional - if wanting to use advanced analysis with python
            "options": {
                "analysis": {
                    "python": {
                        "enabled": True
                    }
                }
            }
        }
    }

    # Call the get_stream function to stream the response
    get_stream(chat_url, chat_payload)

    ################### Multi-turn conversation ###################

    chat_url = f"https://geminidataanalytics.googleapis.com/v1alpha/projects/{billing_project}/locations/global:chat"

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

    data_agent_id = "data_agent_1"

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

      userMessage = {
          "userMessage": {
              "text": msg
          }
      }

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

      # Construct the payload
      chat_payload = {
          "parent": f"projects/{billing_project}/locations/global",
          "messages": conversation_messages,
          # Use a data agent reference
          "data_agent_context": {
              "data_agent": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}",
              # "credentials": looker_credentials
          },
          # Use inline context
          # "inline_context": {
          #     "datasource_references": bigquery_data_sources,
          # }
      }

      # Call the get_stream_multi_turn helper function to stream the response
      get_stream_multi_turn(chat_url, chat_payload, conversation_messages)

    # Send 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?")

다음의 확장 가능한 코드 샘플에는 채팅 응답을 스트리밍하는 데 사용되는 Python 도우미 함수가 포함되어 있습니다.

채팅 응답을 스트리밍하는 도우미 Python 함수

    def is_json(str):
      try:
          json_object = json_lib.loads(str)
      except ValueError as e:
          return False
      return True

    def handle_text_response(resp):
      parts = resp['parts']
      print(''.join(parts))

    def get_property(data, field_name, default = ''):
      return data[field_name] if field_name in data else default

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

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

    def format_bq_table_ref(table_ref):
      return '{}.{}.{}'.format(table_ref['projectId'], table_ref['datasetId'], table_ref['tableId'])

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

    def display_datasource(datasource):
      source_name = ''

      if 'studioDatasourceId' in datasource:
        source_name = datasource['studioDatasourceId']
      elif 'lookerExploreReference' in datasource:
        source_name = format_looker_table_ref(datasource['lookerExploreReference'])
      else:
        source_name = format_bq_table_ref(datasource['bigqueryTableReference'])

      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('Query name: {}'.format(query['name']))
        print('Question: {}'.format(query['question']))
        print('Data sources:')
        for datasource in query['datasources']:
          display_datasource(datasource)
      elif 'generatedSql' in resp:
        display_section_title('SQL generated')
        print(resp['generatedSql'])
      elif 'result' in resp:
        display_section_title('Data retrieved')

        fields = map(lambda field: get_property(field, 'name'), resp['result']['schema']['fields'])
        dict = {}

        for field in fields:
          dict[field] = map(lambda el: get_property(el, field), resp['result']['data'])

        display(pd.DataFrame(dict))

    def handle_chart_response(resp):
      if 'query' in resp:
        print(resp['query']['instructions'])
      elif 'result' in resp:
        vegaConfig = resp['result']['vegaConfig']
        alt.Chart.from_json(json_lib.dumps(vegaConfig)).display();

    def handle_error(resp):
      display_section_title('Error')
      print('Code: {}'.format(resp['code']))
      print('Message: {}'.format(resp['message']))

    def get_stream(url, json):
      s = requests.Session()

      acc = ''

      with s.post(url, json=json, headers=headers, stream=True) as resp:
        for line in resp.iter_lines():
          if not line:
            continue

          decoded_line = str(line, encoding='utf-8')

          if decoded_line == '[{':
            acc = '{'
          elif decoded_line == '}]':
            acc += '}'
          elif decoded_line == ',':
            continue
          else:
            acc += decoded_line

          if not is_json(acc):
            continue

          data_json = json_lib.loads(acc)

          if not 'systemMessage' in data_json:
            if 'error' in data_json:
                handle_error(data_json['error'])
            continue

          if 'text' in data_json['systemMessage']:
            handle_text_response(data_json['systemMessage']['text'])
          elif 'schema' in data_json['systemMessage']:
            handle_schema_response(data_json['systemMessage']['schema'])
          elif 'data' in data_json['systemMessage']:
            handle_data_response(data_json['systemMessage']['data'])
          elif 'chart' in data_json['systemMessage']:
            handle_chart_response(data_json['systemMessage']['chart'])
          else:
            colored_json = highlight(acc, lexers.JsonLexer(), formatters.TerminalFormatter())
            print(colored_json)
            print('\n')
            acc = ''

    def get_stream_multi_turn(url, json, conversation_messages):
        s = requests.Session()

        acc = ''

        with s.post(url, json=json, headers=headers, stream=True) as resp:
            for line in resp.iter_lines():
                if not line:
                    continue

                decoded_line = str(line, encoding='utf-8')

                if decoded_line == '[{':
                    acc = '{'
                elif decoded_line == '}]':
                    acc += '}'
                elif decoded_line == ',':
                    continue
                else:
                    acc += decoded_line

                if not is_json(acc):
                    continue

                data_json = json_lib.loads(acc)
                # Store the response that will be used in the next iteration
                conversation_messages.append(data_json)

                if not 'systemMessage' in data_json:
                    if 'error' in data_json:
                        handle_error(data_json['error'])
                    continue

                if 'text' in data_json['systemMessage']:
                    handle_text_response(data_json['systemMessage']['text'])
                elif 'schema' in data_json['systemMessage']:
                    handle_schema_response(data_json['systemMessage']['schema'])
                elif 'data' in data_json['systemMessage']:
                    handle_data_response(data_json['systemMessage']['data'])
                elif 'chart' in data_json['systemMessage']:
                    handle_chart_response(data_json['systemMessage']['chart'])
                else:
                    colored_json = highlight(acc, lexers.JsonLexer(), formatters.TerminalFormatter())
                    print(colored_json)
                print('\n')
                acc = ''