40.1 Literature, Coding, and Analysis Agents

1. The Four Research Agent Roles

In a single afternoon, a researcher hunting for a new antibiotic lead will scan thirty papers, write a clustering script, run a statistical test on the resulting groups, and then pause to ask whether she picked the right test in the first place. These four cognitive modes recur in every research workflow, from a weekend survey to a multi-year discovery campaign, and each maps to a specialized agent class:

Without specialized agents, research teams routinely miss critical papers, re-run failed experiments they forgot about, or apply the wrong statistical test because the same overloaded prompt handles reading, coding, and analysis at once. Structuring these roles as distinct agents eliminates entire categories of such errors.

A research agent is a large language model (LLM) wrapped in a control loop. This loop grants access to external tools (search APIs, code sandboxes, statistical libraries) and persistent state, so the agent can carry out multi-step scientific tasks autonomously. Research agents matter because modern science produces far more literature, data, and candidate hypotheses than any human team can process manually. A single agent can search thousands of papers, run hundreds of experiment configurations, or screen millions of molecules in the time a researcher spends on a handful. The mechanism is the observe-think-act loop. The agent observes its current state (papers found, results obtained), reasons about what to do next (refine a query, run a test, flag an anomaly), executes an action through a tool call, and repeats until it meets a stopping criterion. Use research agents when the task involves high-volume information processing with domain-specific tool use; for single-turn question answering without tool access, a standard LLM prompt suffices.

Literature agents search, retrieve, and synthesize published knowledge. They query application programming interfaces (APIs) (Semantic Scholar, PubMed, arXiv), download and parse full-text PDFs, extract key findings, and produce structured summaries with full citations. Their system prompt emphasizes accuracy, citation fidelity, and awareness of recency. Their tool set includes search APIs, PDF parsers, and citation managers. The RAG pipeline from Chapter 37 provides the retrieval backbone.

Figure 40.1 below illustrates how these four roles exchange typed data in a research workflow, with feedback loops that allow later stages to trigger re-examination by earlier ones.

From Text to Code to Judgment

Coding agents translate research ideas into executable code. They write data processing pipelines, implement algorithms from papers, set up experiment configurations, and produce visualization scripts. Their system prompt emphasizes correctness, reproducibility, and adherence to coding standards. Their tool set includes code execution sandboxes, package managers, and version control. The vibe coding patterns from Chapter 9 apply directly.

Analysis agents run statistical tests, fit models, interpret results, and assess whether findings support or refute a hypothesis. They select appropriate statistical methods, check assumptions (normality, independence, sample size), compute effect sizes and confidence intervals, and flag potential confounders. Their system prompt emphasizes statistical rigor and honest reporting of uncertainty. Their tool set includes statistical libraries, plotting tools, and the Bayesian methods from Chapter 32.

Reviewer agents critique methodology, check for logical errors, identify missing controls, and assess whether conclusions follow from the evidence. They act as an internal peer reviewer, catching problems before findings are shared. Their system prompt emphasizes skepticism, thoroughness, and constructive feedback. Their tool set includes access to the literature agent's outputs (to verify claims against sources) and the analysis agent's results (to check statistical validity). In short: give each cognitive mode its own agent, its own tools, and its own memory, and the research loop polices itself. Figure 40.1.1 illustrates four research agent roles and their information flow.

2. Implementing a Literature Agent

A literature agent wraps the RAG pipeline from Chapter 37 in an agent loop that can plan multi-step searches, follow citation chains, and synthesize findings into structured reviews. The key architectural decision is how much autonomy to give the search planning: should the agent decide its own queries, or should it follow a pre-defined search protocol?

The answer depends on the task. For a targeted question ("What is the binding affinity of compound X to target Y?"), a pre-defined protocol works well. For an exploratory review ("What approaches have been tried for antibiotic resistance prediction?"), the agent needs autonomy to refine its queries based on what it finds. The following code demonstrates both patterns:

from dataclasses import dataclass, field
from typing import Any
import json


@dataclass
class LiteratureAgentState:
    """Typed state for the literature agent's session."""
    query: str
    papers_found: list[dict] = field(default_factory=list)
    papers_read: list[dict] = field(default_factory=list)
    synthesis: str = ""
    search_history: list[str] = field(default_factory=list)
    iteration: int = 0
    max_iterations: int = 10
    confidence: float = 0.0


