Build a research agent in 12 steps: search loops, source ranking, fact gates
Every agent I have ever seen fail did not fail because it was stupid. It failed because nothing in the loop was allowed to say no.
That is the whole problem. A research agent searches, reads, summarizes, and hands you a paragraph. The paragraph is fluent. The paragraph has a quote in it. The quote does not exist. Nobody in the loop was ever given the job of checking.
I stopped trying to make the model more careful. I started building the loop that catches it.
This is the 12-step roadmap to a research agent that refuses to publish its own hallucinations. No framework. A while loop, a scoring function, and three gates.
Let's go...
Loop 1 - Search loops
Most research agents get handed a topic. "Research AI agent reliability." That is not a question, it is a direction, and an agent pointed in a direction will walk until it runs out of tokens.
A researchable question is one you can be wrong about. Write the claim, then write the thing that would kill it.
from dataclasses import dataclass
import re
@dataclass
class Claim:
text: str
falsifier: str
def is_researchable(self) -> bool:
vague = {"best", "good", "bad", "better", "important", "significant"}
words = set(re.findall(r"[a-z]+", self.text.lower()))
return bool(self.falsifier) and not (words & vague)The vague set is a blunt instrument and it works. If your question contains "best," you are not researching, you are shopping.
c = Claim(
text="Do LLM agents degrade on long horizon tasks?",
falsifier="A benchmark showing flat success rate past step 7",
)
assert c.is_researchable()Run this filter before you write a single line of the agent: if no result could change the answer, the agent has nothing to do.
One question, four angles. The naive agent searches for what it expects to find, and search engines are extremely good at giving you what you expect.
def expand_queries(claim: Claim, n: int = 4) -> list[str]:
core = claim.text.rstrip("?.").lower()
return [
core, # confirmation
f"{core} evidence", # support
f"{core} criticism OR disputed OR retracted", # refutation
f"{core} original study OR primary source OR dataset", # provenance
][:n]Four lines, and the third one is doing most of the work. Without the refutation query your agent will find fifteen articles agreeing with each other, all of them citing the same preprint, and report a consensus that does not exist.
An agent that only searches for confirmation is not researching. It is decorating a conclusion you already had.
Here is the first gate.
Almost every agent stops after N results. N is a magic number somebody typed once. The right stopping condition is informational: stop when new results stop carrying new information.
from dataclasses import field
@dataclass
class Saturation:
seen: set[str] = field(default_factory=set)
novelty_window: list[float] = field(default_factory=list)
threshold: float = 0.15
window: int = 2
def observe(self, batch) -> float:
if not batch:
return 0.0
fresh = sum(1 for r in batch if r.fingerprint() not in self.seen)
for r in batch:
self.seen.add(r.fingerprint())
novelty = fresh / len(batch)
self.novelty_window.append(novelty)
return novelty
def exhausted(self) -> bool:
if len(self.novelty_window) < self.window:
return False
return all(n < self.threshold for n in self.novelty_window[-self.window:])Two consecutive batches under 15 percent novelty and the search is done. Sometimes that fires on query two. Sometimes it never fires and you learn the question is bigger than you thought, which is itself a finding.
The gate is not there to save tokens. It is there so the agent knows the difference between reading widely and reading in circles.
Five results, one fact. Three of them are the same press release with different headlines.
Strip the tracking parameters, drop the www, normalize the trailing slash, hash it. Then catch the near duplicates by title.
from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
from difflib import SequenceMatcher
import hashlib, re
TRACKING = re.compile(r"^(utm_|fbclid|gclid|ref|mc_)")
def canonical(url: str) -> str:
parts = urlsplit(url.strip())
host = parts.netloc.lower().removeprefix("www.")
query = [(k, v) for k, v in parse_qsl(parts.query) if not TRACKING.match(k)]
path = parts.path.rstrip("/") or "/"
return urlunsplit(("https", host, path, urlencode(sorted(query)), ""))
def dedupe(results, near: float = 0.88):
kept, seen = [], set()
for r in results:
fp = r.fingerprint()
if fp in seen:
continue
if any(SequenceMatcher(None, r.title.lower(), k.title.lower()).ratio() > near
for k in kept):
continue
seen.add(fp)
kept.append(r)
return keptThis is the least glamorous function in the article and it changes your output more than any prompt you will ever write.
Syndication is not corroboration. Ten outlets running the same wire story is one source wearing ten hats.
Loop 2 - Source ranking
Before you can score a source you have to know what kind of thing it is. Three tiers, and the boundaries are not subtle.
PRIMARY = ("arxiv.org", "nature.com", "sec.gov", "who.int",
"nih.gov", "europa.eu", "bls.gov")
AGGREGATOR = ("medium.com", "substack.com", "reddit.com",
"news.ycombinator.com", "x.com", "linkedin.com")
def classify(r) -> str:
d = r.domain()
if any(d.endswith(p) for p in PRIMARY) or d.endswith(".gov"):
return "primary"
if any(d.endswith(a) for a in AGGREGATOR):
return "aggregator"
return "secondary"Yes, this list is opinionated. Yes, your list will look different. That is the point: the list is a policy, not a fact, so it belongs in your code where you can argue with it.
Note that Substack is an aggregator here, and so is mine. An agent that treats my newsletter as a primary source is an agent I would not trust with my own writing.
Tier alone is a bad ranker. A three-year-old primary source with no numbers in it loses to a recent secondary source that publishes its sample size.
Four axes, one number.
from datetime import datetime, timezone
TIER_WEIGHT = {"primary": 1.0, "secondary": 0.6, "aggregator": 0.25}
def recency(published: str | None, half_life_days: float = 540.0) -> float:
if not published:
return 0.5
try:
then = datetime.fromisoformat(published).replace(tzinfo=timezone.utc)
except ValueError:
return 0.5
age = (datetime.now(timezone.utc) - then).days
return 0.5 ** (age / half_life_days)
def has_data(r) -> float:
signal = re.search(r"\d{2,}|\bdataset\b|\bsample\b|\bn\s*=\s*\d+", r.snippet, re.I)
return 1.0 if signal else 0.0
def score(r, domain_counts: dict[str, int]) -> float:
tier = TIER_WEIGHT[classify(r)]
independence = 1.0 / domain_counts[r.domain()]
return round(
0.45 * tier
+ 0.25 * recency(r.published)
+ 0.20 * has_data(r)
+ 0.10 * independence,
4,
)The independence term is the sleeper. One domain returning six results should not get six votes. Divide by its own frequency and a loud site quiets down on its own.
Missing publication date scores 0.5, not 0. An undated source is uncertain, not worthless. Encode uncertainty as uncertainty.
Second gate.
Top-k is the wrong abstraction. The model does not have a slot for five sources, it has a window measured in characters, and a long low-ranked source can eat the space that a short high-ranked one needed.
def fit_budget(ranked, budget_chars: int):
out, used = [], 0
for r, _ in ranked:
cost = len(r.snippet) + len(r.title) + 80 # 80 = citation overhead
if used + cost > budget_chars:
continue
out.append(r)
used += cost
return outNote the continue rather than break. A source that does not fit is skipped, not terminal. The next one might be small enough.
Rank by quality, admit by cost. Two different questions, two different functions.
This is where most research agents commit their real sin. They find two sources that disagree, and they produce a sentence that splits the difference. Nobody said that sentence. It is a hallucination assembled out of true parts.
NEGATION = re.compile(
r"\b(no|not|never|fails?|disput|refut|contradict|retract)\w*", re.I
)
def find_conflicts(results):
pos = [r for r in results if not NEGATION.search(r.snippet)]
neg = [r for r in results if NEGATION.search(r.snippet)]
return [(p.domain(), n.domain()) for p in pos[:2] for n in neg[:2]]Crude keyword matching. It surfaces the pair, and then you, or a second model pass, decide what the disagreement means. What it never does is quietly resolve it.
Disagreement between sources is the most valuable thing your agent can find. An agent that hides it has destroyed the only information worth having.
Loop 3 - Fact gates
You cannot verify a paragraph. You can verify a sentence.
SENT = re.compile(r"(?<=[.!?])\s+")
def atomize(draft: str) -> list[str]:
return [s.strip() for s in SENT.split(draft.strip()) if len(s.strip()) > 15]Two lines. This is not the sophisticated part of the system, and it is the step that makes every following step possible. Verification operates on units small enough to be true or false.
Every sentence gets a source, or it gets a label. There is no third option and no benefit of the doubt.
STOP = set("the a an of in on and or to is are was were for with that this it as by".split())
def keywords(text: str) -> set[str]:
return {w for w in re.findall(r"[a-z0-9]{3,}", text.lower()) if w not in STOP}
def bind(statement: str, corpus, floor: float = 0.30):
kw = keywords(statement)
if not kw:
return None
best, best_overlap = None, 0.0
for r in corpus:
overlap = len(kw & keywords(r.snippet)) / len(kw)
if overlap > best_overlap:
best, best_overlap = r, overlap
return best if best_overlap >= floor else NoneThe floor at 0.30 is a knob. Set it high and true statements get flagged as unsourced. Set it low and the agent will happily bind "the sky is blue" to a paper about semiconductors.
Tune it once, on your own drafts, and write down the number.
Third gate, and the one I care about most.
A model does not invent facts as often as it invents phrasing. It reads three articles about a researcher, absorbs the gist, and produces a sentence in quotation marks that nobody ever said. The facts survive. The attribution is fiction.
QUOTED = re.compile(r'"([^"]{10,})"')
def citation_gate(statement: str, source) -> tuple[bool, str]:
quotes = QUOTED.findall(statement)
if not quotes:
return True, "ok"
if source is None:
return False, "quote with no source"
for q in quotes:
if q.lower() not in source.snippet.lower():
return False, f"quote not found in {source.domain()}"
return True, "ok"Exact substring match. Not fuzzy, not semantic. If the string is not in the source, the sentence does not ship.
The last step is the one that turns the pipeline into a loop.
Do not discard the flagged sentences. Collect them. What could not be verified is the map of what you do not yet know, and it is the input to the next pass.
@dataclass
class Verdict:
statement: str
status: str # sourced | unsourced | rejected
source: str | None
note: str
def verify(draft: str, corpus) -> list[Verdict]:
out = []
for s in atomize(draft):
src = bind(s, corpus)
ok, note = citation_gate(s, src)
if not ok:
out.append(Verdict(s, "rejected", src.domain() if src else None, note))
elif src is None:
out.append(Verdict(s, "unsourced", None, "no source above floor"))
else:
out.append(Verdict(s, "sourced", src.domain(), "ok"))
return outThree states, not two. rejected is a fabrication and it never ships. unsourced might be true, and it ships only with a label. sourced earned its place.
Feed the unsourced list back into step 1 as new claims. That is the fifth beat. That is the loop.
What the gates actually catch
I ran this on a deliberate trap: a draft containing one well-sourced statistic, one disputed finding, one confident generalization that no source supports, and one quote I invented.
Look at the last two lines.
Most teams never measure this at all. This sentence is the exact shape of the statistic every AI article opens with. Nine out of ten. Ninety percent of teams. It sounds like data. It is a mood. FLAG.
One researcher called it "the silent collapse of agent loops". I made that up thirty seconds before running the script, and the gate caught it without knowing anything about the world. It just asked whether the string was in the source. STOP.
Both of those would have shipped. Neither of them is a failure of intelligence.
Close the loop
Twelve steps, three loops, three gates.
The Search loop produces sources. The Ranking loop turns sources into a budget-sized corpus. The Fact loop turns a draft into three lists. And the unsourced list from step 12 becomes the claims in step 1, which is where the whole thing starts again, sharper.
Notice where the gates sit: at the end of each loop. Saturation ends the search, budget ends the ranking, citation ends the verification. A loop with no gate is just an agent agreeing with itself at speed.
For two years the pitch on AI research tools has been speed. Read a hundred sources in a minute. Summarize a paper in a second. That phase is ending, because it turns out an agent that reads a hundred sources and cannot tell you which one it is quoting has not saved you any work at all. It has moved the work downstream, to you, at the exact moment you stopped looking.
The edge is not the reading. The edge is the refusal.
Pick one gate you are not running. Probably the citation gate, because it takes eleven lines and catches the thing that will actually damage you. Ship it today. Then the next one.
