커스텀 컴포넌트 만들기

커스텀 컴포넌트 만들기 (Creating Custom Components)

직접 만든 컴포넌트를 단독으로, 또는 파이프라인 안에서 사용해 봐요.

Haystack으로는 결과 필터링부터 외부 소프트웨어와의 연동까지, 다양한 작업을 위한 커스텀 컴포넌트를 쉽게 만들 수 있어요. 이렇게 만든 컴포넌트는 파이프라인에 넣어 재사용하고 공유할 수 있으며, 패키징해서 Haystack Integrations에 제출하면 외부 사용자들에게도 공유할 수 있답니다!

출처: 공식문서

커스텀 컴포넌트의 요구사항

모든 커스텀 컴포넌트가 지켜야 할 요구사항은 다음과 같아요.

  • @component: 이 데코레이터가 클래스를 컴포넌트로 표시해서, 파이프라인에서 쓸 수 있게 해줍니다.
  • run(): 모든 컴포넌트에 반드시 있어야 하는 메서드예요. 입력 인자를 받아 dict를 반환합니다. 입력은 파이프라인이 실행될 때 넘어오거나, connect()로 연결된 다른 컴포넌트의 출력에서 올 수 있어요. run() 메서드는 컴포넌트에 선언된 입출력 정의와 호환되어야 합니다. 동작 방식을 확인하려면 아래 '확장 예시'를 보세요.

입력을 제자리에서 바꾸지 않기 (Avoid in-place input mutation)

커스텀 컴포넌트를 만들 때, 컴포넌트의 입력을 직접 변경하지 마세요. 대신 입력의 복사본이나 새 버전을 만들어서 그걸 수정한 뒤 반환하는 방식이 좋아요. 원래 입력 값은 다른 컴포넌트나 이후 파이프라인 단계에서 재사용될 수 있기 때문이에요. 입력을 직접 변경하면 다른 컴포넌트가 원래 값을 기대하고 있을 때 의도하지 않은 부작용과 버그가 생길 수 있습니다.

입력의 한두 필드만 바꾸면 되는 경우(예: Document의 meta)에는 dataclasses.replace()를 써서 필드가 갱신된 새 인스턴스를 만드는 게 좋아요. 전체 객체를 깊은 복사하는 것보다 더 간단하고 효율적이거든요.

from dataclasses import replace

def run(self, documents):
    updated = [
        replace(doc, meta={**doc.meta, "processed": True})
        for doc in documents
    ]
    return {"documents": updated}

반대로 list나 dict 같은 중첩된 변경 가능한 구조를 수정해야 하거나, 데이터클래스 인스턴스의 여러 필드를 한꺼번에 갱신해야 한다면 전체 깊은 복사(deep copy)를 쓰는 편이 좋아요.

import copy

def run(self, documents):
    documents_copy = copy.deepcopy(documents)
    # 여기서 documents_copy를 안전하게 변경하세요
    return {"documents": documents_copy}

컴포넌트의 입력과 출력 정의하기

컴포넌트의 입력과 출력을 정의해 봅시다.

입력은 세 가지 방법 중 하나를 선택할 수 있어요.

  • set_input_type: 컴포넌트 인스턴스의 단일 입력 소켓을 정의하거나 업데이트합니다. 다른 입력에 영향을 주지 않고 특정 입력 하나만 런타임에 추가·수정할 때 이상적이에요. 특정 조건에 따라 단일 입력을 동적으로 설정해야 할 때 쓰세요.
  • set_input_types: 여러 입력 소켓을 한 번에 정의합니다. 기존 입력을 대체해요. 컴포넌트에 필요한 모든 입력을 미리 알고 있고 일괄로 구성하고 싶을 때 유용합니다. 초기화 중에 여러 입력을 정의하고 싶을 때 쓰세요.
  • run() 메서드에 인자를 직접 선언하기. 컴포넌트의 입력이 정적이고 클래스 정의 시점에 이미 알려져 있을 때 쓰는 방법이에요.