class LiteratureAgent:
    """Agent that searches, retrieves, and synthesizes scientific literature."""

    SYSTEM_PROMPT = """You are a scientific literature agent. Your role is to
    find, read, and synthesize published research papers relevant to a query.

    Rules:
    - Always cite papers with authors, year, and title.
    - Distinguish between claims supported by evidence and speculation.
    - Report the number of papers searched and read.
    - When uncertain, say so explicitly and suggest follow-up queries.
    - Never fabricate citations or invent paper titles.
    """

    def __init__(self, llm_client, search_tools: dict[str, Any]):
        self.llm = llm_client
        self.tools = search_tools  # semantic_scholar, pubmed, arxiv, etc.

    async def search_papers(
        self, query: str, source: str = "semantic_scholar", limit: int = 20
    ) -> list[dict]:
        """Search for papers using the specified source API."""
        search_fn = self.tools[source]
        results = await search_fn(query=query, limit=limit)
        return [
            {
                "paper_id": r["paperId"],
                "title": r["title"],
                "authors": [a["name"] for a in r.get("authors", [])],
                "year": r.get("year"),
                "abstract": r.get("abstract", ""),
                "citation_count": r.get("citationCount", 0),
                "source": source,
            }
            for r in results
        ]

    async def read_paper(self, paper: dict) -> dict:
        """Retrieve and extract key information from a paper."""
        full_text = await self.tools["pdf_reader"](paper["paper_id"])
        extraction = await self.llm.chat(
            system="Extract: (1) main finding, (2) method, "
                   "(3) key results with numbers, (4) limitations.",
            user=f"Paper: {paper['title']}\n\n{full_text[:8000]}",
        )
        return {
            **paper,
            "extraction": extraction.content,
            "read": True,
        }

    async def should_continue_searching(
        self, state: LiteratureAgentState
    ) -> bool:
        """Decide whether to search more or synthesize.

        Uses a simple confidence threshold: stop when the agent
        believes it has sufficient coverage of the topic.
        """
        if state.iteration >= state.max_iterations:
            return False
        if len(state.papers_read) < 3:
            return True  # always read at least 3 papers

        assessment = await self.llm.chat(
            system="Assess search completeness. Return JSON: "
                   '{"confidence": 0.0-1.0, "reason": "...", '
                   '"suggested_query": "..." or null}',
            user=(
                f"Original query: {state.query}\n"
                f"Papers read: {len(state.papers_read)}\n"
                f"Key findings so far: {state.synthesis[:2000]}\n"
                f"Search queries tried: {state.search_history}"
            ),
        )
        result = json.loads(assessment.content)
        state.confidence = result["confidence"]
        return result["confidence"] < 0.8 and result["suggested_query"] is not None

    async def run(self, query: str) -> LiteratureAgentState:
        """Execute the full literature agent loop."""
        state = LiteratureAgentState(query=query)

        while await self.should_continue_searching(state):
            # search
            current_query = query if state.iteration == 0 else (
                await self._refine_query(state)
            )
            state.search_history.append(current_query)
            papers = await self.search_papers(current_query)
            state.papers_found.extend(papers)

            # read top papers (by citation count, deduplicated)
            seen_ids = {p["paper_id"] for p in state.papers_read}
            new_papers = [
                p for p in papers if p["paper_id"] not in seen_ids
            ]
            new_papers.sort(key=lambda p: p["citation_count"], reverse=True)

            for paper in new_papers[:5]:
                read_result = await self.read_paper(paper)
                state.papers_read.append(read_result)

            # update running synthesis
            state.synthesis = await self._synthesize(state)
            state.iteration += 1

        return state

    async def _refine_query(self, state: LiteratureAgentState) -> str:
        """Generate a refined search query based on findings so far."""
        response = await self.llm.chat(
            system="Generate a follow-up search query to fill gaps.",
            user=(
                f"Original: {state.query}\n"
                f"Previous queries: {state.search_history}\n"
                f"Current synthesis: {state.synthesis[:1000]}"
            ),
        )
        return response.content.strip()

    async def _synthesize(self, state: LiteratureAgentState) -> str:
        """Produce a running synthesis of all papers read."""
        paper_summaries = "\n\n".join(
            f"- {p['title']} ({p['year']}): {p['extraction']}"
            for p in state.papers_read
        )
        response = await self.llm.chat(
            system=self.SYSTEM_PROMPT,
            user=(
                f"Query: {state.query}\n\n"
                f"Papers reviewed:\n{paper_summaries}\n\n"
                "Synthesize the findings into a coherent summary "
                "with citations."
            ),
        )
        return response.content

