6장: Haystack -- 모듈러 파이프라인 아키텍처

from haystack import component
 
@component
class TextCleaner:
    """텍스트 전처리 컴포넌트"""
 
    @component.output_types(cleaned_text=str)
    def run(self, text: str) -> dict:
        cleaned = text.strip().lower()
        cleaned = " ".join(cleaned.split())  # 중복 공백 제거
        return {"cleaned_text": cleaned}

from haystack import Pipeline
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder
 
# 컴포넌트 생성
prompt_builder = PromptBuilder(
    template="""
다음 컨텍스트를 바탕으로 질문에 답변해주세요.
 
컨텍스트:
{% for doc in documents %}
{{ doc.content }}
{% endfor %}
 
질문: {{ question }}
답변:
"""
)
generator = OpenAIGenerator(model="gpt-4o")
 
# 파이프라인 구성
pipeline = Pipeline()
pipeline.add_component("prompt_builder", prompt_builder)
pipeline.add_component("generator", generator)
 
# 연결: prompt_builder의 출력 -> generator의 입력
pipeline.connect("prompt_builder", "generator")
 
# 실행
result = pipeline.run(
    {
        "prompt_builder": {
            "documents": documents,
            "question": "Haystack의 핵심 개념은?",
        }
    }
)
print(result["generator"]["replies"][0])

from haystack import Pipeline
from haystack.components.retrievers.in_memory import (
    InMemoryBM25Retriever,
)
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder
from haystack.document_stores.in_memory import InMemoryDocumentStore
 
# 문서 스토어 설정
document_store = InMemoryDocumentStore()
 
# 파이프라인 구성
rag_pipeline = Pipeline()
rag_pipeline.add_component(
    "retriever",
    InMemoryBM25Retriever(document_store=document_store),
)
rag_pipeline.add_component(
    "prompt_builder",
    PromptBuilder(template=rag_template),
)
rag_pipeline.add_component(
    "generator",
    OpenAIGenerator(model="gpt-4o"),
)
 
# 연결
rag_pipeline.connect("retriever.documents", "prompt_builder.documents")
rag_pipeline.connect("prompt_builder", "generator")
 
# 실행
result = rag_pipeline.run(
    {
        "retriever": {"query": "Haystack 파이프라인이란?"},
        "prompt_builder": {"question": "Haystack 파이프라인이란?"},
    }
)

from haystack import AsyncPipeline
 
# AsyncPipeline으로 전환
async_pipeline = AsyncPipeline()
async_pipeline.add_component("bm25_retriever", bm25_retriever)
async_pipeline.add_component("embedding_retriever", embedding_retriever)
async_pipeline.add_component("joiner", DocumentJoiner())
async_pipeline.add_component("prompt_builder", prompt_builder)
async_pipeline.add_component("generator", generator)
 
# bm25와 embedding 검색은 독립적 -> 병렬 실행
async_pipeline.connect("bm25_retriever.documents", "joiner.documents")
async_pipeline.connect("embedding_retriever.documents", "joiner.documents")
async_pipeline.connect("joiner.documents", "prompt_builder.documents")
async_pipeline.connect("prompt_builder", "generator")
 
# 비동기 실행
result = await async_pipeline.run_async(
    {
        "bm25_retriever": {"query": "검색어"},
        "embedding_retriever": {"query": "검색어"},
        "prompt_builder": {"question": "검색어"},
    }
)

from haystack.components.routers import ConditionalRouter
 
routes = [
    {
        "condition": "{{ query|length > 100 }}",
        "output": "{{ query }}",
        "output_name": "long_query",
        "output_type": str,
    },
    {
        "condition": "{{ query|length <= 100 }}",
        "output": "{{ query }}",
        "output_name": "short_query",
        "output_type": str,
    },
]
 
router = ConditionalRouter(routes=routes)
 
pipeline = Pipeline()
pipeline.add_component("router", router)
pipeline.add_component("detailed_retriever", detailed_retriever)
pipeline.add_component("simple_retriever", simple_retriever)
 
