Design AI Research Loops That Stop, Recover and Produce Evidence
AI & Machine LearningLearn how to build an AI research loop with clear states, firm stop conditions, safe retries after failed tool calls, and a final report that rejects unsupported conclusions.
Most AI research agents run in a loop. The model looks at what it knows, picks a tool, reads the result and decides what to do next. Then it does it again.
That loop is where things go wrong. It can run forever. It can crash on one bad tool call and lose an hour of work. And it can end with a confident answer that none of its sources support.
This guide shows how to design the loop itself. You will define clear states, add stop conditions, recover from failed tool calls and reject any conclusion that lacks evidence.
The example task
For each of three vendors, does the public documentation say the product supports SAML single sign-on? If yes, which plan includes it?
Each vendor is a sub-question. The loop is done when every sub-question has an answer backed by evidence, or when it runs out of budget.
The loop uses two MrScraper endpoints:
- Scrape Sitemap with the
mapagent to find pages on each vendor's docs site - Scrape Markdown to read a page as clean text Your own model reads each page and pulls out evidence.
What are the loop states?
A loop without named states is hard to stop, hard to resume and hard to debug. Give every step a name.
| State | What happens | Next state |
|---|---|---|
PLAN |
Pick the next open sub-question | DISCOVER or REPORT |
DISCOVER |
Find candidate pages on the vendor's site | READ |
READ |
Fetch the next unread candidate page | EXTRACT |
EXTRACT |
The model pulls evidence from the page | EVALUATE |
EVALUATE |
Check whether the sub-question is answered | READ, PLAN or REPORT |
REPORT |
Build the final answer from verified evidence | DONE |
There are also two end states:
DONEmeans the loop finished normally.STOPPEDmeans a stop condition ended the loop early. The state records which one.
Write the loop as a state machine
Each state is a small function. It does one thing, updates the job, and returns the name of the next state. The main loop just calls the right function until it reaches an end state.
from dataclasses import dataclass, field
@dataclass
class Job:
job_id: str
sub_questions: list # [{"id": "vendor_a", "domain": "...", "start_url": "..."}]
state: str = "PLAN"
current: str | None = None # The sub-question being worked on
candidates: dict = field(default_factory=dict) # sub_question_id -> [urls]
pages: dict = field(default_factory=dict) # url -> markdown
evidence: list = field(default_factory=list)
answered: set = field(default_factory=set) # Has valid evidence
exhausted: set = field(default_factory=set) # No pages left to try
counters: dict = field(default_factory=lambda: {
"steps": 0, "discover_calls": 0, "pages": 0, "tokens": 0,
"failures_in_a_row": 0, "rounds_without_new_evidence": 0,
})
stop_reason: str | None = None
pending_call: dict | None = None
last_url: str | None = None
def run(job: Job, handlers: dict) -> Job:
while job.state not in ("DONE", "STOPPED"):
reason = check_stop_conditions(job)
if not reason:
job.counters["steps"] += 1
try:
job.state = handlers[job.state](job)
except StopLoop as e:
reason = str(e)
if reason:
job.stop_reason = reason
handlers["REPORT"](job) # Still report what was verified
job.state = "STOPPED"
save_job(job)
return job
save_job is your own function. It writes the job to a database or file. The call after every step matters. If the process dies, you load the last saved job and call run again. It picks up in the state where it stopped.
Write the small states
Here are four of the states. READ comes later, in the section on failed tool calls. extract_evidence is your own model call. It reads one page and returns a list of evidence items in the format shown later in this guide.
from urllib.parse import urlparse
def plan(job: Job) -> str:
done = job.answered | job.exhausted
open_ids = [q["id"] for q in job.sub_questions if q["id"] not in done]
if not open_ids:
return "REPORT"
job.current = open_ids[0]
return "READ" if job.current in job.candidates else "DISCOVER"
def discover(job: Job) -> str:
q = next(q for q in job.sub_questions if q["id"] == job.current)
job.counters["discover_calls"] += 1
try:
result = call_with_retries(
{"url": q["start_url"], "agent": "map", "maxDepth": 2, "maxPages": 3,
"limit": 50, "includePatterns": [".*(sso|saml|security|pricing).*"]},
{},
)
except SkipCall:
job.counters["failures_in_a_row"] += 1
job.exhausted.add(job.current)
return "PLAN"
job.counters["tokens"] += result.get("token_usage") or 0
site = urlparse(q["start_url"]).netloc
urls = (result.get("data") or {}).get("urls", [])
job.candidates[job.current] = [u for u in urls if urlparse(u).netloc == site][:5]
return "READ"
def extract(job: Job) -> str:
page = job.pages.get(job.last_url, "")
found = extract_evidence(page, job.current, job.last_url) if page else []
for item in found:
item["id"] = f"E{len(job.evidence) + 1}"
job.evidence.append(item)
job.counters["rounds_without_new_evidence"] = (
0 if found else job.counters["rounds_without_new_evidence"] + 1
)
return "EVALUATE"
def evaluate(job: Job) -> str:
has_valid = any(
e["sub_question"] == job.current and evidence_is_valid(job, e)[0]
for e in job.evidence
)
if has_valid:
job.answered.add(job.current)
job.counters["rounds_without_new_evidence"] = 0
return "PLAN"
return "READ"
HANDLERS = {"PLAN": plan, "DISCOVER": discover, "READ": read_page,
"EXTRACT": extract, "EVALUATE": evaluate, "REPORT": build_report}
build_report collects the supported and rejected conclusions and returns "DONE". The last section of this guide covers what goes into it.
Which stop conditions prevent runaway work?
A loop needs more than one way to stop. Each rule below catches a different kind of runaway.
Hard limits
These are simple counters. Pick numbers that fit the task.
LIMITS = {
"max_steps": 60,
"max_discover_calls": 6, # Two per vendor
"max_pages": 15,
"max_tokens": 300, # MrScraper tokens for this job
"max_failures_in_a_row": 3,
"max_rounds_without_new_evidence": 3,
}
Signals that work has stalled
Hard limits stop a loop eventually. These rules stop it sooner, when more work is unlikely to help:
- No new evidence. If several pages in a row add nothing, the loop is probably reading the wrong pages.
- Repeated actions. If the model asks for a page it already read, skip it and count it as a round with no new evidence.
- Too many failures. If several tool calls fail in a row, something is wrong outside the loop. Stop and report it.
Check every condition before each step
def check_stop_conditions(job: Job) -> str | None:
c = job.counters
if job.state == "REPORT":
return None # Always let the loop write its report
if c["steps"] >= LIMITS["max_steps"]:
return "step limit reached"
if c["pages"] >= LIMITS["max_pages"] and job.state == "READ":
return "page limit reached"
if c["discover_calls"] >= LIMITS["max_discover_calls"] and job.state == "DISCOVER":
return "discovery limit reached"
if c["tokens"] >= LIMITS["max_tokens"]:
return "token budget used"
if c["failures_in_a_row"] >= LIMITS["max_failures_in_a_row"]:
return "too many failed tool calls in a row"
if c["rounds_without_new_evidence"] >= LIMITS["max_rounds_without_new_evidence"]:
return "no new evidence"
return None
A STOPPED loop is not a failure. It still produces a report from whatever evidence it verified. The report says which stop condition fired and which sub-questions are still open.
Respect your account limits
Your job budget should never be larger than what your MrScraper account has left. The Platform API has an endpoint that reports your token limit and usage. Check it before the loop starts.
import os
import requests
PLATFORM_URL = "https://api.app.mrscraper.com/api/v1"
TOKEN = os.environ["MRSCRAPER_API_TOKEN"]
def tokens_remaining() -> int:
response = requests.get(
f"{PLATFORM_URL}/subscription-accounts",
headers={"x-api-token": TOKEN},
timeout=30,
)
response.raise_for_status()
data = response.json()["data"]
return data["tokenLimit"] - data["tokenUsage"]
LIMITS["max_tokens"] = min(LIMITS["max_tokens"], tokens_remaining())
Budgets exist to keep work inside your plan and inside each source's limits. If a loop needs more, raise the budget on purpose or choose a plan that fits. Do not spread work across extra tokens or accounts to go past a limit.
How do you resume after a failed tool call?
Tool calls fail. A page times out. A server returns an error. The question is what the loop does next.
Sort errors into two groups
| Error | Retry? | What the loop does |
|---|---|---|
| Timeout | Yes, with a wait | Retry up to a small limit, then move on |
| HTTP 5xx | Yes, with a wait | Retry up to a small limit, then move on |
| HTTP 429 | Yes, with a longer wait | Slow down. Never switch tokens to keep going |
| HTTP 401 | No | Stop the loop. The token is missing, invalid or expired |
| Other HTTP 4xx | No | Skip this call and record why |
MrScraper also retries on its side. The maxRetries request field caps those retries, and tokenCap caps the tokens one request and its retries can use. Your loop's retries sit on top of that, so keep both numbers small.
Retry with a limit and a growing wait
import random
import time
MRSCRAPER_URL = "https://api.mrscraper.com/"
class StopLoop(Exception):
pass
class SkipCall(Exception):
pass
def call_with_retries(body: dict, params: dict, attempts: int = 3) -> dict:
for attempt in range(1, attempts + 1):
try:
response = requests.post(
MRSCRAPER_URL,
params={"token": TOKEN, **params},
headers={"x-api-token": TOKEN, "Content-Type": "application/json"},
json={**body, "maxRetries": 1, "tokenCap": 20},
timeout=330,
)
if response.status_code == 401:
raise StopLoop("MrScraper rejected the API token")
if response.status_code == 429 or response.status_code >= 500:
raise requests.HTTPError(f"retryable status {response.status_code}")
if response.status_code >= 400:
raise SkipCall(f"status {response.status_code}")
return response.json()
except (requests.Timeout, requests.ConnectionError, requests.HTTPError):
if attempt == attempts:
raise SkipCall("failed after retries")
time.sleep(min(2 ** attempt, 30) + random.random())
The wait grows after each try: about 2 seconds, then 4. The small random extra keeps many jobs from retrying at the exact same moment.
Write down the call before you make it
A crash in the middle of a tool call is the tricky case. Did the call finish? Did you pay for it? To answer that, save a note about the call before you send it.
import hashlib
import json
def call_key(tool: str, args: dict) -> str:
return hashlib.sha256(json.dumps([tool, args], sort_keys=True).encode()).hexdigest()[:16]
def read_page(job: Job) -> str:
queue = job.candidates.get(job.current, [])
url = next((u for u in queue if u not in job.pages), None)
if url is None:
job.exhausted.add(job.current) # Nothing left to read for this one
return "PLAN"
key = call_key("read_page", {"url": url})
job.pending_call = {"key": key, "tool": "read_page", "url": url}
save_job(job) # Saved before the call goes out
try:
result = call_with_retries({"url": url}, {"markdown": "true"})
job.pages[url] = result.get("markdown", "")
job.counters["pages"] += 1
job.counters["tokens"] += result.get("token_usage") or 0
job.counters["failures_in_a_row"] = 0
job.last_url = url
next_state = "EXTRACT"
except SkipCall:
job.pages[url] = "" # Mark as tried so the loop moves on
job.counters["failures_in_a_row"] += 1
next_state = "READ"
job.pending_call = None
return next_state
When the job loads after a crash, look at pending_call:
- If the URL is already in
job.pages, the call finished. Clear the note and keep going. - If it is not, the call never finished. The loop is still in
READ, so it simply tries that page again. Because reading a page has no side effects, trying it again is safe. For tools that do have side effects, like sending a message, use the call key to check whether the action already happened before you repeat it.
How do you reject unsupported conclusions?
A model can write a conclusion that sounds right but goes beyond its sources. The loop should treat every conclusion as a draft until code checks it.
Store evidence as quotes, not summaries
In the EXTRACT state, ask your model to return evidence in a fixed shape. Each item must quote the page word for word.
{
"sub_question": "vendor_a",
"claim": "SAML SSO is available on the Business plan.",
"value": "Business",
"quote": "SAML single sign-on is included in the Business and Enterprise plans.",
"url": "https://docs.vendor-a.example/security/sso"
}
Check each piece of evidence
def normalize(text: str) -> str:
return " ".join(text.split()).lower()
def evidence_is_valid(job: Job, item: dict) -> tuple[bool, str]:
page = job.pages.get(item.get("url", ""))
if not page:
return False, "cited page was not read in this job"
if normalize(item.get("quote", "")) not in normalize(page):
return False, "quote does not appear on the page"
if normalize(str(item.get("value", ""))) not in normalize(item["quote"]):
return False, "value does not appear in the quote"
return True, "ok"
Then check each conclusion
A conclusion is a sentence in the final answer. It must point to evidence IDs, and the evidence must back it up:
- It cites at least one valid piece of evidence. No citation, no conclusion.
- The cited evidence is about the same sub-question. Evidence about Vendor B cannot support a claim about Vendor A.
- Broad claims need more than one source. A sentence like "all three vendors include SSO on business plans" covers several sub-questions. Each one must have its own valid evidence.
- Conflicts are reported, not hidden. If two valid quotes disagree, the conclusion is marked as disputed, and both sources are shown.
def check_conclusion(job: Job, conclusion: dict, valid_ids: set) -> tuple[bool, str]:
cited = [e for e in job.evidence if e["id"] in conclusion.get("evidence_ids", [])]
if not cited:
return False, "no evidence cited"
if any(e["id"] not in valid_ids for e in cited):
return False, "cites evidence that failed verification"
covered = {e["sub_question"] for e in cited}
missing = set(conclusion["sub_questions"]) - covered
if missing:
return False, f"no evidence for {sorted(missing)}"
values = {normalize(e["value"]) for e in cited if e["sub_question"] in conclusion["sub_questions"]}
if len(conclusion["sub_questions"]) == 1 and len(values) > 1:
return False, "sources disagree; report as disputed"
return True, "supported"
Build the report from what passed
The final report has three parts:
- Supported conclusions, each with its quotes and source URLs
- Rejected conclusions, each with the reason it failed
- Open questions, plus the stop reason if the loop ended early Keep the rejected list in the report, at least for internal readers. It shows where the model went further than its sources. That is the most useful signal you have when you tune the loop.
Hand off to your application
When the loop ends, your app has a saved job with every page read, every piece of evidence and every check result. Store the supported conclusions with their URLs and quotes. Send disputed and open items to a person for review if the answer matters.
FAQs
What are the loop states?
A simple research loop uses six working states: plan, discover, read, extract, evaluate and report. It also has two end states: done and stopped. Each state does one job and names the next state, so the loop is easy to stop, resume and debug.
How do you reject unsupported conclusions?
Store evidence as exact quotes with source URLs. Check that each quote appears on a page the loop actually read and contains the value it claims. Then check that each conclusion cites valid evidence for every sub-question it covers. Mark conflicts as disputed and list rejected conclusions in the report.
Which parts does MrScraper provide, and which parts must my application implement?
MrScraper is a developer-focused web data API. It discovers URLs from a starting page, fetches pages as clean Markdown or HTML, and lets you cap retries and tokens per request. Each response reports token usage, and the Platform API reports your account's token limit and usage.
Your application builds the loop. That includes the states, stop conditions, retry rules, saved job state, the model calls, evidence checks and the final report. Your team also decides which sources it has permission to collect from.
Summarize this post
Open it in your assistant of choice with the prompt ready to send.
Take a Taste of Easy Scraping!
Find more insights here
Build a Source-Led Deep Research Workflow With MrScraper
Build a deep research workflow that turns a question into a source plan, collects evidence from appr…
Choose an Agent Framework for a Web Data Application
Compare LangGraph, the OpenAI Agents SDK, Google ADK, CrewAI, Mastra and the Claude Agent SDK on too…
How to Build a Custom Amazon Price Monitor Without Code
Learn how to build a custom Amazon price tracker in under two minutes without code. Automate price m…