3. Coding and Analysis Agents

While the literature agent operates on text, the coding and analysis agents operate on data and code. The coding agent translates a research plan into executable experiments; the analysis agent interprets the results.

A coding agent for research differs from a general-purpose coding agent (see Chapter 16) in three ways. First, it must produce reproducible code: fixed random seeds, pinned dependency versions, and logged configurations. Second, it must handle scientific data formats such as HDF5 (hierarchical binary data), FITS (astronomy images), SDF/MOL (molecular structures), and FASTA (biological sequences). Third, it must integrate with experiment tracking: every run logs parameters, metrics, and artifacts to a system like MLflow, an open-source platform for tracking experiment runs, parameters, and outputs (see Chapter 22).

from dataclasses import dataclass
from typing import Any


@dataclass
class ExperimentSpec:
    """Specification for a computational experiment."""
    name: str
    hypothesis: str
    data_sources: list[str]
    method: str
    parameters: dict[str, Any]
    expected_output: str
    success_criteria: str


class CodingAgent:
    """Agent that writes and executes reproducible experiment code."""

    SYSTEM_PROMPT = """You are a scientific coding agent. You write Python code
    to run computational experiments.

    Rules:
    - Always set random seeds for reproducibility.
    - Log all parameters and results to MLflow.
    - Include error handling for data loading and computation.
    - Write clear docstrings explaining what each function does.
    - Pin library versions in requirements.
    - Never modify raw data files; write results to separate outputs.
    """

    def __init__(self, llm_client, sandbox, mlflow_client):
        self.llm = llm_client
        self.sandbox = sandbox        # isolated code execution
        self.mlflow = mlflow_client

    async def generate_experiment_code(
        self, spec: ExperimentSpec
    ) -> str:
        """Generate experiment code from a specification."""
        response = await self.llm.chat(
            system=self.SYSTEM_PROMPT,
            user=(
                f"Write a complete Python script for this experiment:\n"
                f"Name: {spec.name}\n"
                f"Hypothesis: {spec.hypothesis}\n"
                f"Data: {spec.data_sources}\n"
                f"Method: {spec.method}\n"
                f"Parameters: {spec.parameters}\n"
                f"Expected output: {spec.expected_output}\n"
                f"Success criteria: {spec.success_criteria}\n\n"
                "Include MLflow logging with mlflow.log_param() and "
                "mlflow.log_metric(). Set random seed 42."
            ),
        )
        return response.content

    async def run_experiment(
        self, spec: ExperimentSpec
    ) -> dict[str, Any]:
        """Generate, validate, and execute experiment code."""
        code = await self.generate_experiment_code(spec)

        # static validation: check for common issues
        validation = await self._validate_code(code)
        if not validation["valid"]:
            code = await self._fix_code(code, validation["issues"])

        # execute in sandbox
        with self.mlflow.start_run(run_name=spec.name):
            self.mlflow.log_params(spec.parameters)
            result = await self.sandbox.execute(code, timeout=300)

            if result.exit_code != 0:
                return {
                    "status": "error",
                    "error": result.stderr,
                    "code": code,
                }

            self.mlflow.log_metric("exit_code", result.exit_code)
            return {
                "status": "success",
                "stdout": result.stdout,
                "artifacts": result.files,
                "code": code,
                "mlflow_run_id": self.mlflow.active_run().info.run_id,
            }

    async def _validate_code(self, code: str) -> dict:
        """Check code for common research pitfalls."""
        checks = {
            "has_random_seed": "random.seed" in code or "np.random.seed" in code
                               or "torch.manual_seed" in code,
            "has_mlflow_logging": "mlflow.log" in code,
            "no_raw_data_mutation": "inplace=True" not in code,
            "has_error_handling": "try:" in code or "except" in code,
        }
        issues = [k for k, v in checks.items() if not v]
        return {"valid": len(issues) == 0, "issues": issues}

    async def _fix_code(self, code: str, issues: list[str]) -> str:
        """Fix identified issues in generated code."""
        response = await self.llm.chat(
            system="Fix the following issues in this experiment code.",
            user=f"Issues: {issues}\n\nCode:\n{code}",
        )
        return response.content

The analysis agent receives the outputs of the coding agent (data frames, metrics, plots) and applies statistical reasoning:

