Adding a Freshness Layer to an Existing RAG Pipeline

Your vector store is right about your documents and wrong about the world. A routing layer fixes that without rebuilding anything.

Profile picture of Serply
Serply
A router directing queries to either a vector store or live search

Every RAG system has the same blind spot. It answers beautifully from the corpus you indexed and confidently wrong about anything that happened since. Worse, it can’t tell the difference — a question about your product’s current pricing retrieves the pricing doc from eight months ago and returns it with full confidence.

You don’t need to rebuild for this. You need a router and a second retriever.

Classifying the query

The router decides where a question should go. Cheap heuristics first, model second:

import re

TEMPORAL = re.compile(
    r"\b(current|currently|latest|newest|recent|recently|today|this (week|month|year|quarter)|"
    r"right now|as of|still|these days|nowadays|up to date|20\d\d)\b", re.I
)
VOLATILE = re.compile(
    r"\b(price|pricing|cost|version|release|available|availability|status|"
    r"news|announced|launch|deprecated|supported)\b", re.I
)


def needs_freshness(query: str) -> bool:
    return bool(TEMPORAL.search(query) or VOLATILE.search(query))

This catches most of it. For the rest, ask the model — but only when the heuristics are ambiguous, so you’re not paying for a classification call on every query:

from anthropic import Anthropic

client = Anthropic()

ROUTE = """Does answering this question require information that could have changed
in the last few months?

Answer with exactly one word: FRESH or STABLE."""


def route(query: str) -> str:
    if needs_freshness(query):
        return "fresh"
    msg = client.messages.create(
        model="claude-haiku-4-5-20251001", max_tokens=10,
        system=ROUTE, messages=[{"role": "user", "content": query}],
    )
    return "fresh" if "FRESH" in msg.content[0].text.upper() else "stable"

A small model is the right call here. The classification is easy and you’re running it on every miss.

The web retriever

import os
import requests
from urllib.parse import quote_plus
from concurrent.futures import ThreadPoolExecutor

API_KEY = os.environ["SERPLY_API_KEY"]


def web_retrieve(query: str, k: int = 5, read_top: int = 3) -> list[dict]:
    resp = requests.get(
        f"https://api.serply.io/v1/search/q={quote_plus(query)}&num={k}",
        headers={"X-Api-Key": API_KEY, "X-Proxy-Location": "US"},
        timeout=30,
    )
    resp.raise_for_status()
    results = resp.json().get("results", [])

    def read(url: str) -> str | None:
        try:
            r = requests.post(
                "https://api.serply.io/v1/request",
                headers={"X-Api-Key": API_KEY, "Content-Type": "application/json"},
                json={"url": url, "response_type": "markdown"},
                timeout=60,
            )
            r.raise_for_status()
            return r.text[:10000]
        except requests.RequestException:
            return None

    urls = [r.get("link") for r in results[:read_top] if r.get("link")]
    with ThreadPoolExecutor(max_workers=read_top) as pool:
        texts = list(pool.map(read, urls))

    docs = []
    for r, t in zip(results, texts + [None] * len(results)):
        docs.append({
            "text": t or f"{r.get('title', '')}\n{r.get('description', '')}",
            "url": r.get("link", ""),
            "title": r.get("title", ""),
            "source": "web",
            "fetched_full": bool(t),
        })
    return docs

Serply’s search endpoint embeds the query in the path (/v1/search/q=...) and returns hits under results. The scraper’s markdown mode returns text as the body, so .text is correct.

Falling back to the snippet when a page fetch fails keeps the document count stable, and the fetched_full flag lets the prompt distinguish a full read from a two-line snippet.

Hybrid, not either-or

Pure routing gets a common case wrong: “how does our refund policy compare to what competitors offer now” needs both stores. Run them together when the question mentions your own domain:

def retrieve(query: str, vector_store) -> list[dict]:
    mode = route(query)

    if mode == "stable":
        return [
            {"text": d.page_content, "url": d.metadata.get("source", ""),
             "title": d.metadata.get("title", ""), "source": "internal"}
            for d in vector_store.similarity_search(query, k=6)
        ]

    internal = [
        {"text": d.page_content, "url": d.metadata.get("source", ""),
         "title": d.metadata.get("title", ""), "source": "internal"}
        for d in vector_store.similarity_search(query, k=3)
    ]
    return internal + web_retrieve(query, k=5, read_top=3)

Keeping three internal documents even on a fresh query costs almost nothing and covers the hybrid case.

Labelling sources in the prompt

This is where most implementations lose the benefit. If both source types arrive as undifferentiated context, the model has no way to prefer the current one when they conflict — and they will conflict, because that’s the whole reason you added this.

def build_prompt(query: str, docs: list[dict]) -> str:
    internal = [d for d in docs if d["source"] == "internal"]
    web = [d for d in docs if d["source"] == "web"]

    parts = []
    if internal:
        parts.append("=== INTERNAL DOCUMENTS (authoritative for our own policies "
                     "and products; may be out of date on external facts) ===")
        parts += [f"[{d['title']}]\n{d['text']}" for d in internal]
    if web:
        parts.append("=== LIVE WEB RESULTS (current as of today; less authoritative "
                     "about our internal details) ===")
        parts += [f"[{d['url']}]\n{d['text']}" for d in web]

    parts.append(
        "\nIf internal documents and live results conflict on a fact about the "
        "outside world, prefer the live results and say the internal doc appears "
        "outdated. If they conflict about our own policies, prefer the internal "
        "document. Cite URLs for web claims."
    )
    parts.append(f"\nQuestion: {query}")
    return "\n\n".join(parts)

That conflict rule is the payoff of the whole exercise. Without it you’ve added a retriever and kept the bug.

Measuring whether it helped

Track two things: how often the router fires, and whether fresh-routed answers actually cite web sources.

def audit(query: str, answer: str, docs: list[dict]) -> dict:
    web_urls = {d["url"] for d in docs if d["source"] == "web" and d["url"]}
    cited = set(re.findall(r"https?://[^\s\)\]]+", answer))
    return {
        "routed_fresh": any(d["source"] == "web" for d in docs),
        "web_cited": bool(cited & web_urls),
        "fabricated": len(cited - web_urls - {d["url"] for d in docs}),
    }

A high routed_fresh rate with a low web_cited rate means you’re paying for search calls the model is ignoring — usually because the source labelling isn’t clear enough, or the internal documents are being retrieved with enough similarity to crowd out the web results in the prompt.