LangChain Integration

Integrate FLTR with LangChain to build powerful RAG (Retrieval-Augmented Generation) applications using LangChain’s framework.

Installation

pip install langchain langchain-openai fltr-python

Quick Start

Basic Retriever

from langchain.schema import Document
from langchain.schema.retriever import BaseRetriever
from typing import List
from fltr import FLTR

class FLTRRetriever(BaseRetriever):
    """FLTR retriever for LangChain."""

    fltr_client: FLTR
    dataset_id: str
    k: int = 4  # Number of documents to retrieve

    def _get_relevant_documents(self, query: str) -> List[Document]:
        """Retrieve documents from FLTR."""
        results = self.fltr_client.query(
            dataset_id=self.dataset_id,
            query=query,
            limit=self.k
        )

        # Convert FLTR chunks to LangChain Documents
        documents = []
        for chunk in results['chunks']:
            doc = Document(
                page_content=chunk['text'],
                metadata={
                    'score': chunk['score'],
                    'document_id': chunk.get('document_id'),
                    **chunk.get('metadata', {})
                }
            )
            documents.append(doc)

        return documents

    async def _aget_relevant_documents(self, query: str) -> List[Document]:
        """Async version."""
        # Use sync for now, implement async later
        return self._get_relevant_documents(query)


# Usage
fltr_client = FLTR(api_key="fltr_sk_...")
retriever = FLTRRetriever(
    fltr_client=fltr_client,
    dataset_id="ds_abc123",
    k=5
)

docs = retriever.get_relevant_documents("How do I reset my password?")
for doc in docs:
    print(f"Score: {doc.metadata['score']:.3f}")
    print(f"Content: {doc.page_content[:100]}...\n")

RetrievalQA Chain

Build a complete question-answering system.

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate

# Initialize components
fltr_client = FLTR(api_key="fltr_sk_...")
retriever = FLTRRetriever(
    fltr_client=fltr_client,
    dataset_id="ds_abc123",
    k=5
)

llm = ChatOpenAI(
    model="gpt-4-turbo",
    temperature=0.3,
    openai_api_key="sk-..."
)

# Custom prompt template
prompt_template = """Use the following pieces of context to answer the question at the end.
If you don't know the answer, say so. Don't make up an answer.

Context:
{context}

Question: {question}

Answer: Let me help you with that."""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# Create QA chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # Stuff all context into prompt
    retriever=retriever,
    chain_type_kwargs={"prompt": PROMPT},
    return_source_documents=True
)

# Ask questions
result = qa_chain({"query": "How do I reset my password?"})

print("Answer:", result['result'])
print("\nSources:")
for i, doc in enumerate(result['source_documents'], 1):
    print(f"{i}. Score: {doc.metadata['score']:.3f}")
    print(f"   {doc.page_content[:100]}...\n")

ConversationalRetrievalChain

Add conversation memory for multi-turn interactions.

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

# Create memory
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="answer"
)

# Create conversational chain
conversation_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=retriever,
    memory=memory,
    return_source_documents=True,
    verbose=True
)

# First question
response1 = conversation_chain({"question": "What are your pricing plans?"})
print("Answer 1:", response1['answer'])

# Follow-up question (uses memory)
response2 = conversation_chain({"question": "Which one is best for startups?"})
print("Answer 2:", response2['answer'])

# Check conversation history
print("\nChat History:")
for msg in memory.chat_memory.messages:
    print(f"{msg.type}: {msg.content}")

Advanced Retriever

Add reranking, filtering, and score thresholds.

class AdvancedFLTRRetriever(BaseRetriever):
    """Advanced FLTR retriever with reranking and filtering."""

    fltr_client: FLTR
    dataset_id: str
    k: int = 10
    rerank: bool = True
    score_threshold: float = 0.5
    filters: dict = None

    def _get_relevant_documents(self, query: str) -> List[Document]:
        """Retrieve with advanced options."""
        results = self.fltr_client.query(
            dataset_id=self.dataset_id,
            query=query,
            limit=self.k,
            rerank=self.rerank,
            filters=self.filters
        )

        # Filter by score threshold
        documents = []
        for chunk in results['chunks']:
            if chunk['score'] >= self.score_threshold:
                doc = Document(
                    page_content=chunk['text'],
                    metadata={
                        'score': chunk['score'],
                        'document_id': chunk.get('document_id'),
                        'chunk_id': chunk.get('id'),
                        **chunk.get('metadata', {})
                    }
                )
                documents.append(doc)

        return documents

    async def _aget_relevant_documents(self, query: str) -> List[Document]:
        return self._get_relevant_documents(query)