@dataclass
class AnalysisResult:
    """Structured output from the analysis agent."""
    test_name: str
    test_statistic: float
    p_value: float
    effect_size: float | None
    confidence_interval: tuple[float, float] | None
    assumptions_met: dict[str, bool]
    interpretation: str
    recommendation: str  # "support", "reject", or "inconclusive"


class AnalysisAgent:
    """Agent that selects and runs appropriate statistical analyses."""

    SYSTEM_PROMPT = """You are a statistical analysis agent. You select
    appropriate tests, verify assumptions, and interpret results.

    Rules:
    - Always check test assumptions before running a test.
    - Report effect sizes, not just p-values.
    - Use confidence intervals alongside point estimates.
    - Distinguish statistical significance from practical significance.
    - When assumptions are violated, use non-parametric alternatives.
    - Never p-hack: pre-register the analysis plan before seeing results.
    """

    def __init__(self, llm_client):
        self.llm = llm_client

    async def select_test(
        self, hypothesis: str, data_description: str
    ) -> dict:
        """Select the appropriate statistical test for a hypothesis."""
        response = await self.llm.chat(
            system=self.SYSTEM_PROMPT,
            user=(
                f"Hypothesis: {hypothesis}\n"
                f"Data: {data_description}\n\n"
                "Select the appropriate statistical test. Return JSON:\n"
                '{"test": "name", "assumptions": ["list"], '
                '"parametric": true/false, '
                '"alternative_if_assumptions_fail": "name"}'
            ),
        )
        return json.loads(response.content)

    async def run_analysis(
        self,
        hypothesis: str,
        data_path: str,
        analysis_plan: dict | None = None,
    ) -> AnalysisResult:
        """Execute a complete analysis pipeline."""
        import pandas as pd
        from scipy import stats

        data = pd.read_csv(data_path)

        # select test if no plan provided
        if analysis_plan is None:
            data_desc = (
                f"Shape: {data.shape}, "
                f"Columns: {list(data.columns)}, "
                f"Types: {data.dtypes.to_dict()}"
            )
            analysis_plan = await self.select_test(hypothesis, data_desc)

        # check assumptions
        assumptions = await self._check_assumptions(
            data, analysis_plan["assumptions"]
        )

        # pick the test (parametric or fallback)
        test_name = analysis_plan["test"]
        if not all(assumptions.values()):
            test_name = analysis_plan.get(
                "alternative_if_assumptions_fail", test_name
            )

        # run the test
        stat, p_value = self._execute_test(test_name, data)

        # compute effect size
        effect_size = self._compute_effect_size(test_name, data)

        # interpret
        interpretation = await self._interpret(
            test_name, stat, p_value, effect_size, hypothesis
        )

        # p < 0.05 is the conventional statistical significance threshold;
        # effect_size > 0.2 corresponds to Cohen's "small" threshold,
        # ensuring the result is not just significant but meaningful.
        recommendation = (
            "support" if p_value < 0.05 and (effect_size or 0) > 0.2
            else "reject" if p_value > 0.1
            else "inconclusive"
        )

        return AnalysisResult(
            test_name=test_name,
            test_statistic=stat,
            p_value=p_value,
            effect_size=effect_size,
            confidence_interval=None,  # computed per test type
            assumptions_met=assumptions,
            interpretation=interpretation,
            recommendation=recommendation,
        )

    def _execute_test(
        self, test_name: str, data: "pd.DataFrame"
    ) -> tuple[float, float]:
        """Run a named statistical test on data."""
        from scipy import stats

        # dispatch to the appropriate scipy function
        test_registry = {
            "t_test_independent": lambda d: stats.ttest_ind(
                d.iloc[:, 0].dropna(), d.iloc[:, 1].dropna()
            ),
            "mann_whitney_u": lambda d: stats.mannwhitneyu(
                d.iloc[:, 0].dropna(), d.iloc[:, 1].dropna()
            ),
            "paired_t_test": lambda d: stats.ttest_rel(
                d.iloc[:, 0].dropna(), d.iloc[:, 1].dropna()
            ),
            "chi_squared": lambda d: stats.chi2_contingency(
                pd.crosstab(d.iloc[:, 0], d.iloc[:, 1])
            )[:2],
            "pearson_correlation": lambda d: stats.pearsonr(
                d.iloc[:, 0].dropna(), d.iloc[:, 1].dropna()
            ),
        }
        fn = test_registry.get(test_name)
        if fn is None:
            raise ValueError(f"Unknown test: {test_name}")
        result = fn(data)
        return float(result[0]), float(result[1])

    def _compute_effect_size(
        self, test_name: str, data: "pd.DataFrame"
    ) -> float | None:
        """Compute Cohen's d or equivalent effect size.

        Cohen's d measures the standardized difference between two group
        means: d = (mean1 - mean2) / pooled_std. Values of 0.2, 0.5, and
        0.8 are conventionally interpreted as small, medium, and large.
        """
        import numpy as np

        if test_name in ("t_test_independent", "paired_t_test"):
            g1 = data.iloc[:, 0].dropna()
            g2 = data.iloc[:, 1].dropna()
            pooled_std = np.sqrt(
                (g1.std() ** 2 + g2.std() ** 2) / 2
            )
            if pooled_std == 0:
                return None
            return float((g1.mean() - g2.mean()) / pooled_std)
        return None

    async def _check_assumptions(
        self, data: "pd.DataFrame", assumptions: list[str]
    ) -> dict[str, bool]:
        """Check statistical assumptions for the selected test."""
        from scipy import stats

        results = {}
        for assumption in assumptions:
            if assumption == "normality":
                col = data.iloc[:, 0].dropna()
                _, p = stats.shapiro(col[:5000])  # shapiro limit
                results["normality"] = p > 0.05
            elif assumption == "equal_variance":
                g1 = data.iloc[:, 0].dropna()
                g2 = data.iloc[:, 1].dropna()
                _, p = stats.levene(g1, g2)
                results["equal_variance"] = p > 0.05
            elif assumption == "independence":
                results["independence"] = True  # assumed by design
            else:
                results[assumption] = True  # default pass
        return results

    async def _interpret(
        self, test_name, stat, p_value, effect_size, hypothesis
    ) -> str:
        """Generate a natural language interpretation of results."""
        response = await self.llm.chat(
            system="Interpret statistical results in plain language.",
            user=(
                f"Test: {test_name}\n"
                f"Statistic: {stat:.4f}\n"
                f"P-value: {p_value:.4f}\n"
                f"Effect size: {effect_size}\n"
                f"Hypothesis: {hypothesis}\n\n"
                "Provide a clear interpretation."
            ),
        )
        return response.content

