Utilizzare un agente

Il codice per interrogare un agente è lo stesso indipendentemente dal fatto che venga eseguito localmente o implementato in remoto. Pertanto, in questa pagina, il termine agent si riferisce in modo intercambiabile a local_agent o remote_agent. Poiché l'insieme di operazioni supportate varia a seconda dei framework, forniamo istruzioni per l'utilizzo di modelli specifici per framework:

Framework Descrizione
Agent Development Kit (anteprima) Progettato in base alle best practice interne di Google per gli sviluppatori che creano applicazioni AI o per i team che devono prototipare e implementare rapidamente soluzioni robuste basate su agenti.
LangChain Più facile da usare per i casi d'uso di base grazie alle configurazioni e alle astrazioni predefinite.
LangGraph Approccio basato su grafici per definire i flussi di lavoro, con funzionalità avanzate human-in-the-loop e di riavvolgimento/riproduzione.
AG2 (in precedenza AutoGen) AG2 fornisce un framework di conversazione multi-agente come astrazione di alto livello per la creazione di workflow LLM.
LlamaIndex (anteprima) La pipeline di query di LlamaIndex offre un'interfaccia di alto livello per la creazione di flussi di lavoro RAG (Retrieval-Augmented Generation).

Per gli agenti personalizzati che non si basano su uno dei modelli specifici del framework, puoi seguire questi passaggi:

  1. Autenticazione utente.
  2. Recupera un'istanza dell'agente.
  3. Cercare le operazioni supportate.
  4. Esegui query sull'agente.
  5. (Se applicabile) Trasmetti in streaming le risposte dell'agente.

Passaggio 1: autenticazione degli utenti

Segui le stesse istruzioni per configurare l'ambiente.

Passaggio 2: ottieni 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 recuperare un'istanza esistente di un agente.

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

SDK Vertex AI per Python

Esegui questo codice:

from vertexai import agent_engines

agent = agent_engines.get("RESOURCE_ID")

In alternativa, puoi fornire il nome completo della risorsa dell'agente:

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

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

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

Passaggio 3: 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 Vertex AI per Python

Esegui questo codice:

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. Di seguito è riportato un esempio dello schema dell'operazione per un'operazione sincrona:

Il seguente comando fornisce un elenco di schemi in formato JSON che corrispondono alle operazioni dell'oggetto remote_app:

agent.operation_schemas()

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.

Passaggio 4: interroga l'agente

Per eseguire query sull'agente utilizzando una delle operazioni supportate (ad es. query):

SDK Vertex AI per Python 

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

Passaggio 5: riproduci in streaming le risposte dell'agente

Se applicabile, puoi riprodurre in streaming una risposta dell'agente utilizzando una delle sue operazioni (ad es. stream_query):

SDK Vertex AI per Python 

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

Vertex AI Agent Engine 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

Passaggio 6: esegui query asincrone dell'agente

Se hai definito un'operazione async_query durante lo sviluppo dell'agente, l'SDK Vertex AI per Python supporta le query asincrone lato client dell'agente.

SDK Vertex AI per 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."}

Passaggio 7: trasmetti in streaming in modo asincrono le risposte dell'agente

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

SDK Vertex AI per 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 generate durante i test locali.

Passaggi successivi