Pathway Retriever

Pathway는 실시간 데이터 소스와 변화하는 데이터를 다루는 오픈소스 데이터 처리 프레임워크예요. 이 노트북에서는 Pathway의 실시간 인덱싱 파이프라인을 LlamaIndex와 함께 사용하는 방법을 보여드릴게요.

출처: 문서

본문

Pathway는 오픈 데이터 처리 프레임워크로, 실시간 데이터 소스와 변화하는 데이터를 다루는 데이터 변환 파이프라인과 Machine Learning 애플리케이션을 쉽게 개발할 수 있게 해줍니다.

이 노트북은 LlamaIndex와 함께 실시간 데이터 인덱싱 파이프라인을 사용하는 방법을 보여줍니다. 제공되는 PathwayRetriever를 이용해 LLM 애플리케이션에서 이 파이프라인의 결과를 질의할 수 있습니다. 내부적으로 Pathway는 데이터가 변경될 때마다 인덱스를 갱신해 항상 최신 답변을 제공합니다.

이 노트북에서 사용할 공개 데모 문서 처리 파이프라인은 다음을 수행합니다:

여러 클라우드 데이터 소스의 데이터 변경을 모니터링합니다.
데이터에 대한 벡터 인덱스를 구축합니다.

자체 문서 처리 파이프라인을 만들려면 호스팅 제공을 확인하거나 이 노트북을 따라 직접 구축하세요.

retrieve 인터페이스를 구현하는 llama_index.retrievers.pathway.PathwayRetriever 리트리버를 사용해 인덱스에 연결합니다.

이 문서에서 설명하는 기본 파이프라인을 이용하면 클라우드 위치에 저장된 파일의 간단한 인덱스를 손쉽게 구축할 수 있습니다. 하지만 Pathway는 서로 다른 데이터 소스 간 groupby-reduction과 join 같은 SQL 유사 연산, 시간 기반 그룹화·윈도잉, 그리고 다양한 커넥터를 포함해 실시간 데이터 파이프라인과 애플리케이션을 구축하는 데 필요한 모든 것을 제공합니다.

Pathway 데이터 수집 파이프라인과 벡터 스토어에 대한 자세한 내용은 vector store pipeline을 참고하세요.

사전 요구 사항 (Prerequisites)

PathwayRetrievier를 사용하려면 llama-index-retrievers-pathway 패키지를 설치해야 합니다.

!pip install llama-index-retrievers-pathway

llama-index용 Retriever 생성

PathwayRetriever를 인스턴스화하고 구성하려면 문서 인덱싱 파이프라인의 url 또는 host와 port를 제공해야 합니다. 아래 코드에서는 공개된 데모 파이프라인을 사용하는데, 그 REST API는 https://demo-document-indexing.pathway.stream에서 접근할 수 있습니다. 이 데모는 Google Drive와 Sharepoint에서 문서를 수집하고 문서 검색용 인덱스를 유지합니다.

from llama_index.retrievers.pathway import PathwayRetriever


retriever = PathwayRetriever(
    url="https://demo-document-indexing.pathway.stream"
)
retriever.retrieve(str_or_query_bundle="what is pathway")

여러분 차례입니다! 파이프라인을 받아 보거나 새 문서를 업로드한 뒤 질의를 다시 실행해 보세요!

Query Engine에서 사용하기

from llama_index.core.query_engine import RetrieverQueryEngine


query_engine = RetrieverQueryEngine.from_args(
    retriever,
)

response = query_engine.query("Tell me about Pathway")
print(str(response))

자체 데이터 처리 파이프라인 구축하기

사전 요구 사항

pathway 패키지를 설치한 뒤 샘플 데이터를 다운로드합니다.

%pip install pathway
%pip install llama-index-embeddings-openai

!mkdir -p 'data/'
!wget 'https://gist.githubusercontent.com/janchorowski/dd22a293f3d99d1b726eedc7d46d2fc0/raw/pathway_readme.md' -O 'data/pathway_readme.md'

Pathway가 추적하는 데이터 소스 정의

Pathway는 로컬 파일, S3 폴더, 클라우드 스토리지, 그리고 모든 데이터 스트림 등 많은 소스를 동시에 수신하며 데이터 변경을 감지할 수 있습니다. 자세한 내용은 pathway-io를 참고하세요.

import pathway as pw


data_sources = []
data_sources.append(
    pw.io.fs.read(
        "./data",
        format="binary",
        mode="streaming",
        with_metadata=True,
    )  # This creates a `pathway` connector that tracks
    # all the files in the ./data directory
)


# This creates a connector that tracks files in Google drive.
# please follow the instructions at https://pathway.com/developers/tutorials/connectors/gdrive-connector/ to get credentials
# data_sources.append(
#     pw.io.gdrive.read(object_id="17H4YpBOAKQzEJ93xmC2z170l0bP2npMy", service_user_credentials_file="credentials.json", with_metadata=True))

문서 인덱싱 파이프라인 생성

문서 인덱싱 파이프라인을 만들어 봅시다. transformations는 Embedding 변환으로 끝나는 TransformComponent들의 리스트여야 합니다.

이 예시에서는 먼저 TokenTextSplitter로 텍스트를 분리한 뒤 OpenAIEmbedding으로 임베딩을 수행합니다.

from pathway.xpacks.llm.vector_store import VectorStoreServer
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import TokenTextSplitter


embed_model = OpenAIEmbedding(embed_batch_size=10)


transformations_example = [
    TokenTextSplitter(
        chunk_size=150,
        chunk_overlap=10,
        separator=" ",
    ),
    embed_model,
]


processing_pipeline = VectorStoreServer.from_llamaindex_components(
    *data_sources,
    transformations=transformations_example,
)


# Define the Host and port that Pathway will be on
PATHWAY_HOST = "127.0.0.1"
PATHWAY_PORT = 8754


# `threaded` runs pathway in detached mode, we have to set it to False when running from terminal or container
# for more information on `with_cache` check out https://pathway.com/developers/api-docs/persistence-api
processing_pipeline.run_server(
    host=PATHWAY_HOST, port=PATHWAY_PORT, with_cache=False, threaded=True
)

커스텀 파이프라인에 리트리버 연결

from llama_index.retrievers.pathway import PathwayRetriever


retriever = PathwayRetriever(host=PATHWAY_HOST, port=PATHWAY_PORT)
retriever.retrieve(str_or_query_bundle="what is pathway")

Pathway Retriever