使用服務專員

無論代理程式是在本機執行或遠端部署，查詢代理程式的程式碼都相同。因此，在本頁面中，agent 一詞可互換地指稱 local_agent 或 remote_agent。由於各架構支援的作業集有所不同，因此我們提供架構專屬範本的使用說明：

架構 說明
Agent Development Kit (預先發布版) 這項服務是根據 Google 內部最佳做法設計，適合開發 AI 應用程式的開發人員，或需要快速設計原型並部署強大代理程式解決方案的團隊。
LangChain 由於預先定義的設定和抽象化，可更輕鬆地用於基本用途。
LangGraph 以圖表為基礎定義工作流程，並提供進階的人機迴圈和倒轉/重播功能。
AG2 (舊稱 AutoGen) AG2 提供多代理程式對話架構，做為建構 LLM 工作流程的高階抽象化。
LlamaIndex (預先發布版) LlamaIndex 的查詢 pipeline 提供高階介面，可建立檢索增強生成 (RAG) 工作流程。

如果是自訂代理程式，且並非以其中一個架構專屬範本為基礎，請按照下列步驟操作：

  1. 使用者驗證。
  2. 取得代理程式例項。
  3. 查詢支援的作業。
  4. 查詢代理程式。
  5. (如適用) 從代理程式串流回應。

步驟 1：使用者驗證

請按照設定環境的相同操作說明操作。

步驟 2：取得代理程式的執行個體

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

如要取得與特定資源 ID 相對應的服務專員：

Python 適用的 Vertex AI SDK

請執行下列程式碼：

from vertexai import agent_engines

agent = agent_engines.get("RESOURCE_ID")

或者，您也可以提供代理程式的完整資源名稱：

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

要求

請執行下列程式碼：

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 的執行個體。

步驟 3：支援的作業

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

Python 適用的 Vertex AI SDK

請執行下列程式碼：

agent.operation_schemas()

要求

請執行下列程式碼：

import json

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

REST

以 spec.class_methods 表示，來自 curl 要求的相關回應。

每項作業的結構定義都是字典，其中記錄了可呼叫的代理程式方法資訊。以下是同步作業的作業結構定義範例：

下列指令會提供 JSON 格式的結構定義清單，對應至 remote_app 物件的作業：

agent.operation_schemas()

舉例來說，以下是 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 是作業的名稱 (即名為 query 的作業的 agent.query)。
  • api_mode 是作業的 API 模式 ("" 代表同步，"stream" 代表串流)。
  • description 是根據方法 docstring 的作業說明。
  • parameters 是 OpenAPI 結構定義格式的輸入引數結構定義。

步驟 4：查詢代理程式

如要使用其中一個支援的作業 (例如 query) 查詢代理程式，請按照下列步驟操作：

Python 適用的 Vertex AI SDK 

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

要求

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."}

步驟 5：從代理程式串流回應

如適用，您可以透過代理程式的其中一項作業 (例如 stream_query) 串流回應：

Python 適用的 Vertex AI SDK 

agent = agent_engines.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)

要求

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?"
  }
}'

Vertex AI Agent Engine 會以迭代生成的物件序列形式，串流傳送回覆。舉例來說，一組三個回覆可能如下所示：

{'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

步驟 6：非同步查詢代理程式

如果您在開發代理程式時定義了 async_query 作業，Python 適用的 Vertex AI SDK 支援代理程式的用戶端非同步查詢。

Python 適用的 Vertex AI SDK 

agent = agent_engines.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."}

步驟 7：從代理程式非同步串流回應

如果您在開發代理程式時定義了 async_stream_query 作業，可以使用其中一項作業 (例如 async_stream_query) 從代理程式非同步串流回應：

Python 適用的 Vertex AI SDK 

agent = agent_engines.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

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

後續步驟