Hybrid Search and Semantic RAG Implementation

This document provides a comprehensive, production-grade architectural breakdown of a Hybrid Search and Semantic Retrieval-Augmented Generation (RAG) system. The architecture integrates Apache Airflow for orchestration, Neon DB (serverless Postgres) for relational metadata storage, OpenSearch as a vector database, FastAPI for the application layer, and LangGraph/LangFuse for agentic workflow orchestration and tracing.

1. Architectural Overview & Infrastructure Layout

The system is designed as a decoupled, multi-container microservices architecture orchestrated via Docker Compose. The infrastructure is split into ingestion, storage, serving, and observability layers.

The execution flow is divided into distinct operational phases:

• Phase 1: Service Setup & Infrastructure Provisioning: Initializing Docker containers, configuring network bridges, and setting up persistent volumes.
• Phase 2: Data Ingestion & Processing: Querying the arXiv API, downloading PDFs, chunking text, and extracting metadata.
• Phase 3: Vector Indexing: Generating embeddings and indexing document chunks into OpenSearch.
• Phase 4: Serving & Retrieval: Exposing endpoints via FastAPI to handle standard, streaming, and agentic RAG queries.

2. Database Layer: Serverless Postgres & OpenSearch

The storage layer uses a hybrid database approach to separate structured metadata from high-dimensional vector embeddings.

Relational Metadata Storage: Neon DB

Apache Airflow requires a relational database backend to manage DAG states, task histories, and schedules. This implementation uses Neon DB, a serverless Postgres solution.

Neon DB also stores raw metadata for ingested academic papers (e.g., titles, authors, categories, and publication dates) before they are chunked and vectorized.

Note: Ingesting data from external sources like the arXiv API requires rate-limiting strategies. To avoid HTTP 429 (Too Many Requests) errors, the ingestion pipeline is configured with backoff limits and a bare-minimum initial ingestion set (e.g., 5-6 foundational papers such as the Attention Is All You Need, Convolutions, and Balanced LoRA papers).

Vector Storage: OpenSearch

OpenSearch serves as the vector database for storing document chunks and performing hybrid search queries. It provides a native dashboard for index management and cluster monitoring.

In the demonstration environment, the OpenSearch cluster runs on an Elastic Kubernetes Service (EKS) cluster, indexing approximately 135 chunks derived from 5 source papers (~2.6 MB of total index data).

3. Automation Layer: Makefiles & Docker Compose Orchestration

To simplify local development and deployment, the entire multi-container application is managed via a unified Makefile wrapping complex docker compose commands.

The Automation Workflow

Developers interact with the system using clean, single-word commands rather than long Docker CLI strings:

# Start all services (Airflow, FastAPI, OpenSearch, Redis) in detached mode
make start

# Stop all running containers and preserve volumes
make stop

# Rebuild and restart all containers
make restart

Production-Grade Makefile Implementation

Below is the structure of the automation Makefile used to manage the container lifecycles:

.PHONY: start stop restart logs status clean

# Variables
COMPOSE_FILE=docker-compose.yml

start:
	@echo "Starting all services..."
	docker compose -f $(COMPOSE_FILE) up --build -d
	@echo "Services started successfully."

stop:
	@echo "Stopping all services..."
	docker compose -f $(COMPOSE_FILE) down
	@echo "Services stopped."

restart:
	@echo "Restarting services..."
	$(MAKE) stop
	$(MAKE) start

logs:
	docker compose -f $(COMPOSE_FILE) logs -f

status:
	docker compose -f $(COMPOSE_FILE) ps

clean:
	@echo "Cleaning up volumes and orphaned containers..."
	docker compose -f $(COMPOSE_FILE) down -v --remove-orphans

4. LLM Providers & Environment Configuration

The application supports multiple LLM providers, allowing seamless switching between local models and cloud-hosted APIs.

Environment Configuration (.env.example)

