A DSPy Retrieval Module Over Live Search

• The retriever
• A multi-hop program
• Compiling
• A caching note

DSPy’s central idea is that you shouldn’t hand-write prompts — you should declare signatures and let an optimizer find the prompts. That’s compelling for generation. It’s more interesting for retrieval, because the thing being optimized becomes how the system searches.

A hand-written agent gets one query-formulation strategy: whatever you wrote in the system prompt. A DSPy program with a compiled query generator gets whatever strategy actually scored best on your dev set.

The retriever

import os
import requests
from urllib.parse import quote_plus
import dspy

API_KEY = os.environ["SERPLY_API_KEY"]


class SerplyRM(dspy.Retrieve):
    """DSPy retriever backed by live Google results."""

    def __init__(self, k: int = 5, proxy_location: str = "US"):
        super().__init__(k=k)
        self.proxy_location = proxy_location

    def forward(self, query_or_queries, k: int | None = None) -> dspy.Prediction:
        queries = (
            [query_or_queries]
            if isinstance(query_or_queries, str)
            else list(query_or_queries)
        )
        k = k or self.k
        passages = []

        for q in queries:
            resp = requests.get(
                f"https://api.serply.io/v1/search/q={quote_plus(q)}&num={k}",
                headers={"X-Api-Key": API_KEY, "X-Proxy-Location": self.proxy_location},
                timeout=30,
            )
            resp.raise_for_status()
            for r in resp.json().get("results", []):
                passages.append(
                    dspy.Example(
                        long_text=f"{r.get('title', '')}\n{r.get('description', '')}",
                        url=r.get("link", ""),
                        title=r.get("title", ""),
                    )
                )
        return dspy.Prediction(passages=passages[:k])

Serply’s search endpoint embeds the query string in the path — /v1/search/q=... — and returns organic hits under results with title, link, description, and position.

Handling both a single string and a list matters: DSPy’s multi-hop patterns pass lists, and a retriever that only accepts strings breaks the moment you try to compile one.

A multi-hop program

class GenerateQuery(dspy.Signature):
    """Write a search query that would find information missing from the context."""

    context = dspy.InputField(desc="Facts gathered so far")
    question = dspy.InputField()
    query = dspy.OutputField(desc="A focused web search query")


class AnswerWithCitations(dspy.Signature):
    """Answer the question using only the given context. Cite URLs."""

    context = dspy.InputField(desc="Retrieved passages with URLs")
    question = dspy.InputField()
    answer = dspy.OutputField(desc="Answer with inline URL citations")


class WebResearch(dspy.Module):
    def __init__(self, hops: int = 2, k: int = 5):
        super().__init__()
        self.hops = hops
        self.retrieve = SerplyRM(k=k)
        self.gen_query = [dspy.ChainOfThought(GenerateQuery) for _ in range(hops)]
        self.answer = dspy.ChainOfThought(AnswerWithCitations)

    def forward(self, question: str):
        context = []
        for hop in range(self.hops):
            q = self.gen_query[hop](
                context="\n".join(c.long_text for c in context) or "(nothing yet)",
                question=question,
            ).query
            passages = self.retrieve(q).passages
            context.extend(passages)

        formatted = "\n\n".join(f"[{c.url}] {c.long_text}" for c in context)
        return self.answer(context=formatted, question=question)

Separate GenerateQuery predictors per hop is intentional. The first hop and the second hop are different jobs — one opens the topic, one fills a specific gap — and giving the optimizer separate parameters for each lets it learn that distinction.

Compiling

from dspy.teleprompt import BootstrapFewShot

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))


def citation_metric(example, pred, trace=None) -> float:
    """Reward answers that are correct and actually cite a URL."""
    has_citation = "http" in (pred.answer or "")
    correct = example.answer.lower() in (pred.answer or "").lower()
    return float(has_citation and correct)


trainset = [
    dspy.Example(
        question="Which HTTP header does Serply use for API authentication?",
        answer="X-Api-Key",
    ).with_inputs("question"),
    # ...more labeled examples
]

optimizer = BootstrapFewShot(metric=citation_metric, max_bootstrapped_demos=4)
compiled = optimizer.compile(WebResearch(hops=2), trainset=trainset)

print(compiled(question="What proxy locations does Serply support?").answer)

The metric is where the real design happens. Rewarding only correctness produces a program that answers from the model’s memory when retrieval is weak, because memory is often right. Adding the has_citation term makes ungrounded answers score zero regardless of correctness, which pushes the optimizer toward query strategies that actually retrieve.

A caching note

Compilation replays the same questions many times. Without caching you’ll burn a startling number of search calls on a handful of dev examples:

from functools import lru_cache


@lru_cache(maxsize=1024)
def _cached_get(url: str) -> str:
    resp = requests.get(url, headers={"X-Api-Key": API_KEY}, timeout=30)
    resp.raise_for_status()
    return resp.text

Cache on the fully-formed URL and the hit rate during optimization is high — often above eighty percent, since bootstrapping revisits the same trajectories. Watch x-ratelimit-requests-remaining on the responses that do go out; a compile run is the easiest way to discover your rate limit the hard way.