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

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

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

샘플 코드 전체 버전은 페이지 끝에 포함되어 있으며 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()

access_token = !gcloud auth application-default print-access-token
headers = {
        "Authorization": f"Bearer {access_token[0]}",
        "Content-Type": "application/json",
        "x-server-timeout": "300", # Custom timeout up to 600s
}

billing_project = 'YOUR-BILLING-PROJECT'
system_instruction = 'YOUR-SYSTEM-INSTRUCTIONS'

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

  • YOUR-BILLING-PROJECT: 필수 API를 사용 설정한 결제 프로젝트의 ID
  • YOUR-SYSTEM-INSTRUCTIONS: 에이전트 동작을 안내하고 데이터 니즈에 맞게 맞춤설정하는 시스템 요청 사항. 예를 들어 시스템 요청 사항을 사용하여 비즈니스 용어를 정의하거나 대답 길이를 제어하거나 데이터 형식을 설정할 수 있습니다. 효과적인 시스템 요청 사항 작성에서 권장하는 YAML 형식을 사용하여 시스템 요청 사항을 정의하여 자세하고 체계화된 안내를 제공하는 것이 좋습니다.

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 값입니다.

데이터 소스에 연결

다음 섹션에서는 에이전트 데이터 소스 연결 세부정보를 정의하는 방법을 보여줍니다. 에이전트는 Looker, BigQuery 또는 Looker Studio의 데이터에 연결될 수 있습니다.

Looker 데이터에 연결

다음 샘플 코드는 Looker Explore에 대한 연결을 정의합니다. Looker 인스턴스와의 연결을 설정하려면 Conversational Analytics API로 데이터 소스를 인증하고 연결에 설명된 대로 Looker API 키를 생성해야 합니다. 대화형 분석 API를 사용하면 한 번에 Looker Explore 최대 5개에 연결할 수 있습니다.

Looker 데이터 소스에 연결할 때는 다음 사항에 유의하세요.

  • 대화에서 포함된 Explore를 쿼리할 수 있습니다.
  • 에이전트는 한 번에 Explore 하나만 쿼리할 수 있습니다. Explore 여러 개에서 동시에 쿼리를 수행할 수는 없습니다.
  • 에이전트는 같은 대화에서 Explore 여러 개를 쿼리할 수 있습니다.

  • 에이전트는 여러 부분으로 구성된 질문이 포함된 대화나 후속 질문이 포함된 대화에서 Explore 여러 개를 쿼리할 수 있습니다.

    예를 들어 사용자가 cat-explore 하나와 dog-explore 하나 등 Explore 2개를 연결합니다. 사용자가 '고양이와 개 중에 어느 것이 더 많아?'라는 질문을 입력합니다. 이렇게 하면 cat-explore의 고양이 수를 세는 쿼리와 dog-explore의 개 수를 세는 쿼리 등 쿼리 2개가 생성됩니다. 에이전트는 두 쿼리 모두 완료한 후 두 쿼리의 숫자를 비교합니다.

looker_instance_uri = "https://my_company.looker.com"
looker_data_source = {
    "looker": {
        "explore_references": {
            "looker_instance_uri": "https://your_company.looker.com"
            "lookml_model": "your_model",
            "explore": "your_explore",
       },
       {
            "looker_instance_uri": looker_instance_uri,
            "lookml_model": "your_model_2",
            "explore": "your_explore_2",
       },
       # Do not include the following line during agent creation
       "credentials": looker_credentials
    }
}

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

  • https://your_company.looker.com: Looker 인스턴스의 전체 URL입니다.
  • your_model: 연결할 Explore가 포함된 LookML 모델의 이름.
  • your_explore: 데이터 에이전트에서 쿼리할 Looker Explore의 이름
  • my_model_2: 연결할 Explore가 포함된 두 번째 LookML 모델의 이름. Explore 최대 5개의 추가 모델에 이 변수를 반복할 수 있습니다.
  • my_explore_2: 데이터 에이전트에서 쿼리할 추가 Looker Explore의 이름. 이 변수를 반복하여 Explore를 최대 5개까지 포함할 수 있습니다.

BigQuery 데이터에 연결

