LangChain Integration Guide (Python)

IndieML integrates natively into LangChain as a custom DocumentTransformer. It intercepts chunked data before the embedding step, scoring and dropping low-substance artifacts to save vector space and compute.

Prerequisite: Request an IndieML API Key

Before starting, you need an API key to authenticate the connection. Click here to get your free API key. Keep the key handy for the configuration step.

Installation

Install the IndieML SDK alongside your LangChain environment:

Bash 
pip install indieml langchain-core
The Drop-in Class

Drop this code directly into your project:

Python 
class IndieMLFilter(BaseDocumentTransformer):
    """Filters LangChain documents based on IndieML substance scores."""
    
    def __init__(self, api_key: str, threshold: float = 0.5):
        self.api = Substance(api_key=api_key)
        self.threshold = threshold
        
    def transform_documents(self, documents: list[Document], **kwargs) -> list[Document]:
        texts = [doc.page_content for doc in documents]
        
        batch_results = self.api.score_batch(texts)
        
        clean_docs = []
        for doc, result in zip(documents, batch_results):
            score = result.get("substance_score", 0.0)
            
            if score >= self.threshold:
                doc.metadata["indieml_score"] = score
                clean_docs.append(doc)
                
        return clean_docs
The Pipeline Implementation

Because IndieML operates as a semantic filter, it should be placed in your ingestion pipeline immediately after chunking and before embedding. By ensuring your chunks are a uniform length, you neutralize any natural verbosity bias in the scoring model.

Python 
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
# ... (assume IndieMLFilter is imported)

# 1. Load your messy data
loader = WebBaseLoader("https://example.com/messy-data")
documents = loader.load()

# 2. Split into uniform chunks to normalize the evaluation
# (150-300 characters is typical for most applications.)
# (The API is capped at 1,000 characters.)
text_splitter = RecursiveCharacterTextSplitter(chunk_size=150, chunk_overlap=15)
split_docs = text_splitter.split_documents(documents)

# 3. Apply the filter
# (Tune the threshold based on your specific document type and chunk size)
indieml = IndieMLFilter(api_key="YOUR_API_KEY", threshold=0.5)
clean_docs = indieml.transform_documents(split_docs)

# 4. Pass clean_docs to the next step in your pipeline.
Other LangChain Patterns

While RAG ingestion is the most common use case, this filter pattern is highly versatile. You can apply the same logic to Agent Memory Filtering (filtering out blank or error-filled returns before they bloat the context window), Prompt Gating (intercepting low-context user inputs before paying for LLM inference), Dataset Curation (automatically dropping low-substance artifacts from raw text before fine-tuning a model), or Web Crawling (discarding boilerplate navigation menus and footer text during live extraction).

Complete Working Example - Web Crawling

This is a fully runnable script that scrapes a real-world essay by Paul Graham, splits it into uniform 150-character chunks, and filters out the low-substance noise. You can copy and paste the code, add your API key, and run this locally to see how the API handles live data.

Python 
# Import the Libraries
import requests
from bs4 import BeautifulSoup
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document, BaseDocumentTransformer
from indieml import Substance

# Configure the Substance Filter
class IndieMLFilter(BaseDocumentTransformer):
    """Filters LangChain documents based on IndieML substance scores."""
    
    def __init__(self, api_key: str, threshold: float = 0.5):
        self.api = Substance(api_key=api_key)
        self.threshold = threshold
        
    def transform_documents(self, documents: list[Document], **kwargs) -> list[Document]:
        texts = [doc.page_content for doc in documents]
        
        batch_results = self.api.score_batch(texts)
        
        clean_docs = []
        for doc, result in zip(documents, batch_results):
            score = result.get("substance_score", 0.0)
            
            if score >= self.threshold:
                doc.metadata["indieml_score"] = score
                clean_docs.append(doc)
                
        return clean_docs

# Retrieve and Process the Data
response = requests.get(
    "https://paulgraham.com/startupideas.html", 
    headers={"User-Agent": "IndieML-TestBot/1.0"}
)
soup = BeautifulSoup(response.content, "html.parser")
text_content = soup.get_text(separator=' ', strip=True)

# Split into chunks
raw_document = Document(page_content=text_content, metadata={"source": "paulgraham.com"})
text_splitter = RecursiveCharacterTextSplitter(chunk_size=150, chunk_overlap=15)
split_docs = text_splitter.split_documents([raw_document])

print(f"Total chunks created: {len(split_docs)}")

# Score and filter
print("Sending batch to live IndieML API...")
indieml = IndieMLFilter(api_key="YOUR_API_KEY", threshold=0.5)
clean_docs = indieml.transform_documents(split_docs)

print(f"\nFinal Result: Kept {len(clean_docs)} out of {len(split_docs)} chunks.")