LangChain has emerged as the leading framework for building applications powered by Large Language Models (LLMs). Whether you're creating chatbots, document analysis systems, or complex AI agents, LangChain provides the tools and abstractions you need to build production-ready applications.
What is LangChain?
LangChain is a framework for developing applications powered by language models. It provides a standardized interface for chains, agents, memory systems, and integrations with various LLM providers and data sources.
Core Philosophy
LangChain is built on several key principles:
- Composability: Build complex applications from simple, reusable components
- Standardization: Consistent interfaces across different LLMs and tools
- Observability: Built-in monitoring and debugging capabilities
- Production-Ready: Tools and patterns for deploying at scale
Core Components of LangChain
1. Models and Prompts
LangChain supports multiple LLM providers with a unified interface:
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain.prompts import ChatPromptTemplate
# Initialize models
openai_model = ChatOpenAI(model="gpt-4")
anthropic_model = ChatAnthropic(model="claude-3-sonnet-20240229")
# Create reusable prompt templates
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful AI assistant specialized in {domain}."),
("human", "{input}")
])
# Use with any model
chain = prompt | openai_model
response = chain.invoke({
"domain": "software engineering",
"input": "Explain microservices architecture"
})
2. Chains - Composing LLM Operations
Chains allow you to combine multiple operations into a single workflow:
from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import PromptTemplate
# Simple chain
summary_template = PromptTemplate(
input_variables=["text"],
template="Summarize the following text:\n\n{text}\n\nSummary:"
)
summary_chain = LLMChain(llm=openai_model, prompt=summary_template)
# Sequential chain - multi-step processing
translate_template = PromptTemplate(
input_variables=["summary"],
template="Translate the following to Spanish:\n\n{summary}"
)
translate_chain = LLMChain(llm=openai_model, prompt=translate_template)
combined_chain = SequentialChain(
chains=[summary_chain, translate_chain],
input_variables=["text"],
output_variables=["summary", "translation"]
)
result = combined_chain.invoke({
"text": "Long article about AI..."
})
3. Agents - Dynamic Action Selection
Agents can use tools and make decisions about which actions to take:
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool
from langchain import hub
# Define tools
def search_database(query: str) -> str:
"""Search the internal database"""
# Implementation
return f"Database results for: {query}"
def calculate(expression: str) -> str:
"""Perform mathematical calculations"""
try:
return str(eval(expression))
except:
return "Invalid expression"
tools = [
Tool(
name="DatabaseSearch",
func=search_database,
description="Search the internal database for information"
),
Tool(
name="Calculator",
func=calculate,
description="Perform mathematical calculations"
)
]
# Create agent
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(openai_model, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5
)
# Use agent
response = agent_executor.invoke({
"input": "Find the revenue for Q4 and calculate the growth percentage"
})
4. Memory Management
LangChain provides multiple memory types for maintaining conversation context:
from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain.chains import ConversationChain
# Buffer memory - stores all messages
buffer_memory = ConversationBufferMemory()
# Summary memory - summarizes old conversations
summary_memory = ConversationSummaryMemory(llm=openai_model)
# Conversation chain with memory
conversation = ConversationChain(
llm=openai_model,
memory=buffer_memory,
verbose=True
)
conversation.invoke({"input": "Hi, I'm working on a Python project"})
conversation.invoke({"input": "What was I just talking about?"})
# Memory maintains context across calls
Advanced LangChain Patterns
Document Loading and Processing
from langchain_community.document_loaders import (
DirectoryLoader,
PDFLoader,
UnstructuredMarkdownLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
# Load documents
loader = DirectoryLoader(
'./docs',
glob="**/*.md",
loader_cls=UnstructuredMarkdownLoader
)
documents = loader.load()
# Split into chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len
)
chunks = text_splitter.split_documents(documents)
# Create vector store
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db"
)
Retrieval-Augmented Generation (RAG)
from langchain.chains import RetrievalQA
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
# Basic RAG
qa_chain = RetrievalQA.from_chain_type(
llm=openai_model,
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 4})
)
# Advanced RAG with compression
compressor = LLMChainExtractor.from_llm(openai_model)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=vectorstore.as_retriever()
)
compressed_qa = RetrievalQA.from_chain_type(
llm=openai_model,
retriever=compression_retriever
)
answer = compressed_qa.invoke({
"query": "What are the main features of the product?"
})
Custom Chains for Complex Workflows
from langchain.chains.base import Chain
from typing import Dict, Any
class CustomAnalysisChain(Chain):
"""Custom chain for document analysis"""
llm: Any
vectorstore: Any
@property
def input_keys(self) -> list[str]:
return ["document", "analysis_type"]
@property
def output_keys(self) -> list[str]:
return ["analysis", "recommendations"]
def _call(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
document = inputs["document"]
analysis_type = inputs["analysis_type"]
# Custom processing logic
relevant_docs = self.vectorstore.similarity_search(
document,
k=3
)
# Perform analysis
analysis_prompt = f"""
Analyze the following document for {analysis_type}:
{document}
Context from similar documents:
{relevant_docs}
"""
analysis = self.llm.invoke(analysis_prompt)
# Generate recommendations
rec_prompt = f"""
Based on this analysis, provide recommendations:
{analysis}
"""
recommendations = self.llm.invoke(rec_prompt)
return {
"analysis": analysis,
"recommendations": recommendations
}
# Use custom chain
custom_chain = CustomAnalysisChain(
llm=openai_model,
vectorstore=vectorstore
)
result = custom_chain.invoke({
"document": "Product specification...",
"analysis_type": "security vulnerabilities"
})
LangChain Expression Language (LCEL)
LCEL provides a declarative way to compose chains:
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
# Simple LCEL chain
chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| openai_model
| StrOutputParser()
)
result = chain.invoke("What is the pricing model?")
# Parallel execution
from langchain_core.runnables import RunnableParallel
chain = RunnableParallel({
"summary": summary_chain,
"sentiment": sentiment_chain,
"keywords": keyword_chain
})
results = chain.invoke({"text": "Article content..."})
Production Deployment Strategies
1. Caching for Performance
from langchain.cache import InMemoryCache, RedisCache
from langchain.globals import set_llm_cache
import redis
# In-memory cache for development
set_llm_cache(InMemoryCache())
# Redis cache for production
redis_client = redis.Redis(host='localhost', port=6379)
set_llm_cache(RedisCache(redis_client))
# Cached calls are much faster
llm.invoke("What is AI?") # Slow - makes API call
llm.invoke("What is AI?") # Fast - returns cached result
2. Error Handling and Retries
from langchain.chains import LLMChain
from tenacity import retry, stop_after_attempt, wait_exponential
class RobustLLMChain:
def __init__(self, chain: LLMChain):
self.chain = chain
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10)
)
def invoke_with_retry(self, inputs: dict) -> str:
try:
return self.chain.invoke(inputs)
except Exception as e:
print(f"Error occurred: {e}")
raise
robust_chain = RobustLLMChain(summary_chain)
result = robust_chain.invoke_with_retry({"text": "Content..."})
3. Monitoring and Observability
from langchain.callbacks import StdOutCallbackHandler
from langsmith import Client
# LangSmith integration for monitoring
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "production-app"
# Custom callback for logging
class MetricsCallback(StdOutCallbackHandler):
def on_llm_start(self, *args, **kwargs):
# Log start time, tokens, etc.
pass
def on_llm_end(self, *args, **kwargs):
# Log completion, calculate costs
pass
chain = prompt | openai_model
result = chain.invoke(
{"input": "Query"},
config={"callbacks": [MetricsCallback()]}
)
4. Streaming Responses
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
# Streaming for real-time output
streaming_llm = ChatOpenAI(
model="gpt-4",
streaming=True,
callbacks=[StreamingStdOutCallbackHandler()]
)
chain = prompt | streaming_llm
# Streams output token by token
for chunk in chain.stream({"input": "Write a long essay"}):
print(chunk.content, end="", flush=True)
Best Practices
1. Prompt Engineering with LangChain
from langchain.prompts import FewShotPromptTemplate, PromptTemplate
# Few-shot learning
examples = [
{"input": "happy", "output": "sad"},
{"input": "tall", "output": "short"},
]
example_prompt = PromptTemplate(
input_variables=["input", "output"],
template="Input: {input}\nOutput: {output}"
)
few_shot_prompt = FewShotPromptTemplate(
examples=examples,
example_prompt=example_prompt,
prefix="Give the opposite of each word:",
suffix="Input: {adjective}\nOutput:",
input_variables=["adjective"]
)
chain = few_shot_prompt | openai_model
2. Cost Management
from langchain.callbacks import get_openai_callback
with get_openai_callback() as cb:
result = chain.invoke({"input": "Query"})
print(f"Total Tokens: {cb.total_tokens}")
print(f"Prompt Tokens: {cb.prompt_tokens}")
print(f"Completion Tokens: {cb.completion_tokens}")
print(f"Total Cost (USD): ${cb.total_cost}")
3. Testing LangChain Applications
import pytest
from langchain.llms.fake import FakeListLLM
def test_chain():
# Use fake LLM for testing
fake_llm = FakeListLLM(
responses=["Expected response 1", "Expected response 2"]
)
chain = prompt | fake_llm
result = chain.invoke({"input": "test"})
assert "Expected" in result
Real-World Use Cases
Customer Support Automation
support_chain = (
{
"context": support_docs_retriever,
"history": lambda x: x["conversation_history"],
"question": lambda x: x["question"]
}
| support_prompt
| openai_model
| StrOutputParser()
)
Document Analysis Pipeline
analysis_pipeline = (
document_loader
| text_splitter
| embeddings
| vectorstore
| retrieval_qa
)
Code Generation Assistant
code_chain = (
{"language": RunnablePassthrough(), "task": RunnablePassthrough()}
| code_prompt
| openai_model
| code_parser
| syntax_validator
)
Conclusion
LangChain provides a comprehensive framework for building LLM applications, from simple chains to complex multi-agent systems. By leveraging its composable architecture, you can:
- Build production-ready applications faster
- Maintain consistency across different LLM providers
- Implement complex workflows with reusable components
- Monitor and optimize performance at scale
Whether you're building chatbots, document analysis systems, or intelligent agents, LangChain provides the tools and patterns you need to succeed.
The framework continues to evolve with new features, better abstractions, and improved performance. Stay engaged with the community, experiment with new patterns, and build the next generation of AI-powered applications.
Ready to build with LangChain? Start with simple chains, experiment with agents, and gradually build more complex systems. The journey to mastering LangChain is iterative - learn, build, and iterate.