7 LangChain Document Transformers (Splitters, Filters, Embeddings)

pip install langchain langchain-openai langchain-community chromadb tiktoken

import os
os.environ["OPENAI_API_KEY"] = "your-key-here"

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

raw_text = """
LangChain is a framework for building applications powered by language models.

It provides tools for chaining together LLM calls, managing prompts, and connecting
to external data sources. The core abstraction is the Chain — a sequence of steps
that each take some input and produce some output.

Retrieval-Augmented Generation (RAG) is one of the most popular patterns built
with LangChain. In a RAG pipeline, documents are indexed in a vector store, and
at query time the most relevant chunks are retrieved and passed to the LLM as
additional context.

Agents extend this further by giving the LLM access to tools — search engines,
calculators, APIs, databases — and allowing it to decide which tools to call
based on the user's request.
"""

splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,           # target chunk size in characters
    chunk_overlap=40,         # overlap between chunks for context continuity
    length_function=len,
    is_separator_regex=False,
)

docs = splitter.create_documents([raw_text])

for i, doc in enumerate(docs):
    print(f"Chunk {i}: {len(doc.page_content)} chars")
    print(doc.page_content[:100])
    print("---")

import tiktoken
from langchain.text_splitter import RecursiveCharacterTextSplitter

def tiktoken_len(text):
    tokenizer = tiktoken.get_encoding("cl100k_base")
    tokens = tokenizer.encode(text, disallowed_special=())
    return len(tokens)

splitter = RecursiveCharacterTextSplitter(
    chunk_size=256,       # now measured in tokens, not characters
    chunk_overlap=32,
    length_function=tiktoken_len,
)

from langchain.text_splitter import RecursiveCharacterTextSplitter, Language

python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=1000,
    chunk_overlap=100,
)

python_code = """
def calculate_embeddings(texts, model="text-embedding-3-small"):
    from openai import OpenAI
    client = OpenAI()
    response = client.embeddings.create(input=texts, model=model)
    return [item.embedding for item in response.data]

class VectorStore:
    def __init__(self, dimension):
        self.dimension = dimension
        self.vectors = []

    def add(self, vector, metadata=None):
        self.vectors.append({"vector": vector, "metadata": metadata or {}})

    def search(self, query_vector, k=5):
        import numpy as np
        scores = []
        for item in self.vectors:
            similarity = np.dot(query_vector, item["vector"])
            scores.append((similarity, item["metadata"]))
        return sorted(scores, reverse=True)[:k]
"""

code_chunks = python_splitter.create_documents([python_code])
print(f"Split into {len(code_chunks)} chunks")
for chunk in code_chunks:
    print(f"  {chunk.page_content[:60]}...")

from langchain.text_splitter import MarkdownTextSplitter

markdown_doc = """
# Introduction to Transformers

Transformer models revolutionized NLP in 2017 with the "Attention Is All You Need" paper.
They replaced recurrent networks with self-attention mechanisms that capture long-range dependencies.

## Self-Attention Mechanism

The attention mechanism allows each token to attend to all other tokens in the sequence.
This enables capturing relationships that RNNs struggled with due to vanishing gradients.

### Scaled Dot-Product Attention

Attention scores are computed as the dot product of queries and keys,
scaled by the square root of the key dimension to prevent gradient saturation.

## Positional Encoding

Since attention has no inherent notion of sequence order, positional encodings are added
to inject position information into the token representations.

## Applications

Transformers are now used in vision, audio, code generation, and multimodal tasks.
The architecture has proven remarkably general across very different data types.
"""

md_splitter = MarkdownTextSplitter(chunk_size=300, chunk_overlap=50)
md_chunks = md_splitter.create_documents([markdown_doc])

for chunk in md_chunks:
    print(f"Content: {chunk.page_content[:150]}")
    print("---")

from langchain.text_splitter import TokenTextSplitter

token_splitter = TokenTextSplitter(
    model_name="gpt-4o",   # uses tiktoken encoding for this model
    chunk_size=512,
    chunk_overlap=64,
)

# Generate a long text to demonstrate
sample_text = " ".join([
    f"This is sentence number {i} with some extra filler content to make it longer."
    for i in range(300)
])