To configure the system, copy the .env.example file to .env and populate the required API keys and endpoints:

# Environment Mode
ENVIRONMENT=development # staging | production

# LLM Provider Configuration (Options: openai | bedrock)
LLM_PROVIDER=bedrock

# AWS Bedrock Configuration
AWS_ACCESS_KEY_ID=your_aws_access_key
AWS_SECRET_ACCESS_KEY=your_aws_secret_key
AWS_DEFAULT_REGION=us-east-1
BEDROCK_MODEL_ID=meta.llama3-70b-instruct-v1:0

# OpenAI Configuration
OPENAI_API_KEY=your_openai_api_key
OPENAI_MODEL_NAME=gpt-4-turbo

# Vector Database (OpenSearch)
OPENSEARCH_HOST=localhost
OPENSEARCH_PORT=9200
OPENSEARCH_USER=admin
OPENSEARCH_PASSWORD=your_secure_password

# Relational Database (Neon Postgres)
DATABASE_URL=postgresql://user:password@subdomain.neon.tech/dbname?sslmode=require

# Caching & Observability
REDIS_URL=redis://your-upstash-redis-url:6379
LANGFUSE_PUBLIC_KEY=lf-pub-...
LANGFUSE_SECRET_KEY=lf-sec-...
LANGFUSE_HOST=https://cloud.langfuse.com

Dynamic Provider Initialization

The application dynamically instantiates the LLM client based on the .env configuration:

import os
from langchain_openai import ChatOpenAI
from langchain_aws import ChatBedrock

def get_llm_client():
    provider = os.getenv("LLM_PROVIDER", "openai").lower()
    
    if provider == "bedrock":
        return ChatBedrock(
            model_id=os.getenv("BEDROCK_MODEL_ID", "meta.llama3-70b-instruct-v1:0"),
            region_name=os.getenv("AWS_DEFAULT_REGION", "us-east-1")
        )
    elif provider == "openai":
        return ChatOpenAI(
            model=os.getenv("OPENAI_MODEL_NAME", "gpt-4-turbo"),
            temperature=0.0
        )
    else:
        raise ValueError(f"Unsupported LLM provider: {provider}")

5. Data Ingestion & PDF Processing Pipeline

The ingestion pipeline extracts academic papers from the arXiv API, parses the PDFs, chunks the text, and writes the resulting vectors to OpenSearch.

Ingestion Implementation

The following script demonstrates how papers are fetched, chunked, and indexed:

import uuid
from langchain_text_splitters import RecursiveCharacterTextSplitter
from opensearchpy import OpenSearch

def process_and_index_pdf(pdf_path: str, metadata: dict, opensearch_client: OpenSearch):
    # 1. Load and extract text (Simulated extraction)
    with open(pdf_path, "r", errors="ignore") as f:
        raw_text = f.read()
    
    # 2. Chunk the text using a recursive splitter
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        length_function=len
    )
    chunks = text_splitter.split_text(raw_text)
    
    # 3. Index each chunk into OpenSearch
    for i, chunk in enumerate(chunks):
        # Generate a mock embedding (replace with actual OpenAI/Bedrock embedding call)
        mock_embedding = [0.1] * 1536  
        
        document = {
            "chunk_id": str(uuid.uuid4()),
            "text": chunk,
            "embedding": mock_embedding,
            "metadata": {
                **metadata,
                "chunk_index": i,
                "total_chunks": len(chunks)
            }
        }
        
        opensearch_client.index(
            index="archive_paper_chunks",
            body=document,
            id=document["chunk_id"],
            refresh=True
        )
    return len(chunks)

6. FastAPI Application Layer & API Endpoints

The FastAPI application serves as the entry point for user queries, exposing endpoints for health checks, standard retrieval, streaming, and agentic workflows.

API Endpoint Architecture

FastAPI Server Implementation

Below is the implementation of the FastAPI application showing the standard and agentic endpoints:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Any

