파이프라인과 비동기 실행

파이프라인과 비동기 실행 (Async Pipelines)

LLM으로 최신 검색 파이프라인을 만들려면 두 가지가 필요해요. 강력한 컴포넌트, 그리고 그것들을 쉽게 조립할 방법이죠. Haystack의 파이프라인은 바로 이 목적을 위해 만들어졌어요. 여러 LLM 호출을 동시에 실행하고, 서로 독립적인 파이프라인 분기를 병렬로 처리하며, 그렇지 않으면 대기 때문에 느려질 I/O 중심 작업을 효율적으로 처리할 수 있게요. 이 문서는 특히 비동기 실행과 스트리밍(asyncio) 에 초점을 맞춰 파이프라인을 살펴봐요.

출처: Pipelines (공식 문서)

유연성 (Flexibility)

단일 기능만 하는 파이프라인도 여전히 만들 수 있어요. 준비된 컴포넌트로 문서를 정리·분할하고 Document Store에 쓰는 인덱싱 파이프라인, 질문을 받아 답을 반환하는 쿼리 파이프라인처럼요. 하지만 Haystack은 그보다 더 나아가, 의사 결정을 통해 여러 사용 사례를 하나의 파이프라인으로 묶을 수 있게 해줘요.

에이전트형 파이프라인 (Agentic Pipelines)

Haystack의 루프와 분기는 에이전트 같은 복잡한 애플리케이션을 만들 수 있게 해줘요. 예시는 이 튜토리얼과 쿡북을 참고하세요.

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

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

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에 컴포넌트 이름 목록을 넘기세요.
소비자가 반복(iteration)을 포기하면 그에 따라 파이프라인 실행이 자동 취소돼요. 끝까지 실행되게 하려면 cancel_on_abandon=False를 넘기면 돼요.
컴포넌트에 streaming_callback을 설정해 두면(init 때든 런타임에 data로든), 청크가 handle로 푸시되는 것과 별개로 각 청크마다 그 콜백도 함께 호출돼요.
스트리밍할 때 컴포넌트는 run_async에서 동기 streaming_callback도 받아들여요. 자세한 내용은 Choosing the Right Generator 가이드를 참고하세요.

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

한 컴포넌트가 오류를 일으켰는데 형제(sibling) 컴포넌트들이 여전히 동시에 실행 중이라면, 파이프라인은 진행 중인 작업(tasks)을 취소하고 정리한 다음 원래 오류를 다시 던져요. 그래서 백그라운드에서 계속 도는 작업이 남지 않아요.

같은 정리가 다음 경우에도 적용돼요. run_async_generator를 일찍 멈출 때(예: 루프를 벗어나거나 제네레이터를 닫을 때), 또는 실행 자체가 취소될 때요.

SuperComponents

코드를 단순화하기 위해, Haystack은 완성된 파이프라인을 감싸서 하나의 컴포넌트처럼 재사용할 수 있게 해주는 SuperComponents를 소개했어요. 자세한 내용과 예시는 문서 페이지를 확인하세요.

파이프라인 만드는 단계

파이프라인 연결 단계에서 한 컴포넌트의 출력을 다음 컴포넌트의 입력에 명시적으로 연결해요. 이때 파이프라인은 컴포넌트를 실행하지 않고 연결을 검증하므로 검증이 아주 빠르게 이뤄져요.

그다음 파이프라인을 .run({"component_1": {"mandatory_inputs": value}})로 실행해요. 파이프라인의 첫 컴포넌트를 지정하고 그 필수 입력을 넘기면 되죠.

파이프라인 생성 전체 예시

작동하는 RAG 파이프라인이 모든 요소를 어떻게 조합하는지 보려면 [Creating Pipelines](https://docs.haystack.deepset.ai/docs/creating-pipelines) 문서의 전체 예시를 참고하세요.

직렬화 (Serialization)

직렬화 덕분에 파이프라인을 저장하고 다시 불러올 수 있어요. 직렬화란 Haystack 파이프라인을 디스크에 저장하거나 네트워크로 전송할 수 있는 형식으로 변환하는 것을 말해요. 특히 이런 상황에서 유용해요.

파이프라인을 편집·저장·공유할 때
파이썬이 아닌 다른 형식으로 기존 파이프라인을 수정할 때

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

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

파이프라인과 비동기 실행