LangChain document loaders for Google Drive and S3

LangChain document loaders for Google Drive and S3 let you pull unstructured data directly into your RAG pipeline without writing custom extraction logic. This guide walks through the complete setup for both sources, shows how to combine them with chunking strategies, and includes verification steps so you know the pipeline works before you hit production.

Step 1: Install dependencies and configure credentials

Start with a clean virtual environment. You need the core LangChain packages plus the community integrations for Google Drive and S3.

python -m venv .venv
source .venv/bin/activate
pip install langchain langchain-community langchain-text-splitters \
    google-api-python-client google-auth-httplib2 google-auth-oauthlib \
    boto3 pypdf python-docx

Google Drive credentials

Create a service account in Google Cloud Console, enable the Drive API, and download the JSON key. Grant the service account access to the specific folders or files you want to load (share the folder with the service account email).

export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json

AWS credentials

Configure AWS credentials with read access to your target S3 buckets. Use IAM roles for EC2/ECS/Lambda, or set environment variables locally:

export AWS_ACCESS_KEY_ID=your_key
export AWS_SECRET_ACCESS_KEY=your_secret
export AWS_DEFAULT_REGION=us-east-1

Verify success: Run python -c "import langchain_community.document_loaders; print('imports ok')". No errors means the environment is ready.

Step 2: Load documents from Google Drive

The GoogleDriveLoader supports three modes: load a single file by ID, load all files in a folder, or load files matching a query. Folder loading is the most common pattern for RAG.

# loaders/gdrive_loader.py
from langchain_community.document_loaders import GoogleDriveLoader

def load_gdrive_folder(folder_id: str, recursive: bool = True) -> list:
    """
    Load all supported documents from a Google Drive folder.
    
    Args:
        folder_id: The folder ID from the Drive URL 
                   (https://drive.google.com/drive/folders/FOLDER_ID)
        recursive: Whether to traverse subfolders
    """
    loader = GoogleDriveLoader(
        folder_id=folder_id,
        recursive=recursive,
        file_types=["document", "sheet", "pdf", "text"],  # filter mime types
    )
    docs = loader.load()
    print(f"Loaded {len(docs)} documents from Google Drive folder {folder_id}")
    return docs

if __name__ == "__main__":
    # Replace with your actual folder ID
    FOLDER_ID = "1ABCdefGHIjklMNOpqrSTUvwxYZ"
    docs = load_gdrive_folder(FOLDER_ID)
    for d in docs[:3]:
        print(f"  - {d.metadata.get('source', 'unknown')}: {len(d.page_content)} chars")

Key metadata fields attached to each document:

• source: File name
• id: Google Drive file ID
• mimeType: MIME type (e.g., application/pdf, application/vnd.google-apps.document)
• modifiedTime: Last modification timestamp
• webViewLink: Direct link to open in browser

Verify success: Run the script. You should see a count > 0 and sample metadata printed. If you get a 403, check that the service account has access to the folder. If you get 0 documents, confirm the file_types filter matches your files — Google Workspace docs (Docs, Sheets) need "document" and "sheet" in the list.

Step 3: Load documents from Amazon S3

The S3FileLoader pulls individual objects, while S3DirectoryLoader recursively loads a prefix. Both use boto3 under the hood and support the same file parsers as the local loaders (PDF, DOCX, TXT, CSV, etc.).

# loaders/s3_loader.py
import boto3
from langchain_community.document_loaders import S3DirectoryLoader
from langchain_community.document_loaders.parsers import PDFMinerParser

def load_s3_prefix(bucket: str, prefix: str, region: str = "us-east-1") -> list:
    """
    Load all supported documents from an S3 prefix.
    
    Args:
        bucket: S3 bucket name
        prefix: Key prefix (e.g., "knowledge-base/quarterly-reports/")
        region: AWS region
    """
    # Configure parser for PDFs — PDFMiner handles tables better than pypdf
    loader = S3DirectoryLoader(
        bucket=bucket,
        prefix=prefix,
        region_name=region,
        parser=PDFMinerParser(),  # optional, improves PDF extraction
    )
    docs = loader.load()
    print(f"Loaded {len(docs)} documents from s3://{bucket}/{prefix}")
    return docs

def load_single_s3_object(bucket: str, key: str, region: str = "us-east-1") -> list:
    """Load a single S3 object when you know the exact key."""
    from langchain_community.document_loaders import S3FileLoader
    loader = S3FileLoader(bucket=bucket, key=key, region_name=region)
    return loader.load()

if __name__ == "__main__":
    BUCKET = "my-company-rag-corpus"
    PREFIX = "contracts/2024/"
    docs = load_s3_prefix(BUCKET, PREFIX)
    for d in docs[:3]:
        print(f"  - {d.metadata['source']}: {len(d.page_content)} chars")

Key metadata fields:

