17 Code: Agent 2 — Literature Review
Folder: 02_agents_code/agent_02_literature_review/ — self-contained; run python agent_02_literature_review/main.py --help.
17.1 README
17.1.1 README.md
# Agent 2 — AI Literature Review Agent ("Autonomous Systematic Review Scientist")
A PRISMA-style pipeline: **protocol → multi-database search → screening → extraction → review draft**, with an auditable trail (CSV/JSON) at every step.
```
question ─► SearchAgent (protocol: criteria + queries)
─► tools.search_all (arXiv · Semantic Scholar · PubMed · Crossref) ─► dedupe
─► ScreeningAgent (include/exclude/maybe + reason, batches of 10)
─► prisma.build_counts (PRISMA numbers computed by code)
─► ExtractionAgent (evidence table rows)
─► WritingAgent (review.md — cites only included IDs)
```
## Files
| File | Purpose |
|------|---------|
| `main.py` | CLI (`--question`, `--max-records`, `--sources`, `--offline`, `--from-evidence`) |
| `agents.py` | Search / Screening / Extraction / Writing agents + Pydantic models |
| `prisma.py` | PRISMA counting & flow rendering (pure Python) |
| `tools.py` | Database search + `Paper` model + de-duplication + offline sample |
| `llm.py` | Thin Claude client (`ask`, `ask_structured`) |
| `outputs/<slug>/` | `protocol.json`, `records.json`, `screening.csv`, `evidence.csv`, `prisma.json`, `review.md` |
## Run
```powershell
python agent_02_literature_review\main.py --question "AI methods for plant disease detection" --max-records 80
python agent_02_literature_review\main.py --question "..." --offline
# human-in-the-loop: edit outputs/<slug>/evidence.csv, then
python agent_02_literature_review\main.py --question "..." --from-evidence "agent_02_literature_review\outputs\<slug>"
```
## What this agent teaches
1. **Batch structured outputs** — screening 10 abstracts per call with a guaranteed decision per paper.
2. **Code for numbers, LLM for judgement** — PRISMA counts are computed, never generated.
3. **Human-in-the-loop** — edit the evidence table and regenerate.
4. **Audit trail** — every decision has a reason and is saved.
## Ideas to extend
- Full-text PDF extraction (`pypdf`), dual screening + Cohen's kappa, meta-analysis (forest plots), BibTeX/Zotero export, Streamlit UI.
---
**Codanics — Python ka Chilla: Build AI Agents**
Course: https://codanics.com/courses/python-ka-chilla-build-ai-agents/ · Website: https://codanics.com · YouTube: [@codanics](https://www.youtube.com/@codanics)
WhatsApp: https://whatsapp.com/channel/0029Va7nRDq3QxRzGqaQvS3r · Free course: https://www.youtube.com/watch?v=UE3o-7KRNXs&list=PLaQmdklCcX4017.2 Source files
17.2.1 agents.py
"""
agents.py — the four sub-agents of the Literature Review Agent (PRISMA pipeline).
1. SearchAgent : review protocol (PICO-style question, criteria, database queries)
2. ScreeningAgent : include / exclude / maybe + reason, in batches (structured)
3. ExtractionAgent : evidence table rows (structured)
4. WritingAgent : the literature review draft (free text, cites only included IDs)
Codanics — Python ka Chilla: Build AI Agents
https://codanics.com/courses/python-ka-chilla-build-ai-agents/
"""
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field
from llm import ask, ask_structured
from prisma import PrismaCounts
from tools import Paper
BRAND = ("Codanics — Python ka Chilla: Build AI Agents | "
"https://codanics.com/courses/python-ka-chilla-build-ai-agents/")
CITE_RULE = ("CITATION RULE: cite ONLY papers given in the input using their exact [id] tags. "
"Never invent papers, numbers or findings.")
# ================================================================ 1. Search / protocol
class Protocol(BaseModel):
review_title: str
research_question: str
pico_or_scope: str = Field(description="Population/Problem, Intervention/Method, Comparison, Outcome (or scope)")
inclusion_criteria: list[str]
exclusion_criteria: list[str]
queries: list[str] = Field(description="3-6 search strings for the databases")
year_from: int | None = None
class SearchAgent:
SYSTEM = ("You are a systematic-review methodologist (PRISMA 2020). Write precise protocols with "
"operational inclusion/exclusion criteria and high-recall search strings.")
def run(self, question: str) -> Protocol:
return ask_structured(self.SYSTEM, f"Review question: {question}", Protocol, effort="low")
# ===================================================================== 2. Screening
class Decision(BaseModel):
paper_id: str
decision: Literal["include", "exclude", "maybe"]
reason: str = Field(description="Short reason, e.g. 'Excluded: not about plant disease' or 'Included: CNN on leaf images'")
class ScreeningBatch(BaseModel):
decisions: list[Decision]
class ScreeningAgent:
SYSTEM = ("You are a careful title/abstract screener for a systematic review. Apply the protocol "
"criteria literally. Use 'maybe' when the abstract is insufficient. Return one decision per paper.")
def run(self, protocol: Protocol, papers: list[Paper], batch_size: int = 10, log=print) -> list[Decision]:
decisions: list[Decision] = []
for i in range(0, len(papers), batch_size):
batch = papers[i:i + batch_size]
corpus = "\n".join(p.short(1200) for p in batch)
res = ask_structured(
self.SYSTEM,
f"Protocol:\n{protocol.model_dump_json(indent=1)}\n\nPapers to screen:\n{corpus}\n\n"
f"Return exactly {len(batch)} decisions with the given paper ids.",
ScreeningBatch, effort="low")
got = {d.paper_id: d for d in res.decisions}
for p in batch: # guarantee one decision per paper even if the model skipped one
decisions.append(got.get(p.id, Decision(paper_id=p.id, decision="maybe",
reason="No decision returned; needs human check")))
log(f" screened {min(i + batch_size, len(papers))}/{len(papers)}")
return decisions
# ==================================================================== 3. Extraction
class EvidenceRow(BaseModel):
paper_id: str
study: str = Field(description="First author + year, e.g. 'Smith 2025'")
objective: str
dataset_or_sample: str
method: str
key_result: str = Field(description="Main quantitative/qualitative result, e.g. '94% accuracy'")
limitations: str
theme: str = Field(description="Short theme label for grouping, e.g. 'CNN on leaf images'")
class EvidenceTable(BaseModel):
rows: list[EvidenceRow]
class ExtractionAgent:
SYSTEM = ("You extract structured evidence from abstracts for a systematic review. If a field is not "
"reported, write 'not reported'. " + CITE_RULE)
def run(self, protocol: Protocol, papers: list[Paper], batch_size: int = 8, log=print) -> list[EvidenceRow]:
rows: list[EvidenceRow] = []
for i in range(0, len(papers), batch_size):
batch = papers[i:i + batch_size]
corpus = "\n".join(p.short(1500) for p in batch)
res = ask_structured(
self.SYSTEM,
f"Review question: {protocol.research_question}\n\nPapers:\n{corpus}\n\n"
f"Return one row per paper ({len(batch)} rows).",
EvidenceTable, effort="medium")
rows.extend(res.rows)
log(f" extracted {min(i + batch_size, len(papers))}/{len(papers)}")
return rows
# ======================================================================= 4. Writing
class WritingAgent:
SYSTEM = ("You are an academic writer producing a systematic literature review in Markdown. "
+ CITE_RULE + " Structure: Title; Abstract; 1 Introduction; 2 Methods (PRISMA: databases, "
"queries, criteria, screening, extraction); 3 Results (3.1 Study characteristics, "
"3.2-3.x Thematic synthesis with an evidence table); 4 Research Gaps; 5 Discussion; "
"6 Conclusion; References (only cited ids, one per line with title). "
"Be specific and quantitative; do not pad.")
def run(self, protocol: Protocol, prisma: PrismaCounts, rows: list[EvidenceRow],
included: list[Paper], echo: bool = False) -> str:
table = "\n".join(r.model_dump_json() for r in rows)
refs = "\n".join(f"[{p.id}] {p.title} ({p.year}) {p.venue}" for p in included)
text = ask(
self.SYSTEM,
f"Protocol:\n{protocol.model_dump_json(indent=1)}\n\nPRISMA counts (use these numbers verbatim):\n"
f"{prisma.model_dump_json(indent=1)}\n\nEvidence table rows:\n{table}\n\n"
f"Included papers (for citation):\n{refs}\n\nWrite the full review.",
max_tokens=32000, effort="high", echo=echo)
return text + f"\n\n---\n*Generated with the AI Literature Review Agent — {BRAND}*\n"17.2.2 llm.py
"""
llm.py — a thin wrapper around the Anthropic Claude API used by this agent.
Two functions are all you need:
ask(system, user) -> free text (streamed)
ask_structured(system, user, Model) -> a validated Pydantic object (JSON structured output)
Why a wrapper? Every agent in this project talks to the model the same way. Keeping the
API details in ONE file per agent means you can swap the model or provider in one place.
Codanics — Python ka Chilla: Build AI Agents
https://codanics.com/courses/python-ka-chilla-build-ai-agents/
"""
from __future__ import annotations
import os
from typing import Type, TypeVar
import anthropic
from dotenv import load_dotenv
from pydantic import BaseModel
load_dotenv() # reads ANTHROPIC_API_KEY / CLAUDE_MODEL from a .env file if present
MODEL = os.getenv("CLAUDE_MODEL", "claude-opus-5")
T = TypeVar("T", bound=BaseModel)
_client: anthropic.Anthropic | None = None
def client() -> anthropic.Anthropic:
"""Lazily create one shared client (credentials come from the environment)."""
global _client
if _client is None:
_client = anthropic.Anthropic()
return _client
def _check(msg: anthropic.types.Message) -> None:
if msg.stop_reason == "refusal":
detail = getattr(msg, "stop_details", None)
raise RuntimeError(f"The model refused this request: {detail}")
if msg.stop_reason == "max_tokens":
raise RuntimeError("Output was cut off (max_tokens). Increase max_tokens and retry.")
def _text(msg: anthropic.types.Message) -> str:
return "".join(b.text for b in msg.content if b.type == "text")
def ask(system: str, user: str, *, max_tokens: int = 16000, effort: str = "medium",
echo: bool = False) -> str:
"""Free-text generation. Streams so long outputs never hit HTTP timeouts."""
with client().messages.stream(
model=MODEL,
max_tokens=max_tokens,
system=system,
messages=[{"role": "user", "content": user}],
thinking={"type": "adaptive"},
output_config={"effort": effort},
) as stream:
if echo:
for chunk in stream.text_stream:
print(chunk, end="", flush=True)
print()
msg = stream.get_final_message()
_check(msg)
return _text(msg)
def ask_structured(system: str, user: str, model_cls: Type[T], *, max_tokens: int = 16000,
effort: str = "medium") -> T:
"""Structured generation: the SDK converts the Pydantic model to a JSON schema, the API
constrains the output to it, and `parsed_output` is a validated instance of `model_cls`."""
with client().messages.stream(
model=MODEL,
max_tokens=max_tokens,
system=system,
messages=[{"role": "user", "content": user}],
thinking={"type": "adaptive"},
output_config={"effort": effort},
output_format=model_cls,
) as stream:
msg = stream.get_final_message()
_check(msg)
if msg.parsed_output is None: # should not happen with a constrained schema; be defensive
return model_cls.model_validate_json(_text(msg))
return msg.parsed_output17.2.3 main.py
"""
AI Literature Review Agent — "Autonomous Systematic Review Scientist"
=====================================================================
question ─► protocol ─► search (4 databases) ─► screen ─► extract ─► PRISMA counts ─► review.md
Usage
python main.py --question "AI methods for plant disease detection"
python main.py --question "..." --max-records 80 --sources arxiv,pubmed
python main.py --question "..." --offline # demo sample, no internet for search
python main.py --question "..." --from-evidence outputs/<slug> # re-write after editing evidence.csv
Outputs: outputs/<slug>/ protocol.json, records.json, screening.csv, evidence.csv, prisma.json, review.md
Codanics — Python ka Chilla: Build AI Agents
https://codanics.com/courses/python-ka-chilla-build-ai-agents/
"""
from __future__ import annotations
import csv
import json
import sys
from pathlib import Path
import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from slugify import slugify
sys.path.insert(0, str(Path(__file__).parent))
if hasattr(sys.stdout, "reconfigure"): # Windows consoles default to cp1252; we print → ✓ etc.
sys.stdout.reconfigure(encoding="utf-8")
from agents import (EvidenceRow, ExtractionAgent, Protocol, ScreeningAgent, # noqa: E402
SearchAgent, WritingAgent)
from prisma import PrismaCounts, build_counts # noqa: E402
from tools import OFFLINE_SAMPLE, Paper, dedupe, search_all # noqa: E402
app = typer.Typer(add_completion=False)
console = Console()
OUT_DIR = Path(__file__).parent / "outputs"
BANNER = """[bold cyan]AI Literature Review Agent[/] — Autonomous Systematic Review Scientist
[dim]Codanics · Python ka Chilla: Build AI Agents · https://codanics.com/courses/python-ka-chilla-build-ai-agents/
YouTube @codanics · WhatsApp https://whatsapp.com/channel/0029Va7nRDq3QxRzGqaQvS3r[/]"""
def write_csv(path: Path, rows: list[dict]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
if not rows:
path.write_text("", encoding="utf-8")
return
with path.open("w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
w.writeheader()
w.writerows(rows)
console.print(f" [green]saved[/] {path.name}")
def write_json(path: Path, data) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
console.print(f" [green]saved[/] {path.name}")
@app.command()
def run(
question: str = typer.Option(..., help="The review question / topic"),
max_records: int = typer.Option(60, help="Max records to screen"),
sources: str = typer.Option("arxiv,semantic_scholar,pubmed,crossref"),
offline: bool = typer.Option(False, help="Use built-in demo records"),
from_evidence: Path = typer.Option(None, help="Existing output folder: skip to writing using its evidence.csv"),
echo: bool = typer.Option(True, help="Stream the review to the console"),
):
console.print(Panel(BANNER, expand=False))
out = from_evidence or (OUT_DIR / slugify(question)[:60])
# --- resume mode: human edited evidence.csv, just re-write the review -------------
if from_evidence:
console.rule("[bold]Re-writing review from edited evidence")
protocol = Protocol.model_validate_json((out / "protocol.json").read_text(encoding="utf-8"))
prisma = PrismaCounts.model_validate_json((out / "prisma.json").read_text(encoding="utf-8"))
rows = [EvidenceRow(**r) for r in csv.DictReader((out / "evidence.csv").open(encoding="utf-8"))]
records = [Paper(**p) for p in json.loads((out / "records.json").read_text(encoding="utf-8"))]
included = [p for p in records if p.id in {r.paper_id for r in rows}]
review = WritingAgent().run(protocol, prisma, rows, included, echo=echo)
(out / "review.md").write_text(review, encoding="utf-8")
console.print(Panel(f"Done → {out / 'review.md'}", style="green"))
return
# 1) Protocol ---------------------------------------------------------------------
console.rule("[bold]1. Search strategy / protocol")
protocol = SearchAgent().run(question)
console.print(f" [bold]{protocol.review_title}[/]\n Q: {protocol.research_question}")
console.print(f" include: {protocol.inclusion_criteria}\n exclude: {protocol.exclusion_criteria}")
write_json(out / "protocol.json", protocol.model_dump())
# 2) Search -----------------------------------------------------------------------
console.rule("[bold]2. Database search")
if offline:
raw = list(OFFLINE_SAMPLE)
else:
src = [s.strip() for s in sources.split(",") if s.strip()]
per = max(5, max_records // (len(protocol.queries) * len(src)) + 2)
raw = []
for q in protocol.queries:
console.print(f" [cyan]query[/]: {q}")
raw.extend(search_all(q, per, src, log=lambda m: console.print(f" {m}")))
if not raw:
console.print("[yellow]Nothing retrieved — using offline sample.[/]")
raw = list(OFFLINE_SAMPLE)
records = dedupe(raw)[:max_records]
write_json(out / "records.json", [p.model_dump() for p in records])
console.print(f" identified {len(raw)} → after de-duplication {len(records)}")
# 3) Screening --------------------------------------------------------------------
console.rule("[bold]3. Title/abstract screening")
decisions = ScreeningAgent().run(protocol, records, log=lambda m: console.print(f" {m}"))
write_csv(out / "screening.csv", [d.model_dump() for d in decisions])
prisma = build_counts(raw, records, decisions)
write_json(out / "prisma.json", prisma.model_dump())
console.print(prisma.flow_markdown())
included_ids = {d.paper_id for d in decisions if d.decision == "include"}
included = [p for p in records if p.id in included_ids]
if not included:
console.print("[red]No papers included. Loosen criteria or increase --max-records.[/]")
raise typer.Exit(1)
# 4) Extraction -------------------------------------------------------------------
console.rule("[bold]4. Knowledge extraction")
rows = ExtractionAgent().run(protocol, included, log=lambda m: console.print(f" {m}"))
write_csv(out / "evidence.csv", [r.model_dump() for r in rows])
t = Table(title="Evidence table (first rows)")
for c in ("study", "dataset_or_sample", "method", "key_result"):
t.add_column(c, max_width=40)
for r in rows[:10]:
t.add_row(r.study, r.dataset_or_sample, r.method, r.key_result)
console.print(t)
# 5) Writing ----------------------------------------------------------------------
console.rule("[bold]5. Writing the review")
review = WritingAgent().run(protocol, prisma, rows, included, echo=echo)
(out / "review.md").write_text(review, encoding="utf-8")
console.print(Panel(f"Done. Outputs in [bold]{out}[/]\nEdit evidence.csv and re-run with "
f"--from-evidence \"{out}\" to regenerate the review.", style="green"))
if __name__ == "__main__":
app()17.2.4 prisma.py
"""
prisma.py — PRISMA flow bookkeeping, done by CODE not by the LLM.
PRISMA (Preferred Reporting Items for Systematic reviews and Meta-Analyses) asks reviewers to
report how many records were identified, de-duplicated, screened, excluded and included.
We count these deterministically so the numbers in the review are always correct.
Codanics — Python ka Chilla: Build AI Agents
https://codanics.com/courses/python-ka-chilla-build-ai-agents/
"""
from __future__ import annotations
from collections import Counter
from pydantic import BaseModel, Field
class PrismaCounts(BaseModel):
identified_per_source: dict[str, int] = Field(default_factory=dict)
identified_total: int = 0
duplicates_removed: int = 0
screened: int = 0
excluded: int = 0
maybe: int = 0
included: int = 0
exclusion_reasons: dict[str, int] = Field(default_factory=dict)
def flow_markdown(self) -> str:
src = ", ".join(f"{k}: {v}" for k, v in self.identified_per_source.items()) or "n/a"
reasons = "\n".join(f" - {r}: {n}" for r, n in self.exclusion_reasons.items()) or " - (none)"
return f"""```
Identification Records identified from databases: {self.identified_total}
({src})
│
Duplicates removed: {self.duplicates_removed}
▼
Screening Records screened (title/abstract): {self.screened}
│
Records excluded: {self.excluded} | Uncertain / needs human check: {self.maybe}
▼
Included Studies included in the review: {self.included}
```
Exclusion reasons:
{reasons}
"""
def build_counts(raw: list, deduped: list, decisions: list) -> PrismaCounts:
"""`decisions` is a list of objects with `.decision` in {include, exclude, maybe} and `.reason`."""
per_source = Counter(getattr(p, "source", "unknown") for p in raw)
dec = Counter(d.decision for d in decisions)
reasons = Counter(d.reason.split(":")[0].strip()[:60] for d in decisions if d.decision == "exclude")
return PrismaCounts(
identified_per_source=dict(per_source),
identified_total=len(raw),
duplicates_removed=len(raw) - len(deduped),
screened=len(decisions),
excluded=dec.get("exclude", 0),
maybe=dec.get("maybe", 0),
included=dec.get("include", 0),
exclusion_reasons=dict(reasons.most_common(10)),
)17.2.5 tools.py
"""
tools.py — real literature-search tools for the Literature Review Agent.
Every function returns a list[Paper] in ONE common shape, no matter which database it came from.
All four sources are free, public APIs (no key needed; keys only raise rate limits):
arXiv https://arxiv.org/help/api
Semantic Scholar https://api.semanticscholar.org/
PubMed https://www.ncbi.nlm.nih.gov/books/NBK25501/ (E-utilities)
Crossref https://api.crossref.org/
Golden rule of this project: the LLM may only cite papers that came out of these tools.
Codanics — Python ka Chilla: Build AI Agents
https://codanics.com/courses/python-ka-chilla-build-ai-agents/
"""
from __future__ import annotations
import os
import re
import time
import xml.etree.ElementTree as ET
from typing import Iterable
import requests
from dotenv import load_dotenv
from pydantic import BaseModel, Field
load_dotenv()
USER_AGENT = "codanics-research-agents/1.0 (https://codanics.com)"
TIMEOUT = 30
class Paper(BaseModel):
"""One normalised bibliographic record."""
id: str = Field(description="Stable identifier, e.g. arxiv:2401.01234, pmid:38012345, doi:10.1000/xyz")
title: str
abstract: str = ""
authors: list[str] = Field(default_factory=list)
year: int | None = None
venue: str = ""
citations: int | None = None
url: str = ""
source: str = ""
def short(self, n_abstract: int = 900) -> str:
"""Compact text block given to the LLM (keeps prompts small and citable)."""
auth = ", ".join(self.authors[:3]) + (" et al." if len(self.authors) > 3 else "")
cites = f" | citations: {self.citations}" if self.citations is not None else ""
abstract = (self.abstract or "")[:n_abstract]
return f"[{self.id}] {self.title} ({self.year}) — {auth} | {self.venue}{cites}\n{abstract}\n"
# ----------------------------------------------------------------------------- arXiv
def search_arxiv(query: str, max_results: int = 25) -> list[Paper]:
import arxiv # pip install arxiv
client = arxiv.Client(page_size=min(max_results, 100), delay_seconds=3, num_retries=3)
search = arxiv.Search(query=query, max_results=max_results,
sort_by=arxiv.SortCriterion.Relevance)
papers: list[Paper] = []
for r in client.results(search):
papers.append(Paper(
id=f"arxiv:{r.get_short_id()}",
title=" ".join(r.title.split()),
abstract=" ".join(r.summary.split()),
authors=[a.name for a in r.authors],
year=r.published.year if r.published else None,
venue=(r.journal_ref or "arXiv"),
url=r.entry_id,
source="arxiv",
))
return papers
# ------------------------------------------------------------------- Semantic Scholar
def search_semantic_scholar(query: str, max_results: int = 25, year_from: int | None = None) -> list[Paper]:
headers = {"User-Agent": USER_AGENT}
key = os.getenv("SEMANTIC_SCHOLAR_API_KEY")
if key:
headers["x-api-key"] = key
params = {
"query": query,
"limit": min(max_results, 100),
"fields": "title,abstract,year,authors,citationCount,externalIds,venue,url",
}
if year_from:
params["year"] = f"{year_from}-"
url = "https://api.semanticscholar.org/graph/v1/paper/search"
for attempt in range(4): # anonymous access is heavily rate-limited: back off on 429
resp = requests.get(url, params=params, headers=headers, timeout=TIMEOUT)
if resp.status_code != 429:
break
time.sleep(5 * (attempt + 1))
resp.raise_for_status()
papers: list[Paper] = []
for d in resp.json().get("data", []):
ext = d.get("externalIds") or {}
if ext.get("DOI"):
pid = f"doi:{ext['DOI']}"
elif ext.get("ArXiv"):
pid = f"arxiv:{ext['ArXiv']}"
elif ext.get("PubMed"):
pid = f"pmid:{ext['PubMed']}"
else:
pid = f"s2:{d.get('paperId')}"
papers.append(Paper(
id=pid,
title=d.get("title") or "",
abstract=d.get("abstract") or "",
authors=[a.get("name", "") for a in d.get("authors") or []],
year=d.get("year"),
venue=d.get("venue") or "",
citations=d.get("citationCount"),
url=d.get("url") or "",
source="semantic_scholar",
))
return papers
# --------------------------------------------------------------------------- PubMed
def search_pubmed(query: str, max_results: int = 25) -> list[Paper]:
base = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"
common = {"tool": "codanics-research-agents", "email": os.getenv("NCBI_EMAIL", "")}
if os.getenv("NCBI_API_KEY"):
common["api_key"] = os.getenv("NCBI_API_KEY")
ids = requests.get(f"{base}/esearch.fcgi", timeout=TIMEOUT, params={
**common, "db": "pubmed", "term": query, "retmax": max_results, "retmode": "json",
"sort": "relevance"}).json().get("esearchresult", {}).get("idlist", [])
if not ids:
return []
xml = requests.get(f"{base}/efetch.fcgi", timeout=TIMEOUT, params={
**common, "db": "pubmed", "id": ",".join(ids), "retmode": "xml"}).text
root = ET.fromstring(xml)
papers: list[Paper] = []
for art in root.findall(".//PubmedArticle"):
pmid = art.findtext(".//PMID") or ""
title = " ".join((art.findtext(".//ArticleTitle") or "").split())
abstract = " ".join(" ".join(t.itertext()) for t in art.findall(".//AbstractText")).strip()
authors = []
for a in art.findall(".//Author"):
last, fore = a.findtext("LastName"), a.findtext("ForeName")
if last:
authors.append(f"{fore} {last}".strip())
year = art.findtext(".//PubDate/Year") or art.findtext(".//ArticleDate/Year")
venue = art.findtext(".//Journal/Title") or ""
doi = None
for el in art.findall(".//ArticleId"):
if el.get("IdType") == "doi":
doi = el.text
papers.append(Paper(
id=f"pmid:{pmid}", title=title, abstract=abstract, authors=authors,
year=int(year) if year and year.isdigit() else None, venue=venue,
url=f"https://doi.org/{doi}" if doi else f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/",
source="pubmed"))
return papers
# ------------------------------------------------------------------------- Crossref
def search_crossref(query: str, max_results: int = 25) -> list[Paper]:
resp = requests.get("https://api.crossref.org/works", timeout=TIMEOUT,
headers={"User-Agent": USER_AGENT},
params={"query": query, "rows": min(max_results, 100),
"filter": "type:journal-article",
"select": "DOI,title,abstract,author,issued,container-title,is-referenced-by-count,URL"})
resp.raise_for_status()
papers: list[Paper] = []
for it in resp.json().get("message", {}).get("items", []):
title = " ".join((it.get("title") or [""])[0].split())
if not title:
continue
abstract = re.sub(r"<[^>]+>", " ", it.get("abstract") or "")
abstract = " ".join(abstract.split())
authors = [f"{a.get('given', '')} {a.get('family', '')}".strip() for a in it.get("author") or []]
parts = (it.get("issued") or {}).get("date-parts") or [[None]]
year = parts[0][0] if parts and parts[0] else None
papers.append(Paper(
id=f"doi:{it.get('DOI')}", title=title, abstract=abstract, authors=authors,
year=year, venue=(it.get("container-title") or [""])[0],
citations=it.get("is-referenced-by-count"), url=it.get("URL") or "", source="crossref"))
return papers
# ---------------------------------------------------------------------- utilities
def _norm_title(t: str) -> str:
return re.sub(r"[^a-z0-9]", "", t.lower())[:80]
def dedupe(papers: Iterable[Paper]) -> list[Paper]:
"""Merge duplicates across databases (same DOI or same normalised title)."""
seen_ids: set[str] = set()
seen_titles: set[str] = set()
out: list[Paper] = []
for p in papers:
nt = _norm_title(p.title)
if p.id in seen_ids or (nt and nt in seen_titles):
continue
seen_ids.add(p.id)
if nt:
seen_titles.add(nt)
out.append(p)
return out
SOURCES = {
"arxiv": search_arxiv,
"semantic_scholar": search_semantic_scholar,
"pubmed": search_pubmed,
"crossref": search_crossref,
}
def search_all(query: str, max_per_source: int = 20,
sources: Iterable[str] = ("arxiv", "semantic_scholar", "pubmed", "crossref"),
log=print) -> list[Paper]:
"""Query every requested source, tolerate failures, and return a de-duplicated list."""
found: list[Paper] = []
for name in sources:
fn = SOURCES[name]
try:
res = fn(query, max_per_source)
log(f" {name:<17} {len(res):>3} records")
found.extend(res)
except Exception as exc: # network / rate limit / parse errors must never kill the run
log(f" {name:<17} FAILED ({type(exc).__name__}: {exc})")
papers = dedupe(found)
log(f" total after dedupe {len(papers):>3}")
return papers
# -------------------------------------------------------------------- offline demo
OFFLINE_SAMPLE: list[Paper] = [
Paper(id="demo:1", title="Deep learning for drought stress detection in wheat using RGB imagery",
abstract="A CNN trained on RGB canopy images classifies drought stress in wheat with 92% accuracy under field conditions. Hyperspectral and genomic data were not used.",
authors=["A. Khan", "B. Ali"], year=2023, venue="Computers and Electronics in Agriculture", citations=41, source="demo"),
Paper(id="demo:2", title="Hyperspectral indices for early drought detection in cereals: a review",
abstract="Reviews vegetation indices derived from hyperspectral sensors for early drought detection; highlights the lack of integration with machine learning and genotype information.",
authors=["C. Zhang"], year=2022, venue="Remote Sensing", citations=88, source="demo"),
Paper(id="demo:3", title="Genomic prediction of drought tolerance in wheat landraces",
abstract="GBLUP and random forest models predict drought tolerance from SNP data across 400 landraces; phenotypic imaging data were unavailable.",
authors=["D. Rehman", "E. Fatima"], year=2024, venue="Theoretical and Applied Genetics", citations=12, source="demo"),
Paper(id="demo:4", title="Explainable machine learning for crop stress phenotyping",
abstract="Applies SHAP to gradient boosting models of crop stress from multispectral UAV data; calls for multimodal fusion with genomic markers.",
authors=["F. Ahmed"], year=2024, venue="Plant Phenomics", citations=9, source="demo"),
Paper(id="demo:5", title="Transformer models for time-series prediction of soil moisture and crop water stress",
abstract="A temporal transformer predicts soil moisture and canopy temperature from weather and satellite data, outperforming LSTM baselines; not validated on wheat genotypes.",
authors=["G. Li", "H. Wang"], year=2023, venue="Agricultural Water Management", citations=27, source="demo"),
Paper(id="demo:6", title="Multimodal data fusion in precision agriculture: challenges and opportunities",
abstract="Surveys fusion of imaging, sensor, weather and omics data; identifies the absence of open benchmark datasets combining phenomics and genomics for drought.",
authors=["I. Hussain"], year=2025, venue="Frontiers in Plant Science", citations=5, source="demo"),
]