"""
Pattern 023: Retrieval-Augmented Generation (RAG)
Description:
Agent retrieves relevant information from a knowledge base before generating responses.
RAG combines the power of retrieval systems with generative models to provide
grounded, factual responses based on external knowledge.
Components:
- Document Loader: Loads documents into the system
- Text Splitter: Chunks documents into manageable pieces
- Embedding Model: Converts text to vector representations
- Vector Store: Stores and retrieves embeddings efficiently
- Retriever: Finds relevant documents
- LLM: Generates response based on retrieved context
Use Cases:
- Question answering over documents
- Knowledge-intensive tasks
- Chat with your data applications
- Customer support with documentation
LangChain Implementation:
Uses LangChain's built-in RAG components with vector stores and retrievers.
"""
import os
from typing import List, Dict, Any
from dotenv import load_dotenv
from langchain_community.document_loaders import TextLoader
from langchain_core.documents import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
# Load environment variables
load_dotenv()
# Sample documents for demonstration
SAMPLE_DOCUMENTS = [
{
"content": """LangChain is a framework for developing applications powered by language models.
It enables applications that are context-aware and can reason. LangChain provides modular
components for working with language models, as well as pre-built chains and agents.""",
"metadata": {"source": "langchain_intro", "topic": "framework"}
},
{
"content": """Vector databases are specialized databases designed to store and query
high-dimensional vectors efficiently. They use techniques like approximate nearest neighbor
search to find similar vectors quickly. Popular vector databases include Pinecone, Weaviate,
and Chroma.""",
"metadata": {"source": "vector_db_guide", "topic": "databases"}
},
{
"content": """Embeddings are numerical representations of text that capture semantic meaning.
Words or sentences with similar meanings have similar embeddings. Common embedding models include
OpenAI's text-embedding-ada-002 and open-source alternatives like Sentence Transformers.""",
"metadata": {"source": "embeddings_explained", "topic": "embeddings"}
},
{
"content": """Retrieval-Augmented Generation (RAG) combines retrieval systems with language models.
First, relevant documents are retrieved from a knowledge base. Then, these documents are provided
as context to the language model, which generates a response grounded in the retrieved information.""",
"metadata": {"source": "rag_overview", "topic": "rag"}
},
{
"content": """Semantic search finds results based on meaning rather than exact keyword matches.
It uses embeddings to represent queries and documents in a vector space, then finds the most
semantically similar documents. This enables finding relevant information even when different
words are used.""",
"metadata": {"source": "semantic_search_guide", "topic": "search"}
}
]
class RAGAgent:
"""Agent that uses Retrieval-Augmented Generation."""
def __init__(
self,
model_name: str = "gpt-3.5-turbo",
embedding_model: str = "text-embedding-ada-002",
chunk_size: int = 500,
chunk_overlap: int = 50,
k_documents: int = 3
):
"""
Initialize the RAG agent.
Args:
model_name: LLM model to use for generation
embedding_model: Embedding model for vector representations
chunk_size: Size of text chunks
chunk_overlap: Overlap between chunks
k_documents: Number of documents to retrieve
"""
self.llm = ChatOpenAI(
model=model_name,
temperature=0,
api_key=os.getenv("OPENAI_API_KEY")
)
self.embeddings = OpenAIEmbeddings(
model=embedding_model,
api_key=os.getenv("OPENAI_API_KEY")
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len
)
self.k_documents = k_documents
self.vectorstore = None
self.retriever = None
def load_documents(self, documents: List[Dict[str, Any]]) -> None:
"""
Load documents into the vector store.
Args:
documents: List of dictionaries with 'content' and 'metadata' keys
"""
# Convert to Document objects
docs = [
Document(page_content=doc["content"], metadata=doc["metadata"])
for doc in documents
]
# Split documents into chunks
splits = self.text_splitter.split_documents(docs)
print(f"Loaded {len(docs)} documents, split into {len(splits)} chunks")
# Create vector store
self.vectorstore = Chroma.from_documents(
documents=splits,
embedding=self.embeddings,
collection_name="rag_demo"
)
# Create retriever
self.retriever = self.vectorstore.as_retriever(
search_kwargs={"k": self.k_documents}
)
def retrieve_documents(self, query: str) -> List[Document]:
"""
Retrieve relevant documents for a query.
Args:
query: Search query
Returns:
List of relevant documents
"""
if not self.retriever:
raise ValueError("No documents loaded. Call load_documents first.")
return self.retriever.get_relevant_documents(query)
def query_with_retrieval_qa(self, query: str) -> Dict[str, Any]:
"""
Query using RetrievalQA chain (simple approach).
Args:
query: Question to answer
Returns:
Dictionary with answer and source documents
"""
if not self.retriever:
raise ValueError("No documents loaded. Call load_documents first.")
# Create RetrievalQA chain
qa_chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type="stuff", # "stuff" puts all docs into prompt
retriever=self.retriever,
return_source_documents=True
)
result = qa_chain.invoke({"query": query})
return {
"query": query,
"answer": result["result"],
"source_documents": result["source_documents"]
}
def query_with_custom_prompt(self, query: str) -> Dict[str, Any]:
"""
Query using custom prompt template (more control).
Args:
query: Question to answer
Returns:
Dictionary with answer and retrieved documents
"""
if not self.retriever:
raise ValueError("No documents loaded. Call load_documents first.")
# Custom prompt template
template = """Answer the question based on the following context.
If you cannot answer the question based on the context, say so clearly.
Context:
{context}
Question: {question}
Answer:"""
prompt = ChatPromptTemplate.from_template(template)
# Create RAG chain using LCEL (LangChain Expression Language)
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{"context": self.retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| self.llm
| StrOutputParser()
)
# Get retrieved documents for transparency
retrieved_docs = self.retrieve_documents(query)
# Generate answer
answer = rag_chain.invoke(query)
return {
"query": query,
"answer": answer,
"retrieved_documents": retrieved_docs,
"num_retrieved": len(retrieved_docs)
}
def query_with_sources(self, query: str) -> Dict[str, Any]:
"""
Query with explicit source citations.
Args:
query: Question to answer
Returns:
Dictionary with answer and cited sources
"""
if not self.retriever:
raise ValueError("No documents loaded. Call load_documents first.")
# Retrieve documents
docs = self.retrieve_documents(query)
# Create context with source labels
context_with_sources = ""
sources = []
for i, doc in enumerate(docs, 1):
source_label = f"[Source {i}: {doc.metadata.get('source', 'unknown')}]"
context_with_sources += f"{source_label}\n{doc.page_content}\n\n"
sources.append({
"id": i,
"source": doc.metadata.get('source', 'unknown'),
"topic": doc.metadata.get('topic', 'unknown'),
"content": doc.page_content
})
# Generate answer with citations
template = """Answer the question based on the following sources.
Cite your sources using [Source N] notation where applicable.
Sources:
{context}
Question: {question}
Answer (with citations):"""
prompt = ChatPromptTemplate.from_template(template)
chain = prompt | self.llm | StrOutputParser()
answer = chain.invoke({
"context": context_with_sources,
"question": query
})
return {
"query": query,
"answer": answer,
"sources": sources
}
def cleanup(self):
"""Clean up resources."""
if self.vectorstore:
try:
self.vectorstore.delete_collection()
except:
pass
def demonstrate_rag_pattern():
"""Demonstrates the RAG pattern with various approaches."""
print("=" * 80)
print("PATTERN 023: Retrieval-Augmented Generation (RAG)")
print("=" * 80)
print()
# Create RAG agent
agent = RAGAgent(k_documents=3)
# Load sample documents
print("Loading documents into vector store...")
agent.load_documents(SAMPLE_DOCUMENTS)
print()
# Test queries
queries = [
"What is LangChain?",
"How do vector databases work?",
"What is RAG and how does it combine retrieval with generation?",
]
# Approach 1: RetrievalQA chain
print("\n" + "=" * 80)
print("APPROACH 1: RetrievalQA Chain (Simple)")
print("=" * 80)
for i, query in enumerate(queries[:2], 1):
print(f"\n{'- ' * 40}")
print(f"Query {i}: {query}")
print('- ' * 40)
try:
result = agent.query_with_retrieval_qa(query)
print(f"\nAnswer:\n{result['answer']}")
print(f"\nSources used: {len(result['source_documents'])} documents")
except Exception as e:
print(f"\n✗ Error: {str(e)}")
# Approach 2: Custom prompt with LCEL
print("\n\n" + "=" * 80)
print("APPROACH 2: Custom Prompt with LCEL")
print("=" * 80)
query = queries[1]
print(f"\n{'- ' * 40}")
print(f"Query: {query}")
print('- ' * 40)
try:
result = agent.query_with_custom_prompt(query)
print(f"\nAnswer:\n{result['answer']}")
print(f"\nRetrieved {result['num_retrieved']} documents")
print("\nRetrieved document topics:")
for doc in result['retrieved_documents']:
print(f" - {doc.metadata.get('topic', 'unknown')}")
except Exception as e:
print(f"\n✗ Error: {str(e)}")
# Approach 3: With source citations
print("\n\n" + "=" * 80)
print("APPROACH 3: Answer with Source Citations")
print("=" * 80)
query = queries[2]
print(f"\n{'- ' * 40}")
print(f"Query: {query}")
print('- ' * 40)
try:
result = agent.query_with_sources(query)
print(f"\nAnswer:\n{result['answer']}")
print(f"\nSources:")
for source in result['sources']:
print(f" [{source['id']}] {source['source']} (Topic: {source['topic']})")
except Exception as e:
print(f"\n✗ Error: {str(e)}")
# Cleanup
agent.cleanup()
# Summary
print("\n\n" + "=" * 80)
print("RAG PATTERN DEMONSTRATION COMPLETE")
print("=" * 80)
print()
print("RAG Components Demonstrated:")
print("1. Document loading and chunking")
print("2. Vector embeddings and storage")
print("3. Semantic retrieval")
print("4. Context-aware generation")
print("5. Source attribution")
print()
print("Key Benefits:")
print("- Grounded responses based on actual data")
print("- Reduced hallucinations")
print("- Up-to-date information")
print("- Source traceability")
print("- Scalable knowledge base")
print()
print("LangChain Components Used:")
print("- RecursiveCharacterTextSplitter: Chunks documents")
print("- OpenAIEmbeddings: Creates vector embeddings")
print("- Chroma: Vector store for efficient retrieval")
print("- RetrievalQA: Pre-built RAG chain")
print("- LCEL: Custom chain composition")
print("- Document retriever: Finds relevant documents")
print()
if __name__ == "__main__":
# Check for API key
if not os.getenv("OPENAI_API_KEY"):
print("Error: OPENAI_API_KEY not found in environment variables.")
print("Please set it in your .env file or environment.")
exit(1)
demonstrate_rag_pattern()
Agentic Design Pattern: Rag
Pattern 023: Rag – Line-by-Line Explanation
Overview
This document provides a detailed line-by-line explanation of the Rag pattern implementation.
Pattern Description
Pattern 023: Retrieval-Augmented Generation (RAG)
Description: Agent retrieves relevant information from a knowledge base before generating responses. RAG combines the power of retrieval systems with generative models to provide grounded, factual responses based on external knowledge.
Components: – Document Loader: Loads documents into the system – Text Splitter: Chunks documents into manageable pieces – Embedding Model: Converts text to vector representations – Vector Store: Stores and retrieves embeddings efficiently – Retriever: Finds relevant documents – LLM: Generates response based on retrieved context
Use Cases: – Question answering over documents – Knowledge-intensive tasks – Chat with your data applications – Customer support with documentation
LangChain Implementation: Uses LangChain’s built-in RAG components with vector stores and retrievers.
File Statistics
- Total Lines: 431
- Classes: 1
- Functions: 1
- Constants: 1
Module Documentation
"""
Pattern 023: Retrieval-Augmented Generation (RAG)
Description:
Agent retrieves relevant information from a knowledge base before generating responses.
RAG combines the power of retrieval systems with generative models to provide
grounded, factual responses based on external knowledge.
Components:
- Document Loader: Loads documents into the system
- Text Splitter: Chunks documents into manageable pieces
- Embedding Model: Converts text to vector representations
- Vector Store: Stores and retrieves embeddings efficiently
- Retriever: Finds relevant documents
- LLM: Generates response based on retrieved context
Use Cases:
- Question answering over documents
- Knowledge-intensive tasks
- Chat with your data applications
- Customer support with documentation
LangChain Implementation:
Uses LangChain's built-in RAG components with vector stores and retrievers.
"""
Purpose: This module implements the Rag pattern, providing a structured approach to agent design and implementation.
Import Statements
Standard Library Imports
import langchain_community.vectorstores.Chroma
import langchain_core.documents.Document
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.runnables.RunnablePassthrough
import os
import typing.Any
import typing.Dict
import typing.List
Purpose: Core Python functionality:
– langchain_community.vectorstores.Chroma: Regular expression operations
– langchain_core.documents.Document: Regular expression operations
– langchain_core.output_parsers.StrOutputParser: Regular expression operations
– langchain_core.prompts.ChatPromptTemplate: Regular expression operations
– langchain_core.runnables.RunnablePassthrough: Regular expression operations
– os: Environment variable access and system operations
– typing.Any: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.List: Type hints for better code documentation (List, Dict, Any, Optional)
LangChain Imports
from langchain.chains.RetrievalQA
from langchain.text_splitter.RecursiveCharacterTextSplitter
from langchain_community.document_loaders.TextLoader
from langchain_openai.ChatOpenAI
from langchain_openai.OpenAIEmbeddings
Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows
Other Imports
from dotenv.load_dotenv
Purpose: Additional dependencies for specialized functionality.
Module Constants
SAMPLE_DOCUMENTS
SAMPLE_DOCUMENTS = [
Line 46: Module-level constant definition.
Classes
Class: RAGAgent
Lines 82-309
Purpose:
Agent that uses Retrieval-Augmented Generation.
Methods:
__init__(self, model_name, embedding_model, chunk_size, chunk_overlap, k_documents)
Lines 85-122
Purpose:
Initialize the RAG agent.
Args:
model_name: LLM model to use for generation
embedding_model: Embedding model for vector representations
chunk_size: Size of text chunks
chunk_overlap: Overlap between chunks
k_documents: Number of documents to retrieve
Parameters:
– model_name: Language model instance
– embedding_model: Language model instance
– chunk_size: Chunk size
– chunk_overlap: Chunk overlap
– k_documents: K documents
load_documents(self, documents)
Lines 124-152
Purpose:
Load documents into the vector store.
Args:
documents: List of dictionaries with 'content' and 'metadata' keys
Parameters:
– documents: Documents
Returns: None
retrieve_documents(self, query)
Lines 154-167
Purpose:
Retrieve relevant documents for a query.
Args:
query: Search query
Returns:
List of relevant documents
Parameters:
– query: Input query or question
Returns: <ast.Subscript object at 0x000001A06D6F7E90>
query_with_retrieval_qa(self, query)
Lines 169-196
Purpose:
Query using RetrievalQA chain (simple approach).
Args:
query: Question to answer
Returns:
Dictionary with answer and source documents
Parameters:
– query: Input query or question
Returns: <ast.Subscript object at 0x000001A06D702390>
query_with_custom_prompt(self, query)
Lines 198-246
Purpose:
Query using custom prompt template (more control).
Args:
query: Question to answer
Returns:
Dictionary with answer and retrieved documents
Parameters:
– query: Input query or question
Returns: <ast.Subscript object at 0x000001A06D702990>
query_with_sources(self, query)
Lines 248-301
Purpose:
Query with explicit source citations.
Args:
query: Question to answer
Returns:
Dictionary with answer and cited sources
Parameters:
– query: Input query or question
Returns: <ast.Subscript object at 0x000001A06D64EB90>
cleanup(self)
Lines 303-309
Purpose:
Clean up resources.
Parameters:
Functions
demonstrate_rag_pattern()
Lines 312-420
Purpose:
Demonstrates the RAG pattern with various approaches.
Key Concepts Summary
Pattern Overview
The Rag pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.
Core Components
- Agent Class: Main implementation of the pattern logic
- Prompt Engineering: Carefully crafted prompts to guide LLM behavior
- Tool Integration: Optional tools for extended functionality
- Memory Management: State handling across interactions
- Output Parsing: Structured response processing
When to Use
✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability
❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)
Best Practices Demonstrated
1. Environment Configuration
- ✅ Use
.envfiles for API keys - ✅ Never hardcode credentials
- ✅ Validate environment variables on startup
2. Error Handling
- ✅ Graceful degradation on failures
- ✅ Informative error messages
- ✅ Logging for debugging
3. Prompt Engineering
- ✅ Clear, specific instructions
- ✅ Structured output formats
- ✅ Few-shot examples when beneficial
4. Code Organization
- ✅ Separation of concerns (class-based design)
- ✅ Type hints for clarity
- ✅ Docstrings for documentation
5. Performance Optimization
- ✅ Appropriate temperature settings
- ✅ Token budget management
- ✅ Caching where applicable
LangChain Components Used
- ChatOpenAI: OpenAI model interface
- Chains: Sequential component composition
Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development
Execution Flow
1. Initialize agent with configuration
↓
2. Load environment and dependencies
↓
3. Prepare prompt with input data
↓
4. Invoke LLM through LangChain
↓
5. Parse and validate response
↓
6. Return formatted result
Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation
Performance Considerations
Computational Cost
- LLM Calls: Primary cost factor (based on tokens)
- Latency: Depends on model and input size
- Memory: Minimal for most patterns
Optimization Strategies
- Prompt Optimization: Minimize tokens while maintaining clarity
- Caching: Cache repeated queries/responses
- Model Selection: Use smaller models when appropriate
- Batch Processing: Group similar requests
- Streaming: For long responses, stream tokens
Cost Estimates (Approximate)
- GPT-3.5-turbo: ~$0.002 per request
- GPT-4: ~$0.03 per request
- GPT-4-turbo: ~$0.01 per request
Actual costs vary based on input/output token counts
Common Pitfalls to Avoid
1. Prompt Design Issues
❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts
2. Error Handling
❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation
3. Token Management
❌ Exceeding model token limits
✅ Truncate inputs, chunk processing
4. Security
❌ Exposing API keys in code
✅ Use environment variables
5. Testing
❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs
6. Cost Management
❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring
Further Improvements
Production Enhancements
- Monitoring: Add logging, metrics, tracing
- Validation: Input/output schema validation
- Rate Limiting: Implement request throttling
- Caching: Redis or in-memory cache for responses
- Async Support: Non-blocking operations for scalability
Feature Extensions
- Multi-Model Support: Switch between providers
- Fallback Models: Graceful degradation strategy
- Custom Tools: Domain-specific function calling
- Advanced Memory: Vector stores for long-term memory
- Evaluation: Automated testing and benchmarking
Code Quality
- Testing: Unit tests, integration tests
- Documentation: API docs, usage examples
- Type Safety: Stricter type hints
- Configuration: Externalized config files
Related Patterns
- Pattern 022
- Pattern 024
- Other Tool Use Patterns (020-025)
Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…
Summary
The Rag pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:
✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns
🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability
📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/
File: 023_rag.py
Total Lines: 431
Pattern Category: AI Agent Design Pattern #023
Generated from actual implementation code
For educational and reference purposes
Leave a Reply