수퍼컴포넌트

수퍼컴포넌트 (SuperComponents)

SuperComponent를 쓰면 완전한 파이프라인을 감싸서 단일 컴포넌트처럼 사용할 수 있어요. 복잡한 파이프라인의 인터페이스를 단순화하고 싶거나, 여러 상황에서 재사용하고 싶거나, 필요한 입력과 출력만 노출하고 싶을 때 유용하죠.

출처: 공식문서

@super_component 데코레이터 (권장)

Haystack은 이제 파이프라인을 컴포넌트로 감싸기 위한 간단한 @super_component 데코레이터를 제공해요. 데코레이터를 붙인 클래스에 pipeline 속성만 포함하면 됩니다.

이 데코레이터를 쓰면 to_dict와 from_dict 직렬화가 선택 사항이 되고, 입력·출력 매핑도 선택 사항이 돼요.

예제 (Example)

아래의 커스텀 HybridRetriever 예제 SuperComponent는 쿼리를 임베딩으로 바꾼 뒤, BM25 검색과 임베딩 기반 검색을 동시에 실행하고, 마지막에 두 결과 세트를 병합해 결합된 문서를 반환합니다.

이 페이지의 예제들은 sentence-transformers-haystack 패키지로 옮겨진 Sentence Transformers 임베더를 사용합니다. 예제를 실행하려면 설치해야 해요.

pip install sentence-transformers-haystack

# pip install haystack-ai datasets sentence-transformers-haystack
from haystack import Document, Pipeline, super_component
from haystack.components.joiners import DocumentJoiner
from haystack_integrations.components.embedders.sentence_transformers import (
    SentenceTransformersTextEmbedder,
)
from haystack.components.retrievers import (
    InMemoryBM25Retriever,
    InMemoryEmbeddingRetriever,
)
from haystack.document_stores.in_memory import InMemoryDocumentStore
from datasets import load_dataset

@super_component
class HybridRetriever:
    def __init__(
        self,
        document_store: InMemoryDocumentStore,
        embedder_model: str = "BAAI/bge-small-en-v1.5",
    ):
        embedding_retriever = InMemoryEmbeddingRetriever(document_store)
        bm25_retriever = InMemoryBM25Retriever(document_store)
        text_embedder = SentenceTransformersTextEmbedder(embedder_model)
        document_joiner = DocumentJoiner()

        self.pipeline = Pipeline()
        self.pipeline.add_component("text_embedder", text_embedder)
        self.pipeline.add_component("embedding_retriever", embedding_retriever)
        self.pipeline.add_component("bm25_retriever", bm25_retriever)
        self.pipeline.add_component("document_joiner", document_joiner)

        self.pipeline.connect("text_embedder", "embedding_retriever")
        self.pipeline.connect("bm25_retriever", "document_joiner")
        self.pipeline.connect("embedding_retriever", "document_joiner")

dataset = load_dataset("HaystackBot/medrag-pubmed-chunk-with-embeddings", split="train")
docs = [
    Document(content=doc["contents"], embedding=doc["embedding"]) for doc in dataset
]

document_store = InMemoryDocumentStore()
document_store.write_documents(docs)

query = "What treatments are available for chronic bronchitis?"
result = HybridRetriever(document_store).run(text=query, query=query)
print(result)

입력 매핑 (Input Mapping)

선택적으로 SuperComponent의 입력 이름을 파이프라인 내부의 실제 소켓에 매핑할 수 있어요.

input_mapping = {"query": ["retriever.query", "prompt.query"]}

출력 매핑 (Output Mapping)

노출하고 싶은 파이프라인의 출력 소켓을 SuperComponent의 출력 이름에 매핑할 수도 있습니다.

output_mapping = {"llm.replies": "replies"}

매핑을 제공하지 않으면 SuperComponent가 자동 감지하려고 시도해요. 여러 컴포넌트가 같은 이름의 출력을 가질 경우에는 충돌을 피하기 위해 output_mapping을 사용하는 걸 권장합니다.

SuperComponent 클래스

Haystack은 SuperComponent 클래스를 상속하는 옵션도 제공해요. 이 경우 to_dict와 from_dict 직렬화뿐 아니라, 위에서 설명한 입력·출력 매핑도 필요합니다.

예제 (Example)

다음은 파이프라인으로 SuperComponent를 초기화하는 간단한 예제예요.

from haystack import Pipeline, SuperComponent

with open("pipeline.yaml", "r") as file:
    pipeline = Pipeline.load(file)

super_component = SuperComponent(pipeline)

아래 예제 파이프라인은 사용자 쿼리를 바탕으로 관련 문서를 검색하고, 그 문서들로 커스텀 프롬프트를 만든 뒤, OpenAIChatGenerator로 보내 답변을 생성합니다. SuperComponent가 파이프라인을 감싸서, 간단한 입력(query)으로 실행하고 깔끔한 출력(replies)을 반환하도록 만들어 줘요.

from haystack import Pipeline, SuperComponent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.builders import ChatPromptBuilder
from haystack.components.retrievers import InMemoryBM25Retriever
from haystack.dataclasses.chat_message import ChatMessage
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.dataclasses import Document

document_store = InMemoryDocumentStore()
documents = [
    Document(content="Paris is the capital of France."),
    Document(content="London is the capital of England."),
]
document_store.write_documents(documents)

prompt_template = [
    ChatMessage.from_user(
        """
    According to the following documents:
    {% for document in documents %}
    {{document.content}}
    {% endfor %}
    Answer the given question: {{query}}
    Answer:
    """
    )
]

prompt_builder = ChatPromptBuilder(template=prompt_template, required_variables="*")

pipeline = Pipeline()
pipeline.add_component(
    "retriever", InMemoryBM25Retriever(document_store=document_store)
)
pipeline.add_component("prompt_builder", prompt_builder)
pipeline.add_component("llm", OpenAIChatGenerator())
pipeline.connect("retriever.documents", "prompt_builder.documents")
pipeline.connect("prompt_builder.prompt", "llm.messages")

# Create a super component with simplified input/output mapping
wrapper = SuperComponent(
    pipeline=pipeline,
    input_mapping={
        "query": ["retriever.query", "prompt_builder.query"],
    },
    output_mapping={"llm.replies": "replies", "retriever.documents": "documents"},
)

# Run the pipeline with simplified interface
result = wrapper.run(query="What is the capital of France?")
print(result)
# >> {'replies': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>,
# >>  _content=[TextContent(text='The capital of France is Paris.')],...)

타입 검사와 정적 코드 분석 (Type Checking and Static Code Analysis)

@super_component 데코레이터로 SuperComponent를 만들면 타입 또는 린트 오류가 발생할 수 있어요. 이런 문제를 피하는 한 가지 방법은 SuperComponent에 노출된 공개 메서드를 추가하는 거예요. 다음은 그 예제입니다.

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    def run(self, *, documents: list[Document]) -> dict[str, list[Document]]: ...
    def warm_up(self) -> None:  # noqa: D102
        ...

이미 만들어진 SuperComponents (Ready-Made SuperComponents)

Haystack에 이미 통합되어 있는 SuperComponent 구현 두 가지를 확인할 수 있어요.

더 알아보기 (Learn more)