53.4 Building a Supervised AI Scientist

1. Architecture Overview

What happens when you give an LLM a research question, a compute budget, and permission to write its own experiments? It generates a hypothesis, codes a script, runs it in a sandbox, evaluates the results, drafts a report, and then waits for a human to decide whether any of it was worth doing.

Without a structured pipeline, teams that hand an LLM a research question and a compute budget routinely burn through dollars on circular hypotheses, unreproducible code, and self-congratulatory evaluations that no human ever reviews. The architecture below exists to prevent exactly those failure modes.

The supervised AI scientist consists of six agents connected by a LangGraph state machine, where LangGraph is a graph-based agent orchestration framework that models workflows as nodes (agents) and edges (transitions), with built-in support for checkpointing and human-in-the-loop interrupts. Three of the transitions are human gates: points where the system pauses and waits for human approval before proceeding. The architecture follows the separation principle from Section 53.2's lessons: no agent evaluates its own output. Figure 53.4 shows the complete pipeline, with human gates highlighted in gold and feedback loops shown as dashed red arrows. Figure 53.4.1 illustrates Supervised AI scientist pipeline with human gates and feedback loops.

Human Gates Defined

A human gate is a programmatic checkpoint where the pipeline halts and transfers control to a human supervisor. The supervisor must explicitly approve or reject the system's output before execution resumes. Human gates matter because fully autonomous research systems can pursue unproductive directions, introduce subtle methodological errors, or waste compute on low-value experiments. A well-placed gate catches these failures before they propagate. Mechanistically, a human gate serializes the pipeline's current state, presents a summary to the supervisor through a user interface (UI) or command-line interface (CLI) prompt, and blocks the state machine transition until the supervisor provides a binary decision. Use human gates (rather than fully automated quality checks) whenever the decision requires value judgments about research direction, safety implications of running generated code, or whether results merit publication. Use automated checks for mechanical validations such as type checking, schema conformance, or resource limit enforcement.

2. State Definition

All data flows between agents through a single shared LangGraph state object. Each agent reads from and writes to this state, so downstream agents can access everything produced upstream. In short: a single shared state object is the pipeline's memory; every agent reads what came before and writes what comes next, so no discovery is ever lost between phases.

from dataclasses import dataclass, field
from enum import Enum
from typing import Any


class Phase(Enum):
    HYPOTHESIS = "hypothesis"
    HYPOTHESIS_GATE = "hypothesis_gate"
    CODING = "coding"
    EXPERIMENT_GATE = "experiment_gate"
    RUNNING = "running"
    EVALUATION = "evaluation"
    REVIEW = "review"
    REPORT_GATE = "report_gate"
    REPORTING = "reporting"
    DONE = "done"


@dataclass
class ResearchState:
    """Shared state for the supervised AI scientist pipeline."""

    # Configuration
    research_domain: str = ""
    max_retries: int = 3
    compute_budget_dollars: float = 50.0

    # Phase tracking
    current_phase: Phase = Phase.HYPOTHESIS
    iteration: int = 0
    spent_dollars: float = 0.0

    # Hypothesis phase
    hypothesis: str = ""
    hypothesis_rationale: str = ""
    novelty_score: float = 0.0
    literature_context: list[str] = field(
        default_factory=list
    )

    # Coding phase
    experiment_code: str = ""
    code_review_notes: str = ""
    sandbox_path: str = ""

    # Experiment phase
    raw_results: dict[str, Any] = field(
        default_factory=dict
    )
    metrics: dict[str, float] = field(
        default_factory=dict
    )
    artifacts: list[str] = field(default_factory=list)

    # Evaluation phase
    statistical_tests: dict[str, Any] = field(
        default_factory=dict
    )
    is_significant: bool = False
    effect_size: float = 0.0

    # Review phase
    review_verdict: str = ""  # "accept", "revise", "reject"
    review_comments: list[str] = field(
        default_factory=list
    )

    # Report phase
    report_markdown: str = ""
    mlflow_run_id: str = ""

    # Human gate decisions
    gates_passed: dict[str, bool] = field(
        default_factory=dict
    )

3. Hypothesis Generation Agent

