Gemini Enterprise Agent Platform 執行階段合約

Agent Runtime 獨立於應用程式框架，如果您選擇使用自訂容器或 Dockerfile 部署代理，容器必須遵守執行階段合約，才能成功處理查詢。

如要進一步瞭解部署方法，請參閱「部署代理程式」。

限制和規定

如要在 Agent Runtime 部署自訂容器，容器必須在通訊埠 8080 監聽 0.0.0.0 的 HTTP 要求。

端點 (選填)

容器可以公開任何自訂 HTTP 端點。如要叫用這些自訂端點，請將要求傳送至已部署代理的基礎 API。詳情請參閱「透過基礎 API 使用已部署的代理」。

雖然這些端點在 API 級別為選用，但實作後可啟用下列整合功能：

  • Python SDK 支援：同時實作 /api/reasoning_engine 和 /api/stream_reasoning_engine，即可透過 Gemini Enterprise Agent Platform Python SDK 使用已部署的代理。
  • Playground 支援：如要透過Google Cloud 控制台 Playground 與代理互動，則必須實作 /api/stream_reasoning_engine。詳情請參閱「Playground 支援」。

如要使用這些功能，請務必實作下列端點：

  • /api/reasoning_engine：用於提供傳送至 reasoningEngines/query REST API 或 Python SDK 同步和非同步查詢方法的查詢。
  • /api/stream_reasoning_engine：用於提供傳送至 reasoningEngines/streamQuery REST API 或 Python SDK 串流查詢方法的查詢。

類別方法和執行模式

部署自訂容器時，您必須在 Deployment 規格的 classMethods 清單中，宣告支援的類別方法。這些類別方法對應您在開發代理時定義的操作 (請參閱「註冊自訂方法」和「使用支援的操作查詢代理」)。每個宣告的方法都有 name 和 api_mode，用於決定轉送方式：

API 模式 執行作業類型 轉送端點
"" (空字串) 或 "async" 一元 (要求-回應) /api/reasoning_engine
"stream"或"async_stream" 串流 /api/stream_reasoning_engine

用戶端叫用方法時，Agent Runtime 服務會將 POST 要求傳送至容器中對應的轉送端點。要求的 JSON 主體包含 class_method 欄位 (與方法名稱相符) 和 input 欄位。

整合所需方法

如要使用 Python SDK 或 Google Cloud 控制台 Playground，容器必須實作這些整合項目預期的方法：

  • SDK 標準查詢：需要 query (模式 "" 或 "async") 和 stream_query (模式 "stream" 或 "async_stream")。
  • Playground：需要 stream_query (模式 "stream" 或 "async_stream")。

ADK 整合

如果您要部署以 Agent Development Kit (ADK) 建構的代理，可以選擇偏好的程式設計語言和伺服器架構，自行建構 Proxy 容器。如要支援完整的 ADK 功能，容器必須實作 ADK 協定定義的方法。詳情請參閱「使用 ADK 代理」和「註冊及管理 ADK 代理」。

您可以參考 AdkApp 範本，瞭解如何實作。詳情請參閱 AdkApp 參考文件和 AdkApp 原始碼。

如果希望 Google 在您更新 ADK 版本時自動處理更新，請使用 ADK 提供的工具部署代理，而非自行編寫 API 伺服器。詳情請參閱 ADK 部署作業說明文件。

API 規格

/api/reasoning_engine 和 /api/stream_reasoning_engine 都會收到 HTTP POST 要求，其中 JSON 主體包含下列欄位：

  • class_method (字串)：要叫用的基礎代理方法名稱 (例如 query 或 stream_query)。
  • input (JSON 物件)：要傳遞至指定類別方法的引數。

/api/reasoning_engine (一元)

  • Request method (要求方法)：POST
  • 要求主體： json { "class_method": "query", "input": { "message": "What is the capital of France?" } }
  • Response：JSON 物件，包含代理執行作業的輸出內容。json { "output": "The capital of France is Paris." }

/api/stream_reasoning_engine (串流)

  • Request method (要求方法)：POST
  • 要求主體： json { "class_method": "stream_query", "input": { "message": "Tell me a short story." } }
  • 回應：以行分隔的 JSON (ndjson) 串流，每行都是 JSON 編碼的回應區塊。 json {"output": "Once"} {"output": " upon"} {"output": " a time..."}

API 伺服器範例 (Python)

以下範例是以 Python 編寫的 FastAPI 伺服器，用於實作 Agent Platform 執行階段合約。這個伺服器會封裝範例代理 (SimpleAgent)，並處理轉送和編碼作業。您可以使用自己的代理程式實作項目取代 SimpleAgent。

如要執行這個範例，請確認已安裝 fastapi、uvicorn 和 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)))