출력은 두 가지 방법 중 하나를 선택할 수 있어요.

  • @component.output_types: 클래스 정의 시점에 출력 타입과 이름을 정의하는 데코레이터예요. 출력 이름과 타입은 run() 메서드가 반환하는 dict와 일치해야 합니다. 출력 타입이 정적이고 미리 알려져 있을 때 쓰세요. 정적 컴포넌트에는 이 데코레이터가 더 깔끔하고 읽기 좋아요.
  • set_output_types: 컴포넌트 인스턴스의 여러 출력 소켓을 런타임에 정의하거나 업데이트합니다. 출력을 동적으로 구성해야 할 유연성이 필요할 때 유용해요. 출력 타입을 런타임에 설정해야 해서 더 큰 유연성이 필요할 때 쓰세요.

가장 단순한 최소 컴포넌트 설정의 예시입니다.

from haystack import component

@component
class WelcomeTextGenerator:
    """A component generating personal welcome message and making it upper case"""

    @component.output_types(welcome_text=str, note=str)
    def run(self, name: str):
        return {
            "welcome_text": f"Hello {name}, welcome to Haystack!".upper(),
            "note": "welcome message is ready",
        }

여기서 커스텀 컴포넌트 WelcomeTextGenerator는 name 문자열 하나를 입력으로 받아 welcome_text와 note 두 출력을 반환합니다.

두 컴포넌트를 만들어 파이프라인에 연결하기

커스텀 컴포넌트 두 개를 만들어 Haystack 파이프라인에 연결하는 예시를 볼게요.

# import necessary dependencies
from haystack import component, Pipeline

# Create two custom components. Note the mandatory @component decorator and @component.output_types, as well as the mandatory run method.
@component
class WelcomeTextGenerator:
    """A component generating personal welcome message and making it upper case"""

    @component.output_types(welcome_text=str, note=str)
    def run(self, name: str):
        return {
            "welcome_text": ("Hello {name}, welcome to Haystack!".format(name=name)).upper(),
            "note": "welcome message is ready",
        }

@component
class WhitespaceSplitter:
    """A component for splitting the text by whitespace"""

    @component.output_types(split_text=list[str])
    def run(self, text: str):
        return {"split_text": text.split()}

# create a pipeline and add the custom components to it
text_pipeline = Pipeline()
text_pipeline.add_component(
    name="welcome_text_generator",
    instance=WelcomeTextGenerator(),
)
text_pipeline.add_component(name="splitter", instance=WhitespaceSplitter())

# connect the components
text_pipeline.connect(
    sender="welcome_text_generator.welcome_text",
    receiver="splitter.text",
)

# define the result and run the pipeline
result = text_pipeline.run({"welcome_text_generator": {"name": "Bilge"}})
print(result["splitter"]["split_text"])

기존 컴포넌트 확장하기

Haystack에 이미 있는 컴포넌트를 확장하려면, 기존 컴포넌트를 서브클래싱하고 @component 데코레이터로 표시하면 됩니다. run() 메서드를 오버라이드하거나 확장해서 입출력을 처리해요. 초기화 문제를 피하려면 파생 클래스의 __init__에서 super()를 파생 클래스 이름과 함께 호출하세요.

class DerivedComponent(BaseComponent):
    def __init__(self):
        super(DerivedComponent, self).__init__()
        # ...

dc = DerivedComponent()  # ok

확장 컴포넌트의 예시로는 LLMEvaluator에서 파생된 Haystack의 FaithfulnessEvaluator가 있어요.

패키징해서 공유하기

패키징해 공유하고 싶은 커스텀 컴포넌트를 만든다면, 준비된 프로젝트 구조를 제공하는 GitHub 템플릿 저장소를 쓸 수 있어요. 패키징·테스트·배포에 필요한 보일러플레이트가 포함되어 있어, 커스텀 컴포넌트를 독립된 Python 패키지로 만들 수 있답니다. 프로젝트를 처음부터 세팅하지 않고도 새 통합이나 재사용 가능한 컴포넌트를 빠르게 스캐폴딩할 수 있어요.

템플릿 사용법을 단계별로 보여주는 영상 워크스루도 확인해 보세요.

🧑‍🍳 쿡북(Cookbooks):

더 알아보기 (Learn more)