인제스천 파이프라인으로 데이터 흐름 만들기

인제스천 파이프라인으로 데이터 흐름 만들기 (Ingestion Pipeline)

문서를 받아서 잘게 쪼개고, 메타데이터를 뽑고, 임베딩까지 한 번에 처리하고 싶다면 IngestionPipeline이 그 전 과정을 하나로 묶어줘요. 이 파이프라인은 입력 데이터에 적용되는 변환(Transformations) 이라는 개념을 사용해요. 변환이 적용된 결과 노드는 반환되거나 (벡터 DB가 주어졌다면) 벡터 DB에 삽입돼요. 그리고 각 (노드, 변환) 쌍은 캐시되기 때문에, 같은 조합을 다시 실행하면 (캐시가 유지된다면) 캐시 결과를 재사용해 시간을 아낄 수 있어요.

출처: 공식문서

사용 패턴

가장 단순한 사용법은 IngestionPipeline을 이렇게 인스턴스화하는 거예요.

from llama_index.core import Document
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor
from llama_index.core.ingestion import IngestionPipeline, IngestionCache


# 변환을 담아 파이프라인 생성
pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        TitleExtractor(),
        OpenAIEmbedding(),
    ]
)


# 파이프라인 실행
nodes = pipeline.run(documents=[Document.example()])

실무에서는 문서를 SimpleDirectoryReader나 Llama Hub의 다른 리더로부터 가져오게 될 거예요.

벡터 DB 연결하기

인제스천 파이프라인을 실행할 때 결과 노드를 원격 벡터 스토어에 자동 삽입하도록 선택할 수도 있어요. 그러면 나중에 그 벡터 스토어로부터 인덱스를 구성할 수 있어요.

from llama_index.core import Document
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor
from llama_index.core.ingestion import IngestionPipeline
from llama_index.vector_stores.qdrant import QdrantVectorStore


import qdrant_client


client = qdrant_client.QdrantClient(location=":memory:")
vector_store = QdrantVectorStore(client=client, collection_name="test_store")


pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        TitleExtractor(),
        OpenAIEmbedding(),
    ],
    vector_store=vector_store,
)


# 벡터 DB에 직접 인제스천
pipeline.run(documents=[Document.example()])


# 인덱스 생성
from llama_index.core import VectorStoreIndex


index = VectorStoreIndex.from_vector_store(vector_store)

파이프라인 내 임베딩 계산

위 예시에서 임베딩은 파이프라인의 일부로 계산돼요. 파이프라인을 벡터 스토어에 연결한다면 임베딩은 반드시 파이프라인의 한 단계여야 해요. 그렇지 않으면 이후 인덱스를 인스턴스화할 때 실패하게 돼요. 벡터 스토어에 연결하지 않는다면, 즉 단지 노드 목록만 만들 거라면 파이프라인에서 임베딩을 생략해도 돼요.

캐싱

IngestionPipeline에서는 각 (노드, 변환) 조합이 해시되어 캐시돼요. 같은 데이터를 쓰는 후속 실행에서 시간을 아껴주죠.

로컬 캐시 관리

파이프라인을 만들고 난 뒤 캐시를 저장·로드하고 싶을 수 있어요.

# 저장
pipeline.persist("./pipeline_storage")


# 로드 및 상태 복원
new_pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        TitleExtractor(),
    ],
)
new_pipeline.load("./pipeline_storage")


# 캐시 덕분에 즉시 실행됨
nodes = pipeline.run(documents=[Document.example()])

캐시가 너무 커지면 비울 수 있어요.

# 캐시의 모든 컨텍스트 삭제
cache.clear()

원격 캐시 관리

캐시를 위한 여러 원격 스토리지 백엔드를 지원해요.

RedisCache
MongoDBCache
FirestoreCache

아래는 RedisCache를 쓰는 예시예요.

from llama_index.core import Document
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor
from llama_index.core.ingestion import IngestionPipeline, IngestionCache
from llama_index.storage.kvstore.redis import RedisKVStore as RedisCache




ingest_cache = IngestionCache(
    cache=RedisCache.from_host_and_port(host="127.0.0.1", port=6379),
    collection="my_test_cache",
)


pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        TitleExtractor(),
        OpenAIEmbedding(),
    ],
    cache=ingest_cache,
)


# 벡터 DB에 직접 인제스천
nodes = pipeline.run(documents=[Document.example()])

지정한 원격 컬렉션에 저장하면서 진행되므로 별도의 persist 단계가 필요 없어요.

비동기 지원

IngestionPipeline은 비동기 동작도 지원해요.

nodes = await pipeline.arun(documents=documents)

문서 관리 (Document Management)

파이프라인에 docstore를 붙이면 문서 관리가 활성화돼요. document.doc_id나 node.ref_doc_id를 기준점으로 삼아 파이프라인이 중복 문서를 적극적으로 찾아내요. 동작 방식은 다음과 같아요.

doc_id → document_hash 매핑을 저장
벡터 스토어가 붙어 있다면:
- 중복 doc_id가 감지되고 해시가 바뀌었다면 → 문서를 다시 처리하고 upsert
- 중복 doc_id가 감지됐지만 해시가 같다면 → 노드를 건너뜀
벡터 스토어만 붙어 있지 않다면 (즉 단독으로는):
- 각 노드의 기존 해시를 모두 확인
- 중복이 발견되면 노드를 건너뜀
- 그렇지 않으면 노드를 처리

참고: 벡터 스토어를 붙이지 않으면 중복 입력을 확인하고 제거하는 것만 가능해요.

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.storage.docstore import SimpleDocumentStore


pipeline = IngestionPipeline(
    transformations=[...], docstore=SimpleDocumentStore()
)

전체 워크스루는 데모 노트북에서, Redis 전체를 인제스천 스택으로 쓰는 가이드도 확인해보세요.

병렬 처리

IngestionPipeline의 run 메서드는 병렬 프로세스로 실행될 수 있어요. multiprocessing.Pool을 사용해 노드 배치를 여러 프로세서에 분산시키는 방식이에요. num_workers를 원하는 프로세스 수로 설정하면 돼요.

from llama_index.core.ingestion import IngestionPipeline


pipeline = IngestionPipeline(
    transformations=[...],
)
pipeline.run(documents=[...], num_workers=4)

인제스천 파이프라인으로 데이터 흐름 만들기