Agent Platform 執行階段合約

Agent Runtime 的設計與應用程式架構無關。如果您選擇使用自訂容器或 Dockerfile 部署代理程式，容器必須遵守執行階段合約，才能順利處理查詢。

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

限制和規定

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

端點 (選填)

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

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

  • 支援 Python SDK：同時實作 /api/reasoning_engine 和 /api/stream_reasoning_engine，即可透過 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 串流查詢方法的查詢。

類別方法和執行模式

部署自訂容器時，您必須在部署規格的 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 控制台遊樂區，您的容器必須實作這些整合功能預期的特定方法：

  • 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 原始碼。

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

API 規格

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

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

/api/reasoning_engine (一元)

  • 要求方法：POST
  • 要求主體： json { "class_method": "query", "input": { "message": "What is the capital of France?" } }
  • 回應：包含代理程式執行輸出內容的 JSON 物件。 json { "output": "The capital of France is Paris." }

/api/stream_reasoning_engine (串流)

  • 要求方法：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)))