대화형 분석 API를 사용하면 연결할 수 있는 BigQuery 테이블 수에 대한 엄격한 제한이 없습니다. 하지만 많은 수의 테이블에 연결하면 정확성이 떨어지거나 모델의 입력 토큰 한도를 초과할 수 있습니다.

다음 샘플 코드는 여러 BigQuery 테이블에 대한 연결을 정의하고 선택적 구조화된 컨텍스트 필드의 예를 포함합니다. 에이전트 성능을 개선하기 위해 테이블 및 열 설명, 동의어, 태그, 예시 쿼리와 같은 BigQuery 테이블의 구조화된 컨텍스트를 선택적으로 제공할 수 있습니다. 자세한 내용은 BigQuery 데이터 소스의 데이터 에이전트 컨텍스트 정의를 참고하세요.

bigquery_data_sources = {
    "bq": {
        "tableReferences": [
            {
                "projectId": "my_project_id",
                "datasetId": "my_dataset_id",
                "tableId": "my_table_id",
                "schema": {
                    "description": "my_table_description",
                    "fields": [{
                        "name": "my_column_name",
                        "description": "my_column_description"
                    }]
                }
            },
            {
                "projectId": "my_project_id_2",
                "datasetId": "my_dataset_id_2",
                "tableId": "my_table_id_2"
            },
            {
                "projectId": "my_project_id_3",
                "datasetId": "my_dataset_id_3",
                "tableId": "my_table_id_3"
            },
        ]
    }
}

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

  • my_project_id: 연결하려는 BigQuery 데이터 세트 및 테이블이 포함된 Google Cloud 프로젝트의 ID입니다. 공개 데이터 세트에 연결하려면 bigquery-public-data를 지정합니다.
  • my_dataset_id: BigQuery 데이터 세트의 ID입니다.
  • my_table_id: BigQuery 테이블의 ID입니다.
  • my_table_description: 테이블의 콘텐츠와 목적에 대한 선택적 설명입니다.
  • my_column_name: 선택적 설명을 제공하는 테이블의 열 이름입니다.
  • my_column_description: 열의 콘텐츠와 목적에 대한 설명(선택사항)입니다.

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으로 고급 분석을 선택적으로 사용 설정할 수도 있습니다. options 파라미터 및 대화에 구성할 수 있는 옵션에 대한 자세한 내용은 REST 리소스: projects.locations.dataAgents를 참조하세요.

data_agent_url = f"https://geminidataanalytics.googleapis.com/v1beta/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/v1beta/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를 사용하여 데이터 에이전트와 대화를 관리하는 방법을 보여줍니다. 다음 태스크를 수행할 수 있습니다.

데이터 에이전트 가져오기

다음 샘플 코드에서는 데이터 에이전트 리소스 URL에 HTTP GET 요청을 전송하여 기존 데이터 에이전트를 가져오는 방법을 보여줍니다.

data_agent_id = "data_agent_1"
data_agent_url = f"{base_url}/v1beta/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

data_agent_response = requests.get(
    data_agent_url, headers=headers
)

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

이전 예시에서 data_agent_1을 가져오려는 데이터 에이전트의 ID로 바꿉니다.

데이터 에이전트 나열

다음 코드에서는 dataAgents 엔드포인트에 HTTP GET 요청을 전송하여 지정된 프로젝트의 모든 데이터 에이전트를 나열하는 방법을 보여줍니다.

모든 에이전트를 나열하려면 프로젝트에 대한 geminidataanalytics.dataAgents.list 권한이 있어야 합니다. 이 권한이 포함된 IAM 역할에 대한 자세한 내용은 사전 정의된 역할 목록을 참조하세요.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
data_agent_url = f"{base_url}/v1beta/projects/{billing_project}/locations/{location}/dataAgents"

data_agent_response = requests.get(
    data_agent_url, headers=headers
)

if data_agent_response.status_code == 200:
    print("Data Agents Listed successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error Listing Data Agents: {data_agent_response.status_code}")

YOUR-BILLING-PROJECT를 결제 프로젝트 ID로 바꿉니다.

액세스 가능한 데이터 에이전트 나열

다음 코드에서는 dataAgents:listAccessible 엔드포인트에 HTTP GET 요청을 전송하여 지정된 프로젝트의 액세스 가능한 모든 데이터 에이전트를 나열하는 방법을 보여줍니다.

