Agent Runtime è progettato per essere indipendente dal framework dell'applicazione. Se scegli di eseguire il deployment dell'agente utilizzando un container personalizzato o un Dockerfile, il container deve rispettare il contratto di runtime per gestire correttamente le query.
Per saperne di più sui metodi di deployment, consulta Eseguire il deployment di un agente.
Vincoli e requisiti
Per eseguire il deployment di un container personalizzato su Agent Runtime, il container deve rimanere in ascolto delle richieste HTTP su 0.0.0.0 sulla porta 8080.
Endpoint (facoltativo)
Il container può esporre qualsiasi endpoint HTTP personalizzato. Puoi richiamare questi endpoint personalizzati inviando richieste all'API sottostante dell'agente di cui è stato eseguito il deployment. Per saperne di più, consulta Utilizzare gli agenti di cui è stato eseguito il deployment tramite l'API sottostante.
Sebbene questi endpoint siano facoltativi a livello di API, la loro implementazione consente le seguenti funzionalità di integrazione:
- Supporto dell'SDK Python: l'implementazione di
/api/reasoning_enginee/api/stream_reasoning_engineconsente di utilizzare l'agente di cui è stato eseguito il deployment tramite l'SDK Python di Agent Platform. - Supporto del playground: l'implementazione di
/api/stream_reasoning_engineè obbligatoria se vuoi interagire con l'agente tramite il Google Cloud playground della console. Per saperne di più, consulta Supporto del playground.
Se vuoi utilizzare queste funzionalità, devi implementare i seguenti endpoint:
/api/reasoning_engine: utilizzato per gestire le query inviate all'API RESTreasoningEngines/queryo ai metodi di query sincroni e asincroni dell'SDK Python./api/stream_reasoning_engine: utilizzato per gestire le query inviate all'API RESTreasoningEngines/streamQueryo ai metodi di query di streaming dell'SDK Python.
Metodi di classe e modalità di esecuzione
Quando esegui il deployment di un container personalizzato, devi dichiarare i metodi di classe supportati nell'elenco classMethods della specifica di deployment. Questi metodi di classe corrispondono alle operazioni che hai definito durante lo sviluppo dell'agente (vedi Registrare metodi personalizzati e Eseguire query sull'agente utilizzando le operazioni supportate). Ogni metodo dichiarato ha un name e un api_mode che determina il modo in cui viene instradato:
| Modalità API | Tipo di esecuzione | Endpoint di routing |
|---|---|---|
"" (stringa vuota) o "async" |
Unario (richiesta-risposta) | /api/reasoning_engine |
"stream" o "async_stream" |
Streaming | /api/stream_reasoning_engine |
Quando un client richiama un metodo, il servizio Agent Runtime invia una richiesta POST all'endpoint di routing corrispondente nel container. Il corpo JSON della richiesta contiene il campo class_method (che corrisponde al nome del metodo) e il campo input.
Metodi richiesti per l'integrazione
Per utilizzare l'SDK Python o il Google Cloud playground della console, il container deve implementare i metodi specifici previsti da queste integrazioni:
- Query standard dell'SDK: richiede
query(modalità""o"async") estream_query(modalità"stream"o"async_stream"). - Playground: richiede
stream_query(modalità"stream"o"async_stream").
Integrazione dell'ADK
Se stai eseguendo il deployment di un agente creato con Agent Development Kit (ADK), puoi creare il tuo container proxy nel linguaggio di programmazione e nel framework del server che preferisci. Per supportare l'insieme completo di funzionalità dell'ADK, il container deve implementare i metodi definiti dal contratto ADK. Per saperne di più, consulta Utilizzare un agente ADK e Registrare e gestire un agente ADK.
Puoi consultare il modello AdkApp come implementazione di riferimento. Per saperne di più, consulta la documentazione di riferimento di AdkApp e il codice sorgente di AdkApp.
Se vuoi che Google gestisca automaticamente gli aggiornamenti quando aggiorni la versione dell'ADK, devi utilizzare gli strumenti forniti dall'ADK per il deployment degli agenti anziché scrivere il tuo server API. Per saperne di più, consulta la documentazione di deployment dell'ADK.
Specifiche dell'API
Sia /api/reasoning_engine sia /api/stream_reasoning_engine ricevono richieste HTTP POST con un corpo JSON contenente i seguenti campi:
class_method(stringa): il nome del metodo dell'agente sottostante da richiamare (ad esempio,queryostream_query).input(oggetto JSON): gli argomenti da passare al metodo di classe specificato.
/api/reasoning_engine (unario)
- Metodo di richiesta:
POST - Corpo della richiesta:
json { "class_method": "query", "input": { "message": "What is the capital of France?" } } - Risposta: un oggetto JSON contenente l'output dell'esecuzione dell'agente.
json { "output": "The capital of France is Paris." }
/api/stream_reasoning_engine (streaming)
- Metodo di richiesta:
POST - Corpo della richiesta:
json { "class_method": "stream_query", "input": { "message": "Tell me a short story." } } - Risposta: uno stream di JSON delimitato da righe (ndjson), in cui ogni riga è un blocco della risposta codificato in JSON.
json {"output": "Once"} {"output": " upon"} {"output": " a time..."}
Esempio di server API (Python)
Di seguito è riportato un esempio di server FastAPI in Python che implementa il contratto di runtime di Agent Platform. Questo server esegue il wrapping di un agente di esempio (SimpleAgent) e gestisce il routing e la codifica. Puoi sostituire SimpleAgent con la tua implementazione dell'agente.
Per eseguire questo esempio, assicurati di aver installato fastapi, uvicorn e pydantic.
import inspect
import json
import logging
import os
import uvicorn
from fastapi import FastAPI, encoders, responses
from pydantic import BaseModel
app = FastAPI()
# Define the request body structure
class QueryRequest(BaseModel):
input: dict | None = None
class_method: str
# Example Agent implementation
class SimpleAgent:
def query(self, message: str) -> str:
return f"Echo: {message}"
async def stream_query(self, message: str):
words = message.split()
for word in words:
yield {"output": word + " "}
agent = SimpleAgent()
def _encode_chunk_to_json(chunk):
"""Encodes a chunk to a JSON string with a newline."""
try:
json_chunk = encoders.jsonable_encoder(chunk)
return json.dumps(json_chunk) + "\n"
except Exception:
logging.exception("Failed to encode chunk")
return None
async def json_generator(output):
async for chunk in output:
encoded_chunk = _encode_chunk_to_json(chunk)
if encoded_chunk is None:
break
yield encoded_chunk
async def _invoke_callable_or_raise(invocation_callable, invocation_payload):
if inspect.iscoroutinefunction(invocation_callable):
return await invocation_callable(**invocation_payload)
else:
return invocation_callable(**invocation_payload)
@app.post("/api/reasoning_engine")
async def query_endpoint(request: QueryRequest) -> responses.JSONResponse:
try:
method = getattr(agent, request.class_method)
except AttributeError:
return responses.JSONResponse(
status_code=400,
content={"error": f"Method {request.class_method} not found on agent"}
)
output = await _invoke_callable_or_raise(method, request.input or {})
try:
json_serialized_content = encoders.jsonable_encoder({"output": output})
except ValueError as encoding_error:
logging.exception("Failed to JSON-encode response: %s", encoding_error)
raise encoding_error
return responses.JSONResponse(content=json_serialized_content)
@app.post("/api/stream_reasoning_engine")
async def stream_query_endpoint(request: QueryRequest) -> responses.StreamingResponse:
try:
method = getattr(agent, request.class_method)
except AttributeError:
return responses.StreamingResponse(
content=iter([json.dumps({"error": f"Method {request.class_method} not found"})]),
status_code=400,
media_type="application/json"
)
output = await _invoke_callable_or_raise(method, request.input or {})
return responses.StreamingResponse(
content=json_generator(output),
media_type="application/json",
)
if __name__ == "__main__":
# The container must listen on 0.0.0.0 and port 8080
uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))