LangChain document loaders for Notion and Confluence

LangChain document loaders for Notion and Confluence let you pull structured knowledge into RAG pipelines without writing custom scrapers. Both loaders handle authentication, pagination, and content extraction, but they differ in how they model content hierarchy and what metadata they surface. This walkthrough covers end-to-end setup, common pitfalls, and chunking strategies that work in production.

Step 1: Install dependencies and configure credentials

Start with a clean virtual environment. You need the community package for both loaders plus the Notion and Confluence SDKs they wrap.

python -m venv .venv && source .venv/bin/activate
pip install langchain-community notion-client atlassian-python-api python-dotenv

Create a .env file to keep secrets out of source control:

# Notion
NOTION_TOKEN=secret_xxxxxxxxxxxxxxxxxxxxxxxx
NOTION_DATABASE_ID=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx  # or page ID

# Confluence
CONFLUENCE_URL=https://your-domain.atlassian.net/wiki
CONFLUENCE_USERNAME=your-email@company.com
CONFLUENCE_API_TOKEN=ATATT3xFfGF0xxxxxxxxxxxxxxxx
CONFLUENCE_SPACE_KEY=ENG  # or comma-separated list

Verify the imports work before proceeding:

# verify_imports.py
from langchain_community.document_loaders import NotionDBLoader, ConfluenceLoader
from notion_client import Client as NotionClient
from atlassian import Confluence

print("Imports successful")

Run it: python verify_imports.py. You should see “Imports successful” with no traceback.

Step 2: Load Notion databases with NotionDBLoader

The Notion loader treats a database as a collection of pages. Each row becomes a Document with the page content as page_content and all properties as metadata.

# load_notion.py
import os
from dotenv import load_dotenv
from langchain_community.document_loaders import NotionDBLoader

load_dotenv()

loader = NotionDBLoader(
    integration_token=os.getenv("NOTION_TOKEN"),
    database_id=os.getenv("NOTION_DATABASE_ID"),
    request_timeout_sec=30,  # Notion API can be slow on large databases
)

docs = loader.load()
print(f"Loaded {len(docs)} documents from Notion")

# Inspect the first document
if docs:
    doc = docs[0]
    print(f"Content length: {len(doc.page_content)} chars")
    print(f"Metadata keys: {list(doc.metadata.keys())}")
    print(f"Sample metadata: {{k: v for k, v in list(doc.metadata.items())[:5]}}")

Run it: python load_notion.py. Expected output shows document count and metadata keys like id, created_time, last_edited_time, title, and any custom properties (Select, Multi-select, Date, etc.).

Common issue: If you get ObjectNotFoundError, the integration token lacks access to the database. In Notion, open the database → ⋮ → Add connections → select your integration.

Performance note: NotionDBLoader fetches pages sequentially. For databases over 500 pages, consider batching with asyncio or using the lower-level NotionClient directly to parallelize.

Step 3: Load Confluence spaces with ConfluenceLoader

ConfluenceLoader pulls pages from one or more spaces. It supports CQL (Confluence Query Language) for filtering, which is essential for large instances.

# load_confluence.py
import os
from dotenv import load_dotenv
from langchain_community.document_loaders import ConfluenceLoader

load_dotenv()

loader = ConfluenceLoader(
    url=os.getenv("CONFLUENCE_URL"),
    username=os.getenv("CONFLUENCE_USERNAME"),
    api_key=os.getenv("CONFLUENCE_API_TOKEN"),
    space_key=os.getenv("CONFLUENCE_SPACE_KEY"),
    # Optional: limit to recently updated pages
    cql="lastmodified >= -30d",
    # Include attachments? Default False. Set True if you need PDFs/images.
    include_attachments=False,
    # Max pages to fetch (0 = no limit). Useful for test runs.
    limit=100,
)

docs = loader.load()
print(f"Loaded {len(docs)} documents from Confluence")

if docs:
    doc = docs[0]
    print(f"Content length: {len(doc.page_content)} chars")
    print(f"Metadata keys: {list(doc.metadata.keys())}")
    print(f"Title: {doc.metadata.get('title')}")
    print(f"Source URL: {doc.metadata.get('source')}")

Run it: python load_confluence.py. Metadata includes title, source (page URL), space_key, page_id, version, created_date, last_modified, and labels.

Authentication gotcha: Cloud instances use email + API token. Server/Data Center instances may require PAT (Personal Access Token) or basic auth with username/password. Adjust the loader instantiation accordingly:

# For Data Center with PAT
loader = ConfluenceLoader(
    url=os.getenv("CONFLUENCE_URL"),
    token=os.getenv("CONFLUENCE_PAT"),  # Personal Access Token
    space_key=os.getenv("CONFLUENCE_SPACE_KEY"),
)

