Agent Platform-Laufzeitvertrag

Agent Runtime ist so konzipiert, dass sie unabhängig vom Anwendungsframework ist. Wenn Sie Ihren Agenten mit einem benutzerdefinierten Container oder einem Dockerfile bereitstellen, muss Ihr Container den Laufzeitvertrag einhalten, damit Anfragen erfolgreich verarbeitet werden können.

Weitere Informationen zu Bereitstellungsmethoden finden Sie unter Agenten bereitstellen.

Einschränkungen und Anforderungen

Wenn Sie einen benutzerdefinierten Container in Agent Runtime bereitstellen möchten, muss der Container auf Port 8080 auf 0.0.0.0 auf HTTP-Anfragen warten.

Endpunkte (optional)

Ihr Container kann beliebige benutzerdefinierte HTTP-Endpunkte bereitstellen. Sie können diese benutzerdefinierten Endpunkte aufrufen, indem Sie Anfragen an die zugrunde liegende API des bereitgestellten Agenten senden. Weitere Informationen finden Sie unter Bereitgestellte Agenten über ihre zugrunde liegende API verwenden.

Diese Endpunkte sind auf API-Ebene optional. Wenn Sie sie implementieren, werden jedoch die folgenden Integrationsfunktionen aktiviert:

  • Python SDK-Unterstützung: Wenn Sie sowohl /api/reasoning_engine als auch /api/stream_reasoning_engine implementieren, können Sie Ihren bereitgestellten Agenten über das Agent Platform Python SDK verwenden.
  • Playground-Unterstützung: Die Implementierung von /api/stream_reasoning_engine ist erforderlich, wenn Sie über den Google Cloud Playground der Console mit Ihrem Agenten interagieren möchten. Weitere Informationen finden Sie unter Playground support.

Wenn Sie diese Funktionen verwenden möchten, müssen Sie die folgenden Endpunkte implementieren:

  • /api/reasoning_engine: Wird verwendet, um Anfragen zu verarbeiten, die an die reasoningEngines/query REST API oder die synchronen und asynchronen Abfragemethoden des Python SDK gesendet werden.
  • /api/stream_reasoning_engine: Wird verwendet, um Anfragen zu verarbeiten, die an die reasoningEngines/streamQuery REST API oder die Streaming-Abfragemethoden des Python SDK gesendet werden.

Klassenmethoden und Ausführungsmodi

Wenn Sie einen benutzerdefinierten Container bereitstellen, müssen Sie die unterstützten Klassenmethoden in der Liste classMethods der Bereitstellungsspezifikation deklarieren. Diese Klassenmethoden entsprechen den Vorgängen, die Sie beim Entwickeln Ihres Agenten definiert haben (siehe Benutzerdefinierte Methoden registrieren und Agenten mit unterstützten Vorgängen abfragen). Jede deklarierte Methode hat einen name und einen api_mode, der bestimmt, wie sie weitergeleitet wird:

API-Modus Ausführungstyp Routing-Endpunkt
"" (leerer String) oder "async" Unär (Anfrage-Antwort) /api/reasoning_engine
"stream" oder "async_stream" Streaming /api/stream_reasoning_engine

Wenn ein Client eine Methode aufruft, sendet der Agent Runtime-Dienst eine POST-Anfrage an den entsprechenden Routing-Endpunkt in Ihrem Container. Der JSON-Text der Anfrage enthält das Feld class_method (entsprechend dem Methodennamen) und das Feld input.

Erforderliche Methoden für die Integration

Wenn Sie das Python SDK oder den Google Cloud Playground der Console verwenden möchten, muss Ihr Container die spezifischen Methoden implementieren, die von diesen Integrationen erwartet werden:

  • Standardabfrage des SDK: Erfordert query (Modus "" oder "async") und stream_query (Modus "stream" oder "async_stream").
  • Playground: Erfordert stream_query (Modus "stream" oder "async_stream").

ADK-Integration

Wenn Sie einen Agenten bereitstellen, der mit dem Agent Development Kit (ADK) erstellt wurde, können Sie Ihren eigenen Proxy-Container in der Programmiersprache und dem Server-Framework Ihrer Wahl erstellen. Damit alle ADK-Funktionen unterstützt werden, muss Ihr Container die im ADK-Vertrag definierten Methoden implementieren. Weitere Informationen finden Sie unter ADK-Agent verwenden und ADK-Agent registrieren und verwalten.

Sie können die Vorlage AdkApp als Referenzimplementierung verwenden. Weitere Informationen finden Sie in der Referenzdokumentation zu AdkApp und im Quellcode von AdkApp.

Wenn Google Updates automatisch verarbeiten soll, wenn Sie Ihre ADK-Version aktualisieren, sollten Sie die von ADK bereitgestellten Tools zum Bereitstellen von Agenten verwenden, anstatt einen eigenen API-Server zu schreiben. Weitere Informationen finden Sie in der ADK-Bereitstellungsdokumentation.

API-Spezifikationen

Sowohl /api/reasoning_engine als auch /api/stream_reasoning_engine empfangen HTTP-POST-Anfragen mit einem JSON-Text, der die folgenden Felder enthält:

  • class_method (String): Der Methodenname des zugrunde liegenden Agenten, der aufgerufen werden soll (z. B. query oder stream_query).
  • input (JSON-Objekt): Die Argumente, die an die angegebene Klassenmethode übergeben werden sollen.

/api/reasoning_engine (unär)

  • Anfragemethode: POST
  • Anfragetext: json { "class_method": "query", "input": { "message": "What is the capital of France?" } }
  • Antwort: Ein JSON-Objekt mit der Ausgabe der Agentenausführung. json { "output": "The capital of France is Paris." }

/api/stream_reasoning_engine (Streaming)

  • Anfragemethode: POST
  • Anfragetext: json { "class_method": "stream_query", "input": { "message": "Tell me a short story." } }
  • Antwort: Ein Stream mit zeilenbegrenztem JSON (ndjson), wobei jede Zeile ein JSON-codierter Teil der Antwort ist. json {"output": "Once"} {"output": " upon"} {"output": " a time..."}

Beispiel für einen API-Server (Python)

Im Folgenden finden Sie ein Beispiel für einen FastAPI-Server in Python, der den Laufzeitvertrag der Agent Platform implementiert. Dieser Server umschließt einen Beispielagenten (SimpleAgent) und verarbeitet Routing und Codierung. Sie können SimpleAgent durch Ihre eigene Agentenimplementierung ersetzen.

Damit Sie dieses Beispiel ausführen können, müssen fastapi, uvicorn und pydantic installiert sein.

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)))