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 Esegui 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 Utilizza gli agenti di cui è stato eseguito il deployment tramite l'API sottostante.
Sebbene questi endpoint siano facoltativi a livello API, la loro implementazione consente di utilizzare le seguenti funzionalità di integrazione:
- Supporto dell'SDK Python: l'implementazione di
/api/reasoning_enginee/api/stream_reasoning_engineti consente di utilizzare l'agente di cui hai eseguito il deployment tramite l'SDK Python di Gemini Enterprise Agent Platform. - Supporto del playground: l'implementazione di
/api/stream_reasoning_engineè obbligatoria se vuoi interagire con l'agente tramite il playground della consoleGoogle Cloud . Per ulteriori informazioni, consulta Assistenza di 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 in streaming dell'SDK Python.
Metodi della classe e modalità di esecuzione
Quando esegui il deployment di un container personalizzato, devi dichiarare i metodi della classe supportati nell'elenco classMethods della specifica del deployment. Questi metodi della classe corrispondono alle operazioni definite durante lo sviluppo dell'agente (vedi Registrazione di metodi personalizzati e Esecuzione di 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" |
Unaria (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 (corrispondente al nome del metodo) e il campo input.
Metodi richiesti per l'integrazione
Per utilizzare l'SDK Python o il playground della console Google Cloud , 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 esegui 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 Utilizza un agente ADK e Registra e gestisci 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 tua versione di ADK, devi utilizzare gli strumenti forniti da ADK per il deployment degli agenti anziché scrivere il tuo server API. Per saperne di più, consulta la documentazione di deployment di ADK.
Specifiche dell'API
Sia /api/reasoning_engine che /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 della 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: un flusso di JSON delimitato da righe (ndjson), in cui ogni riga è un chunk della risposta con codifica JSON.
json {"output": "Once"} {"output": " upon"} {"output": " a time..."}
Server API di esempio (Python)
Di seguito è riportato un esempio di server FastAPI in Python che implementa il contratto di runtime di Agent Platform. Questo server aggrega 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)))