파이프라인

파이프라인 (Pipelines)

LLM으로 현대적인 검색 파이프라인을 만들려면 두 가지가 필요해요. 강력한 컴포넌트와, 그것들을 쉽게 조립할 방법이죠. Haystack 파이프라인은 바로 그 목적을 위해 만들어졌고, LLM과의 상호작용을 설계하고 확장하게 해 줘요. Haystack 파이프라인은 Haystack 컴포넌트와 통합(integration)들의 방향성 있는 다중 그래프(directed multigraph)예요. 이 컴포넌트들을 다양한 방식으로 연결할 자유를 주므로, 파이프라인이 반드시 정보의 연속적인 흐름일 필요는 없어요.

출처: 공식문서

유연성 (Flexibility)

Haystack 파이프라인은 단순한 쿼리·인덱싱 파이프라인 그 이상이에요. 파이프라인이 무엇을 하느냐 — 인덱싱, 쿼리, API에서 가져오기, 전처리 등 — 는 전적으로 파이프라인을 어떻게 설계하고 어떤 컴포넌트를 쓰느냐에 달려 있어요.

물론 단일 기능 파이프라인도 만들 수 있어요. 문서를 정리·분할·저장하는 인덱싱 파이프라인, 쿼리를 받아 답을 반환하는 쿼리 파이프라인 같은 것들이죠. 하지만 Haystack은 ConditionalRouter 같은 결정 컴포넌트를 사용해 여러 유스케이스를 하나의 파이프라인으로 결합하는 것도 허용해요.

에이전틱 파이프라인 (Agentic Pipelines)

Haystack의 루프와 분기(branch)는 에이전트 같은 복잡한 애플리케이션을 만들 수 있게 해 줘요. 만드는 방법 예시는 아래에서 확인할 수 있어요.

분기 (Branching)

파이프라인은 데이터를 동시에 처리하는 여러 분기를 가질 수 있어요. 예를 들어 서로 다른 파일 유형을 처리하려면, 각각 특정 파일 유형을 담당하는 변환기(converter) 여러 개가 있는 파이프라인을 만들면 돼요.

루프 (Loops)

파이프라인의 컴포넌트는 반복 루프로 동작할 수 있고, 원하는 횟수에서 상한을 둘 수 있어요. 자기 수정 루프 같은 시나리오에 유용해요. Generator가 출력을 만들고 validator 컴포넌트가 그 출력이 맞는지 확인하는 방식이죠.

비동기 실행과 스트리밍 (Async Execution and Streaming)

예를 들어 파이프라인은 여러 Retriever나 LLM 호출을 동시에 실행하고, 독립적인 파이프라인 분기를 병렬로 수행하며, 지연을 유발하던 I/O 중심 작업을 효율적으로 처리할 수 있어요.

동시에 실행되는 컴포넌트 수는 비동기 실행 메서드(run_async, run_async_generator, stream)의 concurrency_limit 인자로 상한을 둘 수 있어요. 동기 run 메서드는 컴포넌트를 순차적으로 실행해요.

  • run_async: 파이프라인을 단일 non-blocking 호출로 실행해요. 파이프라인을 더 큰 비동기 애플리케이션이나 서비스에 통합할 때 이상적이에요.
  • run_async_generator: 컴포넌트가 작업을 완료할 때마다 부분 출력을 만들어 내요(yield). 진행 상황 모니터링, 디버깅, 출력을 점증적으로 처리할 때 유용해요.
  • stream: 파이프라인을 실행하고 생성되는 대로 StreamingChunk 객체를 스트리밍하는 핸들을 반환해요. API 엔드포인트 같은 비동기 애플리케이션에서 LLM 출력을 스트리밍하기 편리한 방법이죠.

핸들을 async for로 반복해 청크를 소비해요. 반복이 끝나면 handle.result에 최종 파이프라인 출력(run_async가 반환하는 것과 같은 딕셔너리)이 들어 있어요.

import asyncio

from haystack import Pipeline
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage

pipe = Pipeline()
pipe.add_component(
    "prompt_builder",
    ChatPromptBuilder(template=[ChatMessage.from_user("Tell me about {{topic}}")]),
)
pipe.add_component("llm", OpenAIChatGenerator())
pipe.connect("prompt_builder.prompt", "llm.messages")

async def main():
    handle = pipe.stream(data={"prompt_builder": {"topic": "Italy"}})
    async for chunk in handle:
        print(chunk.content, end="", flush=True)
    return handle.result

result = asyncio.run(main())

기본적으로 스트리밍 가능한 모든 컴포넌트의 청크가 전달돼요. 특정 컴포넌트만 스트리밍하려면 컴포넌트 이름 리스트를 streaming_components로 전달하세요.

소비자가 반복을 포기하면 기본적으로 내부 파이프라인 실행이 자동 취소돼요. 대신 끝까지 실행시키고 싶다면 cancel_on_abandon=False를 전달하세요.

