Utilizzare un agente personalizzato

Prima di iniziare

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

• Crea un agente personalizzato: per creare un agent personalizzato.
• Autenticazione utente per l'autenticazione come utente per interrogare l'agente.
• Importa e inizializza l'SDK per inizializzare il client per ottenere un'istanza di cui è stato eseguito il deployment (se necessario).

Recupera un'istanza di un agente

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

Per ottenere l'agente che corrisponde a un ID risorsa specifico:

SDK Agent Platform

Esegui questo codice:

import vertexai

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

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

print(agent)

dove

• PROJECT_ID è l'ID progetto Google Cloud in cui crei ed esegui il deployment degli agenti.
• LOCATION è una delle regioni supportate.
• RESOURCE_ID è l'ID dell'agente di cui è stato eseguito il deployment come risorsa reasoningEngine.

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 Python, l'oggetto agent corrisponde a una classe AgentEngine 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, vedi Operazioni supportate.
• agent.api_client che consente interazioni di servizio sincrone
• agent.async_api_client che consente interazioni asincrone tra servizi

Il resto di questa sezione presuppone che tu disponga di 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 enumerare 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. Il set di operazioni supportate dipende dal framework che hai utilizzato per sviluppare l'agente:

• Agent Development Kit
• Agent2Agent
• LangChain
• LangGraph
• AG2
• LlamaIndex
• Personalizzato

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 (ad es. agent.query per un'operazione denominata query).
• api_mode è la modalità API dell'operazione ("" per la modalità sincrona, "stream" per lo streaming).
• description è una descrizione dell'operazione basata sulla docstring del metodo.
• parameters è lo schema degli argomenti di input nel formato dello schema OpenAPI.

Eseguire query sull'agente utilizzando le operazioni supportate

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

• query

• stream_query

• async_query

• async_stream_query

Tieni presente che alcuni framework supportano solo query o operazioni di 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

Interroga l'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 dell'agente utilizzando l'operazione stream_query:

SDK Agent Platform

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)

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

Eseguire query asincrone dell'agente

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

SDK Agent Platform Python

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)

La risposta alla query è un dizionario uguale 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."}

Trasmettere in streaming in modo asincrono le risposte dell'agente

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

SDK Agent Platform Python

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)

L'operazione async_stream_query chiama lo stesso endpoint streamQuery in modo asincrono e 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

Le risposte devono essere le stesse di quelle 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 esecuzione prolungata. Per saperne di più, consulta Utilizzare un agente ADK.

Avvia un job di query a lunga esecuzione

Per avviare un job di query a lunga esecuzione:

SDK Agent Platform Python

import vertexai

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

response = client.agent_engines.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"
}'

Controllare lo stato di un job di query a esecuzione prolungata

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

SDK Agent Platform

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

Passaggi successivi

• Utilizzare un agente.
• Valuta un agente.
• Gestisci gli agenti di cui è stato eseguito il deployment.
• Richiedere assistenza.