Introduction to LangChain
LangChain is revolutionizing how developers build AI-powered applications. It provides a framework for creating applications that leverage Large Language Models (LLMs) with advanced capabilities like memory, reasoning, and tool integration.
This comprehensive guide will take you from LangChain basics to building production-ready AI applications.
What is LangChain?
LangChain is a framework designed to simplify the creation of applications using large language models. It provides:
- Chains: Sequences of calls to LLMs or other utilities
- Agents: LLMs that make decisions about actions
- Memory: Persistence between chain/agent calls
- Indexes: Ways to structure documents for LLM interaction
- Tools: Interfaces for LLMs to interact with external systems
Installation and Setup
Python Installation
pip install langchain langchain-openai langchain-community
# For specific integrations
pip install chromadb tiktoken python-dotenv
JavaScript/TypeScript Installation
npm install langchain @langchain/openai @langchain/community
# or
yarn add langchain @langchain/openai @langchain/community
Environment Setup
# .env file
OPENAI_API_KEY=your_openai_api_key_here
ANTHROPIC_API_KEY=your_anthropic_key_here
Basic LangChain Concepts
1. Simple LLM Call (Python)
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
# Initialize the model
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
# Create messages
messages = [
SystemMessage(content="You are a helpful AI assistant specialized in Python programming."),
HumanMessage(content="Explain list comprehensions with an example.")
]
# Get response
response = llm.invoke(messages)
print(response.content)
2. Prompt Templates
from langchain.prompts import ChatPromptTemplate, PromptTemplate
# Simple template
template = PromptTemplate(
input_variables=["language", "topic"],
template="Write a {language} function that {topic}"
)
prompt = template.format(language="Python", topic="sorts a list of integers")
print(prompt)
# Chat template
chat_template = ChatPromptTemplate.from_messages([
("system", "You are an expert {role}."),
("user", "Explain {concept} in simple terms.")
])
messages = chat_template.format_messages(
role="data scientist",
concept="gradient descent"
)
3. Chains: Combining Components
from langchain.chains import LLMChain
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(temperature=0.7)
# Create a chain
chain = LLMChain(llm=llm, prompt=template)
# Run the chain
result = chain.run(language="JavaScript", topic="validates email addresses")
print(result)
Advanced LangChain Features
Memory: Making Conversations Contextual
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
# Initialize memory
memory = ConversationBufferMemory()
# Create conversation chain
conversation = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)
# Have a conversation
response1 = conversation.predict(input="Hi, my name is Alice")
print(response1)
response2 = conversation.predict(input="What's my name?")
print(response2) # Will remember Alice
# Different memory types
from langchain.memory import ConversationSummaryMemory, ConversationBufferWindowMemory
# Summary memory - keeps a running summary
summary_memory = ConversationSummaryMemory(llm=llm)
# Window memory - keeps last N messages
window_memory = ConversationBufferWindowMemory(k=5)
Document Loading and Processing
from langchain.document_loaders import TextLoader, PyPDFLoader, WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Load documents
loader = PyPDFLoader("document.pdf")
documents = loader.load()
# Split into chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len
)
texts = text_splitter.split_documents(documents)
print(f"Split into {len(texts)} chunks")
Vector Stores and Embeddings
from langchain_openai import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
# Create embeddings
embeddings = OpenAIEmbeddings()
# Create vector store
vectorstore = Chroma.from_documents(
documents=texts,
embedding=embeddings,
persist_directory="./chroma_db"
)
# Create retrieval chain
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)
# Ask questions about your documents
query = "What are the main points discussed in the document?"
result = qa_chain.run(query)
print(result)
Building an AI Agent
from langchain.agents import load_tools, initialize_agent, AgentType
from langchain_openai import ChatOpenAI
# Initialize LLM
llm = ChatOpenAI(temperature=0)
# Load tools
tools = load_tools(["serpapi", "llm-math"], llm=llm)
# Initialize agent
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
# Run agent
response = agent.run(
"What was the average temperature in New York City last week? "
"Calculate the difference from the historical average."
)
print(response)
Custom Tools
from langchain.tools import Tool, tool
from langchain.agents import AgentExecutor
@tool
def search_database(query: str) -> str:
"""Search the company database for information."""
# Your database search logic here
return f"Results for: {query}"
@tool
def calculate_metrics(data: str) -> str:
"""Calculate business metrics from data."""
# Your calculation logic here
return "Calculated metrics: ..."
# Create agent with custom tools
tools = [search_database, calculate_metrics]
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
Production-Ready RAG Application
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
class RAGApplication:
def __init__(self, persist_directory="./vectorstore"):
self.llm = ChatOpenAI(model="gpt-4", temperature=0)
self.embeddings = OpenAIEmbeddings()
self.vectorstore = None
self.chain = None
self.persist_directory = persist_directory
def load_documents(self, documents):
"""Load and index documents."""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
texts = text_splitter.split_documents(documents)
self.vectorstore = Chroma.from_documents(
documents=texts,
embedding=self.embeddings,
persist_directory=self.persist_directory
)
# Create conversational chain
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
self.chain = ConversationalRetrievalChain.from_llm(
llm=self.llm,
retriever=self.vectorstore.as_retriever(search_kwargs={"k": 5}),
memory=memory,
verbose=True
)
def query(self, question):
"""Query the RAG system."""
if not self.chain:
raise ValueError("Please load documents first")
result = self.chain({"question": question})
return result["answer"]
# Usage
app = RAGApplication()
app.load_documents(documents)
answer = app.query("What is the main topic of these documents?")
LangChain with Streaming
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain_openai import ChatOpenAI
# Initialize with streaming
llm = ChatOpenAI(
streaming=True,
callbacks=[StreamingStdOutCallbackHandler()],
temperature=0.7
)
# Stream response
response = llm.invoke("Write a short story about a robot learning to cook")
# Custom streaming callback
from langchain.callbacks.base import BaseCallbackHandler
class CustomStreamingCallback(BaseCallbackHandler):
def on_llm_new_token(self, token: str, **kwargs) -> None:
print(f"Token: {token}", end="", flush=True)
llm_with_custom_callback = ChatOpenAI(
streaming=True,
callbacks=[CustomStreamingCallback()]
)
Error Handling and Best Practices
from langchain.callbacks import get_openai_callback
import logging
# Track token usage
with get_openai_callback() as cb:
result = chain.run(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}")
# Error handling
try:
result = chain.run(query)
except Exception as e:
logging.error(f"Error in LangChain execution: {e}")
# Implement fallback logic
Testing LangChain Applications
import pytest
from unittest.mock import Mock, patch
def test_chain_execution():
# Mock LLM responses
mock_llm = Mock()
mock_llm.invoke.return_value = "Mocked response"
chain = LLMChain(llm=mock_llm, prompt=template)
result = chain.run(language="Python", topic="test")
assert result == "Mocked response"
mock_llm.invoke.assert_called_once()
# Integration testing with actual API (use carefully)
@pytest.mark.integration
def test_real_llm_call():
llm = ChatOpenAI(model="gpt-3.5-turbo")
response = llm.invoke("Say 'test passed'")
assert "test passed" in response.content.lower()
Performance Optimization
1. Caching
from langchain.cache import InMemoryCache, SQLiteCache
import langchain
# In-memory cache
langchain.llm_cache = InMemoryCache()
# Persistent cache
langchain.llm_cache = SQLiteCache(database_path=".langchain.db")
2. Batch Processing
# Batch API calls for efficiency
inputs = [
{"language": "Python", "topic": "sorting"},
{"language": "JavaScript", "topic": "async"},
{"language": "Go", "topic": "concurrency"}
]
results = chain.batch(inputs)
Deployment Considerations
- Environment Variables: Use secure secret management
- Rate Limiting: Implement backoff strategies
- Monitoring: Track token usage and costs
- Error Recovery: Implement retry logic with exponential backoff
- Vector Store: Use production-ready databases (Pinecone, Weaviate)
Conclusion
LangChain empowers developers to build sophisticated AI applications with ease. From simple chatbots to complex RAG systems and autonomous agents, LangChain provides the tools you need. Start with basic chains, experiment with agents, and scale to production-ready applications.
The AI development landscape is evolving rapidly—LangChain keeps you at the forefront.