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을 모두 구현하면 배포된 에이전트를 Agent Platform Python SDK를 통해 사용할 수 있습니다.
  • 플레이그라운드 지원: /api/stream_reasoning_engine을 구현해야 콘솔 플레이그라운드를 통해 에이전트와 상호작용할 수 있습니다.Google Cloud 자세한 내용은 플레이그라운드 지원을 참고하세요.

이러한 기능을 사용하려면 다음 엔드포인트를 구현해야 합니다.

  • /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")가 필요합니다.
  • 플레이그라운드: 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으로 인코딩된 응답 청크인 줄로 구분된 JSON (ndjson) 스트림입니다. 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)))