파이프라인 생성

파이프라인 생성 (Creating Pipelines)

파이프라인을 만드는 일반적인 원칙을 다뤄 봐요. 이 페이지의 지침은 인덱싱 파이프라인과 쿼리 파이프라인을 만들 때 모두 그대로 적용할 수 있어요. 설명을 위해 시맨틱 문서 검색(semantic document search) 파이프라인을 예시로 사용할게요.

출처: 공식문서

컴포넌트의 입출력 이름부터 확인하기

파이프라인에 넣고 싶은 각 컴포넌트에 대해, 그 컴포넌트의 입력과 출력 이름을 알아야 해요. 이 이름들은 각 컴포넌트의 문서 페이지나 컴포넌트의 run() 메서드에서 확인할 수 있어요. 자세한 내용은 Components: Input and Output 문서를 참고하세요.

먼저 필요한 의존성들을 전부 임포트합니다. 파이프라인, Document, Document Store, 그리고 파이프라인에 쓸 모든 컴포넌트가 필요해요. 예를 들어 시맨틱 문서 검색 파이프라인을 만드려면 Document 객체, 파이프라인, Document Store, 임베더(Embedder), 그리고 Retriever가 필요하죠.

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

from haystack import Document, Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack_integrations.components.embedders.sentence_transformers import (
    SentenceTransformersTextEmbedder,
)
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever

컴포넌트 초기화하기

구성하고 싶은 파라미터를 넣어 컴포넌트들을 초기화합니다.

document_store = InMemoryDocumentStore(embedding_similarity_function="cosine")
text_embedder = SentenceTransformersTextEmbedder()
retriever = InMemoryEmbeddingRetriever(document_store=document_store)

컴포넌트를 파이프라인에 추가하기

컴포넌트를 하나씩 파이프라인에 추가해요. 이때 추가하는 순서는 중요하지 않아요.

query_pipeline.add_component("component_name", component_type)

# 위 2단계에서 초기화한 컴포넌트를 추가하는 예시:
query_pipeline.add_component("text_embedder", text_embedder)
query_pipeline.add_component("retriever", retriever)

# 미리 초기화하지 않고 바로 추가할 수도 있어요:
query_pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder())
query_pipeline.add_component(
    "retriever",
    InMemoryEmbeddingRetriever(document_store=document_store),
)

컴포넌트 연결하기

한 컴포넌트의 어떤 출력이 다음 컴포넌트의 어떤 입력에 연결될지를 정해서 컴포넌트를 연결해요. 컴포넌트에 입력이나 출력이 하나뿐이라 연결이 명확하다면, 입력·출력 이름을 지정하지 않고 컴포넌트 이름만 넘겨도 돼요.

파이프라인 실행에 어떤 입력이 필요한지 이해하려면 .inputs() 파이프라인 함수를 쓰면 됩니다. 자세한 예시는 아래 '파이프라인 입력(Pipeline Inputs)' 섹션에서 볼게요.

코드 안에서 더 직관적으로 보여드릴게요.

# 컴포넌트를 연결하는 문법이에요. component1의 output1을 component2의 input1에 연결합니다:
pipeline.connect("component1.output1", "component2.input1")

# 두 컴포넌트 모두 출력과 입력이 하나뿐이라면 이름만 넘겨도 됩니다:
pipeline.connect("component1", "component2")

# 한쪽은 출력이 하나뿐인데 다른 쪽은 입력이 여러 개라면,
# 출력이 하나인 컴포넌트는 이름만 넘기고, 입력이 여러 개인 컴포넌트는
# 어느 입력에 연결할지 명시해야 합니다

# 여기서 component1은 출력이 하나뿐이지만, component2는 입력이 여러 개입니다:
pipeline.connect("component1", "component2.input1")

# 예시로 쓰고 있는 시맨틱 문서 검색 파이프라인에서는 이렇게 되겠죠:
pipeline.connect("text_embedder.embedding", "retriever.query_embedding")
# InMemoryEmbeddingRetriever는 입력이 하나뿐이라서 이렇게 써도 됩니다:
pipeline.connect("text_embedder.embedding", "retriever")

이제 모든 컴포넌트를 둘씩 짝지어 점진적으로 연결해 봐요. 조립 중인 파이프라인의 명시적인 예시입니다.

# 이 파이프라인에 text_embedder, retriever, prompt_builder, llm 네 컴포넌트가 있다고 상상해 보세요.
# 이들을 하나의 파이프라인으로 연결하는 방법입니다:

query_pipeline.connect("text_embedder.embedding", "retriever")
query_pipeline.connect("retriever", "prompt_builder.documents")
query_pipeline.connect("prompt_builder", "llm")

파이프라인 실행하기

파이프라인이 컴포넌트와 연결을 검증할 때까지 기다렸다가, 문제가 없다면 이제 파이프라인을 실행하면 됩니다. Pipeline.run()은 두 가지 방법으로 호출할 수 있어요. 컴포넌트 이름과 입력을 담은 딕셔너리를 넘기거나, 입력만 직접 넘기는 방식이죠. 직접 넘기면 파이프라인이 입력을 알맞은 컴포넌트로 알아서 배분합니다.