The hypothesis agent combines gap analysis from Chapter 39 with the novelty filter from Section 53.1. It generates a hypothesis, checks it against existing literature using PaperQA2 (a retrieval-augmented question-answering system that searches and synthesizes scientific papers to ground LLM responses in published literature), and passes it to the first human gate.

import anthropic
import json


class HypothesisAgent:
    """Generate and validate research hypotheses."""

    def __init__(
        self,
        novelty_filter,   # NoveltyFilter from Section 53.1
        literature_agent,  # PaperQA2 wrapper
    ):
        self.client = anthropic.Anthropic()
        self.novelty_filter = novelty_filter
        self.literature = literature_agent

    def run(self, state: ResearchState) -> ResearchState:
        """Generate a novel, literature-grounded hypothesis."""
        # Step 1: Gather literature context
        context_query = (
            f"What are the current open problems and recent "
            f"advances in {state.research_domain}?"
        )
        lit_results = self.literature.search(context_query, k=10)
        state.literature_context = [
            f"{r.title}: {r.summary}" for r in lit_results
        ]

        # Step 2: Generate hypothesis
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            system=(
                "You are a research scientist. Generate a single "
                "testable hypothesis based on the provided literature "
                "context. The hypothesis must be: (1) specific enough "
                "to test with a single experiment, (2) novel relative "
                "to the cited literature, (3) feasible within a $50 "
                "compute budget. Output JSON with 'hypothesis' and "
                "'rationale' fields."
            ),
            messages=[{
                "role": "user",
                "content": (
                    f"Research domain: {state.research_domain}\n\n"
                    f"Literature context:\n"
                    + "\n".join(state.literature_context)
                ),
            }],
        )
        result = json.loads(response.content[0].text)
        state.hypothesis = result["hypothesis"]
        state.hypothesis_rationale = result["rationale"]

        # Step 3: Novelty check
        novelty = self.novelty_filter.check_novelty(
            state.hypothesis
        )
        state.novelty_score = novelty["distance"]

        if not novelty["is_novel"]:
            # Regenerate with explicit instruction to differ
            state.iteration += 1
            if state.iteration < state.max_retries:
                return self.run(state)  # retry

        state.current_phase = Phase.HYPOTHESIS_GATE
        return state

4. Coding Agent with Claude Code SDK

The coding agent uses the Claude Code software development kit (SDK) to write experiment scripts in an agentic loop. Unlike direct API calls, the SDK provides the agent with file system access, terminal execution, and iterative debugging within a sandboxed environment. This is the same capability we used for AI-assisted implementation in Chapter 16, now directed at writing scientific experiments rather than production software.

import subprocess
import json
import tempfile
from pathlib import Path


class CodingAgent:
    """Write experiment code using Claude Code SDK."""

    def __init__(
        self,
        sandbox_dir: str | None = None,
        max_iterations: int = 5,
    ):
        self.sandbox_dir = sandbox_dir or tempfile.mkdtemp(
            prefix="ai_scientist_"
        )
        self.max_iterations = max_iterations

    def run(self, state: ResearchState) -> ResearchState:
        """Generate experiment code for the approved hypothesis."""
        prompt = self._build_prompt(state)

        # Use Claude Code SDK in subprocess mode
        result = subprocess.run(
            [
                "claude",
                "--print",
                "--output-format", "json",
                "--max-turns", str(self.max_iterations),
                "--allowedTools", "Edit,Write,Bash",
                "-p", prompt,
            ],
            capture_output=True,
            text=True,
            cwd=self.sandbox_dir,
            timeout=300,
        )

        response = json.loads(result.stdout)
        state.experiment_code = self._read_generated_code()
        state.sandbox_path = self.sandbox_dir
        state.current_phase = Phase.EXPERIMENT_GATE
        return state

    def _build_prompt(self, state: ResearchState) -> str:
        return f"""You are writing a scientific experiment script.

HYPOTHESIS: {state.hypothesis}
RATIONALE: {state.hypothesis_rationale}

Write a complete Python experiment script (experiment.py) that:
1. Tests the hypothesis with a controlled experiment
2. Includes a baseline comparison
3. Runs 5 random seeds for statistical validity
4. Saves all metrics to results.json
5. Generates at least one visualization (saved as plot.png)
6. Completes within 30 minutes on a single GPU
7. Uses standard libraries (PyTorch, scikit-learn, numpy, matplotlib)

Also write a requirements.txt for any dependencies.

The script must be self-contained and runnable with:
  python experiment.py

Save all outputs to a 'results/' subdirectory."""

    def _read_generated_code(self) -> str:
        code_path = Path(self.sandbox_dir) / "experiment.py"
        if code_path.exists():
            return code_path.read_text()
        return ""

