S3Downloader

S3Downloader

S3Downloader는 AWS S3 버킷에서 로컬 파일시스템으로 파일을 다운로드하고, 문서에 로컬 파일 경로를 추가해 줘요.

출처: 문서

본문

개요

S3Downloader는 AWS S3 버킷에서 로컬 파일시스템으로 파일을 다운로드하고 Document 객체에 로컬 파일 경로를 추가해요. 이 컴포넌트는 S3에 저장된 파일(예: PDF, 이미지, 텍스트 파일)을 처리해야 하는 파이프라인에 유용하죠.

컴포넌트는 기본적으로 환경 변수를 통한 AWS 인증을 지원해요. AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_DEFAULT_REGION 환경 변수를 설정하면 되죠. 또는 Secret API를 사용해 초기화 시 자격 증명을 직접 전달할 수도 있어요:

from haystack.utils import Secret
from haystack_integrations.components.downloaders.s3 import S3Downloader

downloader = S3Downloader(
    aws_access_key_id=Secret.from_token("<your-access-key-id>"),
    aws_secret_access_key=Secret.from_token("<your-secret-access-key>"),
    aws_region_name=Secret.from_token("<your-region>"),
    file_root_path="/path/to/download/directory",
)

컴포넌트는 max_workers 파라미터(기본값 32개 워커)로 여러 파일을 병렬 다운로드해 대규모 문서 세트 처리를 빠르게 해요. 다운로드한 파일은 로컬에 캐시되고, 캐시가 max_cache_size(기본값 100개 파일)를 초과하면 가장 오래 접근한 파일이 자동으로 제거돼요. 이미 다운로드된 파일은 재다운로드 없이 접근 시간만 갱신하도록 터치(touch)되죠.

필수 구성: 컴포넌트는 두 가지 중요한 구성을 요구해요.

• file_root_path 파라미터 또는 FILE_ROOT_PATH 환경 변수: 파일이 다운로드될 위치를 지정해요. 디렉토리가 없으면 생성해요.
• S3_DOWNLOADER_BUCKET 환경 변수: 어느 S3 버킷에서 파일을 다운로드할지 지정해요.

선택적 환경 변수 S3_DOWNLOADER_PREFIX는 생성되는 모든 S3 키에 파일의 접두사를 추가하도록 설정할 수 있어요.

파일 확장자 필터링

file_extensions 파라미터를 사용해 특정 파일 타입만 다운로드할 수 있어요. 불필요한 다운로드와 처리 시간을 줄여 주죠. 예를 들어 file_extensions=[".pdf", ".txt"]는 PDF와 TXT 파일만 다운로드하고 나머지는 건너뛰어요.

커스텀 S3 키 생성

기본적으로 컴포넌트는 Document 메타데이터의 file_name을 S3 키로 사용해요. S3 파일 구조가 메타데이터의 파일 이름과 일치하지 않으면, 선택적 s3_key_generation_function을 제공해 Document 메타데이터에서 S3 키를 생성하는 방식을 커스터마이즈할 수 있어요.

사용법

S3Downloader를 사용하려면 amazon-bedrock-haystack 패키지를 설치해야 해요:

pip install amazon-bedrock-haystack

단독으로 사용하기

예제를 실행하기 전에 필요한 환경 변수가 설정되어 있는지 확인하세요:

export AWS_ACCESS_KEY_ID="<your-access-key-id>"
export AWS_SECRET_ACCESS_KEY="<your-secret-access-key>"
export AWS_DEFAULT_REGION="<your-region>"
export S3_DOWNLOADER_BUCKET="<your-bucket-name>"

S3Downloader로 S3에서 파일을 다운로드하는 방법이에요:

from haystack.dataclasses import Document
from haystack_integrations.components.downloaders.s3 import S3Downloader

# Create documents with file names in metadata
documents = [
    Document(meta={"file_name": "report.pdf"}),
    Document(meta={"file_name": "data.txt"}),
]

# Initialize the downloader
downloader = S3Downloader(file_root_path="/tmp/s3_downloads")

# Download the files
result = downloader.run(documents=documents)

# Access the downloaded files
for doc in result["documents"]:
    print(f"File downloaded to: {doc.meta['file_path']}")

파일 확장자 필터링과 함께:

from haystack.dataclasses import Document
from haystack_integrations.components.downloaders.s3 import S3Downloader

documents = [
    Document(meta={"file_name": "report.pdf"}),
    Document(meta={"file_name": "image.png"}),
    Document(meta={"file_name": "data.txt"}),
]