Rate limiting: Confluence returns 429 with Retry-After header. The loader respects it but adds no jitter. For production, wrap calls with tenacity:

from tenacity import retry, stop_after_attempt, wait_exponential_jitter

@retry(wait=wait_exponential_jitter(initial=1, max=30), stop=stop_after_attempt(5))
def load_with_retry(loader):
    return loader.load()

Step 4: Normalize metadata across sources

Notion and Confluence use different field names for similar concepts. Normalize before chunking so downstream retrieval treats them uniformly.

# normalize.py
from typing import List
from langchain_core.documents import Document

def normalize_metadata(docs: List[Document], source: str) -> List[Document]:
    """Map source-specific metadata to a common schema."""
    normalized = []
    for doc in docs:
        meta = doc.metadata.copy()
        
        if source == "notion":
            meta.update({
                "source_system": "notion",
                "doc_id": meta.get("id"),
                "doc_title": meta.get("title", "Untitled"),
                "created_at": meta.get("created_time"),
                "updated_at": meta.get("last_edited_time"),
                "url": f"https://notion.so/{meta.get('id', '').replace('-', '')}",
            })
        elif source == "confluence":
            meta.update({
                "source_system": "confluence",
                "doc_id": meta.get("page_id"),
                "doc_title": meta.get("title", "Untitled"),
                "created_at": meta.get("created_date"),
                "updated_at": meta.get("last_modified"),
                "url": meta.get("source"),
                "space_key": meta.get("space_key"),
            })
        
        # Keep only normalized fields + any custom ones you need
        allowed = {"source_system", "doc_id", "doc_title", "created_at", "updated_at", "url", "space_key", "labels"}
        clean_meta = {k: v for k, v in meta.items() if k in allowed or k.startswith("custom_")}
        
        normalized.append(Document(page_content=doc.page_content, metadata=clean_meta))
    
    return normalized

Apply it after loading:

notion_docs = normalize_metadata(notion_docs, "notion")
confluence_docs = normalize_metadata(confluence_docs, "confluence")
all_docs = notion_docs + confluence_docs

Step 5: Choose a chunking strategy for each source

Notion pages tend to be semi-structured (databases, toggles, callouts). Confluence pages are often long-form with headers, tables, and macros. One chunker rarely fits both.

For Notion: Preserve property context

Notion content arrives as plain text with property values embedded. Use a header-aware splitter that keeps property context attached to each section.

# chunk_notion.py
from langchain_text_splitters import MarkdownHeaderTextSplitter

# Notion exports headers as #, ##, ### in the text
headers_to_split_on = [
    ("#", "header_1"),
    ("##", "header_2"),
    ("###", "header_3"),
]

splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)

