리소스 객체(Resource Objects)

리소스 객체(Resource Objects)

리소스는 워크플로의 스텝에 주입할 수 있는 외부 의존성이에요.

이벤트로 전달되거나 ctx.store에 저장되기보다는 런타임이 만들어야 하는 객체에 리소스를 쓰세요. LLM 클라이언트, 리트리버, 인덱스, DB 핸들, 모델 설정, 그리고 비싸거나 상태를 가지거나 JSON 직렬화가 안 되는 다른 의존성이 여기 해당돼요.

간단한 예로 LlamaIndex의 Memory를 워크플로에서 써 보겠습니다.

from typing import Annotated

from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent
from workflows.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 SecondEvent(Event):
    msg: str


class WorkflowWithResource(Workflow):
    @step
    async def first_step(
        self,
        ev: StartEvent,
        memory: Annotated[Memory, Resource(get_memory)],
    ) -> SecondEvent:
        print("Memory before step 1", memory)
        await memory.aput(
            ChatMessage(role="user", content="This is the first step")
        )
        print("Memory after step 1", memory)
        return SecondEvent(msg="This is an input for step 2")

    @step
    async def second_step(
        self, ev: SecondEvent, memory: Annotated[Memory, Resource(get_memory)]
    ) -> StopEvent:
        print("Memory before step 2", memory)
        await memory.aput(ChatMessage(role="user", content=ev.msg))
        print("Memory after step 2", memory)
        return StopEvent(result="Messages put into memory")

출처: 공식문서 - Resource Objects

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

memory: Annotated[Memory, Resource(get_memory)]

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

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

팩토리는 동기 또는 비동기가 될 수 있습니다.

설정 기반 리소스(Config-backed Resources)

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

from typing import Annotated
from pydantic import BaseModel
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent
from workflows.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:
        # config is loaded from classifier.json and validated as ClassifierConfig
        return StopEvent(result=f"Using threshold: {config.threshold}")

파라미터

  • config_file: 설정이 담긴 JSON 파일 경로.
  • path_selector: JSON 파일에서 중첩 값을 추출할 선택적 "." 구분 JSON 경로(예: "settings.classifier").
  • label: 워크플로 시각화를 위한 선택적 표시 이름.
  • description: 워크플로 시각화를 위한 선택적 설명.

중첩 값 선택하기

JSON 파일에 설정이 여러 개 있다면 path_selector로 특정 섹션을 추출하세요.

# Given config.json: {"classifier": {"categories": [...], "threshold": 0.8}, "other": {...}}
config: Annotated[
    ClassifierConfig,
    ResourceConfig(config_file="config.json", path_selector="classifier"),
]

시각화에서의 라벨과 설명

디버거나 다른 시각화 도구에서 워크플로를 볼 때 label과 description이 설정을 식별하는 데 도움을 줍니다.

config: Annotated[
    ClassifierConfig,
    ResourceConfig(
        config_file="classifier.json",
        label="Document Classifier",
        description="Categories and confidence threshold for classification",
    ),
]

라벨을 제공하지 않으면 Pydantic 모델의 타입 이름(예: "ClassifierConfig")이 사용됩니다.

리소스 체이닝(Chaining Resources)

Resource와 ResourceConfig는 서로 연결할 수 있어요. Resource 팩토리 함수는 같은 Annotated 패턴으로 다른 리소스에 대한 의존성을 선언할 수 있습니다.

from typing import Annotated
from pydantic import BaseModel
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent
from workflows.resource import Resource, ResourceConfig
from llama_index.llms.anthropic import Anthropic


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


class MyWorkflow(Workflow):
    @step
    async def generate(
        self,
        ev: StartEvent,
        llm: Annotated[Anthropic, Resource(get_llm)],
    ) -> StopEvent:
        response = await llm.acomplete(ev.input)
        return StopEvent(result=response.text)

의존성 체인은 자동으로 해결됩니다. 이 예에서 워크플로가 실행되면:

  1. llm.json이 로드되어 LLMConfig로 파싱된다
  2. get_llm이 그 설정으로 호출되어 LLM 클라이언트를 만든다
  3. 결과 클라이언트가 스텝에 전달된다

이 패턴은 Resource와 ResourceConfig 의존성의 어떤 조합에서든 동작합니다.

더 알아보기