5. Experiment Runner with Sandboxing

The experiment runner executes the generated code in an isolated environment. Sandboxing is essential: AI-generated code should never run with access to the host file system, network, or other experiments' data. We use Docker containers (lightweight virtualization units that package code and dependencies into an isolated runtime environment) for isolation and MLflow, an open-source platform for tracking experiment parameters, metrics, and artifacts across runs, for tracking.

import subprocess
import json
import mlflow
from pathlib import Path


class ExperimentRunner:
    """Execute experiments in sandboxed containers."""

    def __init__(
        self,
        docker_image: str = "python:3.11-slim",
        gpu: bool = True,
        timeout_seconds: int = 1800,
    ):
        self.docker_image = docker_image
        self.gpu = gpu
        self.timeout = timeout_seconds

    def run(self, state: ResearchState) -> ResearchState:
        """Run the experiment in a Docker container."""
        sandbox = Path(state.sandbox_path)

        # Start MLflow run
        with mlflow.start_run() as run:
            state.mlflow_run_id = run.info.run_id

            # Log hypothesis and code as artifacts
            mlflow.log_param("hypothesis", state.hypothesis[:250])
            mlflow.log_param("novelty_score", state.novelty_score)
            mlflow.log_artifact(str(sandbox / "experiment.py"))

            # Build Docker command
            gpu_flag = "--gpus all" if self.gpu else ""
            cmd = (
                f"docker run --rm {gpu_flag} "
                f"-v {sandbox}:/workspace "
                f"-w /workspace "
                f"{self.docker_image} "
                f"bash -c '"
                f"pip install -r requirements.txt "
                f"&& python experiment.py'"
            )

            try:
                result = subprocess.run(
                    cmd,
                    shell=True,
                    capture_output=True,
                    text=True,
                    timeout=self.timeout,
                )

                # Read results
                results_path = sandbox / "results" / "results.json"
                if results_path.exists():
                    state.raw_results = json.loads(
                        results_path.read_text()
                    )
                    # Log metrics to MLflow
                    for key, value in state.raw_results.items():
                        if isinstance(value, (int, float)):
                            mlflow.log_metric(key, value)
                            state.metrics[key] = value

                # Log artifacts (plots, saved models)
                results_dir = sandbox / "results"
                if results_dir.exists():
                    for artifact in results_dir.iterdir():
                        mlflow.log_artifact(str(artifact))
                        state.artifacts.append(str(artifact))

                state.current_phase = Phase.EVALUATION

            except subprocess.TimeoutExpired:
                mlflow.log_param("status", "timeout")
                state.review_verdict = "reject"
                state.review_comments.append(
                    f"Experiment timed out after "
                    f"{self.timeout}s."
                )
                state.current_phase = Phase.REVIEW

            except Exception as e:
                mlflow.log_param("status", "error")
                mlflow.log_param("error", str(e)[:500])
                state.current_phase = Phase.CODING  # retry

        return state

6. Evaluation and Review Agents

The evaluation agent performs statistical analysis on the raw results, testing whether the hypothesis is supported. The review agent then critiques the entire process, checking for methodological flaws. These two agents enforce the separation principle: the system that ran the experiment does not judge its own output.

Mental Model

Think of the separation between experiment execution and evaluation as the relationship between a chef and a restaurant critic. The chef (experiment runner) prepares the dish following a recipe, tasting along the way to get the seasoning right. But the published review comes from the critic (evaluation and review agents), who has no stake in the kitchen's reputation. If the chef also wrote the review, every dish would be "exquisite." The same applies here: if the agent that generated the hypothesis and code also judged whether the results were significant, it would be structurally biased toward confirming its own work. Splitting generation from judgment forces the evaluation to operate on outputs alone, without access to the optimism that produced them.