def chunk_notion_docs(docs: List[Document]) -> List[Document]:
    chunks = []
    for doc in docs:
        # Split by headers first
        header_chunks = splitter.split_text(doc.page_content)
        
        for chunk in header_chunks:
            # Merge metadata: doc-level + header-level
            merged_meta = {**doc.metadata, **chunk.metadata}
            chunks.append(Document(page_content=chunk.page_content, metadata=merged_meta))
    
    # Secondary split for oversized chunks (>1500 tokens)
    from langchain_text_splitters import RecursiveCharacterTextSplitter
    secondary = RecursiveCharacterTextSplitter(
        chunk_size=1200,
        chunk_overlap=150,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    return secondary.split_documents(chunks)

For Confluence: Leverage HTML structure

ConfluenceLoader returns HTML-stripped text by default. If you need structure, fetch raw HTML and split by heading tags.

# chunk_confluence.py
from bs4 import BeautifulSoup
from langchain_text_splitters import RecursiveCharacterTextSplitter

def extract_sections(html: str) -> List[dict]:
    """Parse Confluence HTML into (heading, content) sections."""
    soup = BeautifulSoup(html, "html.parser")
    sections = []
    current_heading = "Introduction"
    current_content = []
    
    for elem in soup.find_all(["h1", "h2", "h3", "h4", "p", "ul", "ol", "table", "pre"]):
        if elem.name in ["h1", "h2", "h3", "h4"]:
            if current_content:
                sections.append({
                    "heading": current_heading,
                    "content": "\n".join(current_content),
                })
            current_heading = elem.get_text(strip=True)
            current_content = []
        else:
            current_content.append(elem.get_text(strip=True))
    
    if current_content:
        sections.append({"heading": current_heading, "content": "\n".join(current_content)})
    
    return sections

def chunk_confluence_docs(docs: List[Document]) -> List[Document]:
    """Requires fetching HTML. Use Confluence REST API directly for raw HTML."""
    # This is a placeholder — see Step 6 for the full implementation
    pass

Unified secondary splitter

After source-specific splitting, apply a consistent secondary splitter to bound token counts for your embedding model.

# final_chunk.py
from langchain_text_splitters import RecursiveCharacterTextSplitter

def final_chunk(docs: List[Document], chunk_size: int = 1000, overlap: int = 120) -> List[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
        length_function=len,  # Use token counter in production
    )
    return splitter.split_documents(docs)

Step 6: Production Confluence HTML fetching

The default ConfluenceLoader strips HTML. For heading-aware chunking, fetch raw storage format via the REST API.

# fetch_confluence_html.py
import os
import requests
from dotenv import load_dotenv
from langchain_core.documents import Document
from bs4 import BeautifulSoup

load_dotenv()

BASE_URL = os.getenv("CONFLUENCE_URL").rstrip("/")
AUTH = (os.getenv("CONFLUENCE_USERNAME"), os.getenv("CONFLUENCE_API_TOKEN"))
SPACE_KEY = os.getenv("CONFLUENCE_SPACE_KEY")

def fetch_pages(limit: int = 50) -> List[dict]:
    """Fetch pages with storage format (HTML) from Confluence REST API."""
    url = f"{BASE_URL}/rest/api/content"
    params = {
        "spaceKey": SPACE_KEY,
        "expand": "body.storage,version,metadata.labels",
        "limit": limit,
    }
    resp = requests.get(url, auth=AUTH, params=params, timeout=30)
    resp.raise_for_status()
    return resp.json()["results"]

def parse_storage_format(page: dict) -> Document:
    html = page["body"]["storage"]["value"]
    soup = BeautifulSoup(html, "html.parser")
    
    # Extract text with heading markers
    text_parts = []
    for elem in soup.find_all(["h1", "h2", "h3", "h4", "p", "ul", "ol", "li", "table", "td", "th", "code", "pre"]):
        if elem.name in ["h1", "h2", "h3", "h4"]:
            level = "#" * int(elem.name[1])
            text_parts.append(f"{level} {elem.get_text(strip=True)}")
        elif elem.name in ["ul", "ol"]:
            for li in elem.find_all("li", recursive=False):
                text_parts.append(f"- {li.get_text(strip=True)}")
        elif elem.name == "table":
            # Simple table flattening
            for row in elem.find_all("tr"):
                cells = [c.get_text(strip=True) for c in row.find_all(["td", "th"])]
                text_parts.append(" | ".join(cells))
        else:
            text = elem.get_text(strip=True)
            if text:
                text_parts.append(text)
    
    content = "\n\n".join(text_parts)
    
    metadata = {
        "source_system": "confluence",
        "doc_id": page["id"],
        "doc_title": page["title"],
        "url": f"{BASE_URL}{page['_links']['webui']}",
        "space_key": SPACE_KEY,
        "version": page["version"]["number"],
        "updated_at": page["version"]["when"],
        "labels": [l["name"] for l in page.get("metadata", {}).get("labels", {}).get("results", [])],
    }
    
    return Document(page_content=content, metadata=metadata)

# Usage
pages = fetch_pages(limit=100)
docs = [parse_storage_format(p) for p in pages]
print(f"Parsed {len(docs)} Confluence pages with HTML structure preserved")

Step 7: Verify the pipeline end to end

Create a verification script that loads, normalizes, chunks, and validates output.

# verify_pipeline.py
import os
from dotenv import load_dotenv
from langchain_community.document_loaders import NotionDBLoader, ConfluenceLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Import your normalization and chunking functions
from normalize import normalize_metadata
from chunk_notion import chunk_notion_docs
from final_chunk import final_chunk

load_dotenv()

def verify_notion():
    print("=== Notion Pipeline ===")
    loader = NotionDBLoader(
        integration_token=os.getenv("NOTION_TOKEN"),
        database_id=os.getenv("NOTION_DATABASE_ID"),
    )
    raw = loader.load()
    print(f"Raw docs: {len(raw)}")
    
    normalized = normalize_metadata(raw, "notion")
    print(f"Normalized: {len(normalized)}")
    
    chunked = chunk_notion_docs(normalized)
    print(f"After header split: {len(chunked)}")
    
    final = final_chunk(chunked)
    print(f"Final chunks: {len(final)}")
    
    # Validate
    assert all(len(c.page_content) > 50 for c in final), "Empty chunks detected"
    assert all(c.metadata.get("source_system") == "notion" for c in final), "Missing source_system"
    assert all(c.metadata.get("doc_title") for c in final), "Missing doc_title"
    
    # Sample
    print(f"\nSample chunk:")
    print(f"  Title: {final[0].metadata.get('doc_title')}")
    print(f"  Length: {len(final[0].page_content)} chars")
    print(f"  Headers: {final[0].metadata.get('header_1', 'N/A')} > {final[0].metadata.get('header_2', 'N/A')}")
    
    return final

def verify_confluence():
    print("\n=== Confluence Pipeline ===")
    loader = ConfluenceLoader(
        url=os.getenv("CONFLUENCE_URL"),
        username=os.getenv("CONFLUENCE_USERNAME"),
        api_key=os.getenv("CONFLUENCE_API_TOKEN"),
        space_key=os.getenv("CONFLUENCE_SPACE_KEY"),
        limit=20,
    )
    raw = loader.load()
    print(f"Raw docs: {len(raw)}")
    
    normalized = normalize_metadata(raw, "confluence")
    print(f"Normalized: {len(normalized)}")
    
    final = final_chunk(normalized)
    print(f"Final chunks: {len(final)}")
    
    assert all(len(c.page_content) > 50 for c in final), "Empty chunks detected"
    assert all(c.metadata.get("source_system") == "confluence" for c in final), "Missing source_system"
    
    print(f"\nSample chunk:")
    print(f"  Title: {final[0].metadata.get('doc_title')}")
    print(f"  Space: {final[0].metadata.get('space_key')}")
    print(f"  Length: {len(final[0].page_content)} chars")
    
    return final

if __name__ == "__main__":
    notion_chunks = verify_notion()
    confluence_chunks = verify_confluence()
    
    all_chunks = notion_chunks + confluence_chunks
    print(f"\n=== Combined: {len(all_chunks)} chunks ready for embedding ===")
    
    # Quick token estimate (rough: 1 token ≈ 4 chars)
    total_chars = sum(len(c.page_content) for c in all_chunks)
    est_tokens = total_chars // 4
    print(f"Estimated tokens: ~{est_tokens:,}")

Run it: python verify_pipeline.py. Success criteria:

• Both pipelines complete without errors
• Assertions pass (no empty chunks, required metadata present)
• Sample output shows meaningful content and preserved headers
• Combined chunk count is reasonable for your embedding budget

Step 8: Incremental updates and scheduling

Production pipelines need incremental loads. Both sources support modified-since filtering.

# incremental.py
from datetime import datetime, timedelta
import json

STATE_FILE = ".loader_state.json"

def load_state() -> dict:
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f)
    return {}

