Build a Multi-Hop Web Search Pipeline in Python
Divya Lath, GTM + Product
Building agents that need the live web? Start free with TinyFish. TinySearch and TinyFetch are free within their limits.
TL;DR
Multi-hop search answers questions that no single search can: you need one fact to know what to search for next. "Which university did the CEO of the company that makes the Cursor code editor attend?" takes three hops: find the company, then its CEO, then where that person studied.
In this post I build a small multi-hop search pipeline in plain Python, about 190 lines with no framework. It:
- plans the question as a chain of dependent hops, where hop 2's query is written from hop 1's answer
- searches with TinySearch and reads pages with TinyFetch
- keeps an evidence ledger: every claim has a URL and a quote, and page quotes are checked against the page text
- stops when every hop is supported, or when its search budget runs out, and then returns an honest partial answer instead of a guess
You'll get the full code, a trace of a complete run, and a trace of a run that ran out of budget.
Disclosure: I work at TinyFish, which makes the search and fetch tools used here. The loop itself is plain Python, and the same structure works with any search API.
Why multi-hop search needs its own loop
Most tutorials on multi-hop search, such as the DSPy multi-hop tutorial or vector-database guides on query decomposition, work over a fixed document collection. On the open web, three things change:
- The second query depends on the first answer. You can't run the hops in parallel. If hop 1 is wrong, every later hop searches for the wrong thing.
- Search results are noisy. A natural-language query can return pages that share words with the question but not its meaning. You'll see exactly that in the trace below.
- You need to show your work. A multi-hop answer is only as good as its weakest hop, so each hop needs its own evidence, not one citation at the end.
So the pipeline needs four things: a plan of dependent hops, a search-and-read step per hop, a ledger of evidence, and a rule for when to stop. If you want the bigger picture on agentic search first, read what agentic search is and how we think about production agentic search architecture.
How the pipeline works
Here's the loop in one pass:
- Plan. An LLM splits the question into the smallest chain of sub-questions and marks dependencies with placeholders: "Which company makes the Cursor AI code editor?", then "Who is the CEO of {1}?", then "Which university did {2} attend?".
- Fill the placeholder. Before each hop runs, {1} or {2} is replaced with the previous hop's answer. That's what makes the searches dependent.
- Search. The model writes a short keyword query, and TinySearch returns ranked results with titles, URLs and snippets.
- Read the snippets. The model answers from the snippets only, and names which results support the answer and which pages are worth fetching.
- Fetch and quote. If the snippets aren't enough, TinyFetch reads the most promising page as Markdown, and the model copies one sentence that answers the question. The code keeps the quote only if it really appears on the page.
- Check. A hop is supported when it has one verbatim quote from a fetched page, or snippets from two different sites.
- Stop. Move to the next hop when this one is supported. If the budget runs out first, stop and report what's established and what's missing.
The evidence rule
The stopping condition is the most important design choice in the whole pipeline, so here it is on its own:
python
def supported(self) -> bool:
if not self.answer:
return False
pages = [e for e in self.evidence if e.kind == "page"]
domains = {urlparse(e.url).netloc for e in self.evidence}
return bool(pages) or len(domains) >= 2A hop counts as answered when either:
- one quote from a fetched page backs it. The code checks the quote against the page text, so the model can't invent it.
- two search snippets from different domains agree on it. Two snippets from the same site don't count, since they're one source.
You can make this stricter (always require a page quote) or looser (accept one strong snippet). What matters is that the rule is explicit and in code, not a feeling the model has about whether it's done.
The budget
Every search and fetch costs time and, past the free limits, money. So the pipeline gets a fixed budget per question, defaulting to 5 searches and 4 fetches:
python
@dataclass
class Budget:
searches: int = 5
fetches: int = 4
used_searches: int = 0
used_fetches: int = 0The loop checks the budget before every call. When it runs out mid-hop, that hop is marked not supported and the pipeline stops, because every later hop depends on it. Searching for "the university of ?" is worse than stopping.
The full code
Install the two dependencies and set your keys:
bash
pip install requests anthropic # or: pip install requests openai
export TINYFISH_API_KEY=... # from your TinyFish dashboard
export ANTHROPIC_API_KEY=... # or OPENAI_API_KEY with LLM_PROVIDER=openai and LLM_MODEL setThen save this as multihop.py:
python
"""multihop.py: a small multi-hop web search pipeline.
TinySearch finds pages, TinyFetch reads them, and an LLM plans the hops,
writes queries and pulls out evidence. Every claim keeps its source.
"""
from __future__ import annotations
import json, os, re, sys
from dataclasses import dataclass, field
from urllib.parse import urlparse
import requests
TF_KEY = os.environ["TINYFISH_API_KEY"]
SEARCH_URL = "https://api.search.tinyfish.ai"
FETCH_URL = "https://api.fetch.tinyfish.ai"
# ---------- data ----------
@dataclass
class Evidence:
claim: str
url: str
quote: str
kind: str # "page" (verbatim from a fetched page) or "snippet" (search result)
@dataclass
class Hop:
question: str
answer: str | None = None
evidence: list = field(default_factory=list)
queries: list = field(default_factory=list)
def supported(self) -> bool:
if not self.answer:
return False
pages = [e for e in self.evidence if e.kind == "page"]
domains = {urlparse(e.url).netloc for e in self.evidence}
return bool(pages) or len(domains) >= 2 # one page quote, or two independent snippets
@dataclass
class Budget:
searches: int = 5
fetches: int = 4
used_searches: int = 0
used_fetches: int = 0
def can_search(self): return self.used_searches < self.searches
def can_fetch(self): return self.used_fetches < self.fetches
def log(msg): print(msg, flush=True)
def norm(text: str) -> str:
return re.sub(r"\s+", " ", re.sub(r"[*_`]", "", text)).strip().lower()
# ---------- web layer: TinySearch + TinyFetch ----------
def search(query: str, budget: Budget) -> list[dict]:
budget.used_searches += 1
r = requests.get(SEARCH_URL, params={"query": query}, headers={"X-API-Key": TF_KEY}, timeout=60)
r.raise_for_status()
results = r.json().get("results", [])
log(f" search {budget.used_searches}/{budget.searches}: {query!r} -> {len(results)} results")
return results
def fetch(url: str, purpose: str, budget: Budget) -> str:
budget.used_fetches += 1
r = requests.post(FETCH_URL, headers={"X-API-Key": TF_KEY}, timeout=120,
json={"urls": [url], "format": "markdown", "purpose": purpose})
r.raise_for_status()
results = r.json().get("results", [])
text = (results[0].get("text") or "") if results else ""
log(f" fetch {budget.used_fetches}/{budget.fetches}: {url} -> {len(text)} chars")
return text
# ---------- LLM layer (swap providers with LLM_PROVIDER) ----------
def llm_json(prompt: str) -> dict:
provider = os.environ.get("LLM_PROVIDER", "anthropic")
if provider == "anthropic":
import anthropic
msg = anthropic.Anthropic().messages.create(
model=os.environ.get("LLM_MODEL", "claude-sonnet-5"), max_tokens=1024,
messages=[{"role": "user", "content": prompt}])
text = msg.content[0].text
else:
from openai import OpenAI
resp = OpenAI().chat.completions.create(
model=os.environ["LLM_MODEL"], response_format={"type": "json_object"},
messages=[{"role": "user", "content": prompt}])
text = resp.choices[0].message.content
return json.loads(re.search(r"\{.*\}", text, re.S).group(0))
def plan_hops(question: str) -> list[str]:
out = llm_json(
"Split this question into the smallest chain of dependent sub-questions. "
"Refer to an earlier answer as {1}, {2}, ... Return JSON {\"hops\": [...]}.\n\n"
f"Question: {question}")
return out["hops"]
def write_query(sub_q: str, tried: list[str]) -> str:
return llm_json(
"Write one short keyword web search query (not a sentence) for this question. "
f"Do not repeat these queries: {tried}. Return JSON {{\"query\": \"...\"}}.\n\nQuestion: {sub_q}")["query"]
def read_snippets(sub_q: str, results: list[dict]) -> dict:
listing = "\n".join(f"[{i}] {r['url']}\n{r.get('title','')}\n{r.get('snippet','')}"
for i, r in enumerate(results[:8]))
return llm_json(
"Answer the question from these search results only. Return JSON with: "
"\"answer\" (short, or null), \"support\" (list of result numbers that state the answer), "
"\"fetch\" (up to two result numbers most likely to confirm it, best first).\n\n"
f"Question: {sub_q}\n\nResults:\n{listing}")
def quote_from_page(sub_q: str, answer: str | None, page: str) -> dict:
return llm_json(
"From this page, copy one sentence word for word that answers the question. "
"Return JSON {\"answer\": short answer or null, \"quote\": exact sentence or null}.\n\n"
f"Question: {sub_q}\nCandidate answer: {answer}\n\nPage:\n{page[:12000]}")
# ---------- the loop ----------
def run_hop(hop: Hop, budget: Budget):
while not hop.supported() and budget.can_search():
query = write_query(hop.question, hop.queries)
hop.queries.append(query)
results = search(query, budget)
read = read_snippets(hop.question, results)
if not read.get("answer"):
log(" no answer in results, rewriting the query")
continue
hop.answer = read["answer"]
for i in read.get("support", []):
if i < len(results):
hop.evidence.append(Evidence(hop.answer, results[i]["url"], results[i].get("snippet", ""), "snippet"))
for i in read.get("fetch") or []:
if hop.supported() or not budget.can_fetch() or i >= len(results):
break
url = results[i]["url"]
page = fetch(url, hop.question, budget)
got = quote_from_page(hop.question, hop.answer, page)
quote = got.get("quote") or ""
if quote and norm(quote) in norm(page): # only keep quotes that really appear on the page
hop.answer = got.get("answer") or hop.answer
hop.evidence.append(Evidence(hop.answer, url, quote, "page"))
else:
log(" no usable quote on that page, evidence discarded")
log(f" hop {'SUPPORTED' if hop.supported() else 'NOT SUPPORTED'}: {hop.answer}")
def answer(question: str, budget: Budget) -> dict:
hops, done = [], []
templates = plan_hops(question)
for template in templates:
sub_q = re.sub(r"\{(\d+)\}", lambda m: done[int(m.group(1)) - 1].answer or "?", template)
hop = Hop(sub_q)
hops.append(hop)
log(f"\nHOP {len(hops)}: {sub_q}")
run_hop(hop, budget)
if not hop.supported():
break # later hops depend on this one, so stop here
done.append(hop)
complete = len(done) == len(hops) and hops and all(h.supported() for h in hops)
return {
"question": question,
"status": "complete" if complete else "incomplete",
"answer": done[-1].answer if complete else None,
"established": [{"q": h.question, "a": h.answer} for h in done],
"missing": [h.question for h in hops if not h.supported()]
+ [re.sub(r"\{(\d+)\}", r"[answer to hop \1]", t) for t in templates[len(hops):]],
"stopped_because": "answer supported" if complete else (
"search budget used up" if not budget.can_search() else "a hop could not be supported"),
"evidence": [vars(e) for h in hops for e in h.evidence],
"budget": f"{budget.used_searches}/{budget.searches} searches, {budget.used_fetches}/{budget.fetches} fetches",
}
if __name__ == "__main__":
q = sys.argv[1]
max_s = int(os.environ.get("MAX_SEARCHES", 5)); max_f = int(os.environ.get("MAX_FETCHES", 4))
print(json.dumps(answer(q, Budget(max_s, max_f)), indent=2))Run it:
bash
python multihop.py "Which university did the CEO of the company that makes the Cursor code editor attend?"A few details worth calling out:
- The web layer is two functions.
search()calls the Search API with a GET and anX-API-Keyheader.fetch()calls the Fetch API with a POST, asks for Markdown, and passes apurposeso the fetch knows what you're looking for. - The LLM layer is swappable.
llm_json()is the only function that knows about a model provider. Every prompt asks for JSON and gets parsed the same way. - Quotes are checked, not trusted.
norm()strips Markdown emphasis and whitespace before comparing, because pages say Anysphere, Inc. in bold and models copy it as plain text. Without this, good quotes get thrown away. - Each hop can try two pages. If the first page is a dud, the second gets a chance before the pipeline spends another search.
Want to try it on your own questions? Get a TinyFish API key: TinySearch and TinyFetch are free within their limits, and new accounts get $8 in Wallet credit for TinyAgent and TinyBrowser.
Trace: a complete run
Here's the pipeline answering the Cursor question. The searches and pages are real TinySearch and TinyFetch responses from 9 November 2026; I trimmed the JSON for length.
text
HOP 1: Which company makes the Cursor AI code editor?
search 1/5: 'company that makes the Cursor AI code editor' -> 9 results
no answer in results, rewriting the query
search 2/5: 'Cursor AI code editor developer company' -> 8 results
fetch 1/4: https://en.wikipedia.org/wiki/Cursor_(company) -> 431 chars
hop SUPPORTED: Anysphere, Inc. (trading as Cursor)
HOP 2: Who is the CEO of Anysphere, Inc. (trading as Cursor)?
search 3/5: 'Anysphere CEO Cursor' -> 4 results
hop SUPPORTED: Michael Truell
HOP 3: Which university did Michael Truell attend?
search 4/5: 'Michael Truell Cursor CEO university education' -> 7 results
fetch 2/4: https://www.forbes.com/profile/michael-truell/ -> 208 chars
no usable quote on that page, evidence discarded
fetch 3/4: https://en.wikipedia.org/wiki/Michael_Truell -> 367 chars
hop SUPPORTED: MIT (BS, computer science and mathematics)Final status:
json
{
"status": "complete",
"answer": "MIT (BS, computer science and mathematics)",
"stopped_because": "answer supported",
"budget": "4/5 searches, 3/4 fetches"
}What happened, hop by hop:
- Hop 1 needed two searches. The first query, "company that makes the Cursor AI code editor", came back with Make.com, America Makes and a dating app. The words matched; the meaning didn't. The model found no answer in those snippets, so the loop rewrote the query as keywords ("Cursor AI code editor developer company") and got Cursor's site, its Wikipedia page and a blog snippet saying "Cursor, developed by Anysphere". A fetch of the Wikipedia page gave the verbatim quote that settled it.
- Hop 2 needed no fetch. "Anysphere CEO Cursor" returned the Wikipedia infobox ("Key people: Michael Truell (CEO)") and a Y Combinator page ("Michael Truell, co-founder and CEO of Anysphere"). Two different sites agreed, so the evidence rule was met from snippets alone.
- Hop 3 hit a bad page. The model's first choice was Forbes's profile of Michael Truell, but the fetched page came back as a feed of unrelated headlines. There was no quote to keep, so that evidence was discarded and the second candidate, Wikipedia, gave the sentence it needed.
The evidence for the final hop looks like this in the output:
json
[
{
"claim": "Massachusetts Institute of Technology (MIT)",
"url": "https://www.crunchbase.com/person/michael-truell-238d",
"quote": "Michael attended the Massachusetts Institute of Technology, where he earned a Bachelor's Degree in Computer Science and Mathematics.",
"kind": "snippet"
},
{
"claim": "MIT (BS, computer science and mathematics)",
"url": "https://en.wikipedia.org/wiki/Michael_Truell",
"quote": "Truell went on to earn a Bachelor of Science degree from MIT, majoring in computer science and mathematics.",
"kind": "page"
}
]Total cost: 4 of 5 searches and 3 of 4 fetches.
Trace: an incomplete run
Now the same question with a tighter budget: 2 searches and 1 fetch.
text
HOP 1: Which company makes the Cursor AI code editor?
search 1/2: 'company that makes the Cursor AI code editor' -> 9 results
no answer in results, rewriting the query
search 2/2: 'Cursor AI code editor developer company' -> 8 results
fetch 1/1: https://en.wikipedia.org/wiki/Cursor_(company) -> 431 chars
hop SUPPORTED: Anysphere, Inc. (trading as Cursor)
HOP 2: Who is the CEO of Anysphere, Inc. (trading as Cursor)?
hop NOT SUPPORTED: Nonejson
{
"status": "incomplete",
"answer": null,
"established": [
{
"q": "Which company makes the Cursor AI code editor?",
"a": "Anysphere, Inc. (trading as Cursor)"
}
],
"missing": [
"Who is the CEO of Anysphere, Inc. (trading as Cursor)?",
"Which university did [answer to hop 2] attend?"
],
"stopped_because": "search budget used up",
"budget": "2/2 searches, 1/1 fetches"
}The first hop still spent two searches, because the first query was a dud, and that used the whole search budget. So hop 2 never ran a search. Instead of guessing a CEO or a university, the pipeline returns:
- status "incomplete" and answer null, so calling code can't mistake this for an answer
- what's established, with its evidence: the company is Anysphere, Inc., trading as Cursor
- what's missing, including the hop that never started, written as "Which university did [answer to hop 2] attend?"
- why it stopped: the search budget was used up
That's the behavior you want from a research step inside a larger agent. The caller can raise the budget, ask the user, or answer the part it can.
What this pipeline doesn't do
A small, readable loop leaves some things out. The ones I'd add next:
- Time. One hop-2 result, a LinkedIn profile, listed the CEO role as "Jan 2022 - Aug 2026", the month the SpaceX acquisition closed. The pipeline didn't notice, because the evidence rule checks agreement, not dates. For anything that changes over time, add the date to the claim and prefer recent sources (TinySearch has
recency_minutes,after_dateandbefore_datefilters). - Replanning. The plan is fixed after step 1. If a hop's answer reveals the question was framed wrong, a better version would replan from that point.
- Parallel hops. Some questions have independent branches ("Compare the founders of X and Y"). Those hops could run at the same time; this loop runs everything in order.
- Source quality. Two blog snippets count the same as two news outlets. You could weight domains or require one primary source per claim.
If your questions are more like "write me a cited report on this market" than "find this chain of facts", a managed research run will do more of this for you. TinyFish's Research API reads many sources and returns a cited report, and costs $2 (Deep) or $3 (Max) per completed run according to the docs.
Tips for multi-hop search that works
- Make the model write keyword queries. The one dud search in the trace was a full sentence. Short noun phrases work better on any web search API.
- Answer from snippets first, fetch second. Hop 2 needed zero fetches. Snippets are cheap. Only read pages when snippets don't settle it.
- Never trust a quote you didn't check. The
quote in pagecheck is one line, and it's the difference between evidence and a plausible sentence. - Stop on the first unsupported hop. Everything after it is built on a guess.
- Log every call against the budget. The trace above is just the
log()lines, and it's the fastest way to see where a run went wrong. For more on keeping the context you pass to the model small and relevant, see give agents better context, not more context.
Where TinyFish fits
The pipeline only needs two web calls, and both come from one TinyFish API key:
- TinySearch returns ranked results as JSON (position, title, snippet and URL), with optional filters for location, language, domains and dates.
- TinyFetch reads up to 10 URLs per request in a real browser, including JavaScript-heavy pages, and returns clean Markdown.
Both are free within their limits; see pricing for the current numbers. If you're comparing options, our roundups of the best web search APIs for AI agents and agentic search tools go deeper, and how to build an AI agent covers the rest of the agent around a loop like this one.
Give your agent the live web. Start free with TinyFish: one API key for search, fetch, browser automation and more.
AI disclosure
Content on this website may be created or refined with the assistance of AI tools and is subject to human editorial review.