컴포넌트에 streaming_callback이 설정되어 있으면(초기화 시 또는 런타임 data로), 그 콜백도 handle로 푸시되는 청크에 더해 각 청크마다 여전히 호출돼요.

스트리밍할 때 컴포넌트는 run_async에서도 동기 streaming_callback을 받아요 — 자세한 내용은 "Choosing the Right Generator" 가이드를 참고하세요.

오류 처리와 작업 취소

형제 컴포넌트가 동시에 실행 중일 때 컴포넌트가 오류를 일으키면, 파이프라인은 원래 오류를 다시 일으키기 전에 진행 중인 작업들을 취소하고 비워요(drain). 그래서 백그라운드에서 작업이 계속 실행되지 않아요.

같은 정리는 run_async_generator 반복을 일찍 멈출 때(예: 루프에서 빠져나와 제너레이터를 닫을 때)나 실행 자체가 취소될 때도 적용돼요.

취소는 네이티브로 비동기인 컴포넌트만 중단시킨다는 점을 기억하세요. 동기 컴포넌트는 워커 스레드로 오프로드되는데, 이는 중단될 수 없고 백그라운드에서 끝까지 실행돼요.

슈퍼컴포넌트 (SuperComponents)

코드를 단순화하기 위해 SuperComponents를 도입했어요. 완전한 파이프라인을 감싸서 단일 컴포넌트로 재사용할 수 있게 해 주죠. 자세한 내용과 예제는 해당 문서 페이지에서 확인하세요.

데이터 흐름 (Data Flow)

파이프라인에서 컴포넌트와 통합을 연결하려면 그 입력·출력 이름을 알아야 해요. 한 컴포넌트의 출력은 다음 컴포넌트의 입력으로 받아들여져야 해요. Pipeline.connect()로 컴포넌트를 연결하면 파이프라인이 입력·출력 타입이 맞는지 검증해요.

스마트 파이프라인 연결 (Smart Pipeline Connections)

이런 동작 덕분에 Joiners나 OutputAdapters 같은 접착(glue) 컴포넌트의 필요성이 줄어들어, 파이프라인을 간결하고 읽기 쉽게 유지할 수 있어요. 자세한 내용과 예제는 Smart Pipeline Connections 문서를 참고하세요.

파이프라인을 만드는 단계

  1. Pipeline()로 파이프라인을 만들어요. 이게 Pipeline 객체를 생성해요.
  2. .add_component(name, component)로 컴포넌트를 하나씩 추가해요. 이 단계는 아직 연결하지 않고 컴포넌트만 추가해요. 루프에 특히 유용한데, 다음 단계에서 컴포넌트들이 이미 파이프라인에 존재하므로 부드럽게 연결할 수 있기 때문이에요.
  3. .connect("producer_component.output_name", "consumer_component.input_name")로 컴포넌트를 연결해요.
  4. .run({"component_1": {"mandatory_inputs": value}})로 파이프라인을 실행해요. 마지막으로 파이프라인의 첫 컴포넌트를 지정하고 필수 입력을 전달하며 실행해요.

검증 (Validation)

  • 컴포넌트가 파이프라인에 존재하는지.
  • 컴포넌트의 출력·입력이 일치하고 명시적으로 표시되는지. 예를 들어 컴포넌트가 두 개의 출력을 만들면, 다른 컴포넌트에 연결할 때 어떤 출력이 어떤 입력에 연결되는지 표시해야 해요.
  • 컴포넌트 타입이 일치하는지.

직렬화 (Serialization)

직렬화 덕분에 파이프라인을 저장하고 다시 불러올 수 있어요. 직렬화는 Haystack 파이프라인을 disk에 저장하거나 네트워크로 보낼 수 있는 형식으로 변환하는 것이에요. 특히 다음에 유용해요.

  • 파이프라인 편집, 저장, 공유.
  • 파이썬과 다른 형식으로 기존 파이프라인 수정.

Haystack 파이프라인은 직렬화를 컴포넌트에 위임해요. 그래서 파이프라인을 직렬화한다는 것은 각 컴포넌트를 연결 정보와 함께 하나씩 차례로 직렬화한다는 뜻이에요.

직렬화 형식 — 현재 Haystack은 YAML 형식만 지원해요. 앞으로 더 많은 형식을 점진적으로 지원할 예정이에요. 직렬화가 가능하려면 컴포넌트가 파이썬 딕셔너리로의 변환을 지원해야 해요. 모든 Haystack 컴포넌트에는 직렬화를 가능하게 하는 두 메서드 from_dict와 to_dict가 있어요. Pipeline 클래스에도 자체 from_dict / to_dict 메서드가 있어서 컴포넌트와 연결을 직렬화해요.

더 알아보기 (Learn more)