token_chunks = token_splitter.create_documents([sample_text])
print(f"Created {len(token_chunks)} chunks from {len(sample_text)} characters")

# Verify actual token counts
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o")
for i, chunk in enumerate(token_chunks[:3]):
    token_count = len(enc.encode(chunk.page_content))
    print(f"Chunk {i}: {token_count} tokens (target was 512)")

from langchain_openai import OpenAIEmbeddings
from langchain.retrievers.document_compressors import EmbeddingsFilter
from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.vectorstores import Chroma

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

sample_docs = [
    Document(page_content="Python is a high-level programming language.", metadata={"source": "wiki"}),
    Document(page_content="The Eiffel Tower is located in Paris, France.", metadata={"source": "wiki"}),
    Document(page_content="Machine learning uses statistical algorithms to find patterns.", metadata={"source": "wiki"}),
    Document(page_content="Python is widely used for data science and machine learning.", metadata={"source": "wiki"}),
    Document(page_content="Neural networks are inspired by biological brain structures.", metadata={"source": "wiki"}),
]

vectorstore = Chroma.from_documents(sample_docs, embeddings)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# Drop docs with similarity below 0.76 to the query
embeddings_filter = EmbeddingsFilter(
    embeddings=embeddings,
    similarity_threshold=0.76
)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=embeddings_filter,
    base_retriever=base_retriever,
)

query = "What programming language is good for AI?"
filtered_docs = compression_retriever.invoke(query)

print(f"Retrieved and filtered: {len(filtered_docs)} relevant documents")
for doc in filtered_docs:
    print(f"  - {doc.page_content}")

from langchain.retrievers.document_compressors import EmbeddingsRedundantFilter

# Simulate retrieved documents with near-duplicates
duplicate_docs = [
    Document(page_content="LangChain helps you build LLM applications easily."),
    Document(page_content="LangChain is a framework for building LLM-powered apps."),      # near-duplicate
    Document(page_content="LangChain makes it easy to create language model applications."), # near-duplicate
    Document(page_content="Vector databases store embeddings for semantic search."),
    Document(page_content="Embeddings are numerical representations of text meaning."),
]

redundancy_filter = EmbeddingsRedundantFilter(
    embeddings=embeddings,
    similarity_threshold=0.90   # drop docs with more than 90% similarity to a kept doc
)

filtered = redundancy_filter.transform_documents(duplicate_docs)
print(f"Before: {len(duplicate_docs)} docs, After: {len(filtered)} docs")
for doc in filtered:
    print(f"  KEPT: {doc.page_content}")

from langchain_community.document_transformers import LongContextReorder

# Documents ordered by relevance score (most relevant first)
ordered_docs = [
    Document(page_content="Most relevant: directly answers the question with specific facts."),
    Document(page_content="Second: very closely related to the query topic."),
    Document(page_content="Third: somewhat related with partial information."),
    Document(page_content="Fourth: tangentially related background context."),
    Document(page_content="Fifth: general context that might help."),
    Document(page_content="Sixth: loosely related information."),
    Document(page_content="Seventh: barely related to the query."),
    Document(page_content="Eighth: least relevant of the retrieved set."),
]

reorder = LongContextReorder()
reordered = reorder.transform_documents(ordered_docs)

print("New order (best docs at edges, worst in middle):")
for i, doc in enumerate(reordered):
    print(f"  Position {i}: {doc.page_content[:55]}")

from langchain.text_splitter import HTMLHeaderTextSplitter

html_content = """
<!DOCTYPE html>
<html>
<body>
<h1>LangChain Complete Guide</h1>
<p>LangChain is a comprehensive framework for building LLM applications.</p>

<h2>Core Components</h2>
<p>The main components are chains, agents, memory, and retrievers.</p>

<h3>Chains</h3>
<p>Chains connect multiple LLM calls together using LCEL syntax.</p>
<p>They can be composed declaratively using the pipe operator.</p>

<h3>Agents</h3>
<p>Agents use LLMs to decide dynamically which tools to call.</p>
<p>They handle open-ended tasks that require multi-step reasoning.</p>

<h2>Getting Started</h2>
<p>Install LangChain with pip install langchain and set your API key.</p>
</body>
</html>
"""

