LangChain CSV and JSON loaders for structured data

LangChain’s CSV and JSON loaders turn raw structured files into documents your RAG pipeline can actually use. This guide walks through the langchain csv json loaders structured data workflow end to end — from basic loading through nested JSON flattening, custom dialect handling, and chunking strategies that preserve row-level semantics.

Step 1: Install the right packages

LangChain splits document loaders into separate packages. You need the community package for both CSV and JSON loaders, plus the text splitters package for chunking.

pip install langchain-community langchain-text-splitters

If you’re using LangChain v0.2+, the core package is already a dependency of langchain-community. Verify the imports work:

from langchain_community.document_loaders import CSVLoader, JSONLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
print("Imports successful")

Step 2: Load a simple CSV with defaults

The CSVLoader reads each row as a separate Document. By default it uses Python’s csv module, treats the first row as headers, and concatenates all column values into page_content with a newline separator.

from langchain_community.document_loaders import CSVLoader

loader = CSVLoader(file_path="data/users.csv")
docs = loader.load()

print(f"Loaded {len(docs)} documents")
print(docs[0].page_content[:200])
print(docs[0].metadata)

Output for a CSV with columns id,name,email,tier:

Loaded 100 documents
id: 1
name: Alice Chen
email: alice@example.com
tier: pro

{'source': 'data/users.csv', 'row': 0}

Each document’s metadata includes source (the file path) and row (zero-based row index). This is useful for citations and debugging.

Verify success

assert len(docs) == 100  # matches your CSV row count minus header
assert all("id:" in d.page_content for d in docs)
assert all(d.metadata["source"] == "data/users.csv" for d in docs)

Step 3: Customize CSV parsing with csv_args

Real-world CSVs often have quirks: different delimiters, quoted fields with embedded newlines, no header row, or encoding issues. Pass csv_args directly to Python’s csv.DictReader.

loader = CSVLoader(
    file_path="data/exports/transactions.tsv",
    csv_args={
        "delimiter": "\t",
        "quotechar": '"',
        "quoting": 1,  # csv.QUOTE_ALL
    },
    encoding="utf-8-sig",  # handles BOM from Excel exports
)
docs = loader.load()

No header row

If your file lacks headers, provide fieldnames in csv_args and set source_column to None (since there’s no column to use as the content source).

loader = CSVLoader(
    file_path="data/logs/raw.log",
    csv_args={
        "delimiter": "|",
        "fieldnames": ["timestamp", "level", "service", "message"],
    },
    source_column=None,
)

Select specific columns for content

By default all columns go into page_content. Use content_columns to pick only what matters for embedding:

loader = CSVLoader(
    file_path="data/products.csv",
    content_columns=["name", "description", "category"],
    # metadata_columns defaults to all other columns
)

The metadata_columns parameter (added in v0.2) lets you explicitly control which columns become metadata. Columns not in either list are dropped.

Step 4: Load JSON with jq schemas

JSONLoader uses jq syntax to extract documents from JSON arrays or objects. This is more powerful than CSV loading but requires understanding jq filters.

Array of objects (most common)

from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="data/tickets.json",
    jq_schema=".[]",           # iterate over top-level array
    content_key="description", # field to use as page_content
    metadata_func=lambda record, meta: {
        **meta,
        "ticket_id": record.get("id"),
        "status": record.get("status"),
        "assignee": record.get("assignee"),
    },
)
docs = loader.load()

Given tickets.json:

[
  {"id": "TKT-1001", "status": "open", "assignee": "alice", "description": "Login fails on Safari"},
  {"id": "TKT-1002", "status": "closed", "assignee": "bob", "description": "Export CSV button missing"}
]

You get two documents with page_content set to the description and metadata enriched with the other fields.

Nested objects — flatten with jq

When your JSON has nested structures, use jq to flatten before LangChain sees it.

loader = JSONLoader(
    file_path="data/orders.json",
    jq_schema="""
        .orders[]
        | {
            order_id: .id,
            customer_email: .customer.email,
            items: [.items[].sku],
            total: .totals.grand_total,
        }
    """,
    content_key="description",
    metadata_func=lambda record, meta: {
        **meta,
        "order_id": record["order_id"],
        "customer_email": record["customer_email"],
        "item_count": len(record["items"]),
        "total": record["total"],
    },
)