from scipy import stats
import numpy as np


class EvaluationAgent:
    """Statistical evaluation of experiment results."""

    def run(self, state: ResearchState) -> ResearchState:
        """Analyze results for statistical significance."""
        results = state.raw_results

        # Extract baseline and treatment results
        baseline = np.array(
            results.get("baseline_scores", [])
        )
        treatment = np.array(
            results.get("treatment_scores", [])
        )

        if len(baseline) < 2 or len(treatment) < 2:
            state.is_significant = False
            state.statistical_tests["error"] = (
                "Insufficient data points for testing"
            )
            state.current_phase = Phase.REVIEW
            return state

        # Paired t-test (if same seeds used)
        if len(baseline) == len(treatment):
            t_stat, p_value = stats.ttest_rel(
                treatment, baseline
            )
        else:
            t_stat, p_value = stats.ttest_ind(
                treatment, baseline
            )

        # Effect size (Cohen's d: standardized difference
        # between two means, where d > 0.8 is "large")
        pooled_std = np.sqrt(
            (np.std(baseline) ** 2 + np.std(treatment) ** 2)
            / 2
        )
        cohens_d = (
            (np.mean(treatment) - np.mean(baseline))
            / pooled_std
            if pooled_std > 0
            else 0.0
        )

        state.statistical_tests = {
            "test": "paired_t_test",
            "t_statistic": float(t_stat),
            "p_value": float(p_value),
            "cohens_d": float(cohens_d),
            "baseline_mean": float(np.mean(baseline)),
            "treatment_mean": float(np.mean(treatment)),
            "n_seeds": len(baseline),
        }

        state.is_significant = p_value < 0.05
        state.effect_size = abs(cohens_d)
        state.current_phase = Phase.REVIEW
        return state


class ReviewAgent:
    """Methodological review of the experiment."""

    def __init__(self):
        self.client = anthropic.Anthropic()

    def run(self, state: ResearchState) -> ResearchState:
        """Critique the experiment methodology and results."""
        review_prompt = f"""You are a rigorous scientific reviewer.
Review the following experiment for methodological flaws.

HYPOTHESIS: {state.hypothesis}

EXPERIMENT CODE (first 2000 chars):
{state.experiment_code[:2000]}

RESULTS:
{json.dumps(state.statistical_tests, indent=2)}

Check for:
1. Data leakage between train and test sets
2. Proper random seed control
3. Fair baseline comparison (same hyperparameter budget)
4. Appropriate statistical test for the data
5. Multiple comparisons correction if needed
6. Effect size interpretation (is it practically meaningful?)
7. Code bugs that could invalidate results

Output JSON with:
- "verdict": "accept", "revise", or "reject"
- "comments": list of specific issues found
- "severity": "minor", "major", or "critical" for each comment
"""

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            messages=[{
                "role": "user", "content": review_prompt,
            }],
        )

        review = json.loads(response.content[0].text)
        state.review_verdict = review["verdict"]
        state.review_comments = review["comments"]

        if review["verdict"] == "revise":
            state.iteration += 1
            if state.iteration < state.max_retries:
                state.current_phase = Phase.CODING
            else:
                state.review_verdict = "reject"
                state.current_phase = Phase.REPORT_GATE
        elif review["verdict"] == "reject":
            state.current_phase = Phase.HYPOTHESIS
        else:
            state.current_phase = Phase.REPORT_GATE

        return state

7. Orchestrating with LangGraph

Wiring the agents together requires a mechanism for forward data flow, backward feedback loops, and execution halts at human gates.