app = FastAPI(title="Hybrid Search & Semantic RAG API", version="1.0.0")

class QueryRequest(BaseModel):
    query: str

class QueryResponse(BaseModel):
    query: str
    response: str
    sources: List[Dict[str, Any]]

@app.get("/health")
async def health_check():
    # In production, verify connections to OpenSearch, Redis, and Neon DB
    return {
        "status": "healthy",
        "services": {
            "opensearch": "connected",
            "redis": "connected",
            "neon_postgres": "connected"
        }
    }

@app.post("/ask", response_model=QueryResponse)
async def ask_standard_rag(request: QueryRequest):
    try:
        # Standard RAG execution flow
        # 1. Retrieve chunks from OpenSearch
        # 2. Format prompt
        # 3. Call LLM
        return QueryResponse(
            query=request.query,
            response="This is a standard RAG response.",
            sources=[{"document": "Attention Is All You Need", "page": 3}]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/agent/ask", response_model=QueryResponse)
async def ask_agentic_rag(request: QueryRequest):
    try:
        # Agentic RAG execution flow via LangGraph
        # 1. Initialize graph state
        # 2. Execute agent nodes (Retrieve -> Evaluate -> Rewrite -> Generate)
        return QueryResponse(
            query=request.query,
            response="This is an agentic RAG response generated via LangGraph.",
            sources=[{"document": "Vector Policy Optimization", "chunk": 12}]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

7. Agentic RAG Execution Flow & Hybrid Search

The system uses an Agentic RAG design pattern powered by LangGraph. Unlike standard RAG, which retrieves documents and generates answers in a single step, the agentic workflow evaluates the retrieved context and can rewrite queries or fetch additional data if needed.

Agentic Execution Flow

Hybrid Search Scoring Mechanics

The hybrid search engine combines lexical matching (BM25) with dense vector retrieval (k-NN) to compute a unified relevance score.

The combined score $S_{hybrid}$ for a document $d$ given a query $q$ is calculated as:

Where:

• $S_{dense}(d, q)$ is the cosine similarity score of the dense vector embeddings.
• $S_{lexical}(d, q)$ is the normalized BM25 score.
• $\alpha$ is a tuning parameter ($0 \le \alpha \le 1$) that controls the weight of semantic versus keyword matching.

8. Production Observability: Pydantic Logfire & LangFuse Tracing

To run a RAG system reliably in production, you need both application-level logging and execution-level tracing.

Application Logging: Pydantic Logfire

Pydantic Logfire provides structured, schema-validated logging. It captures system-level events, database connections, and API request lifecycles.

Execution Tracing: LangFuse

While Logfire tracks application health, LangFuse provides deep tracing for LLM calls and agentic workflows. It maps the entire lifecycle of a user query through LangGraph.

This level of tracing allows developers to pinpoint latency bottlenecks, inspect raw prompt payloads, and track API token usage across different steps of the pipeline.

9. Summary and Key Takeaways

• Decoupled Infrastructure: Running Apache Airflow inside Docker while offloading relational storage to a serverless Postgres provider (Neon DB) keeps the orchestration layer lightweight and scalable.
• Hybrid Vector Storage: OpenSearch acts as a robust vector database, combining traditional keyword search (BM25) with dense vector search (k-NN) to improve retrieval accuracy.
• Flexible LLM Integration: The system can switch between local models (Ollama) and cloud APIs (AWS Bedrock or OpenAI) with a single change in the .env file.
• Automated Environments: Using a Makefile to wrap Docker Compose commands simplifies local development and ensures consistent environments across staging and production.
• Agentic Orchestration: Implementing RAG with LangGraph allows the system to evaluate retrieved context, rewrite queries, and self-correct before generating a final response.
• Dual-Layer Observability: Combining Pydantic Logfire for structured application logging with LangFuse for detailed LLM tracing provides complete visibility into system performance and cost.