똑똑한 파이프라인 연결

똑똑한 파이프라인 연결 (Smart Pipeline Connections)

Haystack 파이프라인은 보일러플레이트를 줄이고 파이프라인 정의를 읽고 유지하기 쉽게 만들어 주는 스마트한 연결 시맨틱을 지원해요. 이 기능들은 컴포넌트의 동작을 바꾸지 않는 범위에서, 컴포넌트가 서로 어떻게 연결되는지를 단순화하는 데 초점을 맞춥니다.

스마트 연결 덕분에 많은 파이프라인에서 Joiner나 OutputAdapter 같은 흔한 접착(glue) 컴포넌트를 없앨 수 있어요.

출처: 공식문서

암묵적 리스트 결합 (Implicit List Joining)

파이프라인은 여러 컴포넌트의 출력을 명시적인 Joiner 컴포넌트 없이도 단일 컴포넌트의 입력에 직접 연결하는 것을 기본 지원해요.

이 방식이 동작하는 조건은 다음과 같습니다.

  • 대상 입력이 list, list | None, 또는 리스트 타입의 유니온(예: list[int] | list[str])으로 타입이 지정된 경우.
  • 연결된 모든 출력이 호환되는 리스트 타입인 경우.

여러 출력이 같은 입력에 연결되면, 파이프라인이 그 출력들의 리스트를 암묵적으로 이어 붙여(concatenate) 하나의 리스트로 만들어 입력에 전달해요.

예제 (Example)

DocumentJoiner를 쓰지 않고도 여러 컨버터가 단일 DocumentWriter에 직접 쓸 수 있습니다.

from haystack import Pipeline
from haystack.components.converters import HTMLToDocument, TextFileToDocument
from haystack.components.routers import FileTypeRouter
from haystack.components.writers import DocumentWriter
from haystack.dataclasses import ByteStream
from haystack.document_stores.in_memory import InMemoryDocumentStore

sources = [
    ByteStream.from_string(text="Text file content", mime_type="text/plain"),
    ByteStream.from_string(
        text="<html><body>Some content</body></html>",
        mime_type="text/html",
    ),
]

doc_store = InMemoryDocumentStore()

pipe = Pipeline()
pipe.add_component("router", FileTypeRouter(mime_types=["text/plain", "text/html"]))
pipe.add_component("txt_converter", TextFileToDocument())
pipe.add_component("html_converter", HTMLToDocument())
pipe.add_component("writer", DocumentWriter(doc_store))
pipe.connect("router.text/plain", "txt_converter.sources")
pipe.connect("router.text/html", "html_converter.sources")
pipe.connect("txt_converter.documents", "writer.documents")
pipe.connect("html_converter.documents", "writer.documents")

result = pipe.run({"router": {"sources": sources}})

이 패턴은 여러 병렬 분기(parallel branch)에 걸쳐 파일, 문서, 또는 결과를 라우팅할 때 특히 유용해요.

유연한 타입 연결 (Flexible Type Connections)

파이프라인 정의를 더 매끄럽게 만들기 위해, Haystack 파이프라인은 연결 시점에 **제한된 암묵적 타입 적응(implicit type adaptation)**을 지원합니다. 덕분에 파이프라인 연결이 더 유연해지고 OutputAdapter 컴포넌트가 필요해지는 경우가 줄어들어요.

지원되는 적응(adaptation)은 다음과 같습니다.

소스 타입 대상 타입 동작
str ChatMessage user 역할을 가진 ChatMessage로 감쌉니다.
ChatMessage str ChatMessage.text를 추출합니다. None이면 PipelineRuntimeError를 발생시킵니다.
T list[T] 항목을 단일 요소 리스트로 감쌉니다.
list[str] 또는 list[ChatMessage] str 또는 ChatMessage 첫 번째 항목을 추출합니다. 리스트가 비어 있으면 PipelineRuntimeError를 발생시킵니다.

모든 적응은 타입 안전성을 위해 연결 시점에 검사되지만, 실제로 적용되는 것은 **파이프라인 실행 시점(runtime)**이에요.

여러 연결이 가능할 때는 암묵적 변환보다 엄격한 타입 일치가 우선됩니다. 이는 유연한 타입 연결을 지원하지 않던 이전 버전의 Haystack과의 하위 호환성을 지키기 위해서예요.

예제 (Example)

Chat Generator의 messages 출력(list[ChatMessage])을 OutputAdapter 없이 retriever의 query 입력(str)에 연결하는 파이프라인입니다.

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

document_store = InMemoryDocumentStore()

documents = [
    Document(content="Bob lives in Paris."),
    Document(content="Alice lives in London."),
    Document(content="Ivy lives in Melbourne."),
    Document(content="Kate lives in Brisbane."),
    Document(content="Liam lives in Adelaide."),
]

document_store.write_documents(documents)

template = """{% message role="user" %}

Rewrite the following query to be used for keyword search.

{{ query }}

{% endmessage %}

"""

p = Pipeline()
p.add_component("prompt_builder", ChatPromptBuilder(template=template))
p.add_component("llm", OpenAIChatGenerator(model="gpt-4.1-mini"))
p.add_component(
    "retriever",
    InMemoryBM25Retriever(document_store=document_store, top_k=3),
)

p.connect("prompt_builder", "llm")
# implicitly converts list[ChatMessage] -> str
p.connect("llm", "retriever")

query = """Someday I'd love to visit Brisbane, but for now I just want
to know the names of the people who live there."""

result = p.run(data={"prompt_builder": {"query": query}})

여전히 Joiner나 OutputAdapter가 필요할 때 (When You Still Need Joiners or OutputAdapters)

명시적인 Joiner나 OutputAdapter는 다음과 같은 경우 여전히 유용합니다.

  • 단순한 리스트 결합을 넘어서는 커스텀 집계 로직이 필요할 때
  • 암묵적 적응으로는 다루지 못하는 타입 변환이 필요할 때
  • 포맷이나 순서에 대한 명시적 제어가 필요할 때

스마트 연결은 접착 컴포넌트의 필요성을 줄여 주지만, 완전히 없애지는 않아요. 잘 모르겠다면, 명시적인 컴포넌트가 명확성과 통제력을 더해 주니 그것을 사용하는 편이 좋습니다.

더 알아보기 (Learn more)