The jq_schema here transforms each order into a flat object with a human-readable description. The metadata_func receives this transformed record.

Verify success

assert len(docs) == 2
assert "Login fails" in docs[0].page_content
assert docs[0].metadata["ticket_id"] == "TKT-1001"
assert docs[1].metadata["status"] == "closed"

Step 5: Handle JSON Lines (newline-delimited JSON)

Many data pipelines emit JSONL. Use jq_schema=".[]" with json_lines=True:

loader = JSONLoader(
    file_path="data/events.jsonl",
    jq_schema=".",
    content_key="message",
    json_lines=True,
    metadata_func=lambda record, meta: {
        **meta,
        "event_type": record.get("event_type"),
        "timestamp": record.get("ts"),
        "user_id": record.get("user_id"),
    },
)

Each line becomes one document. This is memory-efficient for large files since JSONLoader streams line by line when json_lines=True.

Step 6: Chunk structured documents intelligently

Naive chunking destroys the row/record boundary that gives structured data its meaning. A 500-token chunk splitting a single CSV row across two chunks loses the relationship between columns.

Strategy 1: One document per row (no further chunking)

For most CSV/JSONL use cases, don’t chunk at all. Each row is already a semantic unit. Embed the whole row.

# If your rows are short (< 512 tokens), skip chunking entirely
# The loader output is ready for your vector store
vectorstore.add_documents(docs)

Strategy 2: Chunk only long text fields

If a specific column contains long text (e.g., description, transcript, full_text), extract and chunk only that field while preserving other columns as metadata.

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""],
)

chunked_docs = []
for doc in docs:
    # Assume the long field is in page_content, metadata has the rest
    chunks = splitter.split_text(doc.page_content)
    for i, chunk in enumerate(chunks):
        chunked_docs.append(
            Document(
                page_content=chunk,
                metadata={
                    **doc.metadata,
                    "chunk_index": i,
                    "total_chunks": len(chunks),
                }
            )
        )

Strategy 3: Parent document retriever for wide rows

When a CSV row has many columns and you want both full-row context and granular search, use a parent document retriever pattern:

from langchain.storage import InMemoryStore
from langchain.retrievers import ParentDocumentRetriever
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# Small chunks for retrieval
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=20)
# Full row as parent
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=0)

vectorstore = FAISS.from_documents([], OpenAIEmbeddings())
store = InMemoryStore()

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

retriever.add_documents(docs, ids=None)

This stores each full row as a parent document and creates overlapping child chunks for retrieval. At query time you get the relevant chunk but can fetch the full parent row for context.

Step 7: Preserve data types in metadata

LangChain metadata values must be strings, numbers, booleans, or lists thereof. The loaders coerce everything to strings by default. If you need typed metadata for filtering (e.g., numeric range queries, boolean filters), convert explicitly in metadata_func.

def typed_metadata(record, meta):
    return {
        **meta,
        "ticket_id": record.get("id"),
        "priority": int(record.get("priority", 0)),        # numeric
        "is_escalated": record.get("escalated") == "true", # boolean
        "tags": record.get("tags", []),                    # list
        "created_at": record.get("created_at"),            # ISO string, keep as string
    }

loader = JSONLoader(
    file_path="data/tickets.json",
    jq_schema=".[]",
    content_key="description",
    metadata_func=typed_metadata,
)

Your vector store’s metadata filtering (where supported) can now use priority > 3 or is_escalated == true.

Step 8: Stream large files to avoid OOM

Both loaders support lazy loading via .lazy_load() which returns a generator. Use this for files that don’t fit in memory.

# CSV streaming
loader = CSVLoader(file_path="massive_export.csv")
for doc in loader.lazy_load():
    vectorstore.add_documents([doc])
    # optionally checkpoint every N docs

# JSONL streaming (already line-by-line)
loader = JSONLoader(
    file_path="huge_events.jsonl",
    jq_schema=".",
    content_key="message",
    json_lines=True,
)
for doc in loader.lazy_load():
    vectorstore.add_documents([doc])

