Vertrag für die Gemini Enterprise Agent Platform-Laufzeit

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

Weitere Informationen zu Bereitstellungsmethoden finden Sie unter Agent bereitstellen.

Einschränkungen und Anforderungen

Wenn Sie einen benutzerdefinierten Container in der Agent Runtime bereitstellen möchten, muss der Container auf 0.0.0.0 an Port 8080 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 KI-Agenten senden. Weitere Informationen finden Sie unter Bereitgestellte Agenten über die zugrunde liegende API verwenden.

Diese Endpunkte sind auf API-Ebene zwar optional, aber wenn Sie sie implementieren, können Sie die folgenden Integrationsfunktionen nutzen:

• 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 Gemini Enterprise Agent Platform Python SDK verwenden.
• Playground-Unterstützung: Die Implementierung von /api/stream_reasoning_engine ist erforderlich, wenn Sie über denGoogle Cloud Console-Playground mit Ihrem Agenten interagieren möchten. Weitere Informationen finden Sie unter Playground-Unterstützung.

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

• /api/reasoning_engine: Wird für die Bereitstellung von Abfragen verwendet, die an die reasoningEngines/query REST API oder die synchronen und asynchronen Abfragemethoden des Python SDK gesendet werden.
• /api/stream_reasoning_engine: Wird für die Bereitstellung von Abfragen verwendet, 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 bei der Entwicklung 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, die bestimmen, 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 (entspricht dem Methodennamen) und das Feld input.

Erforderliche Methoden für die Integration

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

• SDK-Standardabfrage: 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 KI-Agenten bereitstellen, der mit dem Agent Development Kit (ADK) erstellt wurde, können Sie einen 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 AdkApp-Referenzdokumentation und im AdkApp-Quellcode.

Wenn Sie möchten, dass Google Updates automatisch verarbeitet, wenn Sie Ihre ADK-Version aktualisieren, sollten Sie die vom ADK bereitgestellten Tools zum Bereitstellen von Agenten verwenden, anstatt einen eigenen API-Server zu schreiben. Weitere Informationen finden Sie in der Dokumentation zur ADK-Bereitstellung.

API-Spezifikationen

Sowohl /api/reasoning_engine als auch /api/stream_reasoning_engine erhalten 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, das die Ausgabe der Agentenausführung enthält. 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 von durch Zeilen getrennten JSON-Objekten (ndjson), wobei jede Zeile ein JSON-codierter Block der Antwort ist. json {"output": "Once"} {"output": " upon"} {"output": " a time..."}

Beispiel-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 verpackt einen Beispiel-Agenten (SimpleAgent) und übernimmt das Routing und die 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)))