파이프라인

파이프라인 (Pipelines)

LLM으로 현대적인 검색 파이프라인을 만들려면 두 가지가 필요해요. 강력한 컴포넌트와, 그 컴포넌트를 쉽게 조립할 수 있는 방법이죠. Haystack 파이프라인은 바로 이 목적을 위해 만들어졌고, LLM과의 상호작용을 설계하고 확장할 수 있게 해줘요.

Haystack의 파이프라인은 다양한 Haystack 컴포넌트와 통합으로 이루어진 유향 멀티그래프(directed multigraph)예요. 컴포넌트를 다양한 방식으로 연결할 자유를 줘요. 다시 말해 파이프라인이 연속된 정보의 흐름일 필요가 없어요. 유연한 Haystack 파이프라인으로 동시 실행 흐름(simultaneous flows), 단독 컴포넌트(standalone components), 루프(loops), 그리고 다른 유형의 연결까지 만들 수 있어요.

출처: 공식문서

유연성 (Flexibility)

Haystack 파이프라인은 단순한 쿼리·인덱싱 파이프라인 그 이상이에요. 파이프라인이 하는 일(인덱싱, 쿼리, API에서 가져오기, 전처리 등)은 전적으로 파이프라인을 어떻게 설계하고 어떤 컴포넌트를 쓰느냐에 달려 있어요. 문서를 정리하고 쪼갠 뒤 Document Store에 쓰는 인덱싱 파이프라인 같은 단일 기능 파이프라인을 만들 수도 있고, 쿼리를 받아 답을 반환하는 쿼리 파이프라인도 만들 수 있어요. 하지만 Haystack은 ConditionalRouter 같은 의사결정 컴포넌트와 함께 여러 유스케이스를 하나의 파이프라인으로 결합하는 것도 허용해요.

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

Haystack의 루프와 분기(branch)는 에이전트 같은 복잡한 애플리케이션을 만들 수 있게 해줘요. 몇 가지 예시를 들어 볼게요:

분기 (Branching)

파이프라인은 데이터를 동시에 처리하는 여러 분기를 가질 수 있어요. 예를 들어 서로 다른 파일 유형을 처리하려면, 각 파일 유형을 담당하는 변환기(converter)들을 잔뜩 가진 파이프라인을 만들 수 있어요. 그러면 모든 파일을 파이프라인에 넣기만 하면, 파이프라인이 파일을 적절한 변환기로 한 번에 나누고 라우팅해줘요. 파일을 하나씩 보내는 수고를 덜어주죠.

루프 (Loops)

파이프라인의 컴포넌트는 원하는 횟수로 상한을 둔 반복 루프에서 작업할 수 있어요. 이는 자가 수정 루프(self-correcting loops) 같은 시나리오에 유용해요. 생성기가 어떤 출력을 만들고, 검증기(validator) 컴포넌트가 출력이 올바른지 확인하는 구조죠. 생성기의 출력에 오류가 있으면 검증기 컴포넌트가 생성기로 돌아가 수정된 출력을 요청해요. 출력이 검증을 통과해 파이프라인 아래로 보내질 수 있을 때까지 루프가 계속돼요.

루프가 어떻게 실행·종료되고 안전하게 쓰는지 자세한 설명은 Pipeline Loops를 참고하세요.

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

비동기로 실행하면 파이프라인은 의존성이 허용할 때 컴포넌트를 병렬로 실행해요. 이는 독립적인 작업이 있는 복잡한 파이프라인에서 성능을 향상시켜요. 예를 들어 여러 Retriever나 LLM 호출을 동시에 실행하고, 독립적인 파이프라인 분기를 병렬로 실행하며, 그렇지 않으면 지연을 일으킬 I/O 중심 작업을 효율적으로 처리할 수 있어요. 동시에 실행되는 컴포넌트 수는 비동기 실행 메서드(run_async, run_async_generator, stream)의 concurrency_limit 인자로 제한할 수 있어요. 동기 run 메서드는 컴포넌트를 순차적으로 실행해요.

차단형 run 메서드 외에도 모든 파이프라인은 세 가지 비동기 실행 방식을 제공해요:

  • run_async: 파이프라인을 단일 논블로킹 호출로 실행해요. 파이프라인을 더 큰 비동기 애플리케이션이나 서비스에 통합하기에 이상적이에요.
  • run_async_generator: 컴포넌트가 작업을 마치면 부분 출력을 산출해요. 진행 상황 모니터링, 디버깅, 출력 점진 처리에 유용해요.
  • 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를 전달하세요.

컴포넌트에 (초기화 시나 data로 런타임에) streaming_callback이 설정되어 있으면, 핸들로 푸시되는 청크 외에도 각 청크마다 그 콜백이 여전히 호출돼요. 스트리밍 중에 컴포넌트는 run_async에서도 동기 streaming_callback을 받아들여요. 자세한 내용은 Choosing the Right Generator 가이드를 참고하세요.

오류 처리와 작업 취소 (Error Handling and Task Cancellation)

컴포넌트가 오류를 일으키는 동안 형제 컴포넌트가 여전히 동시에 실행 중이라면, 파이프라인은 원래 오류를 다시 던지기 전에 진행 중인 작업을 취소하고 정리(drain)해요. 그래서 백그라운드에서 계속 실행되는 작업이 없어요. run_async_generator 순회를 일찍 멈출 때(예: 루프에서 벗어나 생성기를 닫을 때)나 실행 자체가 취소될 때도 동일한 정리가 적용돼요.

