Agent Runtime est conçu pour être indépendant du framework d'application. Si vous choisissez de déployer votre agent à l'aide d'un conteneur personnalisé ou d'un fichier Dockerfile, votre conteneur doit respecter le contrat d'exécution pour pouvoir répondre aux requêtes.
Pour en savoir plus sur les méthodes de déploiement, consultez Déployer un agent.
Contraintes et exigences
Pour déployer un conteneur personnalisé sur Agent Runtime, le conteneur doit écouter les requêtes HTTP sur 0.0.0.0 sur le port 8080.
Points de terminaison (facultatif)
Votre conteneur peut exposer n'importe quel point de terminaison HTTP personnalisé. Vous pouvez appeler ces points de terminaison personnalisés en envoyant des requêtes à l'API sous-jacente de l'agent déployé. Pour en savoir plus, consultez Utiliser des agents déployés via leur API sous-jacente.
Bien que ces points de terminaison soient facultatifs au niveau de l'API, leur implémentation permet d'activer les fonctionnalités d'intégration suivantes :
- Compatibilité avec le SDK Python : l'implémentation de
/api/reasoning_engineet de/api/stream_reasoning_enginevous permet d'utiliser votre agent déployé via le SDK Agent Platform pour Python. - Prise en charge du terrain de jeu : l'implémentation de
/api/stream_reasoning_engineest requise si vous souhaitez interagir avec votre agent via le terrain de jeu de la consoleGoogle Cloud . Pour en savoir plus, consultez Assistance pour l'atelier.
Si vous souhaitez utiliser ces fonctionnalités, vous devez implémenter les points de terminaison suivants :
/api/reasoning_engine: utilisé pour traiter les requêtes envoyées à l'API RESTreasoningEngines/queryou aux méthodes de requête synchrones et asynchrones du SDK Python./api/stream_reasoning_engine: utilisé pour répondre aux requêtes envoyées à l'API RESTreasoningEngines/streamQueryou aux méthodes de requête de streaming du SDK Python.
Méthodes de classe et modes d'exécution
Lorsque vous déployez un conteneur personnalisé, vous devez déclarer les méthodes de classe compatibles dans la liste classMethods de la spécification de déploiement. Ces méthodes de classe correspondent aux opérations que vous avez définies lors du développement de votre agent (voir Enregistrer des méthodes personnalisées et Interroger l'agent à l'aide des opérations compatibles). Chaque méthode déclarée comporte un name et un api_mode qui déterminent son routage :
| Mode API | Type d'exécution | Point de terminaison de routage |
|---|---|---|
"" (chaîne vide) ou "async" |
Unaire (requête/réponse) | /api/reasoning_engine |
"stream" ou "async_stream" |
Streaming | /api/stream_reasoning_engine |
Lorsqu'un client appelle une méthode, le service Agent Runtime envoie une requête POST au point de terminaison de routage correspondant dans votre conteneur. Le corps JSON de la requête contient le champ class_method (correspondant au nom de la méthode) et le champ input.
Méthodes requises pour l'intégration
Pour utiliser le SDK Python ou l'atelier de la console Google Cloud , votre conteneur doit implémenter les méthodes spécifiques attendues par ces intégrations :
- Requête standard du SDK : nécessite
query(mode""ou"async") etstream_query(mode"stream"ou"async_stream"). - Bac à sable : nécessite
stream_query(mode"stream"ou"async_stream").
Intégration ADK
Si vous déployez un agent créé avec l'Agent Development Kit (ADK), vous pouvez créer votre propre conteneur proxy dans le langage de programmation et le framework de serveur de votre choix. Pour prendre en charge l'ensemble des fonctionnalités ADK, votre conteneur doit implémenter les méthodes définies par le contrat ADK. Pour en savoir plus, consultez Utiliser un agent ADK et Enregistrer et gérer un agent ADK.
Vous pouvez consulter le modèle AdkApp comme implémentation de référence. Pour en savoir plus, consultez la documentation de référence AdkApp et le code source AdkApp.
Si vous souhaitez que Google gère automatiquement les mises à jour lorsque vous mettez à jour votre version d'ADK, vous devez utiliser les outils fournis par ADK pour déployer des agents au lieu d'écrire votre propre serveur d'API. Pour en savoir plus, consultez la documentation sur le déploiement de l'ADK.
Spécifications de l'API
/api/reasoning_engine et /api/stream_reasoning_engine reçoivent des requêtes HTTP POST avec un corps JSON contenant les champs suivants :
class_method(chaîne) : nom de la méthode de l'agent sous-jacent à appeler (par exemple,queryoustream_query).input(objet JSON) : arguments à transmettre à la méthode de classe spécifiée.
/api/reasoning_engine (unaire)
- Méthode de requête :
POST - Corps de la requête :
json { "class_method": "query", "input": { "message": "What is the capital of France?" } } - Réponse : objet JSON contenant le résultat de l'exécution de l'agent.
json { "output": "The capital of France is Paris." }
/api/stream_reasoning_engine (Streaming)
- Méthode de requête :
POST - Corps de la requête :
json { "class_method": "stream_query", "input": { "message": "Tell me a short story." } } - Réponse : flux JSON délimité par des retours à la ligne (ndjson), où chaque ligne est un bloc de réponse encodé au format JSON.
json {"output": "Once"} {"output": " upon"} {"output": " a time..."}
Exemple de serveur d'API (Python)
Voici un exemple de serveur FastAPI en Python qui implémente le contrat d'exécution de la plate-forme d'agents. Ce serveur encapsule un exemple d'agent (SimpleAgent) et gère le routage et l'encodage. Vous pouvez remplacer SimpleAgent par votre propre implémentation d'agent.
Pour exécuter cet exemple, assurez-vous que fastapi, uvicorn et pydantic sont installés.
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)))