• source: Full S3 URI (s3://bucket/key)
• bucket: Bucket name
• key: Object key
• last_modified: Datetime from S3
• content_type: MIME type from S3 metadata

Verify success: Run the script. Confirm the document count matches your expectation. For large prefixes (>1000 objects), S3DirectoryLoader paginates automatically. If you hit NoCredentialsError, verify your AWS credential chain. If PDFs extract poorly, try PDFMinerParser or PyMuPDFParser (requires fitz).

Step 4: Combine sources and apply chunking

Real pipelines usually merge multiple sources before chunking. Use RecursiveCharacterTextSplitter for general text, or MarkdownHeaderTextSplitter if your sources preserve structure.

# pipeline/chunk_and_merge.py
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from loaders.gdrive_loader import load_gdrive_folder
from loaders.s3_loader import load_s3_prefix

def merge_and_chunk(
    gdrive_folder_id: str,
    s3_bucket: str,
    s3_prefix: str,
    chunk_size: int = 1000,
    chunk_overlap: int = 150,
) -> list[Document]:
    """Load from both sources, merge, and chunk."""
    # Load
    gdrive_docs = load_gdrive_folder(gdrive_folder_id)
    s3_docs = load_s3_prefix(s3_bucket, s3_prefix)
    
    all_docs = gdrive_docs + s3_docs
    print(f"Total raw documents: {len(all_docs)}")
    
    # Add source tag for downstream filtering
    for d in gdrive_docs:
        d.metadata["source_system"] = "google_drive"
    for d in s3_docs:
        d.metadata["source_system"] = "s3"
    
    # Chunk
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
        keep_separator=True,
    )
    chunks = splitter.split_documents(all_docs)
    print(f"Total chunks: {len(chunks)}")
    
    # Attach chunk IDs for traceability
    for i, chunk in enumerate(chunks):
        chunk.metadata["chunk_id"] = i
        chunk.metadata["chunk_size"] = len(chunk.page_content)
    
    return chunks

if __name__ == "__main__":
    chunks = merge_and_chunk(
        gdrive_folder_id="1ABCdefGHIjklMNOpqrSTUvwxYZ",
        s3_bucket="my-company-rag-corpus",
        s3_prefix="contracts/2024/",
    )
    # Inspect first few chunks
    for c in chunks[:5]:
        print(f"Chunk {c.metadata['chunk_id']} | "
              f"source={c.metadata.get('source_system')} | "
              f"chars={c.metadata['chunk_size']} | "
              f"preview={c.page_content[:80]}...")

Chunking strategy notes:

• chunk_size=1000 with overlap=150 works well for OpenAI embeddings (8191 token limit) and most open-source models
• For code or highly structured docs, consider LanguageParser + RecursiveCharacterTextSplitter with language-specific separators
• Keep chunk_overlap at 10-20% of chunk_size to preserve context across boundaries

Verify success: Run the script. You should see raw document count > 0, chunk count > raw count, and each chunk with metadata including source_system, chunk_id, and chunk_size. Spot-check a few chunks — they should read coherently, not cut mid-sentence.

Step 5: Persist to a vector store (verification step)

Loading and chunking are useless if you can’t query the results. This step writes to a local Chroma instance so you can run a quick similarity search and confirm end-to-end correctness.

# pipeline/verify_with_chroma.py
import chromadb
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from pipeline.chunk_and_merge import merge_and_chunk

def verify_pipeline(
    gdrive_folder_id: str,
    s3_bucket: str,
    s3_prefix: str,
    persist_dir: str = "./chroma_verify",
    collection_name: str = "rag_verify",
) -> None:
    """End-to-end verification: load, chunk, embed, query."""
    chunks = merge_and_chunk(gdrive_folder_id, s3_bucket, s3_prefix)
    
    # Embed and persist
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_dir,
        collection_name=collection_name,
    )
    print(f"Persisted {len(chunks)} chunks to {persist_dir}")
    
    # Run a test query against known content
    test_queries = [
        "termination clause",
        "quarterly revenue",
        "data processing agreement",
    ]
    
    for query in test_queries:
        results = vectorstore.similarity_search_with_score(query, k=3)
        print(f"\nQuery: '{query}'")
        for doc, score in results:
            src = doc.metadata.get("source_system", "unknown")
            preview = doc.page_content[:120].replace("\n", " ")
            print(f"  score={score:.3f} | source={src} | {preview}...")

if __name__ == "__main__":
    verify_pipeline(
        gdrive_folder_id="1ABCdefGHIjklMNOpqrSTUvwxYZ",
        s3_bucket="my-company-rag-corpus",
        s3_prefix="contracts/2024/",
    )

Verify success: Run this script. You should see:

1. Chroma persistence completes without error
2. Each test query returns 3 results with similarity scores (lower = more similar for Chroma’s default cosine distance)
3. Results come from both google_drive and s3 sources
4. Preview text looks relevant to the query

