Utilizzare un agente personalizzato

Prima di iniziare

Questo tutorial presuppone che tu abbia letto e seguito le istruzioni riportate in:

Ottieni un'istanza di un agente

Per eseguire query su un agente, devi prima avere un'istanza di un agente. Puoi creare una nuova istanza o recuperare un'istanza esistente di un agente.

Per ottenere l'agente corrispondente a un ID risorsa specifico:

SDK Agent Platform

Esegui questo codice:

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)

dove

richieste

Esegui questo codice:

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

Quando utilizzi l'SDK Agent Platform per Python, l'oggetto agent corrisponde a una classe AgentRuntime che contiene quanto segue:

  • agent.api_resource con informazioni sull'agente di cui è stato eseguito il deployment. Puoi anche chiamare agent.operation_schemas() per restituire l'elenco delle operazioni supportate da agent. Per maggiori dettagli, consulta Operazioni supportate.
  • agent.api_client che consente interazioni sincrone tra i servizi
  • agent.async_api_client che consente interazioni asincrone tra i servizi

Il resto di questa sezione presuppone che tu abbia un'istanza denominata agent.

Elenca le operazioni supportate

Quando sviluppi l'agente localmente, hai accesso e conoscenza delle operazioni che supporta. Per utilizzare un agente di cui è stato eseguito il deployment, puoi elencare le operazioni che supporta:

SDK Agent Platform

Esegui questo codice:

print(agent.operation_schemas())

richieste

Esegui questo codice:

import json

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

REST

Rappresentato in spec.class_methods dalla risposta alla richiesta curl.

Lo schema di ogni operazione è un dizionario che documenta le informazioni di un metodo per l'agente che puoi chiamare. L'insieme di operazioni supportate dipende dal framework utilizzato per sviluppare l'agente:

Ad esempio, di seguito è riportato lo schema per l'operazione query di un LangchainAgent:

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

dove

  • name è il nome dell'operazione (ovvero agent.query per un'operazione denominata query).
  • api_mode è la modalità API dell'operazione ("" per sincrona, "stream" per streaming).
  • description è una descrizione dell'operazione basata sulla docstring del metodo.
  • parameters è lo schema degli argomenti di input nel formato dello schema OpenAPI.

Esegui query sull'agente utilizzando le operazioni supportate

Per gli agenti personalizzati, puoi utilizzare una delle seguenti operazioni di query o streaming definite durante lo sviluppo dell'agente:

Tieni presente che alcuni framework supportano solo operazioni di query o streaming specifiche:

Framework Operazioni di query supportate
Agent Development Kit async_stream_query
LangChain query, stream_query
LangGraph query, stream_query
AG2 query
LlamaIndex query

Esegui una query sull'agente

Esegui una query sull'agente utilizzando l'operazione query:

SDK Agent Platform 

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

richieste 

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

La risposta alla query è una stringa simile all'output di un test dell'applicazione locale:

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

Risposte dinamiche dell'agente

Trasmetti in streaming una risposta dall'agente utilizzando l'operazione stream_query:

SDK Agent Platform 

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)

richieste 

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 trasmette le risposte come una sequenza di oggetti generati in modo iterativo. Ad esempio, un insieme di tre risposte potrebbe avere il seguente aspetto:

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

Esegui query asincrone sull'agente

Se hai definito un'operazione async_query durante la creazione dell'agente, l'SDK Agent Platform per Python supporta le query asincrone lato client dell'agente:

SDK Agent Platform per 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)

La risposta alla query è un dizionario che corrisponde all'output di un test locale:

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

Trasmetti in modo asincrono le risposte dell'agente

Se hai definito un'operazione async_stream_query durante la creazione dell'agente, puoi trasmettere in modo asincrono una risposta dall'agente utilizzando una delle sue operazioni (ad es. async_stream_query):

SDK Agent Platform per 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)

L'operazione async_stream_query chiama lo stesso endpoint streamQuery in background e trasmette in modalità asincrona le risposte come una sequenza di oggetti generati in modo iterativo. Ad esempio, un insieme di tre risposte potrebbe essere il seguente:

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

Le risposte dovrebbero essere le stesse generate durante i test locali.

Job di query a lunga esecuzione

Per le query che possono richiedere molto tempo per essere completate (fino a sette giorni), puoi eseguirle come job a lunga esecuzione. Per saperne di più, consulta Utilizza un agente ADK.

Avvia un job di query a lunga esecuzione

Per avviare un job di query a lunga esecuzione:

SDK Agent Platform per 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"
}'

Controlla lo stato di un job di query a lunga esecuzione

Per controllare lo stato e recuperare i risultati di un job di query a lunga esecuzione:

SDK Agent Platform 

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

Passaggi successivi