from paperqa import Settings, ask

answer = await ask(
    "What ML methods predict antibiotic resistance from genomics?",
    settings=Settings(
        llm="gpt-4o",
        summary_llm="gpt-4o-mini",
        paper_directory="./papers/",
    ),
)
print(answer.formatted_answer)  # cited synthesis
print(f"Sources: {len(answer.contexts)}")  # papers used

4. Agent Memory: Episodic, Semantic, and Procedural

Building specialized agents solves the capability problem, but each session starts from scratch unless the agents can retain what they have learned. A research agent that forgets everything between sessions is like a postdoc who wakes up each morning with no memory of yesterday's experiments. Effective research agents need persistent memory that accumulates across sessions. Following the cognitive science taxonomy used by Park et al. (2023) in their Generative Agents paper, which simulated believable human behavior using LLM-based agents with structured memory, we distinguish three types:

Episodic memory stores what happened: which papers were read, what experiments were run, what results were obtained, and what decisions were made. It is a timestamped log of events, similar to a lab notebook. The agent queries episodic memory to avoid repeating failed experiments, to recall what it found last week, and to maintain continuity across sessions.

Semantic memory stores what is known: domain facts, established methods, known constraints, and relationships between concepts. It is structured knowledge, often represented as a knowledge graph (see Chapter 38). The agent queries semantic memory to check whether a finding is novel, to look up standard protocols, and to ground its reasoning in established science.

Procedural memory stores how to do things: successful prompts, effective search strategies, reliable code patterns, and learned workflows. It captures expertise that the agent develops through experience. The agent queries procedural memory to reuse strategies that worked and to avoid strategies that failed.

from dataclasses import dataclass, field
from datetime import datetime
import numpy as np


@dataclass
class MemoryRecord:
    """A single memory entry with metadata for retrieval."""
    content: str
    memory_type: str  # "episodic", "semantic", "procedural"
    timestamp: datetime = field(default_factory=datetime.now)
    importance: float = 0.5       # 0.0 to 1.0
    access_count: int = 0
    last_accessed: datetime | None = None
    embedding: np.ndarray | None = None
    metadata: dict = field(default_factory=dict)


