When my data‑science team needed instant answers about Stripe’s latest product moves, we hit a wall: pulling the entire blog dump every time we answered a query was both slow and costly, yet a static snapshot quickly went stale. I built a pipeline that watches the Stripe Blog RSS, chunks new articles, updates embeddings in FAISS on the fly, and serves answers with sub‑200 ms latency. If you’re a data‑centric team that wants a live knowledge base without the overhead of full re‑embedding, follow along.
Key Takeaways
- Incremental chunking + FAISS lets you add new documents in milliseconds.
- Separating retrieval (FAISS) from generation (LLM) reduces overall latency to <200 ms for typical queries.
- Simple caching and health checks catch most production hiccups before they affect users.
- Monitoring embedding drift and feed errors is essential for a reliable RAG service.
The Problem
Our analysts needed a “ask‑me‑anything” interface over Stripe’s public blog. The naïve approach—re‑download the feed, re‑chunk every article, re‑embed the entire corpus on each request—costs seconds per query and spikes cloud spend. The challenge is to keep the vector store up‑to‑date with only the newly published posts while guaranteeing low‑latency answers.
Data and Sources
We pull the official Stripe Blog RSS feed: https://stripe.com/blog/feed.rss. The feed is an XML document listing recent posts with title, link, and summary. Data accessed on 2026-11-03.
Step 1 — Fetch & Parse RSS
This step turns the raw XML into a list of article dictionaries, filtering out anything we already processed.
import feedparser
from datetime import datetime
import hashlib
def fetch_new_entries(seen_hashes: set) -> list[dict]:
feed = feedparser.parse('https://stripe.com/blog/feed.rss')
new = []
for entry in feed.entries:
# Use a stable hash of the entry link as the unique identifier
uid = hashlib.sha256(entry.link.encode()).hexdigest()
if uid not in seen_hashes:
new.append({
"id": uid,
"title": entry.title,
"link": entry.link,
"published": datetime(*entry.published_parsed[:6]),
"summary": entry.summary,
})
return new
We compute a SHA‑256 hash of the article URL; this survives minor feed ordering changes and gives us an O(1) membership test.
Step 2 — Incremental Chunking & Embedding
Each new article is split into 300‑token chunks using a simple sentence splitter. We embed only those chunks.
import re
from typing import List
from sentence_transformers import SentenceTransformer
EMBEDDER = SentenceTransformer('all-MiniLM-L6-v2') # lightweight, good for short texts
def chunk_text(text: str, max_len: int = 300) -> List[str]:
sentences = re.split(r'(?<=[.!?])\s+', text)
chunks = []
cur = ""
for s in sentences:
if len(cur) + len(s) < max_len:
cur += " " + s
else:
chunks.append(cur.strip())
cur = s
if cur:
chunks.append(cur.strip())
return chunks
def embed_chunks(chunks: List[str]) -> List[dict]:
vectors = EMBEDDER.encode(chunks, show_progress_bar=False)
return [{"text": txt, "vector": vec} for txt, vec in zip(chunks, vectors)]
The embedder runs locally, avoiding API latency. Because we only call it on fresh chunks, the cost scales with the feed’s publishing rate, not the total corpus size.
Step 3 — Persist Embeddings in FAISS
FAISS gives us fast approximate nearest‑neighbor search. We store the index on disk and load it at startup.
import faiss
import numpy as np
import os
import json
INDEX_PATH = "faiss.index"
META_PATH = "meta.json"
def load_index() -> tuple[faiss.IndexFlatIP, dict]:
if os.path.exists(INDEX_PATH):
index = faiss.read_index(INDEX_PATH)
with open(META_PATH) as f:
meta = json.load(f)
else:
# Dimensionality matches the embedder output (384 for MiniLM‑L6‑v2)
index = faiss.IndexFlatIP(384)
meta = {}
return index, meta
def add_to_index(index: faiss.IndexFlatIP, meta: dict, article: dict, embeddings: List[dict]):
vectors = np.vstack([e["vector"] for e in embeddings]).astype('float32')
index.add(vectors)
# Store mapping from vector offset to article metadata
start = len(meta)
for i, e in enumerate(embeddings):
meta[start + i] = {
"article_id": article["id"],
"title": article["title"],
"link": article["link"],
"chunk": e["text"]
}
def persist_index(index: faiss.IndexFlatIP, meta: dict):
faiss.write_index(index, INDEX_PATH)
with open(META_PATH, "w") as f:
json.dump(meta, f)
We keep a tiny JSON file that maps each vector’s row number to its source article and chunk. This is enough for retrieval without a full relational DB.
Step 4 — Retrieval Layer
Given a user query, we embed it, fetch the top‑k nearest vectors, and reconstruct the context.
def retrieve(query: str, index: faiss.IndexFlatIP, meta: dict, k: int = 5) -> List[dict]:
q_vec = EMBEDDER.encode([query])[0].astype('float32')
distances, ids = index.search(np.array([q_vec]), k)
results = []
for idx, score in zip(ids[0], distances[0]):
if idx == -1:
continue
entry = meta[str(idx)]
entry["score"] = float(score)
results.append(entry)
return results
FAISS’s inner‑product search (`IndexFlatIP`) works well because the embeddings are already L2‑normalized by Sentence‑Transformers.
Step 5 — Prompt Engineering & LLM Call
We feed the retrieved chunks into a concise prompt and call the LLM via OpenAI’s API (or any compatible endpoint). The prompt tells the model to answer using only the provided context.
import os
import openai
openai.api_key = os.getenv("OPENAI_API_KEY")
SYSTEM_PROMPT = (
"You are a knowledgeable assistant answering questions about Stripe's public blog. "
"Use only the supplied excerpts; if the answer is not present, say so."
)
def generate_answer(query: str, context: List[dict]) -> str:
excerpts = "\n---\n".join([c["chunk"] for c in context])
user_msg = f"Question: {query}\n\nContext:\n{excerpts}"
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_msg}
],
temperature=0.0,
max_tokens=300,
)
return response.choices[0].message.content.strip()
Setting temperature to 0 ensures deterministic answers, which is important for auditability in a production setting.
Step 6 — Caching & Performance Optimizations
Two layers of cache keep latency low:
- Query cache (in‑memory LRU) stores recent query → answer pairs.
- Embedding cache persists recently computed article embeddings to avoid re‑embedding on a restart.
from functools import lru_cache
@lru_cache(maxsize=256)
def cached_answer(query: str) -> str:
# Fast path: if the same query was asked recently, skip retrieval
context = retrieve(query, INDEX, META)
return generate_answer(query, context)
We also batch embedding calls when processing a batch of new articles, reducing the per‑article overhead.
Step 7 — Monitoring, Logging, & Alerting
Production reliability hinges on visibility. We emit structured logs and push a few metrics to Prometheus.
import logging
import time
logger = logging.getLogger("rag")
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s'))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
def log_query(query: str, latency: float, hit_rate: float):
logger.info(
"query=%s latency_ms=%.1f hit_rate=%.2f",
query, latency*1000, hit_rate
)
Metrics we track:
- `rag_query_latency_seconds` – overall response time.
- `rag_faiss_recall@5` – average similarity score of the 5th result.
- `rag_feed_errors_total` – count of RSS fetch failures.
Alert on >5% feed‑fetch failures in a 10‑minute window or latency >0.3 s.
Limitations and Tradeoffs
This pipeline assumes the RSS feed is the single source of truth. If Stripe adds pay‑walled articles or moves to a GraphQL endpoint, the ingest logic will need a rewrite. FAISS in‑memory indexes are fast but don’t survive a cold start without persisting to disk; for very large corpora you’d switch to an on‑disk index (e.g., IVF‑PQ) at the cost of higher latency. Finally, the LLM call is the biggest latency contributor; a local LLM or a faster endpoint could push sub‑100 ms responses.
Frequently Asked Questions
How do I avoid re‑embedding the same article after a restart?
Store the article‑to‑vector mapping in a JSON file (as shown) and load it on startup. The embedder only runs on IDs not present in that file.
What if the feed returns duplicate entries?
Our SHA‑256 hash on the article URL guarantees idempotency; duplicates are filtered out by the `seen_hashes` set.
Can I scale this to dozens of feeds?
Yes. Wrap the fetch‑parse‑chunk‑embed logic in a Celery worker per feed, and use a single shared FAISS index with a namespace field to keep vectors separate.
Is inner‑product search always the right choice?
For normalized embeddings, inner product is equivalent to cosine similarity and is faster than L2 distance. If you use non‑normalized vectors, switch to `IndexFlatL2`.
How do I handle rate limits from the LLM provider?
Implement exponential back‑off and a token‑bucket limiter around the `openai.ChatCompletion.create` call. Cache repeated queries to reduce request volume.
What I'd Change
If I were to rebuild this service today, I'd replace the in‑process FAISS index with a managed vector database (e.g., Pinecone or Milvus) to gain automatic sharding, replication, and better on‑disk compression. That would let us ingest multiple high‑volume feeds without worrying about memory caps, and the latency impact would be negligible with proper indexing. I’d also experiment with a locally hosted LLM (e.g., Llama 2 7B) behind an HTTP server to eliminate external API latency and cost, especially for a high‑throughput internal tool.