LangGraph connects the agents into a state machine with conditional edges and human-in-the-loop gates. The interrupt_before parameter pauses execution at each human gate and presents the current state to the supervisor for approval. (As of 2025, LangGraph's API has evolved. Newer versions use langgraph.checkpoint.memory.InMemorySaver rather than MemorySaver, and the interrupt function replaces raw interrupt_before lists for finer-grained control. The architectural pattern shown here remains valid; consult the current LangGraph documentation for updated import paths.)

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver


def build_ai_scientist_graph(
    hypothesis_agent: HypothesisAgent,
    coding_agent: CodingAgent,
    runner: ExperimentRunner,
    evaluation_agent: EvaluationAgent,
    review_agent: ReviewAgent,
) -> StateGraph:
    """Build the supervised AI scientist state graph."""

    graph = StateGraph(ResearchState)

    # Add agent nodes
    graph.add_node("hypothesis", hypothesis_agent.run)
    graph.add_node("coding", coding_agent.run)
    graph.add_node("experiment", runner.run)
    graph.add_node("evaluation", evaluation_agent.run)
    graph.add_node("review", review_agent.run)
    graph.add_node("report", generate_report)

    # Human gate nodes (identity functions; the interrupt
    # happens before the node executes)
    graph.add_node(
        "hypothesis_gate", lambda s: s
    )
    graph.add_node(
        "experiment_gate", lambda s: s
    )
    graph.add_node(
        "report_gate", lambda s: s
    )

    # Entry point
    graph.set_entry_point("hypothesis")

    # Forward edges
    graph.add_edge("hypothesis", "hypothesis_gate")
    graph.add_edge("hypothesis_gate", "coding")
    graph.add_edge("coding", "experiment_gate")
    graph.add_edge("experiment_gate", "experiment")
    graph.add_edge("experiment", "evaluation")
    graph.add_edge("evaluation", "review")

    # Conditional edge from review
    def review_router(state: ResearchState) -> str:
        if state.review_verdict == "accept":
            return "report_gate"
        elif state.review_verdict == "revise":
            return "coding"
        else:  # reject
            return "hypothesis"

    graph.add_conditional_edges(
        "review", review_router,
        {
            "report_gate": "report_gate",
            "coding": "coding",
            "hypothesis": "hypothesis",
        },
    )

    graph.add_edge("report_gate", "report")
    graph.add_edge("report", END)

    # Compile with human-in-the-loop interrupts
    checkpointer = MemorySaver()
    return graph.compile(
        checkpointer=checkpointer,
        interrupt_before=[
            "hypothesis_gate",
            "experiment_gate",
            "report_gate",
        ],
    )


def generate_report(state: ResearchState) -> ResearchState:
    """Generate a markdown research report."""
    client = anthropic.Anthropic()

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        system=(
            "You are a scientific writer. Generate a concise "
            "research report in markdown format covering: "
            "hypothesis, methodology, results, statistical "
            "analysis, limitations, and conclusions."
        ),
        messages=[{
            "role": "user",
            "content": (
                f"Hypothesis: {state.hypothesis}\n\n"
                f"Rationale: {state.hypothesis_rationale}\n\n"
                f"Results: {json.dumps(state.metrics)}\n\n"
                f"Statistical tests: "
                f"{json.dumps(state.statistical_tests)}\n\n"
                f"Review comments: "
                f"{json.dumps(state.review_comments)}"
            ),
        }],
    )

    state.report_markdown = response.content[0].text
    state.current_phase = Phase.DONE
    return state

8. Running the Pipeline

The pipeline pauses at each human gate, prints the current state for review, and waits for approval.

import uuid


