RAG Agent Template
WorkflowRetrieval-augmented generation agent with vector database integration.
RAG Agent Template
Overview
This template provides a complete Retrieval-Augmented Generation (RAG) agent with vector database integration. It enables AI agents to answer questions based on your custom knowledge base, combining the power of LLMs with your proprietary data.
RAG is essential for building AI applications that need to access specific, up-to-date information that wasn't part of the model's training data.
Prerequisites
- Python 3.10+
- Vector database (Chroma, Pinecone, Weaviate, or Qdrant)
- LLM API access (OpenAI, Anthropic, etc.)
Project Structure
rag-agent/
├── agent/
│ ├── __init__.py
│ ├── rag_agent.py # Main RAG agent
│ └── prompts.py # Prompt templates
├── retriever/
│ ├── __init__.py
│ ├── vector_store.py # Vector database operations
│ ├── embeddings.py # Embedding models
│ └── chunking.py # Text chunking strategies
├── indexer/
│ ├── __init__.py
│ ├── document_loader.py # Document loading
│ └── indexer.py # Indexing pipeline
├── data/
│ └── documents/ # Source documents
├── config/
│ └── settings.yaml # Configuration
├── main.py # Entry point
├── requirements.txt
└── README.md
Installation
pip install langchain langchain-openai chromadb tiktoken pyyaml
Configuration
config/settings.yaml:
# RAG Agent Configuration
llm:
provider: openai # openai, anthropic, google
model: gpt-4o
temperature: 0.1
embeddings:
provider: openai
model: text-embedding-3-small
dimensions: 1536
vector_store:
provider: chroma # chroma, pinecone, weaviate, qdrant
persist_directory: ./chroma_db
collection_name: knowledge_base
chunking:
chunk_size: 1000
chunk_overlap: 200
strategy: recursive # recursive, fixed, semantic
retrieval:
top_k: 5
search_type: similarity # similarity, mmr, similarity_score_threshold
score_threshold: 0.7
prompts:
system_prompt: |
You are a helpful assistant. Use the following context to answer questions.
If the answer is not in the context, say so honestly.
user_prompt: |
Context: {context}
Question: {question}
Answer the question based on the context above.
Core Components
Embeddings
retriever/embeddings.py:
from langchain_openai import OpenAIEmbeddings
from langchain.embeddings.base import Embeddings
class EmbeddingManager:
def __init__(self, model: str = "text-embedding-3-small"):
self.embeddings = OpenAIEmbeddings(model=model)
def embed_text(self, text: str) -> list[float]:
"""Embed a single text."""
return self.embeddings.embed_query(text)
def embed_documents(self, texts: list[str]) -> list[list[float]]:
"""Embed multiple documents."""
return self.embeddings.embed_documents(texts)
def embed_query(self, query: str) -> list[float]:
"""Embed a query for search."""
return self.embeddings.embed_query(query)
Chunking Strategies
retriever/chunking.py:
from langchain.text_splitter import (
RecursiveCharacterTextSplitter,
CharacterTextSplitter,
SentenceTransformersTokenTextSplitter,
)
class ChunkingStrategy:
@staticmethod
def recursive(text: str, chunk_size: int = 1000, chunk_overlap: int = 200):
"""Recursive chunking by characters."""
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", ".", " ", ""],
)
return splitter.split_text(text)
@staticmethod
def by_characters(text: str, chunk_size: int = 1000):
"""Fixed-size character chunking."""
splitter = CharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=0,
)
return splitter.split_text(text)
@staticmethod
def by_tokens(text: str, max_tokens: int = 500):
"""Token-based chunking."""
splitter = SentenceTransformersTokenTextSplitter(
chunk_overlap=50,
tokens_per_chunk=max_tokens,
)
return splitter.split_text(text)
@staticmethod
def by_sections(text: str, section_delimiter: str = "\n## "):
"""Section-based chunking (for structured documents)."""
sections = text.split(section_delimiter)
chunks = []
current_chunk = ""
for section in sections:
if len(current_chunk) + len(section) > 1000:
chunks.append(current_chunk.strip())
current_chunk = section_delimiter + section
else:
current_chunk += section_delimiter + section
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
Vector Store
retriever/vector_store.py:
import os
from typing import Optional
from langchain.vectorstores import Chroma, Pinecone, Weaviate
from langchain.schema import Document
class VectorStoreManager:
def __init__(self, config: dict):
self.config = config
self.store = None
def initialize(self, embeddings):
"""Initialize the vector store."""
provider = self.config['provider']
if provider == 'chroma':
self.store = Chroma(
embedding_function=embeddings,
persist_directory=self.config['persist_directory'],
collection_name=self.config['collection_name'],
)
elif provider == 'pinecone':
import pinecone
pinecone.init(
api_key=os.environ['PINECONE_API_KEY'],
environment=os.environ['PINECONE_ENV'],
)
self.store = Pinecone.from_existing_index(
index_name=self.config['collection_name'],
embedding=embeddings,
)
elif provider == 'weaviate':
import weaviate
client = weaviate.Client(os.environ['WEAVIATE_URL'])
self.store = Weaviate(
client=client,
index_name=self.config['collection_name'],
text_key="text",
embedding=embeddings,
)
return self.store
def add_documents(self, documents: list[Document]):
"""Add documents to the vector store."""
if self.store is None:
raise ValueError("Vector store not initialized")
return self.store.add_documents(documents)
def similarity_search(self, query: str, k: int = 5):
"""Search by similarity."""
return self.store.similarity_search(query, k=k)
def similarity_search_with_score(self, query: str, k: int = 5):
"""Search by similarity with scores."""
return self.store.similarity_search_with_score(query, k=k)
def max_marginal_relevance_search(self, query: str, k: int = 5):
"""Search using MMR for diversity."""
return self.store.max_marginal_relevance_search(query, k=k)
def delete_collection(self):
"""Delete the collection (use with caution)."""
if self.store is None:
raise ValueError("Vector store not initialized")
self.store._collection.delete()
Document Loading
indexer/document_loader.py:
from langchain.document_loaders import (
PyPDFLoader,
TextLoader,
UnstructuredMarkdownLoader,
UnstructuredHTMLLoader,
Docx2txtLoader,
CSVLoader,
DirectoryLoader,
)
from pathlib import Path
class DocumentLoader:
@staticmethod
def load_pdf(path: str):
"""Load a PDF document."""
loader = PyPDFLoader(path)
return loader.load()
@staticmethod
def load_text(path: str):
"""Load a plain text file."""
loader = TextLoader(path)
return loader.load()
@staticmethod
def load_markdown(path: str):
"""Load a Markdown file."""
loader = UnstructuredMarkdownLoader(path)
return loader.load()
@staticmethod
def load_html(path: str):
"""Load an HTML file."""
loader = UnstructuredHTMLLoader(path)
return loader.load()
@staticmethod
def load_docx(path: str):
"""Load a Word document."""
loader = Docx2txtLoader(path)
return loader.load()
@staticmethod
def load_csv(path: str):
"""Load a CSV file."""
loader = CSVLoader(path)
return loader.load()
@staticmethod
def load_directory(directory: str, recursive: bool = True):
"""Load all documents from a directory."""
loader = DirectoryLoader(
directory,
recursive=recursive,
show_progress=True,
)
return loader.load()
@staticmethod
def load_from_url(url: str):
"""Load content from a URL."""
from langchain.document_loaders import WebBaseLoader
loader = WebBaseLoader(url)
return loader.load()
RAG Agent
agent/rag_agent.py:
from typing import Optional
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
class RAGAgent:
def __init__(self, vector_store, embeddings, config: dict):
self.vector_store = vector_store
self.embeddings = embeddings
self.config = config
# Initialize LLM
self.llm = ChatOpenAI(
model=config['llm']['model'],
temperature=config['llm']['temperature'],
)
# Initialize prompt
self.prompt = self._create_prompt()
# Initialize chain
self.chain = self._create_chain()
def _create_prompt(self):
"""Create the RAG prompt."""
template = self.config['prompts']['user_prompt']
return ChatPromptTemplate.from_template(template)
def _create_chain(self):
"""Create the RAG chain."""
# Retrieval
retriever = self.vector_store.as_retriever(
search_type=self.config['retrieval']['search_type'],
search_kwargs={
'k': self.config['retrieval']['top_k'],
},
)
# RAG chain
chain = (
{
'context': retriever | self._format_docs,
'question': RunnablePassthrough(),
}
| self.prompt
| self.llm
| StrOutputParser()
)
return chain
def _format_docs(self, docs):
"""Format documents for the prompt."""
return "\n\n".join(doc.page_content for doc in docs)
def query(self, question: str) -> str:
"""Query the RAG system."""
return self.chain.invoke(question)
def query_with_sources(self, question: str) -> dict:
"""Query with source citations."""
retriever = self.vector_store.as_retriever(
search_kwargs={'k': self.config['retrieval']['top_k']},
)
docs = retriever.invoke(question)
answer = self.chain.invoke(question)
return {
'answer': answer,
'sources': [
{
'content': doc.page_content[:200],
'metadata': doc.metadata,
}
for doc in docs
],
}
def stream_query(self, question: str):
"""Stream the query response."""
for chunk in self.chain.stream(question):
yield chunk
Indexing Pipeline
indexer/indexer.py:
from typing import list
from document_loader import DocumentLoader
from chunking import ChunkingStrategy
from retriever.vector_store import VectorStoreManager
class IndexingPipeline:
def __init__(self, vector_store: VectorStoreManager, embeddings, config: dict):
self.vector_store = vector_store
self.embeddings = embeddings
self.config = config
def index_documents(self, source_paths: list[str]):
"""Index documents from source paths."""
all_docs = []
for path in source_paths:
# Load document
docs = DocumentLoader.load(path)
# Chunk document
chunks = []
for doc in docs:
chunk_strategy = self.config['chunking']['strategy']
chunk_size = self.config['chunking']['chunk_size']
chunk_overlap = self.config['chunking']['chunk_overlap']
if chunk_strategy == 'recursive':
chunked = ChunkingStrategy.recursive(
doc.page_content,
chunk_size,
chunk_overlap,
)
elif chunk_strategy == 'fixed':
chunked = ChunkingStrategy.by_characters(
doc.page_content,
chunk_size,
)
# Create chunk documents with metadata
for i, chunk in enumerate(chunked):
chunks.append({
'text': chunk,
'metadata': {
**doc.metadata,
'chunk_id': i,
'total_chunks': len(chunked),
},
})
all_docs.extend(chunks)
# Embed and store
texts = [doc['text'] for doc in all_docs]
metadatas = [doc['metadata'] for doc in all_docs]
self.vector_store.add_documents_with_embeddings(
texts=texts,
metadatas=metadatas,
embeddings=self.embeddings.embed_documents(texts),
)
return len(all_docs)
Main Entry Point
main.py:
import yaml
from retriever.embeddings import EmbeddingManager
from retriever.vector_store import VectorStoreManager
from agent.rag_agent import RAGAgent
from indexer.indexer import IndexingPipeline
def load_config(path: str = 'config/settings.yaml') -> dict:
"""Load configuration from YAML file."""
with open(path, 'r') as f:
return yaml.safe_load(f)
def main():
# Load config
config = load_config()
# Initialize embeddings
embeddings = EmbeddingManager(model=config['embeddings']['model'])
# Initialize vector store
vector_store = VectorStoreManager(config['vector_store'])
vector_store.initialize(embeddings.embeddings)
# Index documents (run once)
indexer = IndexingPipeline(vector_store, embeddings, config)
indexer.index_documents(['data/documents/*.pdf', 'data/documents/*.md'])
# Create RAG agent
agent = RAGAgent(vector_store, embeddings, config)
# Query
question = "What is the company's return policy?"
response = agent.query(question)
print(f"Q: {question}")
print(f"A: {response}")
# Query with sources
result = agent.query_with_sources(question)
print(f"\nSources:")
for i, source in enumerate(result['sources'], 1):
print(f"{i}. {source['content'][:100]}...")
if __name__ == "__main__":
main()
Running the Agent
# Index documents
python main.py --index
# Query
python main.py --query "What is the return policy?"
# Interactive mode
python main.py --interactive