class AgentMemory:
    """Three-store memory system for research agents.

    Retrieval uses a weighted combination of recency, importance,
    and semantic similarity (following Park et al., 2023).
    """

    def __init__(self, embedding_fn, decay_rate: float = 0.995):
        self.stores: dict[str, list[MemoryRecord]] = {
            "episodic": [],
            "semantic": [],
            "procedural": [],
        }
        self.embed = embedding_fn
        self.decay_rate = decay_rate

    async def add(
        self,
        content: str,
        memory_type: str,
        importance: float = 0.5,
        metadata: dict | None = None,
    ) -> MemoryRecord:
        """Store a new memory record."""
        embedding = await self.embed(content)
        record = MemoryRecord(
            content=content,
            memory_type=memory_type,
            importance=importance,
            embedding=embedding,
            metadata=metadata or {},
        )
        self.stores[memory_type].append(record)
        return record

    async def retrieve(
        self,
        query: str,
        memory_type: str | None = None,
        top_k: int = 5,
        recency_weight: float = 1.0,
        importance_weight: float = 1.0,
        relevance_weight: float = 1.0,
    ) -> list[MemoryRecord]:
        """Retrieve memories by weighted recency + importance + relevance.

        Score = alpha * recency + beta * importance + gamma * relevance
        where recency decays exponentially with time.
        """
        query_embedding = await self.embed(query)
        now = datetime.now()

        # gather candidates from specified store(s)
        candidates = []
        stores_to_search = (
            [memory_type] if memory_type else list(self.stores.keys())
        )
        for store_name in stores_to_search:
            candidates.extend(self.stores[store_name])

        if not candidates:
            return []

        scored = []
        for record in candidates:
            # recency: exponential decay by hours since creation
            hours_ago = (now - record.timestamp).total_seconds() / 3600
            recency = self.decay_rate ** hours_ago

            # relevance: cosine similarity (dot product of unit vectors)
            if record.embedding is not None and query_embedding is not None:
                cos_sim = np.dot(record.embedding, query_embedding) / (
                    np.linalg.norm(record.embedding)
                    * np.linalg.norm(query_embedding)
                    + 1e-8
                )
                relevance = float(cos_sim)
            else:
                relevance = 0.0

            score = (
                recency_weight * recency
                + importance_weight * record.importance
                + relevance_weight * relevance
            )
            scored.append((score, record))

        scored.sort(key=lambda x: x[0], reverse=True)

        # update access metadata
        results = []
        for _, record in scored[:top_k]:
            record.access_count += 1
            record.last_accessed = now
            results.append(record)

        return results

    def reflect(self, memories: list[MemoryRecord]) -> str:
        """Combine multiple memories into a higher-level summary.

        This implements the 'reflection' step from Park et al. (2023):
        periodically consolidate low-level observations into
        higher-level insights stored in semantic memory.
        """
        contents = "\n".join(
            f"[{m.memory_type}] {m.content}" for m in memories
        )
        return contents  # in production, pass to LLM for abstraction

The three memory types interact during a research session. When the literature agent finds a new paper, it creates an episodic record ("Read Smith et al. 2024 on Tuesday; method X outperforms method Y by 15%"). If the finding is significant, the agent also updates semantic memory, adding the performance comparison to the domain knowledge graph. A novel search strategy triggers a procedural record ("Searching for 'method X benchmark' on Semantic Scholar yields higher-quality results than PubMed for computational papers").

5. Optimal Stopping: When to Stop Searching

Memory lets an agent accumulate knowledge across sessions, but accumulation without a stopping rule means the agent keeps searching forever. A literature agent can always find one more paper. An analysis agent can always try one more statistical test. The fundamental question is: when should the agent stop gathering information and commit to an action? This is the optimal stopping problem, one of the most studied problems in decision theory.

The classic formulation is the secretary problem (also known as the best-choice problem). You interview \(n\) candidates sequentially. After each interview, you must immediately accept or reject the candidate; you cannot go back. The optimal strategy is to reject the first \(n/e \approx 0.368n\) candidates (the "exploration phase," where you gather information without committing), then accept the first candidate better than all those seen so far (the "exploitation phase," where you act on what you have learned). This gives a probability of selecting the best candidate that converges to \(1/e \approx 0.368\) as \(n \to \infty\).

