> ## Documentation Index
> Fetch the complete documentation index at: https://docs.inceptionlabs.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Search Agent: Mercury + Web Search
> Build an iterative web search agent that plans, searches, and synthesizes cited answers with Mercury.
This cookbook demonstrates how to build an agentic web search loop — `PLAN → SEARCH → ANALYZE → refine or FINISH → ANSWER` with Mercury as the agent. Because each planning turn returns in a few hundred milliseconds, most of the wall clock goes to the search API rather than the model. We walk you through how to build:
1. A pluggable search function and an evidence store with stable citation IDs
2. A JSON action contract and the plan-search-refine loop it drives
3. A fully cited answer with a latency breakdown
Directly via Google Colab or download the `.ipynb` to run it locally in Jupyter
## Dependencies
```python theme={null}
%pip install -q "openai>=1.0" httpx
```
```python theme={null}
import asyncio, getpass, json, os, re, time
import httpx
from openai import AsyncOpenAI
from IPython.display import Markdown, display
```
## API Keys
You need an Inception key plus a key for your search backend. Only the key for the backend you select is prompted for:
| Key | Where |
| ------------------- | -------------------------------------------------------------------------------------------------------- |
| `INCEPTION_API_KEY` | [platform.inceptionlabs.ai](https://platform.inceptionlabs.ai) |
| `EXA_API_KEY` | [dashboard.exa.ai](https://dashboard.exa.ai) — if `SEARCH_BACKEND = "exa"` |
| `PARALLEL_API_KEY` | [docs.parallel.ai](https://docs.parallel.ai/search/search-quickstart) — if `SEARCH_BACKEND = "parallel"` |
```python theme={null}
SEARCH_BACKEND = "exa" # "exa" or "parallel"
BACKEND_KEYS = {"exa": "EXA_API_KEY", "parallel": "PARALLEL_API_KEY"}
for var in ("INCEPTION_API_KEY", BACKEND_KEYS[SEARCH_BACKEND]):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
INCEPTION_BASE_URL = "https://api.inceptionlabs.ai/v1" # OpenAI-compatible
MODEL = "mercury-2"
```
## The Search Backend
The loop only ever sees `{heading: [docs]}`, where a doc is `{url, title, highlights[], publishedDate}`. Keeping that shape fixed means swapping providers — Brave, Tavily, Serper, or your own vector index over internal docs — is one function plus a branch in `search()`.
```python theme={null}
def check(r: httpx.Response) -> httpx.Response:
"""raise_for_status, but keep the server's explanation of what it rejected."""
if r.is_error:
raise RuntimeError(f"{r.request.method} {r.request.url} -> {r.status_code}: {r.text[:400]}")
return r
async def exa_search(http: httpx.AsyncClient, query: str, num_results: int) -> list[dict]:
"""Run one Exa /search with highlights, retrying on rate limits."""
body = {
"query": query,
"type": "auto",
"numResults": num_results,
"contents": {"highlights": {"maxCharacters": 2000}},
}
headers = {"x-api-key": os.environ["EXA_API_KEY"], "Content-Type": "application/json"}
for attempt in range(5):
r = await http.post("https://api.exa.ai/search", json=body, headers=headers)
if (r.status_code == 429 or r.status_code >= 500) and attempt < 4:
await asyncio.sleep(2**attempt) # bursty fanout will hit 429s
continue
return check(r).json().get("results", [])
return []
async def search(http: httpx.AsyncClient, queries: list[str], *, objective: str,
num_results: int, backend: str) -> dict[str, list[dict]]:
"""One round of search -> {transcript heading: [docs]}. Add a provider here."""
if backend == "exa":
per_query = await asyncio.gather(*(exa_search(http, q, num_results) for q in queries))
return dict(zip(queries, per_query))
if backend == "parallel":
return await parallel_search(http, queries, objective=objective, num_results=num_results)
raise ValueError(f"unknown backend: {backend!r} (expected 'exa' or 'parallel')")
```
```python theme={null}
async def parallel_search(http: httpx.AsyncClient, queries: list[str], *,
objective: str, num_results: int) -> dict[str, list[dict]]:
body = {
"objective": objective,
"search_queries": queries,
"mode": "turbo",
"advanced_settings": {"max_results": num_results * len(queries)},
}
headers = {"x-api-key": os.environ["PARALLEL_API_KEY"], "Content-Type": "application/json"}
for attempt in range(5):
r = await http.post("https://api.parallel.ai/v1/search", json=body, headers=headers)
if (r.status_code == 429 or r.status_code >= 500) and attempt < 4:
await asyncio.sleep(2**attempt)
continue
return {" | ".join(queries): [ # one combined transcript heading
{"url": it.get("url"), "title": it.get("title"),
"highlights": it.get("excerpts") or [], "publishedDate": it.get("publish_date")}
for it in check(r).json().get("results", [])
]}
return {}
```
## Evidence Bookkeeping
The agent's memory. Two details are what make multi-round search work:
* **Stable source IDs.** A URL keeps the same `[n]` for the whole run, across rounds. That is what lets the final answer cite `[3]` and have it mean something.
* **Append-only transcript.** Each round is appended, never rewritten, so the prompt prefix stays byte-identical between rounds.
```python theme={null}
class Evidence:
def __init__(self):
self.blocks: list[str] = [] # one formatted block per search round
self.ids: dict[str, int] = {} # url -> stable source id
self.titles: dict[str, str] = {}
def add_round(self, round_no: int, results: dict[str, list[dict]]) -> int:
"""Append one search round. Returns the number of previously unseen sources."""
new, sections = 0, []
for query, docs in results.items():
entries = []
for doc in docs:
url = (doc.get("url") or "").strip()
if not url:
continue
if url not in self.ids: # first sighting -> new id
self.ids[url] = len(self.ids) + 1
self.titles[url] = doc.get("title") or "(untitled)"
new += 1
snippets = "\n".join(f"- {h.strip()}" for h in doc.get("highlights") or []) or "- (none)"
date = f"\nDate: {doc['publishedDate']}" if doc.get("publishedDate") else ""
entries.append(
f"#### Source [{self.ids[url]}]\nURL: {url}\n"
f"Title: {self.titles[url]}{date}\nHighlights:\n{snippets}"
)
sections.append(f"### Query: {query}\n\n" + "\n\n".join(entries))
self.blocks.append(f"## Search round {round_no}\n\n" + "\n\n".join(sections))
return new
def transcript(self) -> str:
return "\n\n".join(self.blocks) if self.blocks else "No searches have been run yet."
def source_list(self) -> str:
by_id = sorted(self.ids.items(), key=lambda kv: kv[1])
return "\n".join(f"[{sid}] {self.titles[url]}: {url}" for url, sid in by_id) or "None."
```
## The Agent Prompt
The prompt the planner sees every round. Three things are doing the work: a strict action contract of exactly one JSON object per turn, an enumerate-then-verify strategy, and hard evidence rules.
```python theme={null}
PLANNER_SYSTEM = "Choose the next search action. Return exactly one JSON object, no prose."
AGENT_PROMPT = """You are a deep search agent. You answer broad, fact-heavy questions by
iteratively searching the web and reasoning over the evidence you retrieve.
## Workflow
1. SEARCH - issue up to {fanout} queries that attack the question from different angles.
2. ANALYZE - decide what is now supported and what is still missing.
3. REFINE - search again only for genuinely missing facts, disambiguation, or verification.
4. FINISH - stop as soon as the evidence can support a complete answer.
## Search strategy
- Enumerate candidates first, then verify each candidate's attributes.
- Prefer official, primary, or list/table sources.
- Never repeat a query string, and avoid synonym-only rewrites.
## Evidence rules
- Answer only from what retrieved sources explicitly state; no outside knowledge.
- If a source does not explicitly support a required fact, treat that fact as missing.
## Action contract - every turn returns exactly one JSON object
- {{"action": "search", "queries": ["...", "..."], "reason": "why these"}}
- {{"action": "finish", "reason": "why the evidence is sufficient"}}
## Question
{query}
## Evidence transcript
{evidence}"""
```
## Model Calls
A plain OpenAI-compatible chat call, with two Mercury-specific notes:
`reasoning_effort` is the speed/quality dial: `instant`, `low`, `medium`, or `high`.
Use `response_format={"type": "json_object"}` for the planner only. The final answer is free-form markdown, and forcing JSON there would coerce a perfectly good table into an empty object.
```python theme={null}
async def call(client, model, system, prompt, *, effort, max_tokens, as_json):
kwargs = {
"model": model,
"messages": [{"role": "system", "content": system},
{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": 0.7,
}
if as_json:
kwargs["response_format"] = {"type": "json_object"}
if effort:
kwargs["extra_body"] = {"reasoning_effort": effort}
t0 = time.perf_counter()
resp = await client.chat.completions.create(**kwargs)
return (resp.choices[0].message.content or ""), int((time.perf_counter() - t0) * 1000)
def parse_action(text: str) -> dict:
"""Planner output -> action dict. Tolerates code fences and stray prose."""
try:
return json.loads(text)
except json.JSONDecodeError:
m = re.search(r"\{.*\}", text, re.S)
return json.loads(m.group(0)) if m else {"action": "finish", "reason": "unparseable plan"}
```
```python theme={null}
def normalize_citations(text: str) -> str:
"""Rewrite full-width citation brackets to ASCII: 【1】 -> [1]."""
return re.sub(r"[【\[]\s*(\d{1,3})\s*[】\]]", r"[\1]", text)
def clean_queries(raw: list, seen: list[set[str]], fanout: int) -> list[str]:
out = []
for item in raw:
query = " ".join(str(item).split())
words = set(re.findall(r"[a-z0-9]+", query.lower()))
if len(query) < 12 or len(words) < 3:
continue # fragment, not a query
if any(len(words & prev) / len(words | prev) > 0.8 for prev in seen):
continue # near-dupe of one we already ran
seen.append(words)
out.append(query)
if len(out) >= fanout:
break
return out
```
Finally, a wall-clock ledger, so a run can show how much of its time was the model and how much was waiting on the web.
```python theme={null}
class Timings:
def __init__(self):
self.t0 = time.perf_counter()
self.plan_ms = self.plan_calls = 0
self.search_ms = self.search_rounds = 0
self.synth_ms = 0
def plan(self, ms: int) -> None:
self.plan_ms += ms; self.plan_calls += 1
def searched(self, ms: int) -> None:
self.search_ms += ms; self.search_rounds += 1
def synth(self, ms: int) -> None:
self.synth_ms = ms
def table(self, *, backend: str, model: str) -> str:
total = max(1, int((time.perf_counter() - self.t0) * 1000))
model_ms = self.plan_ms + self.synth_ms
pct = lambda ms: f"{100 * ms / total:.0f}%"
return (
f"### {total / 1000:.1f} s total\n\n"
f"| stage | calls | time | share |\n|---|---|---|---|\n"
f"| {model} planning | {self.plan_calls} | {self.plan_ms:,} ms "
f"| {pct(self.plan_ms)} |\n"
f"| web search ({backend}) | {self.search_rounds} | {self.search_ms:,} ms "
f"| {pct(self.search_ms)} |\n"
f"| {model} synthesis | 1 | {self.synth_ms:,} ms | {pct(self.synth_ms)} |\n\n"
f"**{model} took {model_ms / 1000:.1f} s of the {total / 1000:.1f} s "
f"({pct(model_ms)}) — the other {pct(total - model_ms)} was spent "
f"waiting on the web.**"
)
```
## The Loop
Plan, fan out searches in parallel, append evidence, and repeat until the agent says `finish` or the round cap is hit. Then one grounded synthesis call:
```python theme={null}
async def deep_search(
query: str,
*,
columns: str | None = None, # e.g. "State,Law name,Effective date" -> markdown table
rounds: int = 3, # max search rounds; the agent may stop earlier
fanout: int = 3, # parallel queries per round
results: int = 5, # search results per query
backend: str | None = None, # "exa" | "parallel"; defaults to SEARCH_BACKEND
model: str = MODEL,
effort: str = "instant",
verbose: bool = True,
) -> str:
searcher = AsyncOpenAI(api_key=os.environ["INCEPTION_API_KEY"], base_url=INCEPTION_BASE_URL)
backend = backend or SEARCH_BACKEND
timings = Timings()
evidence = Evidence()
seen_queries: list[set[str]] = [] # token sets of every query already issued
finish_reason = f"reached the {rounds}-round cap"
async with httpx.AsyncClient(timeout=60) as http:
for round_no in range(1, rounds + 1):
prompt = AGENT_PROMPT.format(
fanout=fanout, query=query, evidence=evidence.transcript()
) + (
f"\n\n## Next action\nRounds used: {round_no - 1} "
f"Rounds left: {rounds - round_no + 1} Max queries this round: {fanout}\n"
"Return exactly one JSON action now."
)
raw, ms = await call(searcher, model, PLANNER_SYSTEM, prompt,
effort=effort, max_tokens=2048, as_json=True)
timings.plan(ms)
action = parse_action(raw)
if action.get("action") != "search":
finish_reason = str(action.get("reason") or "planner chose to finish")
print(f" round {round_no}: finish ({ms} ms) — {finish_reason}")
break
queries = clean_queries(action.get("queries") or [], seen_queries, fanout)
if not queries:
finish_reason = "planner produced no new usable queries"
print(f" round {round_no}: finish ({ms} ms) — {finish_reason}")
break
print(f" round {round_no}: planned in {ms} ms — {len(queries)} queries")
if verbose:
for q in queries:
print(f" · {q}")
t0 = time.perf_counter()
found = await search(http, queries, objective=query,
num_results=results, backend=backend)
search_ms = int((time.perf_counter() - t0) * 1000)
timings.searched(search_ms)
new = evidence.add_round(round_no, found)
print(f" searched in {search_ms} ms via {backend} "
f"— {new} new sources ({len(evidence.ids)} total)")
shape = (
f"Return a markdown table only — no prose before or after it.\n"
f"Columns, exactly these and in this order: {' | '.join(columns.split(','))}\n"
f"Emit one row per item the question asks for; leave a cell blank rather than guessing."
if columns else
"Answer in markdown, and cite the sources you used inline with plain square "
"brackets, e.g. [1] or [2][5]."
)
final_prompt = (
f"Question:\n{query}\n\n## Accumulated evidence\n\n{evidence.transcript()}\n\n"
f"## Search phase conclusion\n\n{finish_reason}\n\n"
f"## Sources\n\n{evidence.source_list()}\n\n## Instructions\n\n{shape}\n"
"Use only the accumulated evidence above."
)
print(f"\n synthesizing with {model} …")
answer, ms = await call(searcher, model, "Answer only from the accumulated evidence.",
final_prompt, effort="", max_tokens=8192, as_json=False)
timings.synth(ms)
print(f" answered in {ms} ms")
return (f"{normalize_citations(answer.strip())}\n\n---\n"
f"{timings.table(backend=backend, model=model)}\n\n---\n"
f"**Sources**\n\n{evidence.source_list()}")
```
## Run It
```python theme={null}
QUESTION = ("Which U.S. states have enacted a comprehensive consumer data privacy law? "
"For each state, give the name of the law and the date it takes effect.")
answer = await deep_search(QUESTION, columns="State,Law name,Effective date", rounds=3)
display(Markdown(answer))
```
Three rounds of planning, 32 sources, and a synthesized table in about 7.5 seconds:
```text theme={null}
round 1: planned in 302 ms — 3 queries
· list of US states comprehensive consumer data privacy law effective date
· state consumer privacy law comprehensive name effective date
· comprehensive consumer data privacy statutes US states effective dates
searched in 1279 ms via exa — 8 new sources (8 total)
round 2: planned in 297 ms — 3 queries
· list of U.S. states with comprehensive consumer data privacy laws effective dates 2026
· official state websites consumer privacy law effective date
· comprehensive consumer privacy law name and effective date per state
searched in 1589 ms via exa — 10 new sources (18 total)
round 3: planned in 324 ms — 3 queries
· official state website California Consumer Privacy Act effective date
· official state website Virginia Consumer Data Protection Act effective date
· official state website Colorado Privacy Act effective date
searched in 1679 ms via exa — 14 new sources (32 total)
synthesizing with mercury-2 …
answered in 1925 ms
[total 7.5s]
```
Planning cost roughly 900 ms across all three rounds. The rest was spent waiting on the web, which is the point: at Mercury speeds, the agent's own deliberation stops being the bottleneck, so you can afford more rounds of it.
## Your Own Question
```python theme={null}
answer = await deep_search(
"What are the main obligations for general-purpose AI models under the EU AI Act, "
"and when do they start applying?",
rounds=2,
verbose=False,
)
display(Markdown(answer))
```