취소는 네이티브 비동기로 실행되는 컴포넌트만 중단시켜요. 동기 컴포넌트는 작업자 스레드로 오프로드되고, 이는 중단할 수 없어 백그라운드에서 끝까지 실행돼요. 그 출력은 버려져서 파이프라인 상태는 일관되게 유지되지만, 컴포넌트의 부작용은 여전히 완료돼요.

수퍼컴포넌트 (SuperComponents)

코드를 단순화하기 위해 SuperComponents가 도입됐어요. 이들은 완전한 파이프라인을 감싸 단일 컴포넌트처럼 재사용할 수 있게 해줘요. 자세한 내용과 예시는 SuperComponents 문서 페이지를 확인하세요.

데이터 흐름 (Data Flow)

데이터(초기 쿼리)가 전체 파이프라인을 통해 흐르는 동안, 개별 값은 연결될 때만 한 컴포넌트에서 다른 컴포넌트로 전달돼요. 따라서 모든 컴포넌트가 모든 데이터에 접근할 수 있는 건 아니에요. 이 접근 방식은 속도와 디버깅 용이성이라는 이점을 제공해요.

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

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

파이프라인은 컴포넌트 연결을 단순화하는 더 똑똑한 연결 시맨틱을 지원해요.

호환되는 출력은 단일 입력에 연결될 때 암시적으로 결합될 수 있어요. 파이프라인은 또한 일부 선택된 타입에 대해 연결 시점에 암시적 타입 적응을 수행해요.

이런 동작들은 Joiners와 OutputAdapters 같은 접착 컴포넌트의 필요성을 줄여, 파이프라인을 간결하고 읽기 쉽게 유지해요.

자세한 내용과 예시는 Smart Pipeline Connections에서 볼 수 있어요.

파이프라인 만드는 단계

모든 컴포넌트를 만들고 파이프라인으로 결합할 준비가 되면, 동작시키기까지 네 단계가 있어요:

  1. Pipeline()으로 파이프라인을 만든다. Pipeline 객체가 생성돼요.
  2. .add_components({name: component})로 컴포넌트를 파이프라인에 추가하거나, .add_component(name, component)로 개별 추가한다. 이 단계에서는 컴포넌트를 연결하지 않고 추가만 해요. 특히 루프에 유용한데, 다음 단계에서 모든 컴포넌트가 이미 파이프라인에 존재하므로 매끄럽게 연결할 수 있어요.
  3. .connect_many([(sender, receiver)])로 여러 컴포넌트 쌍을 연결하거나 .connect("producer_component.output_name", "consumer_component.input_name")으로 한 쌍을 연결한다. 이 단계에서 컴포넌트의 출력 하나를 다음 컴포넌트의 입력 하나에 명시적으로 연결해요. 이때 컴포넌트를 실행하지 않고 연결을 검증하므로 검증이 빠르다는 장점이 있어요.
  4. .run({"component_1": {"mandatory_inputs": value}})로 파이프라인을 실행한다. 마지막으로 파이프라인의 첫 번째 컴포넌트를 지정하고 그 필수 입력을 전달해 파이프라인을 실행해요. 선택적으로 다른 컴포넌트에도 입력을 전달할 수 있어요. 예: .run({"component_1": {"mandatory_inputs": value}, "component_2": {"inputs": value}}).

Creating Pipelines의 전체 파이프라인 예시는 모든 요소가 모여 동작하는 RAG 파이프라인을 만드는 방법을 보여줘요.

파이프라인을 만들고 나면 그래프로 시각화해서 컴포넌트가 어떻게 연결됐는지, 원하는 대로 됐는지 확인할 수 있어요. Mermaid 그래프를 사용할 수 있어요.

검증 (Validation)

검증은 파이프라인에서 컴포넌트를 .connect()로 연결할 때, 컴포넌트를 실행하기 전에 수행되어 더 빠르게 만드는 방식이에요. 파이프라인은 다음을 검증해요:

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

이 모든 검사는 문제를 빨리 고치도록 상세한 오류를 생성해요.

직렬화 (Serialization)

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

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

Haystack 파이프라인은 직렬화를 컴포넌트에 위임해요. 따라서 파이프라인을 직렬화하는 것은 각 컴포넌트를 연결과 함께 하나씩 직렬화하는 것을 의미해요. 파이프라인은 중간 형식으로 작용하는 딕셔너리 형식으로 직렬화되고, 이를 원하는 최종 형식으로 변환할 수 있어요.

직렬화 형식: 현재 Haystack은 YAML 형식만 지원해요. 다른 형식은 점진적으로 출시될 예정이에요.

직렬화가 가능하려면 컴포넌트가 파이썬 딕셔너리로/에서 변환을 지원해야 해요. 모든 Haystack 컴포넌트에는 from_dict와 to_dict라는 두 가지 직렬화 메서드가 있어요. Pipeline 클래스는 컴포넌트와 연결을 직렬화하는 자체 from_dict와 to_dict 메서드를 가져요.

더 알아보기 (Learn more)