使用自訂代理程式

事前準備

本教學課程假設您已閱讀並按照下列文章中的指示操作：

取得代理執行個體

如要查詢代理，您必須先建立代理執行個體。您可以建立新的執行個體，或取得現有的代理執行個體。

如要取得與特定資源 ID 對應的代理，請按照下列步驟操作：

Agent Platform SDK

請執行下列程式碼：

import agentplatform

client = agentplatform.Client(  # For service interactions via client.runtimes
    project="PROJECT_ID",
    location="LOCATION",
)

agent = client.runtimes.get(name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

print(agent)

其中

requests

請執行下列程式碼：

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

response = requests.get(
f"https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    headers={
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": f"Bearer {get_identity_token()}",
    },
)

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID

使用 Agent Platform SDK for Python 時，agent 物件會對應至 AgentRuntime 類別，其中包含下列內容：

本節其餘部分假設您有名為 agent 的執行個體。

列出支援的操作

在本機開發代理時，您可以存取並瞭解代理支援的作業。如要使用已部署的代理，可以列舉其支援的操作：

Agent Platform SDK

請執行下列程式碼：

print(agent.operation_schemas())

requests

請執行下列程式碼：

import json

json.loads(response.content).get("spec").get("classMethods")

REST

在 curl 要求的回應中，以 spec.class_methods 表示。

每項作業的結構定義都是字典，記錄可呼叫的代理方法資訊。支援的操作組合取決於您開發代理時使用的架構：

舉例來說，以下是 LangchainAgent 的 query 作業結構定義：

{'api_mode': '',
 'name': 'query',
 'description': """Queries the Agent with the given input and config.
    Args:
        input (Union[str, Mapping[str, Any]]):
            Required. The input to be passed to the Agent.
        config (langchain_core.runnables.RunnableConfig):
            Optional. The config (if any) to be used for invoking the Agent.
    Returns:
        The output of querying the Agent with the given input and config.
""",            '        ',
 'parameters': {'$defs': {'RunnableConfig': {'description': 'Configuration for a Runnable.',
                                             'properties': {'configurable': {...},
                                                            'run_id': {...},
                                                            'run_name': {...},
                                                            ...},
                                             'type': 'object'}},
                'properties': {'config': {'nullable': True},
                               'input': {'anyOf': [{'type': 'string'}, {'type': 'object'}]}},
                'required': ['input'],
                'type': 'object'}}

其中

  • name 是作業名稱 (例如 agent.query 代表名為 query 的作業)。
  • api_mode 是作業的 API 模式 ("" 代表同步，"stream" 代表串流)。
  • description 是根據方法的 docstring 提供的作業說明。
  • parameters 是輸入引數的結構定義，採用 OpenAPI 結構定義格式。

使用支援的操作查詢代理

如果是自訂代理，您可以使用開發代理時定義的任何查詢或串流作業：

請注意，某些架構僅支援特定查詢或串流作業：

架構 支援的查詢作業
Agent Development Kit async_stream_query
LangChain query、stream_query
LangGraph query、stream_query
AG2 query
LlamaIndex query

查詢代理

使用 query 作業查詢代理：

Agent Platform SDK 

agent.query(input="What is the exchange rate from US dollars to Swedish Krona today?")

requests 

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

requests.post(
    f"https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query",
    headers={
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": f"Bearer {get_identity_token()}",
    },
    data=json.dumps({
        "class_method": "query",
        "input": {
            "input": "What is the exchange rate from US dollars to Swedish Krona today?"
        }
    })
)

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{
  "class_method": "query",
  "input": {
    "input": "What is the exchange rate from US dollars to Swedish Krona today?"
  }
}'

查詢回應是字串，類似於本機應用程式測試的輸出內容：

{"input": "What is the exchange rate from US dollars to Swedish Krona today?",
 # ...
 "output": "For 1 US dollar you will get 10.7345 Swedish Krona."}

串流代理的回覆

使用 stream_query 作業，串流代理的回覆：

Agent Platform SDK 

agent = client.runtimes.get("projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

for response in agent.stream_query(
    input="What is the exchange rate from US dollars to Swedish Krona today?"
):
    print(response)

requests 

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

requests.post(
    f"https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:streamQuery",
    headers={
        "Content-Type": "application/json",
        "Authorization": f"Bearer {get_identity_token()}",
    },
    data=json.dumps({
        "class_method": "stream_query",
        "input": {
            "input": "What is the exchange rate from US dollars to Swedish Krona today?"
        },
    }),
    stream=True,
)

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:streamQuery?alt=sse -d '{
  "class_method": "stream_query",
  "input": {
    "input": "What is the exchange rate from US dollars to Swedish Krona today?"
  }
}'

Agent Platform 會以反覆生成的物件序列形式，串流傳輸回覆。舉例來說，一組三個回覆可能如下所示：

{'actions': [{'tool': 'get_exchange_rate', ...}]}  # first response
{'steps': [{'action': {'tool': 'get_exchange_rate', ...}}]}  # second response
{'output': 'The exchange rate is 11.0117 SEK per USD as of 2024-12-03.'}  # final response

非同步查詢代理

如果您在建立代理時定義了 async_query 作業，Agent Platform SDK for Python 支援用戶端非同步查詢代理：

Agent Platform SDK for Python 

agent = client.runtimes.get("projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

response = await agent.async_query(
    input="What is the exchange rate from US dollars to Swedish Krona today?"
)
print(response)

查詢回應是字典，與本機測試的輸出內容相同：

{"input": "What is the exchange rate from US dollars to Swedish Krona today?",
 # ...
 "output": "For 1 US dollar you will get 10.7345 Swedish Krona."}

以非同步方式串流代理的回覆

如果您在建立代理時定義了 async_stream_query 作業，可以使用其中一項作業 (例如 async_stream_query)，以非同步方式串流代理的回覆：

Agent Platform SDK for Python 

agent = client.runtimes.get("projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

async for response in agent.async_stream_query(
    input="What is the exchange rate from US dollars to Swedish Krona today?"
):
    print(response)

async_stream_query 作業會在背景呼叫相同的 streamQuery 端點，並以非同步方式串流回應，做為一系列反覆生成的物件。舉例來說，一組三個回覆可能如下所示：

{'actions': [{'tool': 'get_exchange_rate', ...}]}  # first response
{'steps': [{'action': {'tool': 'get_exchange_rate', ...}}]}  # second response
{'output': 'The exchange rate is 11.0117 SEK per USD as of 2024-12-03.'}  # final response

回應應與本機測試期間產生的回應相同。

長時間執行的查詢工作

如果查詢需要較長時間才能完成 (最多七天)，可以做為長時間執行的工作執行。詳情請參閱「使用 ADK 代理」。

啟動長時間執行的查詢工作

如要啟動長時間執行的查詢工作，請按照下列步驟操作：

Agent Platform SDK for Python 

import agentplatform

client = agentplatform.Client(
    project="PROJECT_ID",
    location="LOCATION",
)

response = client.runtimes.run_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    config={
        "query": '{ "input": {"product": "Google"} }',
        "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE",
    },
)
print(response)

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1beta1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:asyncQuery -d \
'{
  "input_gcs_uri": "gs://GCS_BUCKET_NAME/INPUT_FILE",
  "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE"
}'

查看長時間執行的查詢工作狀態

如要查看長時間執行的查詢工作狀態並擷取結果，請按照下列步驟操作：

Agent Platform SDK 

response = client.runtimes.check_query_job(
    name="JOB_NAME",
    config={
        "retrieve_result": True,
    },
)
print(response)

後續步驟