Ray Serve 모델 구성 (DeploymentHandle)

Ray Serve 모델 구성 (DeploymentHandle)

여러 ML 모델이나 비즈니스 로직을 담은 deployment들을 하나의 application으로 묶고 싶다면 **모델 구성(composition)**을 쓰면 돼요. DeploymentHandle을 통하면 전처리·추론·후처리 같은 단계를 독립적인 deployment로 쪼개고, 각 단계를 따로 스케일링하고 설정할 수 있어요. 이 가이드에선 DeploymentHandle의 기본 사용법부터 체이닝, 스트리밍 호출까지 다뤄요.

출처: Deploy Compositions of Models — Ray Docs

:::note note Ray 2.10부터 deprecated였던 RayServeHandle와 RayServeSyncHandle API는 완전히 제거됐어요. :::

DeploymentHandle로 deployment 구성하기

application을 만들 때 여러 deployment를 .bind() 하고 서로의 생성자에 전달할 수 있어요. 런타임에서 deployment 코드 안의 bound deployment는 DeploymentHandle로 치환되고, 이 핸들로 다른 deployment의 메서드를 호출할 수 있어요. 따라서 전처리, 모델 추론, 후처리 같은 단계를 독립적인 deployment로 나눠서 각각 독립적으로 스케일링·설정할 수 있어요.

handle.remote로 deployment에 요청을 보내요. 요청에는 일반 Python 인자와 키워드 인자를 넣을 수 있고, DeploymentHandle은 이를 메서드에 그대로 전달해요. 메서드 호출은 결과를 가리키는 퓨처(future)인 DeploymentResponse를 반환해요. 응답을 await해서 결과를 얻거나, 그 결과를 다른 다운스트림 DeploymentHandle 호출에 넘길 수 있어요.

기본 DeploymentHandle 예시

아래 예시는 두 개의 deployment를 가져요.

# File name: hello.py
from ray import serve
from ray.serve.handle import DeploymentHandle


@serve.deployment
class LanguageClassifer:
    def __init__(
        self, spanish_responder: DeploymentHandle, french_responder: DeploymentHandle
    ):
        self.spanish_responder = spanish_responder
        self.french_responder = french_responder

    async def __call__(self, http_request):
        request = await http_request.json()
        language, name = request["language"], request["name"]

        if language == "spanish":
            response = self.spanish_responder.say_hello.remote(name)
        elif language == "french":
            response = self.french_responder.say_hello.remote(name)
        else:
            return "Please try again."

        return await response


@serve.deployment
class SpanishResponder:
    def say_hello(self, name: str):
        return f"Hola {name}"


@serve.deployment
class FrenchResponder:
    def say_hello(self, name: str):
        return f"Bonjour {name}"


spanish_responder = SpanishResponder.bind()
french_responder = FrenchResponder.bind()
language_classifier = LanguageClassifer.bind(spanish_responder, french_responder)

LanguageClassifier는 생성자 인자로 spanish_responder와 french_responder를 받고, 런타임에서 Ray Serve가 이 인자들을 DeploymentHandle로 변환해요. 그러면 LanguageClassifier가 핸들로 두 deployment의 메서드를 호출할 수 있어요. HTTP 요청의 값을 보고 스페인어/프랑스어 응답자를 고르는 형태죠. 호출 형식은 이렇게 표현돼요:

response: DeploymentResponse = self.spanish_responder.say_hello.remote(name)

이 호출을 풀어보면:

• self.spanish_responder는 생성자로 받은 SpanishResponder의 핸들이고,
• say_hello는 호출할 SpanishResponder의 메서드이며,
• remote는 다른 deployment로의 DeploymentHandle 호출임을 나타내고,
• name은 say_hello의 인자예요. 여기에 인자나 키워드 인자를 원하는 만큼 넣을 수 있어요.