def save_state(state: dict):
    with open(STATE_FILE, "w") as f:
        json.dump(state, f)

def get_notion_since(state: dict) -> str:
    """Return ISO timestamp for Notion filter."""
    last = state.get("notion_last_sync")
    if last:
        return last
    return (datetime.utcnow() - timedelta(days=30)).isoformat() + "Z"

def get_confluence_cql(state: dict) -> str:
    """Return CQL for Confluence filter."""
    last = state.get("confluence_last_sync")
    if last:
        return f"lastmodified >= '{last}'"
    return "lastmodified >= -30d"

def update_state(state: dict, source: str, docs: list):
    if not docs:
        return
    latest = max(d.metadata.get("updated_at") or "" for d in docs)
    if latest:
        state[f"{source}_last_sync"] = latest
        save_state(state)

Integrate into your scheduled job (cron, Airflow, Prefect, etc.):

# Daily at 2 AM
0 2 * * * /path/to/.venv/bin/python /path/to/incremental_sync.py >> /var/log/rag_sync.log 2>&1

Step 9: Embedding and indexing considerations

Chunk size must match your embedding model’s context window. For text-embedding-3-small (8191 tokens), 1000-char chunks with 120-char overlap is safe. For text-embedding-3-large or Cohere v3, you can go larger.

# embedding_check.py
from langchain_openai import OpenAIEmbeddings
import tiktoken

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
enc = tiktoken.encoding_for_model("text-embedding-3-small")

def check_chunk_tokens(chunks, max_tokens=8000):
    oversized = []
    for i, chunk in enumerate(chunks):
        tokens = len(enc.encode(chunk.page_content))
        if tokens > max_tokens:
            oversized.append((i, tokens))
    return oversized

# Run after final_chunk()
oversized = check_chunk_tokens(all_chunks)
if oversized:
    print(f"WARNING: {len(oversized)} chunks exceed token limit")
    for idx, tok in oversized[:5]:
        print(f"  Chunk {idx}: {tok} tokens")
else:
    print("All chunks within token limits")

When routing embedding requests through a gateway that handles multiple providers, you can keep the same chunking logic and swap models by changing the embedding client initialization. The gateway handles authentication and fallback transparently.

Step 10: Common failure modes and fixes

Verification checklist

Before considering the pipeline production-ready:

• Both loaders authenticate and return documents
• Normalization produces consistent metadata schema
• Chunking preserves heading hierarchy for retrieval context
• No chunks exceed embedding model token limit
• Incremental sync updates only changed pages
• Rate limiting handled with retry + backoff
• State persisted for resume after failure
• Logs capture document counts and errors per run

You now have a working, maintainable pipeline pulling Notion and Confluence into LangChain. The same pattern extends to other loaders in the ecosystem — normalize early, chunk deliberately, verify continuously.