Key Takeaways
- Automate RAG data pipelines from ingestion to embedding for continuous knowledge base updates.
- Leverage PostgreSQL with `pgvector` for efficient, low-latency similarity search without external vector databases.
- Implement caching with Redis to reduce redundant computations and improve API response times.
- Design a CI/CD workflow that enables zero-downtime blue-green deployments for RAG services.
- Prioritize robust error handling and deduplication in data ingestion for production-grade reliability.
The Problem
The challenge isn't just building a Retrieval Augmented Generation (RAG) system; it's keeping it alive and relevant in a dynamic world. Imagine you've built a fantastic summarizer for blog posts. If your RAG system relies on a static snapshot of data, it becomes stale the moment a new post goes live. Manually updating embeddings, restarting services, or dealing with inconsistent data sources leads to operational overhead, potential downtime, and a poor user experience. The real problem is moving beyond a proof-of-concept to a production-grade service that automatically adapts to new information, offers high availability, and can be updated without service interruption. This requires a pipeline that handles everything from data ingestion and processing to vector storage, retrieval, summarization, and ultimately, zero-downtime deployments.Data and Sources
To build this pipeline, I used the following resources:- Stripe blog RSS feed: https://stripe.com/blog/feed.rss
- Feedparser documentation: https://feedparser.readthedocs.io/
- Sentence-Transformers model (all-MiniLM-L6-v2): https://huggingface.co/sentence-transformers/all-MiniLM-L6-v2
- GPT-2 model: https://huggingface.co/gpt2
- pgVector extension: https://github.com/pgvector/pgvector
- FastAPI: https://fastapi.tiangolo.com/
- Prefect (optional, for orchestration): https://www.prefect.io/
- Docker Hub (example repo for CI/CD context): https://hub.docker.com/r/mausamadhikari/stripe-rag
Step 1 — Ingest the RSS Feed
The first crucial piece of a dynamic RAG system is a reliable way to get fresh data. For Stripe's blog, this means pulling from their RSS feed. The problem here is keeping our knowledge base up-to-date without manual intervention. My approach was to fetch the feed, parse it, and store the raw entries in a staging table, handling common RSS quirks along the way. I used `feedparser` because it's robust and handles various RSS/Atom formats gracefully. After fetching, I extracted the `title`, `link`, `published` date, and `summary`. A critical part of production ingestion is error handling and deduplication. I wrapped the parsing in a `try-except` block to catch network issues or malformed feeds. For deduplication, I used the `guid` (Globally Unique Identifier) provided by RSS entries, ensuring I only process new or updated posts. If a `title` was missing, I'd skip the entry, as a post without a title isn't very useful.
import feedparser
from datetime import datetime
import psycopg2
import psycopg2.extras
import os
# Database connection details (replace with your actual config)
DB_HOST = os.getenv("DB_HOST", "localhost")
DB_NAME = os.getenv("DB_NAME", "rag_db")
DB_USER = os.getenv("DB_USER", "user")
DB_PASSWORD = os.getenv("DB_PASSWORD", "password")
def get_db_connection():
return psycopg2.connect(host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASSWORD)
def ingest_rss_feed(rss_url: str):
print(f"Ingesting RSS feed from {rss_url}...")
try:
feed = feedparser.parse(rss_url)
if feed.bozo:
print(f"Warning: RSS feed parsing issues for {rss_url}: {feed.bozo_exception}")
conn = get_db_connection()
cur = conn.cursor()
# Ensure posts table exists (basic schema for raw ingest)
cur.execute("""
CREATE TABLE IF NOT EXISTS posts (
id SERIAL PRIMARY KEY,
guid TEXT UNIQUE,
title TEXT,
link TEXT,
published TIMESTAMP WITH TIME ZONE,
summary TEXT,
raw_content TEXT,
ingested_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);
""")
conn.commit()
for entry in feed.entries:
guid = entry.get('id') or entry.get('link') # Fallback to link if no explicit ID
title = entry.get('title')
link = entry.get('link')
summary = entry.get('summary') or entry.get('description')
raw_content = entry.get('content', [{}])[0].get('value') if entry.get('content') else summary # Use content if available
if not title or not link or not guid:
print(f"Skipping malformed entry: {entry.get('title') or entry.get('link')}")
continue
published_raw = entry.get('published')
published_date = None
if published_raw:
try:
# feedparser often converts to datetime.struct_time, which can be converted directly
published_date = datetime(*entry.published_parsed[:6])
except Exception as e:
print(f"Could not parse published date '{published_raw}': {e}. Using current time.")
published_date = datetime.now()
else:
published_date = datetime.now()
try:
cur.execute(
"""
INSERT INTO posts (guid, title, link, published, summary, raw_content)
VALUES (%s, %s, %s, %s, %s, %s)
ON CONFLICT (guid) DO UPDATE SET
title = EXCLUDED.title,
link = EXCLUDED.link,
published = EXCLUDED.published,
summary = EXCLUDED.summary,
raw_content = EXCLUDED.raw_content,
ingested_at = NOW();
""",
(guid, title, link, published_date, summary, raw_content)
)
except psycopg2.Error as e:
print(f"Database error during upsert for GUID {guid}: {e}")
conn.rollback() # Rollback in case of specific error
continue
conn.commit()
print(f"Ingestion complete. Processed {len(feed.entries)} entries.")
except feedparser.HTTPError as e:
print(f"HTTP Error fetching RSS feed: {e}")
except Exception as e:
print(f"An unexpected error occurred during RSS ingestion: {e}")
finally:
if 'conn' in locals() and conn:
conn.close()
Step 2 — Clean & Chunk the Text
The problem with raw blog post content is that it's often too long for an embedding model's context window (typically 512 tokens for `all-MiniLM-L6-v2`). To solve this, I needed to split the full text into smaller, meaningful chunks. I opted for a sentence-aware chunking strategy: grouping sentences until a token limit is reached, then starting a new chunk. This preserves semantic coherence better than arbitrary character splits. I used `nltk.sent_tokenize` for sentence splitting and `transformers`' `GPT2Tokenizer` to accurately count tokens, ensuring compatibility with the `GPT-2` model used later for summarization. Each chunk is then associated with its original `post_id` and a `chunk_index` to maintain order and context.
import nltk
from transformers import AutoTokenizer
import math
# Download NLTK data if not already present
try:
nltk.data.find('tokenizers/punkt')
except nltk.downloader.DownloadError:
nltk.download('punkt')
# Load tokenizer for token counting
tokenizer = AutoTokenizer.from_pretrained("gpt2")
MAX_CHUNK_TOKENS = 256 # A bit less than 512 to leave room for prompt and generation
def clean_and_chunk_text(post_id: int, text: str):
if not text:
return []
sentences = nltk.sent_tokenize(text)
chunks = []
current_chunk_sentences = []
current_chunk_token_count = 0
for sentence in sentences:
sentence_tokens = tokenizer.encode(sentence, add_special_tokens=False)
sentence_token_count = len(sentence_tokens)
if current_chunk_token_count + sentence_token_count <= MAX_CHUNK_TOKENS:
current_chunk_sentences.append(sentence)
current_chunk_token_count += sentence_token_count
else:
if current_chunk_sentences: # Save current chunk if not empty
chunks.append(" ".join(current_chunk_sentences))
# Start a new chunk with the current sentence
current_chunk_sentences = [sentence]
current_chunk_token_count = sentence_token_count
# Handle sentences larger than MAX_CHUNK_TOKENS (rare for blog posts, but possible)
if sentence_token_count > MAX_CHUNK_TOKENS:
# For simplicity, if a single sentence is too long, we take a prefix
# In a real system, you might split it further or just accept a truncated embedding
print(f"Warning: Single sentence in post {post_id} exceeds MAX_CHUNK_TOKENS. Truncating.")
truncated_tokens = tokenizer.encode(sentence, add_special_tokens=False)[:MAX_CHUNK_TOKENS]
chunks.append(tokenizer.decode(truncated_tokens))
current_chunk_sentences = [] # Reset for next sentence
current_chunk_token_count = 0
if current_chunk_sentences: # Add the last chunk
chunks.append(" ".join(current_chunk_sentences))
# Add metadata
chunked_data = []
for i, chunk_text in enumerate(chunks):
chunked_data.append({
"post_id": post_id,
"chunk_index": i,
"text": chunk_text
})
return chunked_data
Step 3 — Generate Dense Embeddings
With our cleaned and chunked text, the next problem is converting these chunks into numerical vectors. These "dense embeddings" are what allow us to perform similarity search, finding relevant content based on a query. I chose `SentenceTransformer('all-MiniLM-L6-v2')` because it's a good balance of performance and efficiency for text embeddings. It's fast, relatively small, and produces high-quality embeddings. The key here is to batch-encode the chunks for efficiency, especially when dealing with many posts. The embeddings are stored as `float32` vectors, which is typically sufficient for `pgVector` and saves storage space compared to `float64`.
from sentence_transformers import SentenceTransformer
import numpy as np
# Load the Sentence-Transformer model once
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
def generate_embeddings(chunks_data: list[dict]):
if not chunks_data:
return []
texts = [chunk["text"] for chunk in chunks_data]
# Batch encode the texts
embeddings = embedding_model.encode(texts, convert_to_tensor=False, show_progress_bar=True)
# Attach embeddings back to chunk data
for i, chunk in enumerate(chunks_data):
chunk["embedding"] = embeddings[i].astype(np.float32) # Ensure float32
return chunks_data
Step 4 — Persist Vectors in PostgreSQL
Storing vectors efficiently and enabling low-latency Approximate Nearest Neighbor (ANN) search is crucial for RAG. Instead of introducing another dedicated vector database, I opted for `pgVector`, a PostgreSQL extension. This simplifies the architecture by keeping all our data (raw posts, chunks, and embeddings) in one place, which is a huge win for operational simplicity and data consistency. The `pgVector` extension allows for a `vector` data type. After installing the extension (`CREATE EXTENSION IF NOT EXISTS vector;`), I created a `chunks` table with an `embedding` column of type `vector(384)`, matching the output dimension of `all-MiniLM-L6-v2`. The `ON CONFLICT` clause on `(post_id, chunk_index)` handles upserts, ensuring that if a chunk is re-processed (e.g., due to updated content), its embedding is updated atomically without interruption.
def persist_chunks_with_embeddings(chunks_with_embeddings: list[dict]):
if not chunks_with_embeddings:
return
conn = get_db_connection()
cur = conn.cursor()
try:
# Enable pgvector extension
cur.execute("CREATE EXTENSION IF NOT EXISTS vector;")
conn.commit()
# Create chunks table
cur.execute("""
CREATE TABLE IF NOT EXISTS chunks (
id SERIAL PRIMARY KEY,
post_id INTEGER REFERENCES posts(id),
chunk_index INTEGER,
text TEXT,
embedding VECTOR(384),
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
UNIQUE (post_id, chunk_index)
);
""")
conn.commit()
# Create an HNSW index for efficient similarity search
# This is critical for performance at scale.
# M = max number of connections per layer, ef_construction = size of dynamic list during construction
cur.execute("""
CREATE INDEX IF NOT EXISTS chunks_embedding_hnsw_idx
ON chunks USING HNSW (embedding vector_l2_ops)
WITH (M = 16, ef_construction = 64);
""")
conn.commit()
for chunk_data in chunks_with_embeddings:
try:
cur.execute(
"""
INSERT INTO chunks (post_id, chunk_index, text, embedding)
VALUES (%s, %s, %s, %s)
ON CONFLICT (post_id, chunk_index) DO UPDATE SET
text = EXCLUDED.text,
embedding = EXCLUDED.embedding,
updated_at = NOW();
""",
(chunk_data["post_id"], chunk_data["chunk_index"], chunk_data["text"], chunk_data["embedding"].tolist())
)
except psycopg2.Error as e:
print(f"Database error during upsert for post_id {chunk_data['post_id']}, chunk_index {chunk_data['chunk_index']}: {e}")
conn.rollback() # Rollback specific error
continue
conn.commit()
print(f"Persisted {len(chunks_with_embeddings)} chunks.")
except Exception as e:
print(f"An error occurred while persisting chunks: {e}")
conn.rollback()
finally:
if conn:
conn.close()
Step 5 — Build the Retrieval API
With our knowledge base ingested and vectorized, the next step is to expose a simple HTTP contract for downstream services to retrieve relevant information. This is where FastAPI shines, allowing us to quickly build a high-performance, asynchronous API. Our retrieval API will have a single endpoint, `/summarize`, that accepts a `query` string. Inside this endpoint, we'll embed the user's query using the same `SentenceTransformer` model, then use `pgVector`'s `ORDER BY embedding <=> query_vec LIMIT N` syntax to find the `N` most similar chunks. The `<=>` operator calculates the L2 distance (Euclidean distance) between vectors, which `pgVector` efficiently optimizes with the HNSW index.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import json
import redis
import uvicorn
# Initialize FastAPI app
app = FastAPI()
# Initialize Redis client (replace with your actual config)
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=0)
class SummarizeResponse(BaseModel):
summary: str
sources: list[dict]
@app.get("/summarize", response_model=SummarizeResponse)
async def summarize_query(query: str):
# Check cache first
cache_key = f"summary:{query}"
cached_result = redis_client.get(cache_key)
if cached_result:
print(f"Cache hit for query: {query}")
return json.loads(cached_result)
print(f"Cache miss for query: {query}. Performing retrieval and summarization.")
try:
# Embed the query
query_embedding = embedding_model.encode(query, convert_to_tensor=False).astype(np.float32)
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) # Get dicts instead of tuples
# Perform similarity search
# We cast query_embedding to text for the SQL query, pgvector handles it
cur.execute(
"""
SELECT c.text, p.title, p.link
FROM chunks c
JOIN posts p ON c.post_id = p.id
ORDER BY c.embedding <=> %s::vector
LIMIT 3;
""",
(query_embedding.tolist(),)
)
retrieved_chunks = cur.fetchall()
conn.close()
if not retrieved_chunks:
raise HTTPException(status_code=404, detail="No relevant content found.")
# Concatenate retrieved texts
context_text = "\n".join([chunk["text"] for chunk in retrieved_chunks])
# Prepare sources for response
sources = [{"title": chunk["title"], "link": chunk["link"]} for chunk in retrieved_chunks]
# Summarize (Step 6 will fill this in)
summary_text = await generate_summary(context_text) # Await the async function
response = SummarizeResponse(summary=summary_text, sources=sources)
# Cache the result for 1 hour (3600 seconds)
redis_client.setex(cache_key, 3600, response.json())
return response
except redis.exceptions.ConnectionError as e:
print(f"Redis connection error: {e}. Proceeding without cache.")
# Fallback logic if Redis is down, or just re-raise if cache is critical
raise HTTPException(status_code=500, detail="Internal server error (Redis unavailable).")
except psycopg2.Error as e:
print(f"Database error during retrieval: {e}")
raise HTTPException(status_code=500, detail="Internal server error (database).")
except Exception as e:
print(f"An unexpected error occurred in /summarize: {e}")
raise HTTPException(status_code=500, detail="Internal server error.")
Step 6 — Summarize with a Generative Model
Retrieving relevant chunks is only half the battle; the problem now is converting that raw, often disjointed text into a concise, human-readable summary. For this, I integrated a generative model. While larger, more powerful models like GPT-4 would offer superior quality, for a self-contained, easily deployable RAG system, a smaller local model like `GPT-2` is an excellent choice. It's cheap to run, doesn't require external API calls, and is sufficient for many internal use cases. I loaded `GPT2LMHeadModel` and its corresponding `GPT2Tokenizer`. The trick to getting good summaries from smaller models is a well-crafted prompt. I prepended `"Summarize the following content in 2 sentences:\n"` to the concatenated context. The `max_length` parameter during generation helps control the verbosity of the summary.
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
# Load GPT-2 model and tokenizer once
gpt2_tokenizer = AutoTokenizer.from_pretrained("gpt2")
gpt2_model = AutoModelForCausalLM.from_pretrained("gpt2")
# Add a padding token if it doesn't exist (common for GPT-2 with batching)
if gpt2_tokenizer.pad_token is None:
gpt2_tokenizer.add_special_tokens({'pad_token': '[PAD]'})
gpt2_model.resize_token_embeddings(len(gpt2_tokenizer))
async def generate_summary(context: str) -> str:
prompt = f"Summarize the following content in 2 sentences:\n{context}\nSummary:"
inputs = gpt2_tokenizer(prompt, return_tensors="pt", max_length=1024, truncation=True)
# Generate summary
with torch.no_grad(): # Disable gradient calculation for inference
output = gpt2_model.generate(
**inputs,
max_new_tokens=80, # Generate up to 80 new tokens for the summary
num_beams=1, # Simple greedy search for speed
do_sample=False,
temperature=1.0,
pad_token_id=gpt2_tokenizer.pad_token_id # Use the pad token
)
summary = gpt2_tokenizer.decode(output[0], skip_special_tokens=True)
# Post-process: remove the original prompt from the generated text
# The generation might include the prompt itself if max_length is too short or model behavior
if summary.startswith(prompt):
summary = summary[len(prompt):].strip()
# Basic cleanup to ensure it looks like a summary
if summary.lower().startswith("summary:"):
summary = summary[len("summary:"):].strip()
# Ensure it doesn't end abruptly or with incomplete sentences
# For a small model like GPT-2, this might require more sophisticated post-processing
return summary.split(".")[0] + "." if "." in summary else summary # Simple truncation to first sentence if needed
Step 7 — Cache Results in Redis
A common problem in production APIs is redundant computation. If multiple users ask the exact same question, there's no need to re-run the retrieval and summarization steps every time. This is where caching comes in. I integrated Redis to store the results of previous `/summarize` queries. When a request comes in, the API first checks Redis using the query string as a key. If a result is found and hasn't expired (`TTL`), it's returned immediately. If not, the full retrieval and summarization process runs, and the result is then stored in Redis with a time-to-live (`TTL`). This dramatically reduces latency for frequently asked questions and offloads the database and LLM.
# ... (FastAPI app definition and imports from Step 5) ...
# The caching logic is already integrated into the /summarize endpoint in Step 5.
# Here's a reminder of the key parts:
# @app.get("/summarize", response_model=SummarizeResponse)
# async def summarize_query(query: str):
# cache_key = f"summary:{query}"
# cached_result = redis_client.get(cache_key)
# if cached_result:
# print(f"Cache hit for query: {query}")
# return json.loads(cached_result)
#
# # ... (perform retrieval and summarization) ...
#
# response = SummarizeResponse(summary=summary_text, sources=sources)
# redis_client.setex(cache_key, 3600, response.json()) # Cache for 1 hour
# return response
Step 8 — CI/CD & Zero‑Downtime Deployments
This is where the "zero-downtime" promise of the title truly comes into play. Building the pipeline is one thing; deploying updates without interrupting service is another. The problem is ensuring that new code, new models, or updated data pipelines can be rolled out seamlessly. I designed a CI/CD workflow, typically implemented with GitHub Actions, that handles linting, testing, Docker image building, and a blue-green deployment strategy on Kubernetes. Here's how such a workflow would operate for this RAG service: 1. **Linting and Unit Tests:** On every pull request or push to `main`, run `ruff` for linting and `pytest` for unit tests. This catches basic errors and regressions early. 2. **Docker Image Build & Push:** If tests pass, build a Docker image for the FastAPI application. The image is tagged with the Git SHA (`mausamadhikari/stripe-rag:Complete Script
#!/usr/bin/env python3
import os
import json
import feedparser
from datetime import datetime
import psycopg2
import psycopg2.extras
import numpy as np
import nltk
from transformers import AutoTokenizer, AutoModelForCausalLM
from sentence_transformers import SentenceTransformer
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import redis
import uvicorn
import torch
import logging
# --- Configuration ---
RSS_URL = "https://stripe.com/blog/feed.rss"
MAX_CHUNK_TOKENS = 256 # For SentenceTransformer input, typically up to 512
EMBEDDING_MODEL_NAME = 'all-MiniLM-L6-v2'
GENERATION_MODEL_NAME = 'gpt2'
REDIS_TTL_SECONDS = 3600 # 1 hour
# Database connection details
DB_HOST = os.getenv("DB_HOST", "localhost")
DB_NAME = os.getenv("DB_NAME", "rag_db")
DB_USER = os.getenv("DB_USER", "user")
DB_PASSWORD = os.getenv("DB_PASSWORD", "password")
# Redis connection details
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
# --- Setup Logging ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# --- NLTK Download ---
try:
nltk.data.find('tokenizers/punkt')
except nltk.downloader.DownloadError:
logger.info("Downloading NLTK punkt tokenizer...")
nltk.download('punkt')
# --- Global Model & Tokenizer Instances ---
try:
embedding_model = SentenceTransformer(EMBEDDING_MODEL_NAME)
gpt2_tokenizer = AutoTokenizer.from_pretrained(GENERATION_MODEL_NAME)
gpt2_model = AutoModelForCausalLM.from_pretrained(GENERATION_MODEL_NAME)
if gpt2_tokenizer.pad_token is None:
gpt2_tokenizer.add_special_tokens({'pad_token': '[PAD]'})
gpt2_model.resize_token_embeddings(len(gpt2_tokenizer))
chunking_tokenizer = AutoTokenizer.from_pretrained("gpt2") # Separate tokenizer for token counting
except Exception as e:
logger.error(f"Failed to load models: {e}")
# In a real app, you might want to exit or signal unhealthiness
embedding_model = None
gpt2_tokenizer = None
gpt2_model = None
chunking_tokenizer = None
# --- Database Utilities ---
def get_db_connection():
try:
conn = psycopg2.connect(host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASSWORD)
return conn
except psycopg2.Error as e:
logger.error(f"Database connection error: {e}")
raise
def initialize_db():
conn = None
try:
conn = get_db_connection()
cur = conn.cursor()
cur.execute("CREATE EXTENSION IF NOT EXISTS vector;")
cur.execute("""
CREATE TABLE IF NOT EXISTS posts (
id SERIAL PRIMARY KEY,
guid TEXT UNIQUE,
title TEXT,
link TEXT,
published TIMESTAMP WITH TIME ZONE,
summary TEXT,
raw_content TEXT,
ingested_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS chunks (
id SERIAL PRIMARY KEY,
post_id INTEGER REFERENCES posts(id),
chunk_index INTEGER,
text TEXT,
embedding VECTOR(384),
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
UNIQUE (post_id, chunk_index)
);
""")
# Create an HNSW index for efficient similarity search
cur.execute("""
CREATE INDEX IF NOT EXISTS chunks_embedding_hnsw_idx
ON chunks USING HNSW (embedding vector_l2_ops)
WITH (M = 16, ef_construction = 64);
""")
conn.commit()
logger.info("Database initialized successfully.")
except psycopg2.Error as e:
logger.error(f"Error initializing database: {e}")
raise
finally:
if conn:
conn.close()
# --- Step 1: Ingest RSS Feed ---
def ingest_rss_feed_to_db(rss_url: str):
logger.info(f"Ingesting RSS feed from {rss_url}...")
conn = None
try:
feed = feedparser.parse(rss_url)
if feed.bozo:
logger.warning(f"RSS feed parsing issues for {rss_url}: {feed.bozo_exception}")
conn = get_db_connection()
cur = conn.cursor()
processed_count = 0
for entry in feed.entries:
guid = entry.get('id') or entry.get('link') # Fallback to link if no explicit ID
title = entry.get('title')
link = entry.get('link')
summary = entry.get('summary') or entry.get('description')
raw_content = entry.get('content', [{}])[0].get('value') if entry.get('content') else summary # Use content if available
if not title or not link or not guid:
logger.warning(f"Skipping malformed entry: {entry.get('title') or entry.get('link')}")
continue
published_raw = entry.get('published')
published_date = None
if published_raw:
try:
published_date = datetime(*entry.published_parsed[:6])
except Exception as e:
logger.warning(f"Could not parse published date '{published_raw}': {e}. Using current time.")
published_date = datetime.now()
else:
published_date = datetime.now()
try:
cur.execute(
"""
INSERT INTO posts (guid, title, link, published, summary, raw_content)
VALUES (%s, %s, %s, %s, %s, %s)
ON CONFLICT (guid) DO UPDATE SET
title = EXCLUDED.title,
link = EXCLUDED.link,
published = EXCLUDED.published,
summary = EXCLUDED.summary,
raw_content = EXCLUDED.raw_content,
ingested_at = NOW()
RETURNING id;
""",
(guid, title, link, published_date, summary, raw_content)
)
post_id = cur.fetchone()[0] # Get the ID of the inserted/updated post
processed_count += 1
except psycopg2.Error as e:
logger.error(f"Database error during upsert for GUID {guid}: {e}")
conn.rollback()
continue
conn.commit()
logger.info(f"Ingestion complete. Processed {processed_count} new/updated entries.")
except feedparser.HTTPError as e:
logger.error(f"HTTP Error fetching RSS feed: {e}")
except Exception as e:
logger.error(f"An unexpected error occurred during RSS ingestion: {e}")
finally:
if conn:
conn.close()
# --- Step 2: Clean & Chunk Text ---
def clean_and_chunk_text(post_id: int, text: str):
if not text:
return []
sentences = nltk.sent_tokenize(text)
chunks = []
current_chunk_sentences = []
current_chunk_token_count = 0
for sentence in sentences:
sentence_tokens = chunking_tokenizer.encode(sentence, add_special_tokens=False)
sentence_token_count = len(sentence_tokens)
if current_chunk_token_count + sentence_token_count <= MAX_CHUNK_TOKENS:
current_chunk_sentences.append(sentence)
current_chunk_token_count += sentence_token_count
else:
if current_chunk_sentences:
chunks.append(" ".join(current_chunk_sentences))
current_chunk_sentences = [sentence]
current_chunk_token_count = sentence_token_count
if sentence_token_count > MAX_CHUNK_TOKENS:
logger.warning(f"Single sentence in post {post_id} exceeds MAX_CHUNK_TOKENS. Truncating.")
truncated_tokens = chunking_tokenizer.encode(sentence, add_special_tokens=False)[:MAX_CHUNK_TOKENS]
chunks.append(chunking_tokenizer.decode(truncated_tokens))
current_chunk_sentences = []
current_chunk_token_count = 0
if current_chunk_sentences:
chunks.append(" ".join(current_chunk_sentences))
chunked_data = []
for i, chunk_text in enumerate(chunks):
chunked_data.append({
"post_id": post_id,
"chunk_index": i,
"text": chunk_text
})
return chunked_data
# --- Step 3: Generate Dense Embeddings ---
def generate_embeddings(chunks_data: list[dict]):
if not chunks_data or embedding_model is None:
return []
texts = [chunk["text"] for chunk in chunks_data]
embeddings = embedding_model.encode(texts, convert_to_tensor=False, show_progress_bar=False) # No progress bar in API
for i, chunk in enumerate(chunks_data):
chunk["embedding"] = embeddings[i].astype(np.float32)
return chunks_data
# --- Step 4: Persist Vectors in PostgreSQL ---
def persist_chunks_with_embeddings(chunks_with_embeddings: list[dict]):
if not chunks_with_embeddings:
return
conn = None
try:
conn = get_db_connection()
cur = conn.cursor()
for chunk_data in chunks_with_embeddings:
try:
cur.execute(
"""
INSERT INTO chunks (post_id, chunk_index, text, embedding)
VALUES (%s, %s, %s, %s)
ON CONFLICT (post_id, chunk_index) DO UPDATE SET
text = EXCLUDED.text,
embedding = EXCLUDED.embedding,
updated_at = NOW();
""",
(chunk_data["post_id"], chunk_data["chunk_index"], chunk_data["text"].strip(), chunk_data["embedding"].tolist())
)
except psycopg2.Error as e:
logger.error(f"Database error during upsert for post_id {chunk_data['post_id']}, chunk_index {chunk_data['chunk_index']}: {e}")
conn.rollback()
continue
conn.commit()
logger.info(f"Persisted {len(chunks_with_embeddings)} chunks.")
except Exception as e:
logger.error(f"An error occurred while persisting chunks: {e}")
if conn:
conn.rollback()
finally:
if conn:
conn.close()
# --- Step 6: Summarize with a Generative Model (Async for FastAPI) ---
async def generate_summary(context: str) -> str:
if gpt2_model is None or gpt2_tokenizer is None:
raise RuntimeError("Generative model not loaded.")
prompt = f"Summarize the following content in 2 sentences:\n{context}\nSummary:"
inputs = gpt2_tokenizer(prompt, return_tensors="pt", max_length=1024, truncation=True)
with torch.no_grad():
output = gpt2_model.generate(
**inputs,
max_new_tokens=80,
num_beams=1,
do_sample=False,
temperature=1.0,
pad_token_id=gpt2_tokenizer.pad_token_id
)
summary = gpt2_tokenizer.decode(output[0], skip_special_tokens=True)
if summary.startswith(prompt):
summary = summary[len(prompt):].strip()
if summary.lower().startswith("summary:"):
summary = summary[len("summary:"):].strip()
# Simple post-processing to ensure a complete sentence or two
sentences = summary.split('.')
if len(sentences) > 1 and sentences[1].strip():
return sentences[0].strip() + ". " + sentences[1].strip() + "."
elif sentences[0].strip():
return sentences[0].strip() + "."
return summary.strip()
# --- Step 5 & 7: Build Retrieval API & Cache Results in Redis ---
app = FastAPI()
try:
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=0)
redis_client.ping() # Test connection
logger.info("Connected to Redis successfully.")
except redis.exceptions.ConnectionError as e:
logger.error(f"Could not connect to Redis: {e}. Caching will be disabled.")
redis_client = None
class SummarizeResponse(BaseModel):
summary: str
sources: list[dict]
@app.get("/summarize", response_model=SummarizeResponse)
async def summarize_query(query: str):
if embedding_model is None:
raise HTTPException(status_code=503, detail="Embedding model not loaded. Service unavailable.")
if gpt2_model is None:
raise HTTPException(status_code=503, detail="Generative model not loaded. Service unavailable.")
cache_key = f"summary:{query}"
if redis_client:
try:
cached_result = redis_client.get(cache_key)
if cached_result:
logger.info(f"Cache hit for query: {query}")
return json.loads(cached_result)
except redis.exceptions.ConnectionError as e:
logger.warning(f"Redis error during cache lookup: {e}. Proceeding without cache.")
redis_client = None # Disable Redis for this request if connection fails
logger.info(f"Cache miss for query: {query}. Performing retrieval and summarization.")
conn = None
try:
query_embedding = embedding_model.encode(query, convert_to_tensor=False).astype(np.float32)
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cur.execute(
"""
SELECT c.text, p.title, p.link
FROM chunks c
JOIN posts p ON c.post_id = p.id
ORDER BY c.embedding <=> %s::vector
LIMIT 3;
""",
(query_embedding.tolist(),)
)
retrieved_chunks = cur.fetchall()
if not retrieved_chunks:
raise HTTPException(status_code=404, detail="No relevant content found for your query.")
context_text = "\n".join([chunk["text"] for chunk in retrieved_chunks])
sources = [{"title": chunk["title"], "link": chunk["link"]} for chunk in retrieved_chunks]
summary_text = await generate_summary(context_text)
response = SummarizeResponse(summary=summary_text, sources=sources)
if redis_client:
try:
redis_client.setex(cache_key, REDIS_TTL_SECONDS, response.json())
except redis.exceptions.ConnectionError as e:
logger.warning(f"Redis error during cache set: {e}. Result not cached.")
redis_client = None
return response
except psycopg2.Error as e:
logger.error(f"Database error during retrieval: {e}")
raise HTTPException(status_code=500, detail="Internal server error (database issue).")
except RuntimeError as e: # Catch model loading errors
logger.error(f"Model error: {e}")
raise HTTPException(status_code=500, detail="Internal server error (model issue).")
except Exception as e:
logger.error(f"An unexpected error occurred in /summarize: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error.")
finally:
if conn:
conn.close()
# --- Main Orchestration Function ---
def run_ingestion_pipeline():
initialize_db()
ingest_rss_feed_to_db(RSS_URL)
conn = None
try:
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
# Select posts that have raw_content and haven't been processed into chunks recently,
# or where the post itself was recently updated.
# This is a simplified logic; in production, you'd track processed state more carefully.
cur.execute("""
SELECT id, raw_content FROM posts
WHERE raw_content IS NOT NULL AND raw_content != ''
AND (
NOT EXISTS (SELECT 1 FROM chunks WHERE chunks.post_id = posts.id)
OR posts.ingested_at > (SELECT MAX(updated_at) FROM chunks WHERE chunks.post_id = posts.id)
)
ORDER BY posts.ingested_at DESC
LIMIT 10; -- Process a batch to avoid overwhelming resources
""")
posts_to_process = cur.fetchall()
logger.info(f"Found {len(posts_to_process)} posts to chunk and embed.")
all_chunks_to_persist = []
for post in posts_to_process:
chunks = clean_and_chunk_text(post["id"], post["raw_content"])
chunks_with_embeddings = generate_embeddings(chunks)
all_chunks_to_persist.extend(chunks_with_embeddings)
if all_chunks_to_persist:
persist_chunks_with_embeddings(all_chunks_to_persist)
else:
logger.info("No new chunks to persist.")
except Exception as e:
logger.error(f"Error in ingestion pipeline orchestration: {e}")
finally:
if conn:
conn.close()
if __name__ == "__main__":
# To run the ingestion pipeline:
# python your_script_name.py --ingest
# To run the FastAPI server:
# python your_script_name.py --serve
# Or simply: uvicorn your_script_name:app --host 0.0.0.0 --port 8000
import argparse
parser = argparse.ArgumentParser(description="Stripe RAG Pipeline and API")
parser.add_argument("--ingest", action="store_true", help="Run the data ingestion and embedding pipeline.")
parser.add_argument("--serve", action="store_true", help="Run the FastAPI server.")
args = parser.parse_args()
if args.ingest:
logger.info("Starting ingestion pipeline...")
run_ingestion_pipeline()
logger.info("Ingestion pipeline finished.")
elif args.serve:
logger.info("Starting FastAPI server...")
uvicorn.run(app, host="0.0.0.0", port=8000)
else:
logger.info("No action specified. Use --ingest or --serve.")
logger.info("For ingestion: python script.py --ingest")
logger.info("For API server: python script.py --serve or uvicorn script:app --host 0.0.0.0 --port 8000")
Expected Output
When you run the ingestion pipeline (`python your_script_name.py --ingest`), you should see logs similar to this:
2026-10-27 10:00:00,000 - INFO - Database initialized successfully.
2026-10-27 10:00:00,005 - INFO - Ingesting RSS feed from https://stripe.com/blog/feed.rss...
2026-10-27 10:00:02,500 - INFO - Ingestion complete. Processed 50 new/updated entries.
2026-10-27 10:00:02,510 - INFO - Found 10 posts to chunk and embed.
2026-10-27 10:00:03,100 - INFO - Persisted 250 chunks.
2026-10-27 10:00:03,105 - INFO - Ingestion pipeline finished.
INFO: Started server process [PID]
INFO: Waiting for application startup.
INFO: Connected to Redis successfully.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
{
"summary": "Stripe's latest survey reveals key trends shaping how SaaS platforms compete in the AI era. AI is driving increased competition and platforms are feeling pressure to evolve their strategies.",
"sources": [
{
"title": "Stripe survey: Four trends reshaping how SaaS platforms compete in the AI era",
"link": "https://stripe.com/blog/saas-ai-trends"
},
{
"title": "The “alien” economics of AI monetization",
"link": "https://stripe.com/blog/ai-monetization-economics"
}
]
}
INFO: Cache miss for query: Stripe AI trends. Performing retrieval and summarization.
INFO: 127.0.0.1:54321 - "GET /summarize?query=Stripe%20AI%20trends HTTP/1.1" 200 OK
Limitations and Tradeoffs
This full-stack RAG pipeline is powerful, but it comes with inherent tradeoffs and limitations: * **Local GPT-2 Quality:** While convenient and cost-effective, a local `GPT-2` model produces lower-quality and less nuanced summaries compared to state-of-the-art models like GPT-4 or Claude. For high-stakes applications or nuanced content, an external, more powerful LLM API might be necessary, introducing API costs and external dependency management. * **pgVector Scalability:** `pgVector` is excellent up to a few million vectors, especially with proper indexing. Beyond that, or for very high QPS (queries per second) requirements, dedicated ANN services like Milvus, Weaviate, or Pinecone might offer better performance and features (e.g., distributed indexing, advanced filtering). The simplicity of `pgVector` is a strong advantage for smaller to medium-sized datasets. * **RSS Feed Fragility:** RSS feeds can change schema or disappear without notice. The ingestion step must be robust to `feedparser` errors, but significant changes to the content structure (e.g., moving from `summary` to `description` or completely restructuring the article content within `raw_content`) might require code updates. * **Zero-Downtime Complexity:** True zero-downtime deployments require careful management of API contract compatibility between old and new container versions. Breaking API changes need a more complex migration strategy, potentially involving parallel deployments and gradual traffic shifting, which adds significant complexity beyond a simple blue-green switch. * **Chunking Strategy:** The sentence-aware chunking is good, but not perfect. It might split semantically related paragraphs or include irrelevant sentences if a single sentence pushes a chunk over the token limit. More advanced chunking (e.g., recursive chunking, or using semantic boundaries) could improve retrieval quality. * **Model Freshness:** While data ingestion is dynamic, the embedding and generative models themselves are static. Updating `all-MiniLM-L6-v2` or `GPT-2` would require rebuilding the Docker image and re-deploying, which is handled by the CI/CD but is not "dynamic" in the same way as data ingestion.Frequently Asked Questions
How often should I run the ingestion pipeline?
The frequency depends on how quickly your source data changes and how "fresh" your RAG system needs to be. For a blog, running it every few hours (e.g., via a daily cron job or a Prefect flow scheduled every 4-6 hours) is usually sufficient. For real-time updates, you'd need a push-based system (e.g., webhooks) rather than pull-based RSS.
What if a blog post is updated? Will the RAG system reflect the changes?
Yes, due to the `ON CONFLICT (guid) DO UPDATE` logic in the `posts` table and `ON CONFLICT (post_id, chunk_index) DO UPDATE` in the `chunks` table. When the ingestion pipeline runs, if an RSS entry's `guid` matches an existing post but its content has changed, the `raw_content` will be updated. The subsequent chunking and embedding process (triggered by `posts.ingested_at > (SELECT MAX(updated_at) FROM chunks...)` in `run_ingestion_pipeline`) will then re-process and re-embed the updated content, overwriting the old chunks and embeddings.
Can I use a larger language model like GPT-4 instead of GPT-2?
Absolutely. For higher quality summaries, you would replace the `GPT2LMHeadModel` and `GPT2Tokenizer` logic with calls to an external API (e.g., OpenAI, Anthropic). This would involve managing API keys, handling rate limits, and potentially adjusting the `generate_summary` function to be an `async` HTTP client call instead of a local model inference. The rest of the RAG pipeline (ingestion, embedding, retrieval, caching) would remain largely the same.
How can I monitor the health of this RAG pipeline in production?
You would integrate standard MLOps monitoring tools. For the API, this means Prometheus/Grafana for metrics (request latency, error rates, cache hit ratio), and Loki/Elasticsearch for centralized logging. For the ingestion pipeline, monitor job success/failure, data volume ingested, number of new/updated chunks, and embedding generation times. Health checks on the FastAPI endpoint (e.g., `/health`) that verify database and Redis connectivity are also crucial for Kubernetes to manage deployments.
What if I have multiple RSS feeds or other data sources?
The `ingest_rss_feed_to_db` function can be easily adapted to accept a list of RSS URLs. For other data sources (e.g., web scraping, internal wikis, PDFs), you would create separate ingestion modules, each responsible for extracting text and metadata from its specific source, and then feeding that content into the same chunking, embedding, and persistence pipeline. Normalizing the extracted `title`, `link`, and `content` across sources is key.
What I'd Change
If I were to take this system to the next level for a truly high-scale, enterprise-grade RAG, my first priority would be to **decouple the ingestion and embedding process from the API server completely**. Currently, `run_ingestion_pipeline` is a standalone script, but in a more robust setup, it would...leading to answers that often miss the nuance or fail to capture the full context of what the user is truly asking. This is a common pitfall I've observed when trying to build a reliable RAG system on a constantly evolving content source like the Stripe Engineering Blog. The challenge isn't just indexing new content quickly, but intelligently surfacing the *right* content when queries are ambiguous, complex, or require synthesizing information across multiple, distinct articles. This is where a multi-stage retrieval pipeline becomes indispensable. We need to go beyond simple vector similarity and introduce mechanisms that adapt the query itself and then re-evaluate the retrieved candidates. Today, I want to walk you through how I built a more robust RAG system for the Stripe blog posts, focusing on two key techniques: **intelligent query rewriting** and **semantic re-ranking**, which together form a powerful adaptive retrieval strategy.Key Takeaways
- Building robust RAG for dynamic, heterogeneous content requires more than just embedding and retrieval; it demands an adaptive, multi-stage pipeline.
- Intelligent query rewriting, often powered by an LLM, can significantly improve retrieval by clarifying ambiguous user intent or expanding on underspecified queries.
- Semantic re-ranking using a cross-encoder model is crucial for refining initial retrieval results, prioritizing documents that are most relevant to the query's true meaning, even if their initial vector distance wasn't the absolute closest.
- Combining hybrid retrieval (vector + keyword) with query rewriting and re-ranking forms a powerful, flexible RAG architecture that performs well across diverse query types.
from transformers import pipeline
# Placeholder for an LLM that rewrites queries
# In a real scenario, this would be a more sophisticated model or API call
def rewrite_query_with_llm(original_query: str) -> list[str]:
"""
Simulates an LLM rewriting a query into multiple variations.
In production, this would use a robust LLM API.
"""
if "fintech global expansion" in original_query.lower():
return [
original_query,
"How can fintech companies expand their operations internationally?",
"What are the strategies for global market entry for financial technology firms?",
"New methods for fintechs to move money across borders."
]
elif "ai monetization" in original_query.lower():
return [
original_query,
"How are AI companies generating revenue?",
"What are the pricing models for AI products and services?",
"Economic aspects of monetizing artificial intelligence."
]
else:
# Fallback to just the original query if no specific rewrite rule
return [original_query]
# Example usage (not part of final script, just illustration)
# user_query = "Fintech global expansion"
# rewritten_queries = rewrite_query_with_llm(user_query)
# print(rewritten_queries)
import feedparser
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer
import re
# Initialize embedding model (using a smaller, faster model for demonstration)
# In production, consider 'all-MiniLM-L6-v2' or 'multi-qa-MiniLM-L6-cos-v1'
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
def create_chunks_and_embeddings(entries):
chunks = []
for entry in entries:
title = entry.title
summary = getattr(entry, 'summary', '') # Use getattr for safety
link = entry.link
# Simple chunking: title + summary. In production, more sophisticated chunking
# strategies would be used, potentially splitting long summaries.
content = f"Title: {title}\nSummary: {summary}\nLink: {link}"
chunks.append({"content": content, "metadata": {"title": title, "link": link}})
# Generate embeddings for all chunks
chunk_contents = [chunk["content"] for chunk in chunks]
embeddings = embedding_model.encode(chunk_contents, convert_to_tensor=True).cpu().numpy()
return chunks, embeddings
def build_faiss_index(embeddings):
dimension = embeddings.shape[1]
index = faiss.IndexFlatL2(dimension)
index.add(embeddings)
return index
def vector_search(query_embedding, index, k=5):
D, I = index.search(np.array([query_embedding]), k)
return I[0], D[0] # Return indices and distances
def keyword_search(query: str, chunks: list[dict], top_n=5) -> list[int]:
"""
Performs a simple keyword search across chunk content.
Returns indices of matching chunks.
"""
matches = []
query_terms = query.lower().split()
for i, chunk in enumerate(chunks):
content_lower = chunk["content"].lower()
# Simple heuristic: count how many query terms are in the chunk
score = sum(1 for term in query_terms if term in content_lower)
if score > 0:
matches.append((score, i))
matches.sort(key=lambda x: x[0], reverse=True)
return [m[1] for m in matches[:top_n]]
# (Further code for combining results will be in the complete script)
from sentence_transformers import CrossEncoder
# Initialize a cross-encoder model for re-ranking
# 'cross-encoder/ms-marco-MiniLM-L-6-v2' is a good general-purpose re-ranker
cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
def re_rank_documents(query: str, candidate_chunks: list[dict], top_k=5) -> list[dict]:
"""
Re-ranks candidate chunks based on their relevance to the query using a cross-encoder.
"""
if not candidate_chunks:
return []
# Prepare pairs for the cross-encoder: [(query, chunk_content), ...]
sentences = [(query, chunk["content"]) for chunk in candidate_chunks]
# Get relevance scores
scores = cross_encoder.predict(sentences)
# Pair chunks with their scores and sort
scored_chunks = sorted(zip(scores, candidate_chunks), key=lambda x: x[0], reverse=True)
# Return top_k re-ranked chunks
return [chunk for score, chunk in scored_chunks[:top_k]]
# (This function will be integrated into the complete script)
Data Sources & References
- Stripe Blog RSS Feed: https://stripe.com/blog/feed.rss
feedparserlibrary: https://pypi.org/project/feedparser/sentence-transformersdocumentation: https://www.sbert.net/faiss-cpudocumentation: https://github.com/facebookresearch/faiss- Cross-Encoder Models: https://www.sbert.net/docs/pretrained_models.html#cross-encoders
Data accessed from Stripe Blog RSS on 2024-07-28.
import feedparser
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer, CrossEncoder
import re
import os
import requests
# --- Configuration ---
STRIPE_BLOG_RSS_URL = 'https://stripe.com/blog/feed.rss'
EMBEDDING_MODEL_NAME = 'all-MiniLM-L6-v2' # Smaller, faster bi-encoder
CROSS_ENCODER_MODEL_NAME = 'cross-encoder/ms-marco-MiniLM-L-6-v2' # Re-ranker
NUM_INITIAL_RETRIEVAL = 10 # Number of candidates from vector/keyword search
NUM_RE_RANKED_RESULTS = 5 # Final number of results after re-ranking
# --- Model Initialization (global to avoid re-loading) ---
embedding_model = None
cross_encoder_model = None
def initialize_models():
global embedding_model, cross_encoder_model
if embedding_model is None:
print(f"Loading embedding model: {EMBEDDING_MODEL_NAME}...")
embedding_model = SentenceTransformer(EMBEDDING_MODEL_NAME)
if cross_encoder_model is None:
print(f"Loading cross-encoder model: {CROSS_ENCODER_MODEL_NAME}...")
cross_encoder_model = CrossEncoder(CROSS_ENCODER_MODEL_NAME)
print("Models loaded.")
# --- Data Fetching and Processing ---
def fetch_and_parse_rss(url: str) -> list:
"""Fetches and parses an RSS feed."""
print(f"Fetching RSS feed from {url}...")
try:
# Using requests to handle potential connection issues more robustly
response = requests.get(url, timeout=10)
response.raise_for_status() # Raise an exception for HTTP errors
feed = feedparser.parse(response.text)
if feed.bozo:
print(f"Warning: RSS feed parsing issues detected: {feed.bozo_exception}")
print(f"Found {len(feed.entries)} entries.")
return feed.entries
except requests.exceptions.RequestException as e:
print(f"Error fetching RSS feed: {e}")
return []
except Exception as e:
print(f"An unexpected error occurred during RSS parsing: {e}")
return []
def create_chunks_and_embeddings(entries: list):
"""
Creates document chunks from RSS entries and generates their embeddings.
A more robust chunking strategy would be used in production.
"""
chunks = []
chunk_contents = []
for entry in entries:
title = entry.title
summary = getattr(entry, 'summary', '') # Use getattr for safety
link = entry.link
# Simple chunking: title + summary.
content = f"Title: {title}\nSummary: {summary}\nLink: {link}"
chunks.append({"content": content, "metadata": {"title": title, "link": link}})
chunk_contents.append(content)
if not chunk_contents:
return [], None
print(f"Generating embeddings for {len(chunk_contents)} chunks...")
embeddings = embedding_model.encode(chunk_contents, convert_to_tensor=True).cpu().numpy()
print("Embeddings generated.")
return chunks, embeddings
def build_faiss_index(embeddings: np.ndarray):
"""Builds a FAISS index from embeddings."""
if embeddings is None or embeddings.size == 0:
print("No embeddings to build FAISS index.")
return None
dimension = embeddings.shape[1]
index = faiss.IndexFlatL2(dimension) # Using L2 distance
index.add(embeddings)
print(f"FAISS index built with {index.ntotal} vectors.")
return index
# --- Retrieval Functions ---
def rewrite_query_with_llm(original_query: str) -> list[str]:
"""
Simulates an LLM rewriting a query into multiple variations.
In a real system, this would be an API call to a powerful LLM.
"""
print(f"Rewriting query: '{original_query}'")
# This is a hardcoded simulation for demonstration.
# In production, use a real LLM like OpenAI's GPT-4, Anthropic's Claude, etc.
# Example prompt for an LLM:
# "You are a helpful assistant. Rewrite the following user query into 3-5 diverse,
# semantically equivalent queries to improve retrieval from a knowledge base about
# technology and finance. Original query: '{original_query}'"
rewritten_queries = [original_query] # Always include original
if "fintech global expansion" in original_query.lower() or "fintechs expand globally" in original_query.lower():
rewritten_queries.extend([
"How can financial technology companies expand their operations internationally?",
"What are the strategies for global market entry for fintech firms?",
"New methods for fintechs to move money across borders or internationally."
])
elif "ai monetization" in original_query.lower():
rewritten_queries.extend([
"How are AI companies generating revenue?",
"What are the pricing models for AI products and services?",
"Economic aspects of monetizing artificial intelligence solutions."
])
elif "stripe data pipeline" in original_query.lower():
rewritten_queries.extend([
"How to sync real-time Stripe data to Postgres?",
"What is Stripe Data Pipeline and how does it work?",
"Integrating Stripe data with databases for applications."
])
print(f"Rewritten queries: {rewritten_queries}")
return list(set(rewritten_queries)) # Remove duplicates
def vector_search(query_embedding: np.ndarray, index: faiss.Index, k: int = 5) -> list[int]:
"""Performs vector search and returns indices of top k results."""
if index is None:
return []
D, I = index.search(np.array([query_embedding]), k)
return I[0].tolist()
def keyword_search(query: str, chunks: list[dict], top_n: int = 5) -> list[int]:
"""
Performs a simple keyword search across chunk content.
Returns indices of matching chunks.
In production, consider TF-IDF, BM25, or a dedicated search engine.
"""
matches = []
query_terms = [term for term in re.findall(r'\b\w+\b', query.lower()) if len(term) > 2] # Extract words, filter short ones
if not query_terms:
return []
for i, chunk in enumerate(chunks):
content_lower = chunk["content"].lower()
score = sum(1 for term in query_terms if term in content_lower)
if score > 0:
matches.append((score, i))
matches.sort(key=lambda x: x[0], reverse=True)
return [m[1] for m in matches[:top_n]]
def retrieve_candidates(queries: list[str], all_chunks: list[dict], faiss_index: faiss.Index, k_per_query: int = 5) -> list[dict]:
"""
Combines vector and keyword search results from multiple queries,
deduplicates, and returns a list of unique candidate chunks.
"""
candidate_indices = set()
for query in queries:
# Vector search
query_embedding = embedding_model.encode(query, convert_to_tensor=True).cpu().numpy()
vector_result_indices = vector_search(query_embedding, faiss_index, k=k_per_query)
candidate_indices.update(vector_result_indices)
# Keyword search
keyword_result_indices = keyword_search(query, all_chunks, top_n=k_per_query)
candidate_indices.update(keyword_result_indices)
# Convert indices to actual chunks and remove any invalid indices
unique_candidates = []
for idx in candidate_indices:
if 0 <= idx < len(all_chunks):
unique_candidates.append(all_chunks[idx])
print(f"Retrieved {len(unique_candidates)} unique candidate chunks.")
return unique_candidates
def re_rank_documents(original_query: str, candidate_chunks: list[dict], top_k: int = 5) -> list[dict]:
"""
Re-ranks candidate chunks based on their relevance to the original query
using a cross-encoder.
"""
if not candidate_chunks:
return []
print(f"Re-ranking {len(candidate_chunks)} candidates for query: '{original_query}'")
# Prepare pairs for the cross-encoder: [(query, chunk_content), ...]
sentences = [(original_query, chunk["content"]) for chunk in candidate_chunks]
# Get relevance scores
scores = cross_encoder_model.predict(sentences)
# Pair chunks with their scores and sort
scored_chunks = sorted(zip(scores, candidate_chunks), key=lambda x: x[0], reverse=True)
# Return top_k re-ranked chunks
re_ranked_results = [chunk for score, chunk in scored_chunks[:top_k]]
print(f"Finished re-ranking. Top {len(re_ranked_results)} results selected.")
return re_ranked_results
# --- Main Pipeline ---
def main():
initialize_models()
# 1. Fetch and process RSS data
entries = fetch_and_parse_rss(STRIPE_BLOG_RSS_URL)
if not entries:
print("No RSS entries to process. Exiting.")
return
all_chunks, embeddings = create_chunks_and_embeddings(entries)
if embeddings is None:
print("No embeddings created. Exiting.")
return
faiss_index = build_faiss_index(embeddings)
if faiss_index is None:
print("FAISS index could not be built. Exiting.")
return
# Example user queries
user_queries = [
"What are the new ways for fintechs to move money and expand globally?",
"Tell me about AI monetization strategies.",
"How can I sync Stripe data to Postgres?"
]
for user_query in user_queries:
print("\n--- Processing New Query ---")
print(f"User Query: {user_query}")
# 2. Query Rewriting
rewritten_queries = rewrite_query_with_llm(user_query)
# 3. Hybrid Retrieval
candidate_chunks = retrieve_candidates(rewritten_queries, all_chunks, faiss_index, k_per_query=NUM_INITIAL_RETRIEVAL)
if not candidate_chunks:
print("No candidate chunks found for retrieval.")
continue
# 4. Semantic Re-ranking
final_results = re_rank_documents(user_query, candidate_chunks, top_k=NUM_RE_RANKED_RESULTS)
# 5. Display Results
print(f"\n--- Top {len(final_results)} Re-ranked Results for '{user_query}' ---")
if not final_results:
print("No relevant results found after re-ranking.")
else:
for i, chunk in enumerate(final_results):
print(f"Result {i+1}:")
print(f" Title: {chunk['metadata']['title']}")
print(f" Link: {chunk['metadata']['link']}")
# print(f" Content snippet: {chunk['content'][:200]}...") # Optional: print snippet
print("-" * 20)
if __name__ == '__main__':
# Ensure models are downloaded/cached in a specific directory
# os.environ['SENTENCE_TRANSFORMERS_HOME'] = './.cache/sbert'
# os.makedirs(os.environ['SENTENCE_TRANSFORMERS_HOME'], exist_ok=True)
main()
Q: Why not just use a larger embedding model instead of rewriting and re-ranking?
A: While larger embedding models can capture more nuance, they still struggle with highly ambiguous queries or those that require a significant rephrasing to match the optimal representation in the vector space. Query rewriting directly addresses this by generating multiple "optimal" queries. Re-ranking, on the other hand, refines the *precision* among candidates, which is a different problem than the initial recall challenge that embedding models face.
Q: How do I choose the right LLM for query rewriting?
A: For production, consider models that are strong in instruction following, text generation, and have a good understanding of the domain (e.g., finance, technology). Models like GPT-4, Claude 3, or even fine-tuned open-source models (like Llama 3) can work well. The key is to craft a precise prompt that guides the LLM to generate diverse and relevant query reformulations without hallucinating.
Q: Is FAISS the only option for the vector store?
A: No, FAISS is a popular choice for in-memory or on-disk vector search due to its speed and efficiency, especially for similarity search. In a distributed production environment, you might use dedicated vector databases like Pinecone, Weaviate, Milvus, Qdrant, or even cloud-managed services like Azure Cognitive Search or Amazon OpenSearch with vector capabilities. The choice depends on scale, latency requirements, and infrastructure preferences.
Q: When should I *not* use this multi-stage approach?
A: If your RAG system handles very simple, keyword-rich queries, operates on a highly uniform dataset, or has extremely strict real-time latency requirements (e.g., sub-100ms), a simpler vector search might be more appropriate. The added complexity and latency of query rewriting and re-ranking are best justified when dealing with complex, ambiguous, or diverse queries against heterogeneous content where precision is paramount.