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를 통해 배포된 에이전트를 사용할 수 있습니다.
  • 플레이그라운드 지원:Google Cloud 콘솔 플레이그라운드를 통해 에이전트와 상호작용하려면 /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 통합

에이전트 개발 키트 (ADK)로 빌드된 에이전트를 배포하는 경우 원하는 프로그래밍 언어와 서버 프레임워크로 자체 프록시 컨테이너를 빌드할 수 있습니다. 전체 ADK 기능을 지원하려면 컨테이너가 ADK 계약에 정의된 메서드를 구현해야 합니다. 자세한 내용은 ADK 에이전트 사용 및 ADK 에이전트 등록 및 관리를 참고하세요.

AdkApp 템플릿을 참조 구현으로 참고할 수 있습니다. 자세한 내용은 AdkApp 참고 문서 및 AdkApp 소스 코드를 참고하세요.

ADK 버전을 업데이트할 때 Google에서 업데이트를 자동으로 처리하도록 하려면 자체 API 서버를 작성하는 대신 ADK에서 제공하는 도구를 사용하여 에이전트를 배포해야 합니다. 자세한 내용은 ADK 배포 문서를 참고하세요.

API 사양

/api/reasoning_engine 및 /api/stream_reasoning_engine는 모두 다음 필드가 포함된 JSON 본문이 있는 HTTP POST 요청을 수신합니다.

  • 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)

다음은 Agent Platform 런타임 계약을 구현하는 Python의 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)))