Mental Model

Think of optimal stopping like apartment hunting in a new city. You have one month of viewings scheduled. For the first ~11 days (37% of your month), you tour apartments with no intention of signing a lease; you are calibrating your sense of what is available at your price point. After that calibration window, you commit to the first apartment that is better than everything you saw during the calibration phase. Without the calibration window, you would either grab the first decent place (missing better options) or keep looking forever (losing good options to other renters). A research agent's exploration phase works the same way: early search iterations calibrate the quality landscape, so the agent knows what "good enough" looks like before it commits to synthesizing or acting on a finding.

The analogy is imperfect, yet the core lesson transfers. An agent searching for papers, hypotheses, or analysis methods must eventually commit. Optimal stopping theory prescribes a specific exploration length: spend roughly 37% of the search budget exploring before exploiting the best finding so far.

In practice, research agents use a softer version: a confidence threshold combined with a diminishing returns detector. The agent tracks how much new information each search iteration provides. When the marginal information gain, the fraction of genuinely new information added by the latest iteration relative to the total accumulated so far, drops below a threshold, the agent stops:

class OptimalStoppingCriterion:
    """Decides when an agent should stop searching and start acting.

    Combines the 37% rule with a diminishing-returns detector:
    - Explore for at least 37% of the budget (exploration phase).
    - After that, stop when marginal gain falls below threshold.
    """

    def __init__(
        self,
        budget: int,
        gain_threshold: float = 0.05,
        exploration_fraction: float = 1.0 / 2.718,
    ):
        self.budget = budget
        self.gain_threshold = gain_threshold
        self.explore_until = int(budget * exploration_fraction)
        self.gains: list[float] = []

    def record_gain(self, new_information: float, total_information: float):
        """Record the marginal gain from the latest search iteration."""
        if total_information > 0:
            marginal = new_information / total_information
        else:
            marginal = 1.0
        self.gains.append(marginal)

    def should_stop(self, iteration: int) -> bool:
        """Apply the optimal stopping criterion."""
        # never stop during exploration phase
        if iteration < self.explore_until:
            return False

        # stop if budget exhausted
        if iteration >= self.budget:
            return True

        # stop if last 2 iterations had low marginal gain
        if len(self.gains) >= 2:
            recent_avg = sum(self.gains[-2:]) / 2
            if recent_avg < self.gain_threshold:
                return True

        return False


# usage in the literature agent loop:
# stopper = OptimalStoppingCriterion(budget=10)
# for i in range(10):
#     new_papers = search(...)
#     stopper.record_gain(len(new_papers), len(all_papers))
#     if stopper.should_stop(i):
#         break

6. The Reviewer Agent: Closing the Loop

Optimal stopping tells an agent when to act, but it says nothing about whether the action was correct. The reviewer agent is the quality gate of the research pipeline. It receives the outputs of the other three agents (literature synthesis, code, analysis results) and checks for:

• Methodological errors: Was the right statistical test used? Were assumptions checked? Is the sample size sufficient?
• Logical consistency: Do the conclusions follow from the evidence? Are there unstated assumptions?
• Citation accuracy: Do the cited papers actually say what the literature agent claims they say?
• Reproducibility: Is the code complete, are dependencies specified, can the experiment be re-run?
• Novelty claims: Is the finding actually novel, or does existing literature already report it?

@dataclass
class ReviewResult:
    """Structured output from a reviewer agent."""
    verdict: str        # "accept", "revise", "reject"
    score: float        # 0.0 to 1.0
    issues: list[dict]  # [{"severity": "critical", "description": "..."}]
    suggestions: list[str]
    missing_references: list[str]
    reproducibility_score: float  # 0.0 to 1.0


