Create embeddings and vector store

May 10, 2026 · View on GitHub

🐍 Python Ecosystem for RAG

Python is a dominant language for building RAG systems, offering a rich ecosystem of libraries and frameworks. This section covers essential Python tools for AI/LLM-powered RAG implementations.

Core Libraries

LLM & AI Model Integration

Embedding & Vector Operations

  • NumPy: Fundamental library for numerical computing and vector operations
  • SciPy: Scientific computing library with distance metrics and optimization algorithms
  • scikit-learn: Machine learning library with vector similarity metrics, clustering, and dimensionality reduction
  • Sentence Transformers: Framework for computing sentence embeddings using transformer models
  • Instructor: Structured outputs for LLMs with Pydantic validation
  • Embedchain: Framework to create LLM-powered bots over any dataset

Document Processing

  • PyPDF2: PDF manipulation library for extracting text and metadata
  • pdfplumber: Advanced PDF parsing with table extraction capabilities
  • python-docx: Library for reading and writing Microsoft Word documents
  • BeautifulSoup4: HTML/XML parsing for web content extraction
  • Unstructured: Open-source library for extracting structured data from documents (PDFs, Word, HTML, etc.)
  • pypandoc: Python wrapper for Pandoc document converter
  • markdown: Python implementation of Markdown for processing markdown documents
  • markitdown: Python tool from Microsoft for converting files and office documents to Markdown.

Async & Performance

  • asyncio: Built-in library for asynchronous I/O operations
  • aiohttp: Async HTTP client/server framework for concurrent API calls
  • httpx: Modern async HTTP client with sync and async support
  • multiprocessing: Built-in library for parallel processing
  • joblib: Library for parallel computing and caching
  • tqdm: Progress bars for loops and long-running operations

RAG Frameworks

  • LangChain: Comprehensive framework for building LLM applications with RAG support

    • Document loaders, text splitters, vector stores, and retrieval chains
    • Integration with 100+ data sources and vector databases
    • LangChain Python Documentation

  • LlamaIndex: Data framework for LLM applications with advanced RAG capabilities

  • Haystack: End-to-end NLP framework with RAG pipelines

  • RAGatouille: RAG framework built on top of ColBERT for efficient retrieval

  • RAGFlow: Open-source RAG engine with document parsing and knowledge extraction

  • RAGAS: Framework for evaluating RAG pipelines with metrics

Utilities & Tools

Data Processing

  • Pandas: Data manipulation and analysis library for structured data
  • Polars: Fast DataFrame library implemented in Rust with Python bindings
  • DuckDB: In-process analytical database with Python interface
  • data load tool: Open-source Python library for building reliable extract and load data pipelines with automatic schema handling and incremental, stateful loading. Useful for keeping RAG knowledge sources up to date before embedding and indexing.

Configuration & Environment

  • Pydantic: Data validation using Python type annotations
  • python-dotenv: Load environment variables from .env files
  • Hydra: Framework for elegantly configuring complex applications

Monitoring & Observability

  • LangSmith: Platform for debugging, testing, and monitoring LLM applications
  • LangFuse: Open-source LLM engineering platform
  • Weights & Biases: Experiment tracking and visualization
  • MLflow: Platform for managing ML lifecycle including LLM experiments

Testing & Evaluation

  • pytest: Testing framework for Python applications
  • pytest-asyncio: Pytest plugin for testing async code
  • RAGAS: Evaluation framework for RAG pipelines
  • TruLens: LLM evaluation and observability library

Best Practices for RAG

Code Organization

# Recommended project structure
rag_project/
├── src/
│   ├── data_processing/
│   │   ├── chunking.py
│   │   ├── embeddings.py
│   │   └── document_loaders.py
│   ├── retrieval/
│   │   ├── vector_store.py
│   │   └── retrieval_strategies.py
│   ├── generation/
│   │   ├── llm_client.py
│   │   └── prompt_templates.py
│   └── pipeline/
│       └── rag_pipeline.py
├── tests/
├── config/
│   └── settings.yaml
└── requirements.txt

Async Patterns

  • Use asyncio for concurrent API calls to embedding and LLM services
  • Implement connection pooling for database connections
  • Use async context managers for resource cleanup

Error Handling

  • Implement retry logic with exponential backoff for API calls
  • Use circuit breakers for external service failures
  • Log errors with structured logging (e.g., structlog)

Performance Optimization

  • Use batch processing for embedding generation
  • Implement caching for frequently accessed embeddings
  • Leverage multiprocessing for CPU-intensive tasks
  • Use generators for memory-efficient document processing

Type Safety

  • Use type hints throughout your codebase
  • Leverage Pydantic for data validation
  • Use mypy for static type checking

Implementation Examples

Basic RAG Pipeline

from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA

# Load and chunk documents
loader = PyPDFLoader("document.pdf")
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = text_splitter.split_documents(documents)

# Create embeddings and vector store
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# Create RAG chain
llm = OpenAI()
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(),
    return_source_documents=True
)

# Query
result = qa_chain({"query": "What is the main topic?"})

Async Embedding Generation

import asyncio
from sentence_transformers import SentenceTransformer

async def generate_embeddings_async(texts, model_name="all-MiniLM-L6-v2"):
    model = SentenceTransformer(model_name)
    loop = asyncio.get_event_loop()
    embeddings = await loop.run_in_executor(
        None, model.encode, texts
    )
    return embeddings

# Usage
texts = ["Document 1", "Document 2", "Document 3"]
embeddings = await generate_embeddings_async(texts)

Resources & Tutorials

Python-Specific Best Practices

  • Virtual Environments: Use venv or conda to manage dependencies and Python versions
  • Dependency Management: Use requirements.txt or pyproject.toml with version pinning
  • Code Quality: Use black for formatting, flake8 or ruff for linting, and mypy for type checking
  • Testing: Write unit tests with pytest, integration tests for RAG pipelines, and use mocking for external APIs
  • Logging: Use structured logging with structlog or loguru for better observability
  • Error Handling: Implement comprehensive error handling with custom exceptions and retry logic
  • Performance: Profile code with cProfile or py-spy, optimize bottlenecks, and use async/await for I/O-bound operations