billing_project = "YOUR-BILLING-PROJECT"
creator_filter = "YOUR-CREATOR-FILTER"
location = "global"
data_agent_url = f"{base_url}/v1beta/projects/{billing_project}/locations/{location}/dataAgents:listAccessible"

params = {
    "creator_filter": creator_filter
}

data_agent_response = requests.get(
    data_agent_url, headers=headers, params=params
)

if data_agent_response.status_code == 200:
    print("Accessible Data Agents Listed successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error Listing Accessible Data Agents: {data_agent_response.status_code}")

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

  • YOUR-BILLING-PROJECT: 결제 프로젝트 ID
  • YOUR-CREATOR-FILTER: 데이터 에이전트 생성자를 기반으로 적용할 필터. 가능한 값에는 NONE(기본값), CREATOR_ONLY, NOT_CREATOR_ONLY가 포함됩니다.

데이터 에이전트 업데이트

다음 샘플 코드에서는 데이터 에이전트 리소스 URL에 HTTP PATCH 요청을 전송하여 데이터 에이전트를 업데이트하는 방법을 보여줍니다. 요청 페이로드에는 변경하려는 필드의 새 값이 포함되고 요청 파라미터에는 업데이트할 필드를 지정하는 updateMask 파라미터가 포함됩니다.

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

data_agent_url = f"{base_url}/v1beta/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

payload = {
    "description": "Updated description of the data agent.",
    "data_analytics_agent": {
        "published_context": {
            "datasource_references": bigquery_data_sources,
            "system_instruction": system_instruction
        }
    },
}

fields = ["description", "data_analytics_agent"]
params = {
    "updateMask": ",".join(fields)
}

data_agent_response = requests.patch(
    data_agent_url, headers=headers, params=params, json=payload
)

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

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

  • data_agent_1: 업데이트할 데이터 에이전트의 ID
  • YOUR-BILLING-PROJECT: 결제 프로젝트 ID
  • Updated description of the data agent.: 데이터 에이전트에 대한 새로운 설명

데이터 에이전트에 대한 IAM 정책 가져오기

다음 샘플 코드에서는 데이터 에이전트 URL에 HTTP POST 요청을 전송하여 데이터 에이전트에 대한 IAM 정책을 가져오는 방법을 보여줍니다. 요청 페이로드에는 데이터 에이전트 경로가 포함됩니다.

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

data_agent_url = f"{base_url}/v1beta/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}:getIamPolicy"

# Request body
payload = {
    "resource": f"projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"
}

data_agent_response = requests.post(
    data_agent_url, headers=headers, json=payload
)

if data_agent_response.status_code == 200:
    print("IAM Policy fetched successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error fetching IAM policy: {data_agent_response.status_code}")
    print(data_agent_response.text)

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

  • YOUR-BILLING-PROJECT: 결제 프로젝트 ID
  • data_agent_1: IAM 정책을 가져오려는 데이터 에이전트의 ID

데이터 에이전트에 대한 IAM 정책 설정

에이전트를 공유하려면 setIamPolicy 메서드를 사용하여 특정 에이전트의 사용자에게 IAM 역할을 할당하면 됩니다. 다음 샘플 코드에서는 바인딩이 포함된 페이로드를 사용하여 데이터 에이전트 URL로 POST 호출을 하는 방법을 보여줍니다. 바인딩은 어떤 사용자에게 어떤 역할을 할당해야 하는지 지정합니다.

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

data_agent_url = f"{base_url}/v1beta/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}:setIamPolicy"

# Request body
payload = {
    "policy": {
        "bindings": [
            {
                "role": role,
                "members": [
                    f"user:{i.strip()}" for i in users.split(",")
                ]
            }
        ]
    }
}

data_agent_response = requests.post(
    data_agent_url, headers=headers, json=payload
)

if data_agent_response.status_code == 200:
    print("IAM Policy set successfully!")
    print(json.dumps(data_agent_response.json(), indent=2))
else:
    print(f"Error setting IAM policy: {data_agent_response.status_code}")
    print(data_agent_response.text)

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

  • YOUR-BILLING-PROJECT: 결제 프로젝트 ID
  • data_agent_1: IAM 정책을 설정하려는 데이터 에이전트의 ID
  • 222larabrown@gmail.com, cloudysanfrancisco@gmail.com: 지정된 역할을 부여하려는 쉼표로 구분한 사용자 이메일 목록