class ReviewerAgent:
    """Agent that reviews research outputs for quality and correctness."""

    SYSTEM_PROMPT = """You are a scientific reviewer agent. You evaluate
    research outputs for correctness, rigor, and completeness.

    You are constructively skeptical: assume nothing is correct until verified.
    Check every statistical claim against the raw numbers. Verify that every
    citation supports the claim it is attached to. Flag missing controls,
    alternative explanations, and unstated assumptions.

    Output a structured review with:
    1. Verdict: accept, revise, or reject
    2. Score: 0.0 to 1.0
    3. Issues: list with severity (critical/major/minor) and description
    4. Suggestions: actionable improvements
    5. Missing references: papers that should have been cited
    6. Reproducibility score: 0.0 to 1.0
    """

    def __init__(self, llm_client, literature_agent):
        self.llm = llm_client
        self.lit_agent = literature_agent  # for citation verification

    async def review(
        self,
        synthesis: str,
        code: str | None = None,
        analysis: "AnalysisResult | None" = None,
    ) -> ReviewResult:
        """Review a complete research output."""
        # step 1: check citations
        citation_issues = await self._verify_citations(synthesis)

        # step 2: check methodology (if analysis provided)
        method_issues = []
        if analysis is not None:
            method_issues = await self._check_methodology(analysis)

        # step 3: check reproducibility (if code provided)
        repro_score = 1.0
        repro_issues = []
        if code is not None:
            repro_result = await self._check_reproducibility(code)
            repro_score = repro_result["score"]
            repro_issues = repro_result["issues"]

        # step 4: overall assessment
        all_issues = citation_issues + method_issues + repro_issues
        critical_count = sum(
            1 for i in all_issues if i["severity"] == "critical"
        )

        if critical_count > 0:
            verdict = "reject"
            score = 0.3
        elif len(all_issues) > 3:
            verdict = "revise"
            score = 0.6
        else:
            verdict = "accept"
            score = 0.9

        return ReviewResult(
            verdict=verdict,
            score=score,
            issues=all_issues,
            suggestions=[i["suggestion"] for i in all_issues if "suggestion" in i],
            missing_references=[],
            reproducibility_score=repro_score,
        )

    async def _verify_citations(self, text: str) -> list[dict]:
        """Spot-check citations against source papers."""
        response = await self.llm.chat(
            system=(
                "Extract all citations from this text. For each, assess "
                "whether the claim attributed to the citation is plausible "
                "based on the paper title and context. Flag suspicious ones."
            ),
            user=text,
        )
        # in production, cross-reference with the literature agent's
        # stored paper extractions
        return json.loads(response.content)

    async def _check_methodology(
        self, analysis: AnalysisResult
    ) -> list[dict]:
        """Check statistical methodology for common errors."""
        issues = []
        if not analysis.assumptions_met.get("normality", True):
            if "t_test" in analysis.test_name:
                issues.append({
                    "severity": "critical",
                    "description": (
                        f"Used {analysis.test_name} but normality "
                        "assumption is violated."
                    ),
                    "suggestion": "Switch to Mann-Whitney U test.",
                })
        if analysis.p_value < 0.05 and (analysis.effect_size or 0) < 0.2:
            issues.append({
                "severity": "major",
                "description": (
                    "Statistically significant but effect size is trivial "
                    f"(d={analysis.effect_size:.3f})."
                ),
                "suggestion": (
                    "Report as 'statistically significant but with "
                    "negligible practical effect.'"
                ),
            })
        return issues

    async def _check_reproducibility(self, code: str) -> dict:
        """Score code for reproducibility criteria."""
        checks = {
            "random_seed": "seed" in code.lower(),
            "version_pinning": "==" in code or "requirements" in code.lower(),
            "data_path_not_hardcoded": "/home/" not in code
                                       and "C:\\" not in code,
            "mlflow_logging": "mlflow" in code,
            "docstrings": '"""' in code or "'''" in code,
        }
        score = sum(checks.values()) / len(checks)
        issues = [
            {"severity": "minor", "description": f"Missing: {k}"}
            for k, v in checks.items() if not v
        ]
        return {"score": score, "issues": issues}

Exercises

1. Conceptual: Draw a diagram showing the information flow between the four research agent roles (literature, coding, analysis, reviewer) for a hypothesis testing task. Label each edge with the data type being passed (e.g., "paper summaries", "code", "statistical results", "review feedback"). Identify which edges create feedback loops and explain why those loops are necessary. (See Figure 40.1 for the reference architecture.)
2. Coding: Extend the AgentMemory class with a consolidate() method that groups related episodic memories (same topic, within 24 hours), summarizes them using an LLM, and stores the summary as a new semantic memory record. Test with a sequence of 20 episodic memories about reading papers on the same topic.
3. Analysis: Compare the optimal stopping criterion with a simpler approach: stop after a fixed number of iterations (e.g., always search exactly 5 times). Run the literature agent on 10 different queries with each stopping strategy. Measure coverage (fraction of relevant papers found, as judged by a human expert) and cost (total API tokens consumed). When does optimal stopping outperform fixed stopping?