For regular JSON arrays, JSONLoader loads the entire file into memory to parse with jq. If you have a multi-gigabyte JSON array, preprocess it to JSONL first:

jq -c '.[]' huge_array.json > huge_array.jsonl

Then use the JSONL loader with json_lines=True.

Step 9: Handle missing and malformed data

Real data is messy. Add defensive code in your metadata functions and content extraction.

def safe_metadata(record, meta):
    return {
        **meta,
        "id": str(record.get("id", "unknown")),
        "score": float(record.get("score", 0.0)) if record.get("score") else 0.0,
        "category": record.get("category", "uncategorized"),
        "tags": record.get("tags") if isinstance(record.get("tags"), list) else [],
    }

def safe_content(record):
    # Combine multiple fields with fallbacks
    parts = []
    if record.get("title"):
        parts.append(f"Title: {record['title']}")
    if record.get("body"):
        parts.append(f"Body: {record['body']}")
    if record.get("error_message"):
        parts.append(f"Error: {record['error_message']}")
    return "\n".join(parts) if parts else "Empty record"

loader = JSONLoader(
    file_path="data/messy.json",
    jq_schema=".[]",
    content_key=None,  # we'll build content in metadata_func
    metadata_func=lambda r, m: {**safe_metadata(r, m), "content": safe_content(r)},
)

# Post-process to move content to page_content
docs = []
for doc in loader.load():
    content = doc.metadata.pop("content", "")
    docs.append(Document(page_content=content, metadata=doc.metadata))

Step 10: End-to-end verification checklist

Before wiring into your RAG pipeline, run these checks:

def verify_docs(docs, expected_count=None, required_metadata_keys=None):
    """Validate loaded documents meet basic quality bar."""
    assert len(docs) > 0, "No documents loaded"
    
    if expected_count:
        assert len(docs) == expected_count, f"Expected {expected_count}, got {len(docs)}"
    
    # Every doc has content
    empty_content = [d for d in docs if not d.page_content.strip()]
    assert not empty_content, f"{len(empty_content)} documents have empty content"
    
    # Metadata consistency
    if required_metadata_keys:
        for key in required_metadata_keys:
            missing = [d for d in docs if key not in d.metadata]
            assert not missing, f"Metadata key '{key}' missing from {len(missing)} docs"
    
    # No duplicate row IDs (if applicable)
    if "row" in docs[0].metadata:
        rows = [d.metadata["row"] for d in docs]
        assert len(rows) == len(set(rows)), "Duplicate row indices found"
    
    # Token length sanity check
    from tiktoken import encoding_for_model
    enc = encoding_for_model("text-embedding-3-small")
    token_counts = [len(enc.encode(d.page_content)) for d in docs]
    print(f"Token stats: min={min(token_counts)}, max={max(token_counts)}, "
          f"mean={sum(token_counts)/len(token_counts):.0f}")
    assert max(token_counts) < 8000, "Some documents exceed embedding model limit"
    
    print("All verification checks passed")

# Usage
verify_docs(
    docs,
    expected_count=100,
    required_metadata_keys=["source", "row", "ticket_id", "status"]
)

Common pitfalls and fixes

When to use each loader

• CSVLoader: Tabular exports, spreadsheets, any data with consistent columns. Fast, streaming-friendly, minimal dependencies.
• JSONLoader: Nested structures, API responses, logs with variable fields, any data requiring jq transformations.
• JSONLinesLoader (via json_lines=True): High-volume event streams, append-only logs, datasets too large for memory.

Both loaders produce Document objects compatible with every LangChain vector store, retriever, and chain. The langchain csv json loaders structured data pattern works identically whether you’re indexing into FAISS, Pinecone, Weaviate, or a Postgres-backed store.

Next steps

• Add a metadata-based filter to your retriever for tenant isolation or tier-based access
• Experiment with ParentDocumentRetriever for wide rows where you need both precision and context
• Profile embedding latency — structured data often embeds faster than prose because token counts are lower and more predictable
• Consider a scheduled reload job that uses lazy_load() and upserts by metadata["id"] to keep the index fresh without full rebuilds