Introduction: What is LangChain?
LangChain is an open-source framework designed to simplify the development of applications powered by large language models (LLMs). Rather than working directly with raw API calls and manual prompt engineering, LangChain provides a structured, modular approach that lets you chain together different components—prompts, models, memory, tools, and agents—into coherent, production-ready pipelines.
At its core, LangChain solves a fundamental problem: LLMs are incredibly powerful but unpredictable. They hallucinate, forget context, and struggle with structured reasoning. LangChain wraps these models in a layer of software engineering best practices, making them more reliable, composable, and easier to integrate into real-world applications.
Why LangChain Matters
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →Before LangChain, building an LLM application meant writing hundreds of lines of boilerplate code to handle prompt formatting, output parsing, conversation memory, and tool integration. Every developer was reinventing the wheel. LangChain standardizes these patterns. Here's why it matters:
- Composability – Every component (prompts, models, retrievers, tools) is treated as a modular building block that can be chained, swapped, and reused.
- Memory Management – Built-in memory classes handle conversation history, summarization, and context window limits automatically.
- Tool Integration – LLMs can call external APIs, search databases, run code, or scrape the web through a unified tool interface.
- Retrieval Augmented Generation (RAG) – First-class support for connecting LLMs to vector stores and document retrievers, grounding responses in real data.
- Agents and Reasoning – The framework enables autonomous agents that can plan, use tools, and iterate toward a goal.
Core Concepts: The Building Blocks
Before diving into code, let's understand the fundamental abstractions in LangChain:
1. Models (LLMs and Chat Models)
The model interface wraps any LLM provider—OpenAI, Anthropic, Cohere, or local models—behind a consistent API. You can swap models without changing the rest of your pipeline.
2. Prompts and Prompt Templates
Prompt templates let you define reusable, parameterized prompts. They handle variable injection, formatting, and can include few-shot examples, system messages, and role-based structures for chat models.
3. Chains
A chain is a sequence of operations—prompt formatting, model invocation, output parsing, and post-processing—bundled into a single runnable unit. The most basic is the LLMChain, which combines a prompt template with a model.
4. Memory
Memory classes store conversation state across multiple turns. They manage the context window, summarize older messages, and inject relevant history back into prompts.
5. Retrievers and Vector Stores
Retrievers query external knowledge sources (like vector databases) and return relevant documents. When combined with a chain, this creates RAG—answers grounded in your actual data, not the model's training data.
6. Agents
Agents use an LLM as a reasoning engine to decide which actions to take, which tools to call, and in what order. They observe results and iterate until they've satisfied the user's request.
Setting Up Your Environment
Let's start by installing LangChain and its dependencies. Create a fresh project directory and set up a virtual environment:
# Create and activate a virtual environment
python -m venv langchain_env
source langchain_env/bin/activate # On Windows: langchain_env\Scripts\activate
# Install LangChain core
pip install langchain langchain-community
# Install OpenAI integration (if using OpenAI models)
pip install langchain-openai
# For vector stores and embeddings
pip install chromadb sentence-transformers
# For environment variable management
pip install python-dotenvSet your API keys in a .env file:
# .env file
OPENAI_API_KEY=your-openai-api-key-hereYour First Chain: Prompt → Model → Output
Let's build the simplest possible LangChain application—a chain that takes a topic, formats a prompt, calls the model, and returns a response:
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
# Load environment variables
load_dotenv()
# 1. Define the model
model = ChatOpenAI(
model="gpt-3.5-turbo",
temperature=0.7,
max_tokens=200
)
# 2. Create a prompt template
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant that explains technical concepts clearly."),
("human", "Explain {topic} in 3 sentences suitable for a beginner.")
])
# 3. Create the output parser
output_parser = StrOutputParser()
# 4. Build the chain using the | (pipe) operator
chain = prompt | model | output_parser
# 5. Invoke the chain
result = chain.invoke({"topic": "vector databases"})
print(result)This example demonstrates LangChain's elegant pipe syntax (| operator). Each component passes its output to the next. The prompt formats the input variables, the model generates a response, and the parser extracts the string content. This composability is the heart of LangChain.
Adding Conversation Memory
LLMs are stateless—they don't remember previous turns. LangChain's memory classes solve this by storing conversation history and injecting it into prompts. Here's a conversational chain with memory:
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationSummaryMemory
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema.runnable import RunnablePassthrough, RunnableLambda
from langchain.schema.output_parser import StrOutputParser
from langchain_core.messages import HumanMessage, AIMessage
import os
from dotenv import load_dotenv
load_dotenv()
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
# Memory that summarizes older messages to save tokens
memory = ConversationSummaryMemory(
llm=model,
return_messages=True,
max_token_limit=400
)
# Prompt with a placeholder for conversation history
prompt = ChatPromptTemplate.from_messages([
("system", "You are a knowledgeable cooking assistant."),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
def get_history(_):
"""Retrieve conversation history from memory."""
return memory.load_memory_variables({})["history"]
# Build the chain with memory retrieval
chain = (
RunnablePassthrough.assign(history=get_history)
| prompt
| model
| StrOutputParser()
)
# First turn
response1 = chain.invoke({"input": "How do I make a perfect omelette?"})
print("Assistant:", response1)
# Store the exchange in memory
memory.save_context(
{"input": "How do I make a perfect omelette?"},
{"output": response1}
)
# Second turn — the model now remembers the context
response2 = chain.invoke({"input": "What temperature should the pan be for that?"})
print("Assistant:", response2)
# Store again
memory.save_context(
{"input": "What temperature should the pan be for that?"},
{"output": response2}
)
# Third turn — still remembers
response3 = chain.invoke({"input": "Can I add cheese without ruining it?"})
print("Assistant:", response3)The ConversationSummaryMemory class is particularly useful because it compresses older messages into a summary, preventing the context window from overflowing during long conversations. The MessagesPlaceholder in the prompt template dynamically inserts the conversation history at inference time.
Retrieval Augmented Generation (RAG)
RAG is one of the most powerful patterns in LangChain. It grounds the model's responses in your own documents, reducing hallucinations and enabling answers about proprietary information. Here's how to build a complete RAG pipeline:
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
from langchain_community.document_loaders import TextLoader
import os
from dotenv import load_dotenv
load_dotenv()
# 1. Load documents
loader = TextLoader("company_policies.txt")
documents = loader.load()
# 2. Split documents into chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\n\n", "\n", ". ", " ", ""]
)
chunks = text_splitter.split_documents(documents)
# 3. Create embeddings and store in vector database
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
collection_name="company_policies"
)
# 4. Create a retriever
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 4} # Retrieve top 4 most relevant chunks
)
# 5. Define the RAG prompt
rag_prompt = ChatPromptTemplate.from_messages([
("system", """You are an assistant that answers questions based ONLY on
the provided context. If the answer isn't in the context, say you don't know.
Context:
{context}"""),
("human", "{question}")
])
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 6. Build the RAG chain
def format_docs(docs):
"""Combine retrieved documents into a single context string."""
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{
"context": lambda x: format_docs(retriever.get_relevant_documents(x["question"])),
"question": lambda x: x["question"]
}
| rag_prompt
| model
| StrOutputParser()
)
# 7. Query the RAG system
query = "What is the company's remote work policy for employees living in different states?"
answer = rag_chain.invoke({"question": query})
print(answer)This pipeline does several things: it loads a document, splits it into overlapping chunks (to preserve context at boundaries), embeds each chunk, stores them in Chroma (a lightweight vector database), retrieves the most relevant chunks for a query, and feeds them as context into the prompt. The model now answers based on actual company policy, not its training data.
Building Autonomous Agents
Agents are the most sophisticated pattern in LangChain. Instead of following a fixed chain, an agent uses the LLM to decide what to do next—which tool to call, what arguments to pass, and when to stop. Here's an agent that can search the web and do math:
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import tool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.tools import DuckDuckGoSearchRun
import os
from dotenv import load_dotenv
load_dotenv()
# Define custom tools using the @tool decorator
@tool
def multiply(a: int, b: int) -> int:
"""Multiply two integers and return the result."""
return a * b
@tool
def reverse_string(text: str) -> str:
"""Reverse a string and return the reversed version."""
return text[::-1]
# Use a built-in tool for web search
search_tool = DuckDuckGoSearchRun()
# Combine all tools
tools = [multiply, reverse_string, search_tool]
# Create the model with tool-calling capability
model = ChatOpenAI(
model="gpt-4-turbo",
temperature=0,
model_kwargs={"tools": None} # Let LangChain handle tool binding
)
# Create the agent prompt
agent_prompt = ChatPromptTemplate.from_messages([
("system", """You are a helpful assistant with access to tools.
Use tools when appropriate. If a tool fails, try another approach.
Always explain your reasoning before using a tool."""),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
# Create the agent
agent = create_openai_tools_agent(
llm=model,
tools=tools,
prompt=agent_prompt
)
# Wrap the agent in an executor
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # Print the agent's reasoning steps
max_iterations=10,
handle_parsing_errors=True
)
# Run the agent
response = agent_executor.invoke({
"input": "What is the population of Tokyo multiplied by 3? Then reverse the name 'Tokyo'."
})
print("\nFinal Answer:", response["output"])When you run this, the agent will reason through the problem, call the search tool to find Tokyo's population, use the multiply tool on the result, then call the reverse_string tool. The verbose=True flag prints each step so you can see the agent's thought process. The agent_scratchpad stores intermediate tool call results for the model to reason over.
Custom Chains and Advanced Composition
Sometimes you need more control than the built-in chains provide. LangChain's Runnable interface lets you build custom pipelines with branching logic, parallel execution, and fallbacks. Here's a sophisticated example with routing and fallback:
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnableBranch, RunnablePassthrough
import os
from dotenv import load_dotenv
load_dotenv()
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# Different prompts for different question types
coding_prompt = ChatPromptTemplate.from_messages([
("system", "You are an expert programmer. Provide clean, well-commented code."),
("human", "{question}")
])
general_prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful general assistant."),
("human", "{question}")
])
# Router chain that classifies the input
router_prompt = ChatPromptTemplate.from_messages([
("system", "Classify the following question as 'coding' or 'general'. Respond with ONLY one word."),
("human", "{question}")
])
classify_chain = router_prompt | model | StrOutputParser()
# Branch based on classification
branch = RunnableBranch(
(lambda x: "coding" in x["classification"].lower(),
coding_prompt | model | StrOutputParser()),
(lambda x: "general" in x["classification"].lower(),
general_prompt | model | StrOutputParser()),
# Default fallback
general_prompt | model | StrOutputParser()
)
# Full pipeline: classify → route → respond
full_chain = (
RunnablePassthrough.assign(
classification=lambda x: classify_chain.invoke({"question": x["question"]})
)
| RunnablePassthrough.assign(
response=lambda x: branch.invoke(x)
)
)
# Test with different question types
result1 = full_chain.invoke({"question": "Write a Python function to reverse a linked list"})
print("Coding question response:", result1["response"][:200], "...\n")
result2 = full_chain.invoke({"question": "What is the capital of Bhutan?"})
print("General question response:", result2["response"])The RunnableBranch acts like a switch statement—it evaluates conditions in order and executes the first matching branch. The RunnablePassthrough.assign method adds new keys to the state dictionary without modifying existing ones, enabling a clean data flow through the pipeline.
Working with Structured Output
Raw text responses are often insufficient. You need structured data—JSON, typed dictionaries, or Pydantic models. LangChain provides output parsers that coerce model outputs into structured formats:
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List
import os
from dotenv import load_dotenv
load_dotenv()
# Define the structured output schema
class Person(BaseModel):
name: str = Field(description="The person's full name")
age: int = Field(description="Age in years")
profession: str = Field(description="Current job title")
skills: List[str] = Field(description="List of 3-5 key skills")
class AnalysisResponse(BaseModel):
summary: str = Field(description="One-paragraph summary of the person")
person: Person = Field(description="Structured information about the person")
confidence: float = Field(description="Confidence score between 0 and 1")
# Create parser from the Pydantic model
parser = PydanticOutputParser(pydantic_object=AnalysisResponse)
# Build prompt with formatting instructions injected
prompt = ChatPromptTemplate.from_messages([
("system", """You are a data extraction assistant.
Extract information from the text and format it according to the schema.
{format_instructions}"""),
("human", "{text}")
])
# Inject the parser's format instructions into the prompt
prompt = prompt.partial(format_instructions=parser.get_format_instructions())
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# Build the chain
chain = prompt | model | parser
# Invoke with some unstructured text
text = """Sarah Chen, a 34-year-old software architect at Google, has been
leading the cloud infrastructure team for 5 years. She's an expert in
distributed systems, Kubernetes, and has strong leadership skills."""
result = chain.invoke({"text": text})
# The result is a typed Pydantic object
print(f"Name: {result.person.name}")
print(f"Age: {result.person.age}")
print(f"Profession: {result.person.profession}")
print(f"Skills: {', '.join(result.person.skills)}")
print(f"Confidence: {result.confidence}")
print(f"\nSummary: {result.summary}")
The PydanticOutputParser automatically generates formatting instructions that tell the model exactly how to structure its JSON output. The parser then validates and parses the response into a typed Pydantic object. If the model produces malformed JSON, the parser catches it and can trigger retry logic.
Streaming and Async Operations
For production applications, you'll want streaming responses (to show output token-by-token) and async support (for concurrent requests). LangChain handles both elegantly:
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
import asyncio
import os
from dotenv import load_dotenv
load_dotenv()
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7, streaming=True)
prompt = ChatPromptTemplate.from_messages([
("system", "You are a creative storyteller."),
("human", "Write a short story about {topic}")
])
# Synchronous streaming
print("=== Synchronous Streaming ===")
chain = prompt | model
for chunk in chain.stream({"topic": "a robot learning to paint"}):
# Each chunk contains a token of the response
if hasattr(chunk, 'content'):
print(chunk.content, end="", flush=True)
print("\n")
# Asynchronous invocation
async def async_example():
"""Demonstrate async invocation."""
model_async = ChatOpenAI(
model="gpt-3.5-turbo",
temperature=0.7,
streaming=True
)
chain_async = prompt | model_async
print("=== Async Streaming ===")
async for chunk in chain_async.astream({"topic": "a cat who discovers time travel"}):
if hasattr(chunk, 'content'):
print(chunk.content, end="", flush=True)
print("\n")
# Run async example
asyncio.run(async_example())
# Async batch processing
async def batch_example():
"""Process multiple inputs concurrently."""
model_batch = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
chain_batch = prompt | model_batch
topics = [
{"topic": "a dragon opening a bakery"},
{"topic": "a programmer debugging magic spells"},
{"topic": "an AI becoming a gardener"}
]
# abatch processes all inputs concurrently
results = await chain_batch.abatch(topics)
for i, result in enumerate(results):
print(f"Story {i+1}: {result.content[:100]}...\n")
asyncio.run(batch_example())Streaming gives users immediate feedback, which is essential for good UX. Async operations let you handle multiple requests concurrently, dramatically improving throughput in web applications. The astream and abatch methods mirror their synchronous counterparts but work with Python's async/await syntax.
Error Handling and Retry Logic
LLMs are unreliable—they sometimes produce invalid JSON, exceed token limits, or return empty responses. LangChain provides robust error handling mechanisms:
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnableLambda
import time
import os
from dotenv import load_dotenv
load_dotenv()
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# Custom retry logic
def with_retry(chain_func, max_retries=3, delay=1):
"""Wrap a chain function with retry logic."""
def wrapped(input_data):
for attempt in range(max_retries):
try:
return chain_func(input_data)
except Exception as e:
if attempt == max_retries - 1:
raise # Re-raise on last attempt
print(f"Attempt {attempt+1} failed: {e}. Retrying in {delay}s...")
time.sleep(delay)
delay *= 2 # Exponential backoff
return None
return wrapped
# Create a chain with fallback
primary_prompt = ChatPromptTemplate.from_messages([
("system", "You are a precise assistant."),
("human", "{question}")
])
# Fallback chain for when primary fails
fallback_chain = (
ChatPromptTemplate.from_messages([
("system", "Provide a simple, direct answer."),
("human", "{question}")
])
| ChatOpenAI(model="gpt-3.5-turbo", temperature=0, max_tokens=50)
| StrOutputParser()
)
# Build robust chain with fallback
robust_chain = (
primary_prompt
| model.with_fallbacks([fallback_chain])
| StrOutputParser()
)
# Invoke with error handling
try:
result = robust_chain.invoke({"question": "Explain quantum computing"})
print(result)
except Exception as e:
print(f"Chain failed after fallbacks: {e}")
The with_fallbacks method specifies alternative chains to try if the primary one fails. Combined with custom retry logic, you can build resilient pipelines that gracefully handle model failures, rate limits, and malformed outputs.
Best Practices for Production LangChain Applications
After working with LangChain across dozens of production deployments, here are the most important best practices I've learned:
- Start simple, iterate – Begin with a basic chain. Add memory, retrieval, and tools only when you need them. Premature complexity kills projects.
- Version your prompts – Treat prompts like code. Store them in version control, test changes systematically, and roll back when performance degrades.
- Use LangSmith for observability – LangSmith (LangChain's companion platform) lets you trace every chain invocation, debug failures, and monitor costs. It's essential for production.
- Guard your context window – Use
ConversationSummaryMemoryor trim messages aggressively. A bloated context window leads to higher costs, slower responses, and degraded reasoning. - Always set temperature to 0 for structured output – When parsing JSON or extracting data, use deterministic outputs. Save creativity for open-ended tasks.
- Implement proper error boundaries – Wrap chain invocations in try/except blocks, use fallbacks, and log failures comprehensively.
- Cache embeddings – Embedding computation is expensive. Cache embeddings locally or use a persistent vector store to avoid re-computing them on every run.
- Test your chains – Write unit tests for prompt templates, output parsers, and full chains. Use a small, deterministic model (or mock) for fast test suites.
- Monitor token usage – Each chain invocation has a cost. Track tokens per request, set alerts for anomalies, and optimize prompts to reduce waste.
- Keep agent iterations bounded – Always set
max_iterationson agent executors. An agent stuck in a loop can burn through your API budget in minutes.
Building a Complete Application: Document Q&A Bot
Let's bring everything together into a complete, working application—a document Q&A bot with memory, streaming, structured logging, and error handling. This is production-quality code:
import os
import logging
from typing import List, Optional
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationSummaryMemory
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
from langchain.callbacks import StdOutCallbackHandler
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
load_dotenv()
class DocumentQABot:
"""A complete RAG-based Q&A bot with conversation memory."""
def __init__(
self,
document_path: str,
collection_name: str = "default_collection",
model_name: str = "gpt-3.5-turbo",
temperature: float = 0.3,
chunk_size: int = 800,
chunk_overlap: int = 100
):
self.document_path = document_path
self.collection_name = collection_name
logger.info(f"Initializing DocumentQABot with documents from {document_path}")
# Initialize model
self.model = ChatOpenAI(
model=model_name,
temperature=temperature,
streaming=True
)
# Initialize embeddings
self.embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")
# Set up vector store
self.vectorstore = self._create_vectorstore(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap
)
# Create retriever
self.retriever = self.vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 5}
)
# Set up memory
self.memory = ConversationSummaryMemory(
llm=ChatOpenAI(model="gpt-3.5-turbo", temperature=0),
return_messages=True,
max_token_limit=500
)
# Build the chain
self.chain = self._build_chain()
logger.info("DocumentQABot initialized successfully")
def _load_documents(self):
"""Load documents from the specified path."""
try:
loader = DirectoryLoader(
self.document_path,
glob="**/*.txt",
loader_cls=TextLoader,
show_progress=True
)
documents = loader.load()
logger.info(f"Loaded {len(documents)} documents")
return documents
except Exception as e:
logger.error(f"Failed to load documents: {e}")
raise
def _create_vectorstore(self, chunk_size: int, chunk_overlap: int):
"""Create a vector store from documents."""
documents = self._load_documents()
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", ". ", " ", ""],
length_function=len,
)
chunks = text_splitter.split_documents(documents)
logger.info(f"Split into {len(chunks)} chunks")
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=self.embeddings,
collection_name=self.collection_name,
persist_directory="./chroma_db"
)
logger.info("Vector store created and persisted")
return vectorstore
def _build_chain(self):
"""Build the RAG chain with memory."""
prompt = ChatPromptTemplate.from_messages([
("system", """You are a knowledgeable assistant that answers questions
based on the provided context. If the answer is not in the context,
say so honestly. Use conversation history to maintain context.
Relevant context:
{context}"""),
MessagesPlaceholder(variable_name="history"),
("human", "{question}")
])
def get_history(_):
return self.memory.load_memory_variables({}).get("history", [])
def format_docs(docs):
return "\n\n---\n\n".join(doc.page_content for doc in docs)
chain = (
RunnablePassthrough.assign(
context=lambda x: format_docs(
self.retriever.get_relevant_documents(x["question"])
),
history=get_history
)
| prompt
| self.model
| StrOutputParser()
)
return chain
def ask(self, question: str, stream: bool = True) -> str:
"""Ask a question and get an answer."""
logger.info(f"Processing question: {question[:100]}...")
try:
if stream:
# Stream the response token by token
full_response = []
for chunk in self.chain.stream({"question": question}):
print(chunk, end="", flush=True)
full_response.append(chunk)
print() # New line after streaming
response_text = "".join(full_response)
else:
response_text = self.chain.invoke({"question": question})
print(response_text)
# Save to memory
self.memory.save_context(
{"input": question},
{"output": response_text}
)
return response_text
except Exception as e:
logger.error(f"Error processing question: {e}")
fallback_msg = "I'm sorry, I encountered an error. Please try again."
print(fallback_msg)
return fallback_msg
def reset_memory(self):
"""Clear conversation history."""
self.memory.clear()
logger.info("Memory cleared")
# Usage example
if __name__ == "__main__":
# Create a directory with some text files before running
# mkdir docs && echo "Your documents here" > docs/sample.txt
bot = DocumentQABot(
document_path="./docs",
collection_name="my_documents"
)
print("Document Q&A Bot ready. Type 'exit' to quit, 'reset' to clear memory.\n")
while True:
user_input = input("\nYou: ")
if user_input.lower() == 'exit':
print("Goodbye!")
break
elif user_input.lower() == 'reset':
bot.reset_memory()
print("Memory cleared.")
continue
print("\nBot: ", end="")
bot.ask(user_input, stream=True)
This complete application demonstrates production patterns: class-based organization, comprehensive logging, persistent vector storage, streaming responses, conversation memory, error handling with fallbacks, and a clean interactive interface. You can extend this by adding more document formats (PDF, HTML), switching to a cloud vector store like Pinecone, or deploying it behind a REST API with FastAPI.
Conclusion
LangChain transforms LLM development from an artisanal craft into an engineering discipline. By providing standardized abstractions for prompts, models, memory, retrieval, and agents, it lets you focus on your application logic rather than reinventing infrastructure. The pipe operator (|) and Runnable interface make composition feel natural and Pythonic.
The journey from a simple prompt chain to a full RAG agent with memory and streaming is surprisingly short—yet each component can be customized, swapped, or extended independently. Start with the basics, add complexity as your requirements grow, and always instrument your chains with proper logging and observability. The patterns you've learned in this tutorial—chains, memory, retrieval, agents, structured output, streaming, and error handling—form the complete toolkit for building reliable, production-grade LLM applications with LangChain.