def run_supervised_ai_scientist(
    research_domain: str,
    compute_budget: float = 50.0,
) -> ResearchState:
    """Run the full supervised AI scientist pipeline."""
    # Initialize components
    novelty_filter = NoveltyFilter(threshold=0.3)
    # ... build index from existing papers ...

    hypothesis_agent = HypothesisAgent(
        novelty_filter=novelty_filter,
        literature_agent=PaperQA2Wrapper(),
    )
    coding_agent = CodingAgent()
    runner = ExperimentRunner(gpu=True)
    evaluation_agent = EvaluationAgent()
    review_agent = ReviewAgent()

    graph = build_ai_scientist_graph(
        hypothesis_agent,
        coding_agent,
        runner,
        evaluation_agent,
        review_agent,
    )

    # Initial state
    state = ResearchState(
        research_domain=research_domain,
        compute_budget_dollars=compute_budget,
    )

    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    # Run until first gate
    result = graph.invoke(state, config)

    # Human gate loop
    while result.current_phase != Phase.DONE:
        gate_name = result.current_phase.value
        print(f"\n{'='*60}")
        print(f"HUMAN GATE: {gate_name}")
        print(f"{'='*60}")

        if gate_name == "hypothesis_gate":
            print(f"Hypothesis: {result.hypothesis}")
            print(f"Novelty score: {result.novelty_score:.3f}")
            print(f"Rationale: {result.hypothesis_rationale}")
        elif gate_name == "experiment_gate":
            print(f"Code length: {len(result.experiment_code)} chars")
            print("Review the code at: "
                  f"{result.sandbox_path}/experiment.py")
        elif gate_name == "report_gate":
            print(f"Verdict: {result.review_verdict}")
            print(f"P-value: "
                  f"{result.statistical_tests.get('p_value', 'N/A')}")
            print(f"Effect size: {result.effect_size:.3f}")

        approval = input("\nApprove? (y/n): ").strip().lower()

        if approval == "y":
            result.gates_passed[gate_name] = True
            result = graph.invoke(None, config)
        else:
            # Route back to appropriate phase
            if gate_name == "hypothesis_gate":
                result.current_phase = Phase.HYPOTHESIS
            elif gate_name == "experiment_gate":
                result.current_phase = Phase.CODING
            else:
                result.current_phase = Phase.REVIEW
            result = graph.invoke(result, config)

    print(f"\nPipeline complete. MLflow run: {result.mlflow_run_id}")
    return result


# Launch
final_state = run_supervised_ai_scientist(
    research_domain="efficient attention mechanisms for "
    "small-scale vision transformers",
    compute_budget=50.0,
)

9. Discovery Workbench Integration

Running the pipeline from the command line demonstrates the mechanics, but a production deployment needs to plug into the broader infrastructure where hypotheses, data, and evaluation tools already live.

The supervised AI scientist integrates into the Discovery Workbench (Chapter 6) as a high-level orchestration component. The Workbench provides three services that the AI scientist consumes:

• Knowledge layer. The hypothesis agent queries the Workbench's knowledge graph (Chapter 38) for gap analysis, and its retrieval-augmented generation (RAG) pipeline (Chapter 37) for literature context.
• Experiment layer. The experiment runner logs to the Workbench's experiment registry (Chapter 47), which provides provenance tracking, artifact storage, and reproducibility metadata.
• Evaluation layer. The review agent uses the Workbench's claim validation pipeline (Chapter 41) to check whether the experiment's conclusions are consistent with existing knowledge.

10. Current Limitations and Future Directions

The supervised AI scientist we built in this section has several known limitations:

1. ML-domain bias. The pipeline works best for ML research where experiments are cheap, fast, and deterministic. Extending to biology, chemistry, or physics requires domain-specific coding agents, different evaluation metrics, and (for wet-lab work) hardware integration as in Coscientist (the LLM-driven robotic chemistry platform introduced in Section 53.3).
2. Narrow novelty. The embedding-based novelty filter catches obvious duplicates but cannot assess whether an idea is interestingly novel versus superficially different. A hypothesis that changes one hyperparameter passes the novelty filter but does not advance science.
3. Reproducibility gaps. Despite multi-seed evaluation, the pipeline does not address all reproducibility concerns: hardware differences, library version drift, and dataset preprocessing variations can still cause results to differ across environments.
4. Evaluation circularity. The review agent uses an LLM to judge LLM-generated work. While the separation of generation and evaluation agents reduces bias, it does not eliminate it. Future work should incorporate domain-specific automated checks (unit tests for code, formal verification for mathematical claims) alongside LLM-based review.

Chapter 54 addresses the first limitation by scaling from single-agent to multi-agent discovery systems where specialized AI scientists collaborate across domains. The self-driving laboratories of Chapter 55 address the physical integration challenge. And the evaluation frameworks of Chapter 56 provide systematic methods for assessing discovery system quality beyond what automated review can offer.