headers_to_split_on = [
    ("h1", "Header 1"),
    ("h2", "Header 2"),
    ("h3", "Header 3"),
]

html_splitter = HTMLHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
html_chunks = html_splitter.split_text(html_content)

for chunk in html_chunks:
    print(f"Metadata: {chunk.metadata}")
    print(f"Content: {chunk.page_content[:120]}")
    print("---")

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.retrievers.document_compressors import (
    EmbeddingsRedundantFilter,
    EmbeddingsFilter,
)
from langchain_community.document_transformers import LongContextReorder
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
import os

os.environ["OPENAI_API_KEY"] = "your-key-here"
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")


def preprocess_documents(raw_documents: list) -> list:
    """Full preprocessing pipeline for RAG ingestion."""

    # Step 1: Split into appropriately-sized chunks
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=512,
        chunk_overlap=64,
        separators=["\n\n", "\n", ". ", " ", ""]
    )
    split_docs = splitter.transform_documents(raw_documents)
    print(f"After splitting: {len(split_docs)} chunks")

    # Step 2: Remove near-duplicate chunks before indexing
    dedup_filter = EmbeddingsRedundantFilter(
        embeddings=embeddings,
        similarity_threshold=0.92
    )
    unique_docs = dedup_filter.transform_documents(split_docs)
    print(f"After deduplication: {len(unique_docs)} chunks")

    return unique_docs


def retrieve_with_pipeline(
    query: str,
    vectorstore,
    k_initial: int = 20,
    k_final: int = 6
) -> list:
    """Retrieve, filter by relevance, reorder for LLM consumption."""

    # Over-retrieve to give the filter enough candidates
    initial_docs = vectorstore.similarity_search(query, k=k_initial)

    # Keep only docs actually relevant to this query
    relevance_filter = EmbeddingsFilter(
        embeddings=embeddings,
        similarity_threshold=0.75,
        k=k_final
    )
    relevant_docs = relevance_filter.compress_documents(initial_docs, query)
    print(f"After relevance filter: {len(relevant_docs)} docs")

    # Reorder to combat the lost-in-the-middle effect
    reorder = LongContextReorder()
    final_docs = reorder.transform_documents(relevant_docs)

    return final_docs


# Build the index
sample_documents = [
    Document(
        page_content="""
        Vector databases store high-dimensional embeddings for efficient similarity search.
        Popular options include Chroma, Pinecone, Weaviate, and Qdrant.
        Chroma is excellent for local development — no API key required.
        Pinecone is a managed service that scales to billions of vectors automatically.
        Weaviate supports hybrid search combining dense and sparse retrieval.
        """,
        metadata={"source": "db_guide.txt", "category": "infrastructure"}
    ),
]

processed = preprocess_documents(sample_documents)
vectorstore = Chroma.from_documents(
    processed,
    embeddings,
    persist_directory="./chroma_db"
)

# Query with the full pipeline
results = retrieve_with_pipeline(
    "Which vector database should I use for a production app?",
    vectorstore
)
print(f"\nFinal context for LLM: {len(results)} documents")

import hashlib
from langchain.text_splitter import RecursiveCharacterTextSplitter


def split_with_rich_metadata(
    text: str,
    source: str,
    category: str,
    chunk_size: int = 512
) -> list:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=64,
    )

    chunks = splitter.create_documents([text])

    for i, chunk in enumerate(chunks):
        chunk.metadata.update({
            "source": source,
            "category": category,
            "chunk_index": i,
            "total_chunks": len(chunks),
            "char_count": len(chunk.page_content),
            "chunk_hash": hashlib.md5(chunk.page_content.encode()).hexdigest()[:8],
            "is_first": i == 0,
            "is_last": i == len(chunks) - 1,
        })

    return chunks


# Retrieve only from a specific category later
engineering_results = vectorstore.similarity_search(
    query="installation procedure",
    k=5,
    filter={"category": "engineering"}
)

# Efficient — single API call for all documents
texts = [doc.page_content for doc in documents]
all_embeddings = embeddings.embed_documents(texts)  # batched automatically