Contrato de entorno de ejecución de Agent Platform

Agent Runtime está diseñado para ser independiente del framework de la aplicación. Si eliges implementar tu agente con un contenedor personalizado o un Dockerfile, tu contenedor debe cumplir con el contrato de tiempo de ejecución para responder consultas correctamente.

Para obtener más información sobre los métodos de implementación, consulta Implementa un agente.

Restricciones y requisitos

Para implementar un contenedor personalizado en Agent Runtime, el contenedor debe detectar solicitudes HTTP en 0.0.0.0 en el puerto 8080.

Extremos (opcional)

Tu contenedor puede exponer cualquier extremo HTTP personalizado. Puedes invocar estos extremos personalizados enviando solicitudes a la API subyacente del agente implementado. Para obtener más información, consulta Cómo usar agentes implementados a través de su API subyacente.

Si bien estos endpoints son opcionales a nivel de API, su implementación habilita las siguientes funciones de integración:

  • Compatibilidad con el SDK de Python: La implementación de /api/reasoning_engine y /api/stream_reasoning_engine te permite usar tu agente implementado a través del SDK de Python de Agent Platform.
  • Compatibilidad con Playground: Es necesario implementar /api/stream_reasoning_engine si quieres interactuar con tu agente a través delGoogle Cloud Playground de la consola. Para obtener más información, consulta Compatibilidad con Playground.

Si quieres usar estas funciones, debes implementar los siguientes extremos:

  • /api/reasoning_engine: Se usa para entregar las consultas enviadas a la API de REST de reasoningEngines/query o a los métodos de consulta síncronos y asíncronos del SDK de Python.
  • /api/stream_reasoning_engine: Se usa para entregar respuestas a las consultas enviadas a la API de REST de reasoningEngines/streamQuery o a los métodos de consulta de transmisión del SDK de Python.

Métodos de clase y modos de ejecución

Cuando implementas un contenedor personalizado, debes declarar los métodos de clase admitidos en la lista classMethods de la especificación de implementación. Estos métodos de clase corresponden a las operaciones que definiste cuando desarrollaste tu agente (consulta Cómo registrar métodos personalizados y Cómo consultar el agente con operaciones compatibles). Cada método declarado tiene un name y un api_mode que determinan cómo se enruta:

Modo de API Tipo de ejecución Extremo de enrutamiento
"" (cadena vacía) o "async" Unario (solicitud-respuesta) /api/reasoning_engine
"stream" o "async_stream" Transmisión /api/stream_reasoning_engine

Cuando un cliente invoca un método, el servicio de Agent Runtime envía una solicitud POST al extremo de enrutamiento correspondiente en tu contenedor. El cuerpo JSON de la solicitud contiene el campo class_method (que coincide con el nombre del método) y el campo input.

Métodos obligatorios para la integración

Para usar el SDK de Python o el área de pruebas de la consola de Google Cloud , tu contenedor debe implementar los métodos específicos que esperan estas integraciones:

  • Consulta estándar del SDK: Requiere query (modo "" o "async") y stream_query (modo "stream" o "async_stream").
  • Playground: Requiere stream_query (modo "stream" o "async_stream").

Integración del ADK

Si implementas un agente creado con el Kit de desarrollo de agentes (ADK), puedes compilar tu propio contenedor de proxy en el lenguaje de programación y el framework de servidor que elijas. Para admitir el conjunto completo de funciones del ADK, tu contenedor debe implementar los métodos definidos por el contrato del ADK. Para obtener más información, consulta Cómo usar un agente del ADK y Cómo registrar y administrar un agente del ADK.

Puedes consultar la plantilla AdkApp como implementación de referencia. Para obtener más información, consulta la documentación de referencia de AdkApp y el código fuente de AdkApp.

Si deseas que Google controle automáticamente las actualizaciones cuando actualices tu versión del ADK, debes usar las herramientas que proporciona el ADK para implementar agentes en lugar de escribir tu propio servidor de API. Para obtener más información, consulta la documentación sobre la implementación del ADK.

Especificaciones de la API

Tanto /api/reasoning_engine como /api/stream_reasoning_engine reciben solicitudes HTTP POST con un cuerpo JSON que contiene los siguientes campos:

  • class_method (cadena): Es el nombre del método del agente subyacente que se invocará (por ejemplo, query o stream_query).
  • input (objeto JSON): Son los argumentos que se pasarán al método de clase especificado.

/api/reasoning_engine (unario)

  • Método de solicitud: POST
  • Cuerpo de la solicitud: json { "class_method": "query", "input": { "message": "What is the capital of France?" } }
  • Respuesta: Es un objeto JSON que contiene el resultado de la ejecución del agente. json { "output": "The capital of France is Paris." }

/api/stream_reasoning_engine (Transmisión)

  • Método de solicitud: POST
  • Cuerpo de la solicitud: json { "class_method": "stream_query", "input": { "message": "Tell me a short story." } }
  • Respuesta: Es un flujo de JSON delimitado por líneas (ndjson), en el que cada línea es un fragmento de la respuesta codificado en JSON. json {"output": "Once"} {"output": " upon"} {"output": " a time..."}

Ejemplo de servidor de API (Python)

A continuación, se muestra un ejemplo de un servidor de FastAPI en Python que implementa el contrato de tiempo de ejecución de Agent Platform. Este servidor encapsula un agente de ejemplo (SimpleAgent) y controla el enrutamiento y la codificación. Puedes reemplazar SimpleAgent por tu propia implementación del agente.

Para ejecutar este ejemplo, asegúrate de tener instalados fastapi, uvicorn y 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)))