# Only download PDF files
downloader = S3Downloader(file_root_path="/tmp/s3_downloads", file_extensions=[".pdf"])
result = downloader.run(documents=documents)

# Only report.pdf is downloaded
print(f"Downloaded {len(result['documents'])} file(s)")
# >> Downloaded 1 file(s)

커스텀 S3 키 생성과 함께:

from haystack.dataclasses import Document
from haystack_integrations.components.downloaders.s3 import S3Downloader

def custom_s3_key_function(document: Document) -> str:
    """Generate S3 key from custom metadata."""
    folder = document.meta.get("folder", "default")
    file_name = document.meta.get("file_name")
    if not file_name:
        raise ValueError("Document must have 'file_name' in metadata")
    return f"{folder}/{file_name}"

documents = [
    Document(meta={"file_name": "report.pdf", "folder": "reports/2025"}),
]

downloader = S3Downloader(
    file_root_path="/tmp/s3_downloads",
    s3_key_generation_function=custom_s3_key_function,
)
result = downloader.run(documents=documents)

파이프라인에서 사용하기

문서 처리 파이프라인에서 S3Downloader를 사용하는 예시예요:

from haystack import Pipeline
from haystack.components.converters import PDFMinerToDocument
from haystack.components.routers import DocumentTypeRouter
from haystack.dataclasses import Document
from haystack_integrations.components.downloaders.s3 import S3Downloader

# Create a pipeline
pipe = Pipeline()

# Add S3Downloader to download files from S3
pipe.add_component(
    "downloader",
    S3Downloader(file_root_path="/tmp/s3_downloads", file_extensions=[".pdf", ".txt"]),
)

# Route documents by file type
pipe.add_component(
    "router",
    DocumentTypeRouter(
        file_path_meta_field="file_path",
        mime_types=["application/pdf", "text/plain"],
    ),
)

# Convert PDFs to documents
pipe.add_component("pdf_converter", PDFMinerToDocument())

# Connect components
pipe.connect("downloader.documents", "router.documents")
pipe.connect("router.application/pdf", "pdf_converter.documents")

# Create documents with S3 file names
documents = [
    Document(meta={"file_name": "report.pdf"}),
    Document(meta={"file_name": "summary.txt"}),
]

# Run the pipeline
result = pipe.run({"downloader": {"documents": documents}})

이미지 처리와 LLM을 포함한 더 복잡한 예시예요:

from haystack import Pipeline
from haystack.components.builders import ChatPromptBuilder
from haystack.components.converters.image import DocumentToImageContent
from haystack.components.routers import DocumentTypeRouter
from haystack.dataclasses import Document
from haystack_integrations.components.downloaders.s3 import S3Downloader
from haystack_integrations.components.generators.amazon_bedrock import (
    AmazonBedrockChatGenerator,
)

# Create documents with file names
documents = [
    Document(meta={"file_name": "chart.png"}),
    Document(meta={"file_name": "report.pdf"}),
]

# Create pipeline
pipe = Pipeline()

# Download files from S3
pipe.add_component("downloader", S3Downloader(file_root_path="/tmp/s3_downloads"))

# Route by document type
pipe.add_component(
    "router",
    DocumentTypeRouter(
        file_path_meta_field="file_path",
        mime_types=["image/png", "application/pdf"],
    ),
)

# Convert images for LLM
pipe.add_component("image_converter", DocumentToImageContent(detail="auto"))

# Create chat prompt with template
template = """{% message role="user" %}Answer the question based on the provided images.
Question: {{ question }}
{% for image in image_contents %}{{ image | templatize_part }}{% endfor %}{% endmessage %}"""
pipe.add_component("prompt_builder", ChatPromptBuilder(template=template))

# Generate response
pipe.add_component(
    "llm",
    AmazonBedrockChatGenerator(model="anthropic.claude-3-haiku-20240307-v1:0"),
)

# Connect components
pipe.connect("downloader.documents", "router.documents")
pipe.connect("router.image/png", "image_converter.documents")
pipe.connect("image_converter.image_contents", "prompt_builder.image_contents")
pipe.connect("prompt_builder.prompt", "llm.messages")

# Run pipeline
result = pipe.run(
    {
        "downloader": {"documents": documents},
        "prompt_builder": {"question": "What information is shown in the chart?"},
    },
)

더 알아보기 (Learn more)

• API reference: S3Downloader
• GitHub: amazon_bedrock 통합