리소스 객체 — 워크플로에 의존성 주입하기

리소스 객체 — 워크플로에 의존성 주입하기

스텝마다 LLM 클라이언트나 인덱스, DB 핸들을 어떻게 넘겨줄까요? 이벤트에 실어 나르기엔 무겁고, 상태 저장소에 두기엔 직렬화가 안 되죠. 리소스(Resource) 는 이런 외부 의존성을 워크플로의 스텝에 주입하는 공식적인 방법이에요.

출처: 공식문서

리소스가 필요한 대상

리소스는 런타임이 만들어서 스텝에 넣어줘야 하는 객체를 위한 자리예요. LLM 클라이언트, 리트리버, 인덱스, 데이터베이스 핸들, 모델 설정처럼 비싸거나, 상태를 갖거나, JSON 직렬화가 안 되는 의존성을 여기에 둡니다.

간단한 예로 LlamaIndex의 Memory를 워크플로에서 쓰는 경우를 볼게요.

from typing import Annotated
from llama_index.core.workflow import Workflow, step
from llama_index.core.workflow.events import Event, StartEvent, StopEvent
from llama_index.core.workflow.resource import Resource
from llama_index.core.llms import ChatMessage
from llama_index.core.memory import Memory

def get_memory() -> Memory:
    return Memory.from_defaults("user_id_123", token_limit=60000)

class WorkflowWithResource(Workflow):
    @step
    async def first_step(self, ev: StartEvent,
                         memory: Annotated[Memory, Resource(get_memory)]) -> Event:
        await memory.aput(ChatMessage(role="user", content="첫 번째 스텝"))
        return Event(msg="2번 스텝으로 넘길 입력")

    @step
    async def second_step(self, ev: Event,
                          memory: Annotated[Memory, Resource(get_memory)]) -> StopEvent:
        await memory.aput(ChatMessage(role="user", content=ev.msg))
        return StopEvent(result="메모리에 담았어요")

리소스를 주입하려면 스텝 시그니처에 파라미터를 추가하고, 타입을 Annotated로 감싼 뒤 Resource()에 팩토리를 넘기면 돼요.

memory: Annotated[Memory, Resource(get_memory)]

팩토리의 반환 타입은 어노테이션된 파라미터 타입과 일치해야 해요. 기본적으로 리소스는 워크플로 실행 동안 캐시되어, 두 스텝이 같은 Memory 객체를 받고 팩토리는 한 번만 호출돼요. 스텝마다 새 객체가 필요하다면 cache=False를 넘기면 됩니다. 팩토리는 동기일 수도 비동기일 수도 있어요.

memory: Annotated[Memory, Resource(get_memory, cache=False)]

설정 기반 리소스: ResourceConfig

JSON 파일에 저장된 설정 데이터는 Resource 대신 ResourceConfig를 써요. JSON 파일을 로드해서 Pydantic 모델로 파싱해 줍니다.

from typing import Annotated
from pydantic import BaseModel
from llama_index.core.workflow import Workflow, step
from llama_index.core.workflow.events import StartEvent, StopEvent
from llama_index.core.workflow.resource import ResourceConfig

class ClassifierConfig(BaseModel):
    categories: list[str]
    threshold: float

class DocumentClassifier(Workflow):
    @step
    async def classify(self, ev: StartEvent,
                       config: Annotated[ClassifierConfig,
                                         ResourceConfig(config_file="classifier.json")]) -> StopEvent:
        return StopEvent(result=f"Using threshold: {config.threshold}")

주요 파라미터는 이래요.

  • config_file — 설정이 담긴 JSON 파일 경로.
  • path_selector — JSON 파일에서 중첩 값을 뽑을 "."으로 구분된 경로(예: "settings.classifier"). 파일 안에 설정이 여러 개면 특정 구간만 추출해요.
  • label / description — 디버거·시각화 도구에서 설정을 식별하는 데 쓰는 표시 이름과 설명. 생략하면 Pydantic 모델의 타입 이름을 써요.

리소스 체이닝

리소스와 ResourceConfig는 서로 이어 붙일 수 있어요. Resource 팩토리 함수가 같은 Annotated 패턴으로 다른 리소스에 의존함을 선언할 수 있죠.

class LLMConfig(BaseModel):
    model: str
    temperature: float
    max_tokens: int

def get_llm(config: Annotated[LLMConfig, ResourceConfig(config_file="llm.json")]) -> Anthropic:
    return Anthropic(model=config.model, temperature=config.temperature,
                     max_tokens=config.max_tokens)

워크플로가 실행되면 의존성 체인이 자동으로 풀려요. llm.json을 로드해 LLMConfig로 파싱하고, 그 설정으로 get_llm을 호출해 LLM 클라이언트를 만든 뒤, 그 클라이언트를 스텝에 넘겨줍니다. Resource와 ResourceConfig의 어떤 조합이든 이 패턴이 동작해요.

더 알아보기