Building a research agent that can rank filter search results agents is harder than wiring up a search API. The raw hits are noisy, duplicated, and often off-topic; you need a pipeline that scores, filters, and presents only what the agent can act on.
Step 1: Fetch raw results from a search source
Most agents start with a call to a web search provider. Keep the integration thin. You want a function that returns a normalized list of candidate documents, not a tangle of provider-specific fields. Never trust a single provider’s ranking—their relevance signal optimizes for ad revenue, not your agent’s task. Pull from two sources if you can; the dedupe step merges them.
import requests
def fetch_search_results(query: str, api_key: str, count: int = 20) -> list[dict]:
# Example using Bing Web Search; swap for SerpAPI, Google, or your internal index
endpoint = "https://api.bing.microsoft.com/v7.0/search"
headers = {"Ocp-Apim-Subscription-Key": api_key}
params = {"q": query, "count": count, "textDecorations": False}
resp = requests.get(endpoint, headers=headers, params=params, timeout=10)
resp.raise_for_status()
web_pages = resp.json().get("webPages", {}).get("value", [])
return [
{
"url": item["url"],
"title": item.get("name", ""),
"snippet": item.get("snippet", ""),
"source": "bing",
}
for item in web_pages
]If you don’t have a key, stub the function with a local JSON file. The rest of the pipeline doesn’t care where the bytes came from. Add pagination only when you see the top 20 missing obvious sources; premature pagination just multiplies noise.
Step 2: Normalize and deduplicate
Search providers return near-duplicates: same article on two domains, or AMP mirrors. Collapse them before spending tokens. Hash the normalized URL (strip query params, lowercase host) and keep the variant with the longest snippet.
from urllib.parse import urlparse, urlunparse
def normalize_url(u: str) -> str:
p = urlparse(u)
clean = p._replace(query="", fragment="")
return urlunparse(clean).rstrip("/").lower()
def dedupe(results: list[dict]) -> list[dict]:
seen = {}
for r in results:
key = normalize_url(r["url"])
if key not in seen or len(r["snippet"]) > len(seen[key]["snippet"]):
seen[key] = r
return list(seen.values())After this step you should have a manageable set—typically under 30 items. If you pull 100 raw hits, expect 40–60% reduction from dedupe alone. Store the original source field so you can debug later why a domain was kept.
Step 3: Pre-filter with embedding similarity
LLM calls are expensive and slow. Cut the list further with a cheap cosine similarity against the query embedding. Use a local model like all-MiniLM-L6-v2 to avoid network round-trips and to keep latency under 50ms per batch.
from sentence_transformers import SentenceTransformer
import numpy as np
model = SentenceTransformer("all-MiniLM-L6-v2")
def embed(texts: list[str]) -> np.ndarray:
return model.encode(texts, normalize_embeddings=True)
def cosine(a: np.ndarray, b: np.ndarray) -> float:
return float(np.dot(a, b))
def prefilter(query: str, results: list[dict], threshold: float = 0.25) -> list[dict]:
q_emb = embed([query])[0]
docs = [f"{r['title']} {r['snippet']}" for r in results]
d_embs = embed(docs)
scored = []
for r, d_emb in zip(results, d_embs):
score = cosine(q_emb, d_emb)
if score >= threshold:
r["emb_score"] = round(score, 3)
scored.append(r)
return scoredThis step drops clearly irrelevant pages. Tune the threshold on a labeled sample; 0.25 is a sane start for MiniLM on short snippets. If you scale past a few thousand candidates, move to FAISS or a vector DB instead of looping in Python.
Step 4: Use an LLM to rank filter search results agents
The embedding pre-filter catches topical mismatch, but not trustworthiness, freshness, or agent-specific utility. This is where you rank filter search results agents with a language model. Send a compact prompt with the query and the surviving candidates, and ask for a JSON array of judgments.
We point the OpenAI client at an OpenAI-compatible gateway so we can swap models without code changes. n4n.ai exposes one endpoint that fronts 240+ models and falls back automatically when a provider is degraded, which keeps the ranking step from becoming a single point of failure.
from openai import OpenAI
import json
client = OpenAI(
base_url="https://api.n4n.ai/v1", # OpenAI-compatible, 240+ models
api_key="YOUR_KEY",
)
def llm_rank(query: str, results: list[dict], model: str = "anthropic/claude-3.5-sonnet") -> list[dict]:
# limit to top 15 by embedding score to stay within token limits
candidates = sorted(results, key=lambda x: x.get("emb_score", 0), reverse=True)[:15]
payload = {
"query": query,
"candidates": [
{"id": i, "title": r["title"], "snippet": r["snippet"], "url": r["url"]}
for i, r in enumerate(candidates)
],
}
sys = "You are a strict research triage agent. Score each candidate 0-10 for relevance, recency, and authority. Drop anything below 6. Return JSON: [{'id':int,'score':int,'keep':bool,'reason':str}]."
resp = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": sys},
{"role": "user", "content": json.dumps(payload)},
],
response_format={"type": "json_object"},
extra_headers={"x-routing": "cost-optimized"}, # gateway honors client directives
)
judgments = json.loads(resp.choices[0].message.content)
kept = []
for j in judgments:
if j.get("keep") and 0 <= j["id"] < len(candidates):
r = candidates[j["id"]]
r["llm_score"] = j["score"]
r["reason"] = j["reason"]
kept.append(r)
kept.sort(key=lambda x: x["llm_score"], reverse=True)
return keptThe gateway forwards provider cache-control hints, so repeated queries with overlapping snippets hit cache and cut cost. Treat the LLM output as untrusted: validate id bounds and types before indexing back into your list. If the model returns malformed JSON, retry once with a stricter system prompt rather than crashing the agent loop.
Step 5: Apply deterministic filters
Model judgment is not enough. Enforce hard rules: blocklisted domains, minimum content length, and maximum age. These are faster and more predictable than asking the model to remember your policies.
from datetime import datetime, timezone
from urllib.parse import urlparse
BLOCKED = {"spam.example", "content-farm.example"}
MIN_SNIPPET = 40
def apply_rules(results: list[dict], max_age_days: int = 730) -> list[dict]:
now = datetime.now(timezone.utc)
out = []
for r in results:
host = urlparse(r["url"]).netloc.lower()
if host in BLOCKED:
continue
if len(r["snippet"]) < MIN_SNIPPET:
continue
if r.get("date"):
age = (now - r["date"]).days
if age > max_age_days:
continue
out.append(r)
return outWhy deterministic rules beat model memory
A model will occasionally let a known content farm through because its snippet looked plausible. A blocklist check is three lines and never hallucinates. Keep the blocklist in configuration, not code. Research agents degrade silently when a content farm sneaks into the top results; a 5-line rule saves the agent from citing garbage.
Step 6: Package results for the agent loop
The agent needs a stable schema. Emit a list of objects with the fields your tool caller expects, plus the scoring metadata so the agent can explain its choices.
def build_context(results: list[dict]) -> dict:
return {
"count": len(results),
"sources": [
{
"url": r["url"],
"title": r["title"],
"summary": r["snippet"],
"relevance": r.get("llm_score", r.get("emb_score", 0)),
"note": r.get("reason", ""),
}
for r in results
],
}Feed build_context output into the agent’s retrieval tool. The agent should never see raw provider JSON. If you stream results to a UI, include the note field so a human can audit why a source was kept.
Step 7: Monitor and tune in production
Shipping the pipeline is the start. Log the score distribution at each stage: raw count, post-dedupe, post-embedding, post-LLM. A sudden drop in post-embedding survival means your query drifted from the indexed corpus or the threshold is too high.
import logging
def log_stage(name: str, n: int):
logging.info(f"stage={name} surviving={n}")Wire these logs to your metrics dashboard. When the LLM step fails due to provider outage, the gateway fallback should still return a response from a secondary model; log the model field from the response to confirm which one served. For local dev, stub the LLM call with a fake that returns fixed judgments to keep tests fast.
Verify the pipeline
You can’t ship a ranking filter blind. Write a fixture with 50 known documents—mix of on-topic, off-topic, duplicates, and blocklisted. Run the pipeline and assert:
- Duplicates collapsed (count drops).
- Embedding pre-filter removes at least the obvious off-topic items.
- LLM step returns only
keep:truewith scores ≥6. - Blocked domains absent from final output.
pytest tests/test_rank_filter.py -qIf the LLM step fails due to provider outage, the gateway fallback should still return a response from a secondary model; log the model field from the response to confirm which one served. For local dev, stub the LLM call with a fake that returns fixed judgments to keep tests fast.
A research agent is only as good as the documents it sees. The work to rank filter search results agents is plumbing, but it’s the difference between a demo that works on the happy path and a system that survives real queries.