Kontrak runtime Agent Platform

Agent Runtime dirancang agar independen dari framework aplikasi. Jika Anda memilih untuk men-deploy agen menggunakan container kustom atau Dockerfile, container Anda harus mematuhi kontrak runtime agar berhasil menayangkan kueri.

Untuk mengetahui informasi selengkapnya tentang metode deployment, lihat Men-deploy agen.

Batasan dan persyaratan

Untuk men-deploy container kustom di Agent Runtime, container harus memproses permintaan HTTP di 0.0.0.0 di port 8080.

Endpoint (Opsional)

Container Anda dapat mengekspos endpoint HTTP kustom apa pun. Anda dapat memanggil endpoint kustom ini dengan mengirimkan permintaan ke API dasar agen yang di-deploy. Untuk mengetahui informasi selengkapnya, lihat Menggunakan agen yang di-deploy melalui API dasarnya.

Meskipun endpoint ini bersifat opsional di tingkat API, penerapan endpoint ini akan mengaktifkan fitur integrasi berikut:

  • Dukungan Python SDK: Dengan menerapkan /api/reasoning_engine dan /api/stream_reasoning_engine, Anda dapat menggunakan agen yang di-deploy melalui Agent Platform Python SDK.
  • Dukungan Playground: Penerapan /api/stream_reasoning_engine diperlukan jika Anda ingin berinteraksi dengan agen melalui Google Cloud playground konsol. Untuk mengetahui informasi selengkapnya, lihat Dukungan Playground.

Jika ingin menggunakan fitur ini, Anda harus menerapkan endpoint berikut:

  • /api/reasoning_engine: Digunakan untuk menayangkan kueri yang dikirim ke reasoningEngines/query REST API atau metode kueri sinkron dan asinkron Python SDK.
  • /api/stream_reasoning_engine: Digunakan untuk menayangkan kueri yang dikirim ke reasoningEngines/streamQuery REST API atau metode kueri streaming Python SDK.

Metode class dan mode eksekusi

Saat men-deploy container kustom, Anda harus mendeklarasikan metode class yang didukung dalam daftar classMethods spesifikasi deployment. Metode class ini sesuai dengan operasi yang Anda tentukan saat mengembangkan agen (lihat Mendaftarkan metode kustom dan Mengkueri agen menggunakan operasi yang didukung). Setiap metode yang dideklarasikan memiliki name dan api_mode yang menentukan cara peruteannya:

Mode API Jenis Eksekusi Endpoint Perutean
"" (string kosong) atau "async" Unary (Permintaan-Respons) /api/reasoning_engine
"stream" atau "async_stream" Streaming /api/stream_reasoning_engine

Saat klien memanggil metode, layanan Agent Runtime akan mengirimkan permintaan POST ke endpoint perutean yang sesuai di container Anda. Isi JSON permintaan berisi kolom class_method (yang cocok dengan nama metode) dan kolom input.

Metode yang diperlukan untuk integrasi

Untuk menggunakan Python SDK atau Google Cloud playground konsol, container Anda harus menerapkan metode spesifik yang diharapkan oleh integrasi ini:

  • Kueri standar SDK: Memerlukan query (mode "" atau "async") dan stream_query (mode "stream" atau "async_stream").
  • Playground: Memerlukan stream_query (mode "stream" atau "async_stream").

Integrasi ADK

Jika men-deploy agen yang dibuat dengan Agent Development Kit (ADK), Anda dapat membuat container proxy sendiri dalam bahasa pemrograman dan framework server pilihan Anda. Untuk mendukung rangkaian lengkap fitur ADK, container Anda harus menerapkan metode yang ditentukan oleh kontrak ADK. Untuk mengetahui informasi selengkapnya, lihat Menggunakan agen ADK dan Mendaftarkan dan mengelola agen ADK.

Anda dapat melihat template AdkApp sebagai implementasi referensi. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi AdkApp dan kode sumber AdkApp.

Jika Anda ingin Google menangani pembaruan secara otomatis saat Anda memperbarui versi ADK, Anda harus menggunakan alat yang disediakan oleh ADK untuk men-deploy agen, bukan menulis server API Anda sendiri. Untuk mengetahui informasi selengkapnya, lihat dokumentasi deployment ADK.

Spesifikasi API

Baik /api/reasoning_engine maupun /api/stream_reasoning_engine menerima permintaan POST HTTP dengan isi JSON yang berisi kolom berikut:

  • class_method (string): Nama metode agen dasar yang akan dipanggil (misalnya, query atau stream_query).
  • input (objek JSON): Argumen yang akan diteruskan ke metode class yang ditentukan.

/api/reasoning_engine (Unary)

  • Metode permintaan: POST
  • Isi permintaan: json { "class_method": "query", "input": { "message": "What is the capital of France?" } }
  • Respons: Objek JSON yang berisi output eksekusi agen. json { "output": "The capital of France is Paris." }

/api/stream_reasoning_engine (Streaming)

  • Metode permintaan: POST
  • Isi permintaan: json { "class_method": "stream_query", "input": { "message": "Tell me a short story." } }
  • Respons: Streaming JSON yang dibatasi baris (ndjson), dengan setiap baris adalah bagian respons yang dienkode JSON. json {"output": "Once"} {"output": " upon"} {"output": " a time..."}

Contoh server API (Python)

Berikut adalah contoh server FastAPI di Python yang menerapkan kontrak runtime Agent Platform. Server ini menggabungkan agen contoh (SimpleAgent) dan menangani perutean dan encoding. Anda dapat mengganti SimpleAgent dengan implementasi agen Anda sendiri.

Untuk menjalankan contoh ini, pastikan Anda telah menginstal fastapi, uvicorn, dan 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)))