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