이 호출은 결과 자체가 아니라 결과에 대한 참조인 DeploymentResponse 객체를 반환해요. 이 패턴 덕에 비동기로 실행할 수 있어요. 실제 결과를 얻으려면 await로 응답을 기다리면 되는데, async 호출이 끝날 때까지 블록됐다가 결과를 반환해요.

:::warning 경고 remote DeploymentHandle 호출 결과를 얻는 데 response.result() 메서드를 쓸 수 있어요. 다만 deployment 내부에서 .result()를 호출하는 건 피해야 해요 — 원격 메서드 호출이 끝날 때까지 다른 코드를 실행하지 못하고 블록되거든요. await를 쓰면 원격 호출을 기다리는 동안 다른 요청도 처리할 수 있어요. deployment 안에서는 .result()보다 await를 쓰세요. :::

위의 hello.py를 복사해서 serve run으로 실행할 수 있어요. hello.py가 있는 디렉터리에서 실행해야 스크립트를 찾을 수 있어요:

$ serve run hello:language_classifier

이 클라이언트 스크립트로 예시와 상호작용할 수 있어요:

# File name: hello_client.py
import requests

response = requests.post(
    "http://localhost:8000", json={"language": "spanish", "name": "Dora"}
)
greeting = response.text
print(greeting)

serve run이 실행되는 동안 별도의 터미널에서 스크립트를 실행하면 Hola Dora가 출력돼요.

:::note 참고 구성(composition)을 쓰면 application을 쪼개서 각 부분을 독립적으로 스케일링할 수 있어요. 예를 들어 LanguageClassifier 요청의 75%가 스페인어고 25%가 프랑스어라고 해볼게요. 그러면 SpanishResponder는 레플리카 3개, FrenchResponder는 레플리카 1개로 설정해서 워크로드 수요를 맞출 수 있어요. 이 유연성은 CPU·GPU 같은 리소스 예약과 deployment마다 설정할 수 있는 다른 모든 구성에도 동일하게 적용돼요.

구성을 쓰면 서로 다른 유형·양의 리소스를 쓰는 모델과 비즈니스 로직 단계를 서빙할 때 application 차원의 병목을 피할 수 있어요. :::

DeploymentHandle 호출 체이닝

Ray Serve는 DeploymentHandle이 반환한 DeploymentResponse 객체를 다른 DeploymentHandle 호출에 직접 넘겨 파이프라인의 여러 단계를 연결할 수 있어요. 첫 응답을 await하지 않아도 되고, Ray Serve가 내부적으로 await 동작을 관리해요. 첫 호출이 끝나면 DeploymentResponse 객체가 아니라 첫 호출의 출력을 두 번째 호출에 직접 전달해요.

아래 예시는 application에 세 개의 deployment를 정의해요:

• 설정한 증가분만큼 값을 더하는 Adder deployment
• 설정한 배수만큼 값을 곱하는 Multiplier deployment
• adder와 multiplier 호출을 체이닝하고 최종 응답을 반환하는 Ingress deployment

Adder 핸들의 응답이 Multiplier 핸들에 직접 전달되는데, multiplier 내부에서 입력 인자는 Adder 호출의 출력으로 해석돼요.

# File name: chain.py
from ray import serve
from ray.serve.handle import DeploymentHandle, DeploymentResponse


@serve.deployment
class Adder:
    def __init__(self, increment: int):
        self._increment = increment

    def __call__(self, val: int) -> int:
        return val + self._increment


@serve.deployment
class Multiplier:
    def __init__(self, multiple: int):
        self._multiple = multiple

    def __call__(self, val: int) -> int:
        return val * self._multiple


@serve.deployment
class Ingress:
    def __init__(self, adder: DeploymentHandle, multiplier: DeploymentHandle):
        self._adder = adder
        self._multiplier = multiplier

    async def __call__(self, input: int) -> int:
        adder_response: DeploymentResponse = self._adder.remote(input)
        # Pass the adder response directly into the multiplier (no `await` needed).
        multiplier_response: DeploymentResponse = self._multiplier.remote(
            adder_response
        )
        # `await` the final chained response.
        return await multiplier_response