# run() 메서드를 호출하는 방법 중 하나
results = pipeline.run({"component1": {"input1_value": value1, "input2_value": value2}})

# 컴포넌트 이름을 지정하지 않고 입력만 직접 넘길 수도 있어요
results = pipeline.run({"input1_value": value1, "input2_value": value2})

# 예시로 쓰는 시맨틱 문서 검색 파이프라인의 실행 방법입니다:
query = "Here comes the query text"
results = query_pipeline.run({"text_embedder": {"text": query}})

파이프라인 입력 (Pipeline Inputs)

파이프라인 실행에 어떤 컴포넌트 입력이 필요한지 이해해야 할 때, Haystack은 유용한 .inputs() 파이프라인 함수를 제공합니다. 이 함수는 컴포넌트들이 요구하는 모든 필수 입력을 나열해 줘요. 이렇게 동작합니다.

# 웹페이지를 문서로 변환하는 짧은 파이프라인 예시
from haystack import Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.fetchers import LinkContentFetcher
from haystack.components.converters import HTMLToDocument
from haystack.components.writers import DocumentWriter

document_store = InMemoryDocumentStore()
fetcher = LinkContentFetcher()
converter = HTMLToDocument()
writer = DocumentWriter(document_store=document_store)

pipeline = Pipeline()
pipeline.add_component(instance=fetcher, name="fetcher")
pipeline.add_component(instance=converter, name="converter")
pipeline.add_component(instance=writer, name="writer")

pipeline.connect("fetcher.streams", "converter.sources")
pipeline.connect("converter.documents", "writer.documents")

# 필요한 입력 목록 요청하기
pipeline.inputs()

# {'fetcher': {'urls': {'type': typing.List[str], 'is_mandatory': True}},
# 'converter': {'meta': {'type': typing.Union[typing.Dict[str, typing.Any], typing.List[typing.Dict[str, typing.Any]], NoneType],
# 'is_mandatory': False,
# 'default_value': None},
# 'extraction_kwargs': {'type': typing.Optional[typing.Dict[str, typing.Any]],
# 'is_mandatory': False,
# 'default_value': None}},
# 'writer': {'policy': {'type': typing.Optional[haystack.document_stores.types.policy.DuplicatePolicy],
# 'is_mandatory': False,
# 'default_value': None}}}

위 응답을 보면 LinkContentFetcher에는 urls 입력이 **필수(mandatory)**라는 걸 알 수 있어요. 그러면 이 파이프라인은 이렇게 실행하면 됩니다.

pipeline.run(
    data={"fetcher": {"urls": ["https://docs.haystack.deepset.ai/docs/pipelines"]}},
)

RAG 파이프라인 예시

다음 예시는 RAG 파이프라인을 만드는 과정을 안내합니다.

# import necessary dependencies
from haystack import Pipeline, Document
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.retrievers import InMemoryBM25Retriever
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.builders import ChatPromptBuilder
from haystack.utils import Secret
from haystack.dataclasses import ChatMessage

# create a document store and write documents to it
document_store = InMemoryDocumentStore()
document_store.write_documents(
    [
        Document(content="My name is Jean and I live in Paris."),
        Document(content="My name is Mark and I live in Berlin."),
        Document(content="My name is Giorgio and I live in Rome."),
    ],
)

# A prompt corresponds to an NLP task and contains instructions for the model. Here, the pipeline will go through each Document to figure out the answer.
prompt_template = [
    ChatMessage.from_system(
        """
        Given these documents, answer the question.
        Documents:
        {% for doc in documents %}
            {{ doc.content }}
        {% endfor %}
        Question:
        """,
    ),
    ChatMessage.from_user("{{question}}"),
    ChatMessage.from_system("Answer:"),
]

# create the components adding the necessary parameters
retriever = InMemoryBM25Retriever(document_store=document_store)
prompt_builder = ChatPromptBuilder(template=prompt_template, required_variables="*")
llm = OpenAIChatGenerator(
    api_key=Secret.from_env_var("OPENAI_API_KEY"),
    model="gpt-4o-mini",
)

# Create the pipeline and add the components to it. The order doesn't matter.
# At this stage, the Pipeline validates the components without running them yet.
rag_pipeline = Pipeline()
rag_pipeline.add_component("retriever", retriever)
rag_pipeline.add_component("prompt_builder", prompt_builder)
rag_pipeline.add_component("llm", llm)

# Arrange pipeline components in the order you need them. If a component has more than one inputs or outputs, indicate which input you want to connect to which output using the format ("component_name.output_name", "component_name, input_name").
rag_pipeline.connect("retriever", "prompt_builder.documents")
rag_pipeline.connect("prompt_builder", "llm")

# Run the pipeline by specifying the first component in the pipeline and passing its mandatory inputs. Optionally, you can pass inputs to other components.
question = "Who lives in paris?"
results = rag_pipeline.run(
    {
        "retriever": {"query": question},
        "prompt_builder": {"question": question},
    },
)

print(results["llm"]["replies"])

이 파이프라인을 시각화한 Mermaid 그래프로 보면 다음과 같은 모습이에요.

더 알아보기 (Learn more)