# Usage with filtering
retriever = AdvancedFLTRRetriever(
    fltr_client=fltr_client,
    dataset_id="ds_abc123",
    k=10,
    rerank=True,
    score_threshold=0.7,
    filters={
        "metadata.category": "technical-docs",
        "metadata.version": "v2"
    }
)

Multi-Query Retriever

Generate multiple query variations for better recall.

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI

# Base retriever
fltr_retriever = FLTRRetriever(
    fltr_client=fltr_client,
    dataset_id="ds_abc123",
    k=3
)

# LLM for query generation
llm = ChatOpenAI(temperature=0, model="gpt-4")

# Multi-query retriever
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=fltr_retriever,
    llm=llm
)

# This will generate variations and retrieve for each
docs = multi_query_retriever.get_relevant_documents(
    "How do I upload documents?"
)

# Deduplicates and returns combined results
print(f"Retrieved {len(docs)} unique documents")

Ensemble Retriever

Combine FLTR with other retrievers.

from langchain.retrievers import EnsembleRetriever, BM25Retriever
from langchain.schema import Document

# FLTR retriever (semantic)
fltr_retriever = FLTRRetriever(
    fltr_client=fltr_client,
    dataset_id="ds_abc123",
    k=5
)

# Load local documents for BM25
local_docs = [
    Document(page_content="Document 1 content..."),
    Document(page_content="Document 2 content..."),
]

# BM25 retriever (keyword-based)
bm25_retriever = BM25Retriever.from_documents(local_docs)
bm25_retriever.k = 5

# Combine retrievers
ensemble_retriever = EnsembleRetriever(
    retrievers=[fltr_retriever, bm25_retriever],
    weights=[0.7, 0.3]  # 70% FLTR, 30% BM25
)

# Retrieve from both
docs = ensemble_retriever.get_relevant_documents("query")

Agent with Tools

Use FLTR as a tool in a LangChain agent.

from langchain.agents import initialize_agent, Tool, AgentType
from langchain_openai import ChatOpenAI

def search_docs(query: str) -> str:
    """Search documentation."""
    retriever = FLTRRetriever(
        fltr_client=fltr_client,
        dataset_id="ds_abc123",
        k=3
    )

    docs = retriever.get_relevant_documents(query)

    if not docs:
        return "No relevant documentation found."

    # Format results
    results = []
    for i, doc in enumerate(docs, 1):
        results.append(f"[{i}] {doc.page_content[:200]}...")

    return "\n\n".join(results)

# Create tool
tools = [
    Tool(
        name="SearchDocs",
        func=search_docs,
        description="Search the documentation for answers. Use this when users ask about features, how-tos, or troubleshooting."
    )
]

# Initialize agent
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.OPENAI_FUNCTIONS,
    verbose=True
)

# Run agent
response = agent.run("How do I reset my password?")
print(response)

Streaming Responses

Stream answers token-by-token.

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

