使用代理

无论代理是在本地运行还是远程部署，用于查询代理的代码都是相同的。因此，在本页中，术语 agent 可互换地指 local_agent 或 remote_agent。由于支持的一组操作因框架而异，因此我们提供了特定于框架的模板的使用说明：

框架 说明
LangChain 由于具有预定义的配置和抽象，因此更易于用于基本用例。
LangGraph 基于图表的方法来定义工作流，具有高级人机协同和快退/重放功能。
AG2（以前称为 AutoGen） AG2 提供了多代理对话框架，作为构建 LLM 工作流的概要抽象。

对于自定义代理（并非基于特定框架的模板），您可以按照以下步骤操作：

  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/v1beta1/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

在 curl 请求响应中以 spec.class_methods 表示。

每个操作的架构都是一个字典，用于记录您可以调用的代理的方法信息。以下是同步操作的操作架构示例：

以下命令提供了 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 是基于方法的文档字符串对操作的说明。
  • parameters 是输入参数的架构，采用 OpenAPI 架构格式。

第 4 步：查询代理

如需使用其支持的某个操作（例如 query）查询代理，请执行以下操作：

Python 版 Vertex AI SDK 

agent.query(input={"messages": [
    ("user", "What is the exchange rate from US dollars to Swedish currency?")
]})

请求

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({"input": {
        "input": {"messages": [
            ("user", "What is the exchange rate from US dollars to Swedish currency?")
        ]},
    }})
)

REST

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

查询响应是一个字符串，类似于本地应用测试的输出：

{"input": "What is the exchange rate from US dollars to Swedish currency?",
 # ...
 "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")

agent.stream_query(input={"messages": [
    ("user", "What is the exchange rate from US dollars to Swedish currency?")
]})

请求

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({"input": {
        "input": {"messages": [
            ("user", "What is the exchange rate from US dollars to Swedish currency?")
        ]},
    }}),
    stream=True,
)

REST

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

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

后续步骤