데이터 에이전트 삭제

다음 샘플 코드에서는 데이터 에이전트 리소스 URL에 HTTP DELETE 요청을 전송하여 데이터 에이전트를 소프트 삭제하는 방법을 보여줍니다. 소프트 삭제는 에이전트가 삭제되지만 30일 이내에 계속 검색될 수 있음을 의미합니다.

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

data_agent_url = f"{base_url}/v1beta/projects/{billing_project}/locations/{location}/dataAgents/{data_agent_id}"

data_agent_response = requests.delete(
    data_agent_url, headers=headers
)

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

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

  • YOUR-BILLING-PROJECT: 결제 프로젝트 ID
  • data_agent_1: 삭제하려는 데이터 에이전트의 ID

대화 가져오기

다음 샘플 코드에서는 대화 리소스 URL에 HTTP GET 요청을 전송하여 기존 대화를 가져오는 방법을 보여줍니다.

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

conversation_url = f"{base_url}/v1beta/projects/{billing_project}/locations/{location}/conversations/{conversation_id}"

conversation_response = requests.get(conversation_url, headers=headers)

# Handle the response
if conversation_response.status_code == 200:
    print("Conversation fetched successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error while fetching conversation: {conversation_response.status_code}")
    print(conversation_response.text)

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

  • YOUR-BILLING-PROJECT: 결제 프로젝트 ID
  • conversation_1: 가져오려는 대화의 ID

대화 나열

다음 샘플 코드에서는 conversations 엔드포인트에 HTTP GET 요청을 전송하여 지정된 프로젝트의 대화를 나열하는 방법을 보여줍니다.

기본적으로 이 메서드는 개발자가 만든 대화를 반환합니다. 관리자(cloudaicompanion.topicAdmin IAM 역할이 있는 사용자)는 프로젝트 내 모든 대화를 볼 수 있습니다.

billing_project = "YOUR-BILLING-PROJECT"
location = "global"
conversation_url = f"{base_url}/v1beta/projects/{billing_project}/locations/{location}/conversations"

conversation_response = requests.get(conversation_url, headers=headers)

# Handle the response
if conversation_response.status_code == 200:
    print("Conversation fetched successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error while fetching conversation: {conversation_response.status_code}")
    print(conversation_response.text)

YOUR-BILLING-PROJECT를 필수 API를 사용 설정한 결제 프로젝트의 ID로 바꿉니다.

대화의 메시지 나열

다음 샘플 코드에서는 대화의 messages 엔드포인트에 HTTP GET 요청을 전송하여 대화의 모든 메시지를 나열하는 방법을 보여줍니다.

메시지를 나열하려면 대화에 대한 cloudaicompanion.topics.get 권한이 있어야 합니다.

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

conversation_id = "conversation_1"

conversation_url = f"{base_url}/v1beta/projects/{billing_project}/locations/{location}/conversations/{conversation_id}/messages"

conversation_response = requests.get(conversation_url, headers=headers)

# Handle the response
if conversation_response.status_code == 200:
    print("Conversation fetched successfully!")
    print(json.dumps(conversation_response.json(), indent=2))
else:
    print(f"Error while fetching conversation: {conversation_response.status_code}")
    print(conversation_response.text)

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

  • YOUR-BILLING-PROJECT: 결제 프로젝트 ID
  • conversation_1: 메시지를 나열하려는 대화의 ID

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/v1beta/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/v1beta/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으로 고급 분석을 선택적으로 사용 설정할 수도 있습니다. options 파라미터 및 대화에 구성할 수 있는 옵션에 대한 자세한 내용은 REST 리소스: projects.locations.dataAgents 페이지를 참조하세요.

chat_url = f"https://geminidataanalytics.googleapis.com/v1beta/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/v1beta/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",
        "x-server-timeout": "300", # Custom timeout up to 600s
    }

    ################### 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",
        },
        # Do not include the following line during agent creation
        # "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/v1beta/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/v1beta/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/v1beta/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/v1beta/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/v1beta/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/v1beta/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 = ''