If scores are all ~1.0 (dissimilar), check that your embeddings model matches the one used in production. If results are empty, verify the chunk content actually contains the query terms.

Step 6: Production hardening

The scripts above work for development. Before deploying, address these concerns:

Incremental loading

Don’t reload everything on every run. Track processed file IDs and modification times.

# production/incremental.py
import json
from pathlib import Path
from googleapiclient.discovery import build

STATE_FILE = Path(".gdrive_state.json")

def load_state() -> dict:
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {"files": {}}

def save_state(state: dict) -> None:
    STATE_FILE.write_text(json.dumps(state, indent=2))

def get_new_or_modified_files(folder_id: str, state: dict) -> list:
    """Return Drive file metadata for files not in state or with newer modifiedTime."""
    service = build("drive", "v3", credentials=...)  # your creds
    query = f"'{folder_id}' in parents and trashed=false"
    results = service.files().list(
        q=query,
        fields="files(id, name, mimeType, modifiedTime)",
        pageSize=1000,
    ).execute()
    
    files = results.get("files", [])
    new_or_modified = []
    for f in files:
        fid = f["id"]
        mtime = f["modifiedTime"]
        if fid not in state["files"] or state["files"][fid] != mtime:
            new_or_modified.append(f)
            state["files"][fid] = mtime
    return new_or_modified

For S3, use S3DirectoryLoader with aws_access_key_id/aws_secret_access_key rotation via IAM roles, and consider S3 Event Notifications → SQS → Lambda for true streaming ingestion.

Error handling and retries

Wrap loader calls with tenacity:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    wait=wait_exponential(multiplier=1, min=2, max=30),
    stop=stop_after_attempt(3),
)
def safe_load_gdrive(folder_id: str):
    return GoogleDriveLoader(folder_id=folder_id).load()

Metadata normalization

Different loaders produce different metadata schemas. Normalize before chunking:

def normalize_metadata(doc: Document) -> Document:
    """Ensure consistent metadata keys across sources."""
    meta = doc.metadata
    normalized = {
        "source_uri": meta.get("source") or meta.get("webViewLink") or f"s3://{meta.get('bucket')}/{meta.get('key')}",
        "source_system": meta.get("source_system", "unknown"),
        "title": meta.get("name") or meta.get("title") or meta.get("key", "").split("/")[-1],
        "modified_at": meta.get("modifiedTime") or meta.get("last_modified"),
        "content_type": meta.get("mimeType") or meta.get("content_type"),
        "original_metadata": meta,  # preserve everything for debugging
    }
    doc.metadata = normalized
    return doc

Monitoring

Emit structured logs for each pipeline stage:

• documents_loaded (count, source, duration_ms)
• chunks_created (count, avg_chunk_size)
• embeddings_generated (count, model, duration_ms)
• vectorstore_upserted (count, collection, duration_ms)

Ship these to your observability stack (Datadog, Honeycomb, CloudWatch) and alert on latency spikes or zero-document runs.

Step 7: Running in a scheduled job

Wrap the incremental loader + chunking + upsert into a single entrypoint for Airflow, Prefect, or a cron job.

# pipeline/daily_ingest.py
import logging
from pipeline.incremental import load_state, save_state, get_new_or_modified_files
from pipeline.chunk_and_merge import merge_and_chunk
from pipeline.verify_with_chroma import verify_pipeline  # reuse embedding logic

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def daily_ingest():
    state = load_state()
    
    # 1. Detect changes
    new_files = get_new_or_modified_files("1ABCdefGHIjklMNOpqrSTUvwxYZ", state)
    logger.info(f"Found {len(new_files)} new/modified files in Google Drive")
    
    if not new_files:
        logger.info("No changes detected. Exiting.")
        return
    
    # 2. Load only changed files (simplified — extend for S3)
    # In practice, you'd filter the loader by file IDs
    chunks = merge_and_chunk(
        gdrive_folder_id="1ABCdefGHIjklMNOpqrSTUvwxYZ",
        s3_bucket="my-company-rag-corpus",
        s3_prefix="contracts/2024/",
    )
    
    # 3. Upsert to vector store (reuse verification logic but with upsert)
    # Chroma.from_documents replaces; for upsert use vectorstore.add_documents()
    
    # 4. Save state
    save_state(state)
    logger.info("Ingest complete")

if __name__ == "__main__":
    daily_ingest()

Verify success: Deploy to staging, trigger manually, and confirm:

• Logs show expected document counts
• Vector store record count increases by the expected delta
• A sample query against new content returns relevant results
• State file updates with new modifiedTime values

You now have a complete, verifiable pipeline for loading documents from Google Drive and S3 using LangChain document loaders, chunking them appropriately, and persisting to a vector store with incremental update support. The same patterns extend to SharePoint, Confluence, Notion, or any other source with a LangChain loader — swap the loader class, normalize metadata, and the rest of the pipeline stays identical.