pipeline.connect("router.long_query", "detailed_retriever.query")
pipeline.connect("router.short_query", "simple_retriever.query")

@component
class QualityChecker:
    """응답 품질을 검증하는 컴포넌트"""
 
    @component.output_types(
        approved=str,
        rejected=str,
    )
    def run(self, reply: str) -> dict:
        # 품질 검증 로직
        if self.is_quality_sufficient(reply):
            return {"approved": reply}
        return {"rejected": reply}
 
    def is_quality_sufficient(self, reply: str) -> bool:
        return len(reply) > 100 and "?" not in reply
 
pipeline = Pipeline()
pipeline.add_component("generator", generator)
pipeline.add_component("checker", QualityChecker())
pipeline.add_component("retry_prompt", retry_prompt_builder)
 
pipeline.connect("generator.replies", "checker.reply")
pipeline.connect("checker.rejected", "retry_prompt.previous_reply")
pipeline.connect("retry_prompt", "generator")  # 루프
# checker.approved는 파이프라인의 최종 출력

from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack import Document
 
# 인메모리 스토어 (개발/테스트용)
store = InMemoryDocumentStore()
 
# 문서 추가
documents = [
    Document(
        content="Haystack은 오픈소스 AI 프레임워크입니다.",
        meta={"source": "docs", "language": "ko"},
    ),
    Document(
        content="파이프라인은 컴포넌트의 조합입니다.",
        meta={"source": "tutorial", "language": "ko"},
    ),
]
store.write_documents(documents)
 
# 필터링 검색
from haystack.document_stores.types import FilterPolicy
 
results = store.filter_documents(
    filters={
        "operator": "AND",
        "conditions": [
            {"field": "meta.source", "operator": "==", "value": "docs"},
            {"field": "meta.language", "operator": "==", "value": "ko"},
        ],
    }
)

# Pinecone
from haystack_integrations.document_stores.pinecone import (
    PineconeDocumentStore,
)
 
pinecone_store = PineconeDocumentStore(
    index="my-index",
    namespace="production",
)
 
# Weaviate
from haystack_integrations.document_stores.weaviate import (
    WeaviateDocumentStore,
)
 
weaviate_store = WeaviateDocumentStore(url="http://localhost:8080")
 
# Qdrant
from haystack_integrations.document_stores.qdrant import (
    QdrantDocumentStore,
)
 
qdrant_store = QdrantDocumentStore(
    url="http://localhost:6333",
    index="my-collection",
)

from haystack import Pipeline
from haystack.components.converters import PyPDFToDocument
from haystack.components.preprocessors import (
    DocumentCleaner,
    DocumentSplitter,
)
from haystack.components.embedders import (
    OpenAIDocumentEmbedder,
)
from haystack.components.writers import DocumentWriter
 
# 인제스천 파이프라인
ingestion = Pipeline()
ingestion.add_component("converter", PyPDFToDocument())
ingestion.add_component("cleaner", DocumentCleaner())
ingestion.add_component(
    "splitter",
    DocumentSplitter(split_by="sentence", split_length=5),
)
ingestion.add_component(
    "embedder",
    OpenAIDocumentEmbedder(model="text-embedding-3-small"),
)
ingestion.add_component(
    "writer",
    DocumentWriter(document_store=document_store),
)
 
ingestion.connect("converter", "cleaner")
ingestion.connect("cleaner", "splitter")
ingestion.connect("splitter", "embedder")
ingestion.connect("embedder", "writer")
 
# PDF 파일 처리
ingestion.run(
    {"converter": {"sources": ["./report.pdf", "./guide.pdf"]}}
)

# 파이프라인을 YAML로 저장
with open("rag_pipeline.yaml", "w") as f:
    pipeline.dump(f)
 
# YAML에서 파이프라인 로드
with open("rag_pipeline.yaml", "r") as f:
    loaded_pipeline = Pipeline.load(f)
 
result = loaded_pipeline.run(...)