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 的流式查询方法的查询。

类方法和执行模式

部署自定义容器时，您必须在部署规范的 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 集成

如果您要部署使用智能体开发套件 (ADK) 构建的智能体，则可以使用您选择的编程语言和服务器框架构建自己的代理容器。为了支持全套 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（一元）

  • 请求方法：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 中实现 Agent Platform 运行时协定的 FastAPI 服务器示例。此服务器封装了一个示例智能体 (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)))