# LLM with streaming
llm = ChatOpenAI(
    model="gpt-4-turbo",
    temperature=0.3,
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

# QA chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# This will stream output to stdout
result = qa_chain({"query": "How do I upload a document?"})

Custom Chain

Build a custom chain with citations.

from langchain.chains.base import Chain
from typing import Dict, Any

class FLTRQAChain(Chain):
    """Custom QA chain with FLTR citations."""

    fltr_client: FLTR
    dataset_id: str
    llm: ChatOpenAI

    @property
    def input_keys(self) -> List[str]:
        return ["question"]

    @property
    def output_keys(self) -> List[str]:
        return ["answer", "sources"]

    def _call(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        question = inputs["question"]

        # Retrieve from FLTR
        results = self.fltr_client.query(
            dataset_id=self.dataset_id,
            query=question,
            limit=5,
            rerank=True
        )

        # Build context
        context_parts = []
        for i, chunk in enumerate(results['chunks'], 1):
            context_parts.append(f"[{i}] {chunk['text']}")

        context = "\n\n".join(context_parts)

        # Generate answer
        prompt = f"""Answer the question based on the context below.
Cite sources using [N] notation.

Context:
{context}

Question: {question}

Answer:"""

        response = self.llm.predict(prompt)

        return {
            "answer": response,
            "sources": results['chunks']
        }

# Usage
chain = FLTRQAChain(
    fltr_client=fltr_client,
    dataset_id="ds_abc123",
    llm=ChatOpenAI(model="gpt-4-turbo", temperature=0.3)
)

result = chain({"question": "What are the pricing plans?"})
print("Answer:", result['answer'])
print("\nSources:", len(result['sources']))

Expression Language (LCEL)

Use LangChain Expression Language for composable chains.

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

# Setup retriever
retriever = FLTRRetriever(
    fltr_client=fltr_client,
    dataset_id="ds_abc123",
    k=5
)

# Prompt template
template = """Answer the question based only on the following context:

{context}

Question: {question}
"""

prompt = ChatPromptTemplate.from_template(template)

# LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.3)

# Build chain with LCEL
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# Invoke chain
answer = rag_chain.invoke("How do I reset my password?")
print(answer)

LangSmith Integration

Monitor and debug with LangSmith.

import os

# Set LangSmith API key
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls_..."
os.environ["LANGCHAIN_PROJECT"] = "fltr-qa"

# Create chain (automatically traced)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=retriever,
    return_source_documents=True
)

# This run will be logged to LangSmith
result = qa_chain({"query": "How do I upload documents?"})

# View in LangSmith dashboard:
# - Retrieval latency
# - Number of chunks retrieved
# - LLM tokens used
# - Full trace of the chain

Best Practices

1. Adjust k Based on Use Case

# High precision (fewer, better results)
retriever_precise = FLTRRetriever(k=3, rerank=True)

# High recall (more results, might include noise)
retriever_recall = FLTRRetriever(k=20, rerank=True)

2. Use Score Thresholds

class ThresholdRetriever(BaseRetriever):
    def _get_relevant_documents(self, query: str) -> List[Document]:
        results = self.fltr_client.query(...)

        # Only return high-quality results
        return [
            Document(page_content=c['text'], metadata={'score': c['score']})
            for c in results['chunks']
            if c['score'] >= 0.7  # Adjust threshold
        ]

3. Add Metadata Filtering

# Filter by document type
retriever = AdvancedFLTRRetriever(
    filters={"metadata.type": "user-guide"}
)

# Filter by date range
retriever = AdvancedFLTRRetriever(
    filters={
        "metadata.published_date": {
            "$gte": "2024-01-01",
            "$lte": "2024-12-31"
        }
    }
)

4. Cache Retrieval Results

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

# Cache LLM calls
set_llm_cache(InMemoryCache())

# LLM calls with same input will be cached

Error Handling

from langchain.schema import Document

class RobustFLTRRetriever(BaseRetriever):
    def _get_relevant_documents(self, query: str) -> List[Document]:
        try:
            results = self.fltr_client.query(
                dataset_id=self.dataset_id,
                query=query,
                limit=self.k
            )

            if not results.get('chunks'):
                # Return empty list if no results
                return []

            return [
                Document(
                    page_content=chunk['text'],
                    metadata=chunk.get('metadata', {})
                )
                for chunk in results['chunks']
            ]

        except Exception as e:
            print(f"Error retrieving documents: {e}")
            # Return empty list on error
            return []

Additional Resources

RAG Best Practices

Production RAG patterns

MCP API

FLTR query endpoint

First Integration

Build your first RAG app

Authentication

Secure your integration

Questions?

Need help with LangChain integration? Contact support. Support: support@fltr.com LangChain Docs: https://python.langchain.com/docs/

Getting Started

Authentication

Integrations

Support

LangChain Integration

LangChain Integration

Installation

Quick Start

Basic Retriever

RetrievalQA Chain

ConversationalRetrievalChain

Advanced Retriever

Multi-Query Retriever

Ensemble Retriever

Agent with Tools

Streaming Responses

Custom Chain

Expression Language (LCEL)

LangSmith Integration

Best Practices

1. Adjust k Based on Use Case

2. Use Score Thresholds

3. Add Metadata Filtering

4. Cache Retrieval Results

Error Handling

Additional Resources

RAG Best Practices

MCP API

First Integration

Authentication

Questions?

Getting Started

Authentication

Integrations

Support

​LangChain Integration

​Installation

​Quick Start

​Basic Retriever

​RetrievalQA Chain

​ConversationalRetrievalChain

​Advanced Retriever

​Multi-Query Retriever

​Ensemble Retriever

​Agent with Tools

​Streaming Responses

​Custom Chain

​Expression Language (LCEL)

​LangSmith Integration

​Best Practices

​1. Adjust k Based on Use Case

​2. Use Score Thresholds

​3. Add Metadata Filtering

​4. Cache Retrieval Results

​Error Handling

​Additional Resources

RAG Best Practices

MCP API

First Integration

Authentication

​Questions?

LangChain Integration

Installation

Quick Start

Basic Retriever

RetrievalQA Chain

ConversationalRetrievalChain

Advanced Retriever

Multi-Query Retriever

Ensemble Retriever

Agent with Tools

Streaming Responses

Custom Chain

Expression Language (LCEL)

LangSmith Integration

Best Practices

1. Adjust k Based on Use Case

2. Use Score Thresholds

3. Add Metadata Filtering

4. Cache Retrieval Results

Error Handling

Additional Resources

Questions?