app = Ingress.bind(
    Adder.bind(increment=1),
    Multiplier.bind(multiple=2),
)

handle: DeploymentHandle = serve.run(app)
response = handle.remote(5)
assert response.result() == 12, "(5 + 1) * 2 = 12"

DeploymentHandle 스트리밍 호출

DeploymentHandle로 여러 출력을 반환하는 스트리밍 메서드 호출도 할 수 있어요. 스트리밍 호출을 만들려면 메서드가 제너레이터(generator)여야 하고 handle.options(stream=True)를 설정해야 해요. 그러면 핸들 호출이 단일 DeploymentResponse 대신 DeploymentResponseGenerator를 반환해요. DeploymentResponseGenerator는 async for 코드 블록에서처럼 동기·비동기 제너레이터로 쓸 수 있어요. DeploymentResponse.result()처럼, deployment 안에서 DeploymentResponseGenerator를 동기 제너레이터로 쓰는 건 피해야 해요 — 그 레플리카에서 다른 요청이 동시에 실행되는 걸 막거든요. 그리고 DeploymentResponseGenerator는 다른 핸들 호출에 전달할 수 없어요.

# File name: stream.py
from typing import AsyncGenerator, Generator

from ray import serve
from ray.serve.handle import DeploymentHandle, DeploymentResponseGenerator


@serve.deployment
class Streamer:
    def __call__(self, limit: int) -> Generator[int, None, None]:
        for i in range(limit):
            yield i


@serve.deployment
class Caller:
    def __init__(self, streamer: DeploymentHandle):
        self._streamer = streamer.options(
            # Must set `stream=True` on the handle, then the output will be a
            # response generator.
            stream=True,
        )

    async def __call__(self, limit: int) -> AsyncGenerator[int, None]:
        # Response generator can be used in an `async for` block.
        r: DeploymentResponseGenerator = self._streamer.remote(limit)
        async for i in r:
            yield i


app = Caller.bind(Streamer.bind())

handle: DeploymentHandle = serve.run(app).options(
    stream=True,
)

# Response generator can also be used as a regular generator in a sync context.
r: DeploymentResponseGenerator = handle.remote(10)
assert list(r) == list(range(10))

고급: DeploymentResponse를 Ray ObjectRef로 변환하기

내부적으로 각 DeploymentResponse는 Ray ObjectRef(스트리밍 호출은 ObjectRefGenerator)에 대응해요. DeploymentHandle 호출을 Ray Actor나 Task와 구성하려면 응답을 ObjectRef로 해석해야 할 수 있어요. 이를 위해 DeploymentResponse._to_object_ref와 DeploymentResponse._to_object_ref_sync 개발자 API를 쓸 수 있어요.

# File name: response_to_object_ref.py
import ray
from ray import serve
from ray.serve.handle import DeploymentHandle, DeploymentResponse


@ray.remote
def say_hi_task(inp: str):
    return f"Ray task got message: '{inp}'"


@serve.deployment
class SayHi:
    def __call__(self) -> str:
        return "Hi from Serve deployment"


@serve.deployment
class Ingress:
    def __init__(self, say_hi: DeploymentHandle):
        self._say_hi = say_hi

    async def __call__(self):
        # Make a call to the SayHi deployment and pass the result ref to
        # a downstream Ray task.
        response: DeploymentResponse = self._say_hi.remote()
        response_obj_ref: ray.ObjectRef = await response._to_object_ref()
        final_obj_ref: ray.ObjectRef = say_hi_task.remote(response_obj_ref)
        return await final_obj_ref


app = Ingress.bind(SayHi.bind())
handle: DeploymentHandle = serve.run(app)
assert handle.remote().result() == "Ray task got message: 'Hi from Serve deployment'"

더 알아보기

• Ray Serve 핵심 개념 — Deployment, Application, DeploymentHandle
• Model Multiplexing으로 여러 모델 서빙하기
• DeploymentHandle API 문서
• DeploymentResponse API 문서