10.3 Building a DSPy Literature Program

"First I chained four prompts together with string concatenation and duct tape. Then I learned about type systems. The duct tape did not survive."

A Vibe Coder Who Read the Specification

1. The Starting Point: A Brittle Prompt Chain

Imagine you paste a research question into a script you wrote last Tuesday. The script crashes because the model used bullet points instead of numbered lines. You patch the parser and re-run, but now the synthesis contradicts itself because two upstream calls silently returned empty strings. Most large language model (LLM)-powered research tools begin this way. They chain prompts together with Python string formatting, creating fragility that surfaces only at runtime. In short: If your LLM pipeline breaks because of formatting, you have a type system problem, not a prompting problem. Here is a typical example, representative of what a vibe coding session might produce:

import anthropic

client = anthropic.Anthropic()


def brittle_literature_review(question: str) -> str:
    """A brittle prompt chain for literature review.

    Problems: no type safety, no error handling, no optimization,
    prompt text mixed with logic, not testable.
    """
    # Step 1: Generate search queries
    resp1 = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"Generate 3 search queries for: {question}\n"
                       f"Return them as a numbered list."
        }],
    )
    queries_text = resp1.content[0].text

    # Step 2: Pretend we searched (in reality, parse and call an API)
    # The parsing here is fragile: what if the model uses bullets?
    queries = [
        line.strip().lstrip("0123456789.)")
        for line in queries_text.strip().split("\n")
        if line.strip()
    ]

    # Step 3: For each "result," extract claims
    all_claims = []
    for q in queries[:3]:
        resp2 = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"Given this search query: {q}\n"
                           f"List the key scientific claims that papers "
                           f"matching this query would typically make. "
                           f"Be specific and cite methodologies."
            }],
        )
        all_claims.append(resp2.content[0].text)

    # Step 4: Synthesize
    claims_blob = "\n---\n".join(all_claims)
    resp3 = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Research question: {question}\n\n"
                       f"Claims from literature:\n{claims_blob}\n\n"
                       f"Write a synthesis paragraph. Note contradictions "
                       f"and gaps."
        }],
    )
    return resp3.content[0].text

This code has at least seven problems:

1. Outputs are untyped strings that require ad-hoc parsing.
2. The parsing logic (splitting on newlines, stripping numbers) breaks if the model changes its formatting.
3. Prompt text is mixed with program logic, making both hard to change.
4. There is no quality metric, so there is no way to know if the output is good.
5. There is no error handling for malformed responses.
6. The program cannot be optimized; improving it requires manual prompt tweaking.
7. It is not testable: you cannot write a unit test that asserts properties of the synthesis without parsing free-form text.

Let us fix all seven problems by refactoring this into a typed DSPy program.

2. Step 1: Define Signatures

When a prompt chain silently returns an empty claim list, every downstream synthesis draws on missing evidence, and no type error or exception alerts you to the failure. Defining explicit signatures for each LLM task prevents this class of silent corruption entirely.

The first step is to identify the distinct LLM tasks in the chain and define a signature for each. Our literature review has three tasks: generate search queries, extract claims from an abstract, and synthesize claims into a finding.

What a signature does

A DSPy signature declares a single LLM task as a Python class. Its typed fields define the exact inputs the model receives and the structured outputs it must produce. Signatures replace hand-written prompt strings with machine-readable contracts. The DSPy compiler reads the field names, types, and descriptions, then generates and optimizes the actual prompt text automatically. At compile time, DSPy serializes the signature's input fields into a prompt template. It then parses the model's response back into the declared output types. You never write or debug prompt formatting code. Use signatures instead of raw prompt strings whenever a pipeline step can be described as "given these typed inputs, produce these typed outputs." Reserve raw prompts for one-off exploratory queries where declaring a class adds unnecessary overhead.

import dspy
from typing import Optional


class GenerateQueries(dspy.Signature):
    """Generate precise search queries for a scientific literature
    review. Each query should target a specific aspect of the
    research question and be suitable for Semantic Scholar or
    PubMed search."""

    research_question: str = dspy.InputField(
        desc="The research question to investigate"
    )
    num_queries: int = dspy.InputField(
        desc="Number of queries to generate", default=3
    )

    queries: list[str] = dspy.OutputField(
        desc="Search queries, each targeting a different aspect"
    )
    aspects: list[str] = dspy.OutputField(
        desc="The aspect of the question each query targets"
    )


class ExtractClaims(dspy.Signature):
    """Extract specific, falsifiable scientific claims from a
    paper abstract. Each claim should be a standalone statement
    that could be independently verified."""

    abstract: str = dspy.InputField(desc="Paper abstract text")
    title: str = dspy.InputField(desc="Paper title")
    research_question: str = dspy.InputField(
        desc="The guiding research question for relevance filtering"
    )

    claims: list[str] = dspy.OutputField(
        desc="Specific scientific claims from the abstract"
    )
    methodology: str = dspy.OutputField(
        desc="Primary methodology: experimental, computational, "
        "theoretical, review, or meta-analysis"
    )
    relevance: str = dspy.OutputField(
        desc="How this paper relates to the research question"
    )


class SynthesizeFindings(dspy.Signature):
    """Synthesize scientific claims from multiple papers into a
    coherent analysis. Identify agreements, contradictions,
    and gaps in the literature."""

    research_question: str = dspy.InputField()
    claims_by_paper: list[str] = dspy.InputField(
        desc="Claims grouped by source paper, each prefixed "
        "with the paper title"
    )
    num_papers: int = dspy.InputField(
        desc="Total number of papers reviewed"
    )

    synthesis: str = dspy.OutputField(
        desc="Coherent synthesis paragraph covering key findings"
    )
    agreements: list[str] = dspy.OutputField(
        desc="Points where multiple papers agree"
    )
    contradictions: list[str] = dspy.OutputField(
        desc="Points where papers disagree or conflict"
    )
    gaps: list[str] = dspy.OutputField(
        desc="Important questions not addressed by the reviewed papers"
    )
    confidence: str = dspy.OutputField(
        desc="Overall confidence in the synthesis: high, medium, or low"
    )

Notice how each signature maps to one step in the original chain, but with explicit types and descriptions instead of ad-hoc string formatting. The GenerateQueries signature outputs a list of strings, not a formatted text block that needs parsing. The ExtractClaims signature includes the research question as context, enabling relevance filtering. The SynthesizeFindings signature has structured outputs for agreements, contradictions, and gaps, making the results directly usable by downstream systems.

3. Step 2: Compose into a Module

With signatures defined, we compose them into a dspy.Module (the base class that groups multiple predictors and defines their execution order through a forward method) that implements the full pipeline. The module's forward method defines the program flow, using standard Python control structures. Figure 10.3 shows how the three stages connect: typed outputs from one signature feed directly into the typed inputs of the next, with a conventional search API call bridging the query generation and claim extraction stages.

import dspy
from pydantic import BaseModel, Field


class PaperResult(BaseModel):
    """A paper retrieved from the search API."""
    title: str
    abstract: str
    doi: str
    year: int
    authors: list[str]


class LiteratureReviewResult(BaseModel):
    """The complete output of a literature review."""
    question: str
    papers_reviewed: int
    synthesis: str
    agreements: list[str]
    contradictions: list[str]
    gaps: list[str]
    confidence: str
    claims_by_paper: dict[str, list[str]]


class LiteratureReviewProgram(dspy.Module):
    """A three-stage literature review pipeline.

    Stage 1: Generate targeted search queries.
    Stage 2: Extract claims from retrieved papers.
    Stage 3: Synthesize findings across all papers.
    """

    def __init__(self):
        # ChainOfThought wraps a signature so the model reasons
        # step by step before producing typed outputs.
        self.generate_queries = dspy.ChainOfThought(GenerateQueries)
        self.extract_claims = dspy.ChainOfThought(ExtractClaims)
        self.synthesize = dspy.ChainOfThought(SynthesizeFindings)

    def forward(self, research_question: str,
                search_fn=None, max_papers: int = 10):
        """Execute the literature review pipeline.

        Args:
            research_question: The question to investigate.
            search_fn: Callable that takes a query string and returns
                a list of PaperResult objects. If None, uses a mock.
            max_papers: Maximum papers to process.
        """
        # Stage 1: Generate search queries
        query_result = self.generate_queries(
            research_question=research_question,
            num_queries=3
        )

        # Stage 2: Retrieve and process papers
        if search_fn is None:
            search_fn = _mock_search  # for testing

        all_papers = []
        seen_dois = set()
        for query in query_result.queries:
            papers = search_fn(query)
            for paper in papers:
                if paper.doi not in seen_dois:
                    seen_dois.add(paper.doi)
                    all_papers.append(paper)

        # Limit to max_papers
        all_papers = all_papers[:max_papers]

        # Extract claims from each paper
        claims_by_paper = {}
        for paper in all_papers:
            extraction = self.extract_claims(
                abstract=paper.abstract,
                title=paper.title,
                research_question=research_question,
            )
            claims_by_paper[paper.title] = extraction.claims

        # Format claims for synthesis
        formatted_claims = [
            f"[{title}]: {'; '.join(claims)}"
            for title, claims in claims_by_paper.items()
        ]

        # Stage 3: Synthesize all findings
        synthesis = self.synthesize(
            research_question=research_question,
            claims_by_paper=formatted_claims,
            num_papers=len(all_papers),
        )

        # Return a dspy.Prediction (a lightweight container whose
        # attributes are accessible by name, similar to a named tuple)
        return dspy.Prediction(
            synthesis=synthesis.synthesis,
            agreements=synthesis.agreements,
            contradictions=synthesis.contradictions,
            gaps=synthesis.gaps,
            confidence=synthesis.confidence,
            claims_by_paper=claims_by_paper,
            papers_reviewed=len(all_papers),
        )


def _mock_search(query: str) -> list[PaperResult]:
    """Mock search for testing. Replace with Semantic Scholar API."""
    return [
        PaperResult(
            title=f"Paper on {query[:40]}",
            abstract=f"We investigate {query}. Our results show...",
            doi=f"10.1234/mock.{hash(query) % 10000}",
            year=2024,
            authors=["A. Researcher"],
        )
    ]

4. Step 3: Write the Quality Metric

A well-structured module with typed signatures and injected dependencies gives us a testable pipeline, but testing requires a definition of success.

The metric defines what "good" means. For a literature review, quality has multiple dimensions that we must capture in a single score.

def literature_review_metric(example, prediction, trace=None):
    """Multi-dimensional quality metric for literature reviews.

    Dimensions:
    - Claim specificity: are claims concrete and falsifiable?
    - Coverage: does the synthesis address the question?
    - Structure: are agreements, contradictions, gaps populated?
    - Grounding: do claims come from the actual paper abstracts?
    """
    scores = {}

    # 1. Claim specificity (heuristic: specific claims are longer)
    all_claims = []
    if hasattr(prediction, 'claims_by_paper'):
        for paper_claims in prediction.claims_by_paper.values():
            all_claims.extend(paper_claims)

    if all_claims:
        # Specific claims tend to have 10+ words and contain numbers
        specific_count = sum(
            1 for c in all_claims
            if len(c.split()) >= 10
        )
        scores['specificity'] = specific_count / len(all_claims)
    else:
        scores['specificity'] = 0.0

    # 2. Coverage: synthesis should be substantive
    synthesis = getattr(prediction, 'synthesis', '')
    if synthesis and len(synthesis.split()) >= 80:
        scores['coverage'] = 1.0
    elif synthesis and len(synthesis.split()) >= 40:
        scores['coverage'] = 0.5
    else:
        scores['coverage'] = 0.0

    # 3. Structure: all output fields should be populated
    has_agreements = bool(getattr(prediction, 'agreements', []))
    has_gaps = bool(getattr(prediction, 'gaps', []))
    has_confidence = getattr(prediction, 'confidence', '') in {
        'high', 'medium', 'low'
    }
    scores['structure'] = (
        (0.4 * has_agreements) +
        (0.3 * has_gaps) +
        (0.3 * has_confidence)
    )

    # 4. Grounding: check against expected claims if available
    if hasattr(example, 'expected_claims') and example.expected_claims:
        expected = set(example.expected_claims)
        found = set(all_claims)
        # Soft match: check if expected claim keywords appear
        grounded = 0
        for exp in expected:
            exp_keywords = set(exp.lower().split())
            for claim in found:
                claim_keywords = set(claim.lower().split())
                overlap = len(exp_keywords & claim_keywords)
                if overlap >= 3:  # at least 3 keyword matches
                    grounded += 1
                    break
        scores['grounding'] = grounded / len(expected)
    else:
        scores['grounding'] = 0.5  # no ground truth available

    # Weighted combination
    weights = {
        'specificity': 0.25,
        'coverage': 0.25,
        'structure': 0.25,
        'grounding': 0.25,
    }
    return sum(scores[k] * weights[k] for k in scores)

5. Step 4: Compile and Evaluate

With the program and metric defined, compilation (where the optimizer runs multiple trials to discover the best prompt wording and few-shot demonstrations for each signature) requires three decisions: which optimizer to use, how much training data to collect, and how to split that data.

import dspy

# Configure the LM
lm = dspy.LM("anthropic/claude-sonnet-4-20250514")
dspy.configure(lm=lm)

# Prepare training data
trainset = [
    dspy.Example(
        research_question=(
            "How do graph neural networks perform for "
            "molecular property prediction?"
        ),
        expected_claims=[
            "Graph neural networks (GNNs) outperform traditional fingerprint methods",
            "Message-passing networks capture local structure",
            "3D-aware models improve binding affinity prediction",
        ],
    ).with_inputs("research_question"),  # marks which fields are inputs;
    # remaining fields (expected_claims) become the target outputs for the optimizer
    dspy.Example(
        research_question=(
            "What are the failure modes of large language models "
            "in mathematical reasoning?"
        ),
        expected_claims=[
            "LLMs struggle with multi-step arithmetic",
            "Chain-of-thought improves but does not eliminate errors",
            "Models show sensitivity to problem framing",
        ],
    ).with_inputs("research_question"),
    # ... 15-20 more examples for robust optimization
]

# Split into train and validation
train, val = trainset[:15], trainset[15:]

# Compile with MIPROv2
program = LiteratureReviewProgram()

# MIPROv2 is DSPy's multi-prompt instruction proposal optimizer:
# it searches over candidate instructions AND bootstrapped demos.
optimizer = dspy.MIPROv2(
    metric=literature_review_metric,
    num_candidates=7,
    max_bootstrapped_demos=3,
    max_labeled_demos=3,
    num_trials=20,
)

compiled_program = optimizer.compile(
    program,
    trainset=train,
    valset=val,
    requires_permission_to_run=False,
)

# Evaluate on a held-out test set
# dspy.Evaluate runs the program on every example in devset,
# scores each with the metric, and returns the mean score.
evaluator = dspy.Evaluate(
    devset=test_set,
    metric=literature_review_metric,
    num_threads=4,
    display_progress=True,
)

# Compare uncompiled vs compiled
baseline_score = evaluator(program)
compiled_score = evaluator(compiled_program)
print(f"Baseline: {baseline_score:.3f}")
print(f"Compiled: {compiled_score:.3f}")
# Illustrative range (results vary by task and training data): 0.45 -> 0.72

6. Step 5: Integrate with the Discovery Workbench

The compiled program is a Python object that can be saved, loaded, and called like any function. Integrating it with the Discovery Workbench means wrapping it in the Workbench's typed artifact system and provenance graph.

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional
import json


class WorkbenchLitReviewRequest(BaseModel):
    """Request schema for the Workbench literature review endpoint."""
    research_question: str = Field(
        description="The question to investigate"
    )
    max_papers: int = Field(default=20, ge=5, le=100)
    search_sources: list[str] = Field(
        default=["semantic_scholar"],
        description="APIs to search: semantic_scholar, pubmed, openalex"
    )


class WorkbenchLitReviewResult(BaseModel):
    """Result schema registered as a Workbench artifact."""
    request: WorkbenchLitReviewRequest
    synthesis: str
    agreements: list[str]
    contradictions: list[str]
    gaps: list[str]
    confidence: str
    papers_reviewed: int
    claims_by_paper: dict[str, list[str]]
    quality_score: float
    model_id: str
    compiled_program_version: str
    timestamp: datetime


class LitReviewService:
    """Workbench service that wraps the compiled DSPy program.

    Handles: loading the compiled program, connecting to search APIs,
    logging provenance, and returning typed results.
    """

    def __init__(self, program_path: str, model_id: str):
        import dspy

        self.model_id = model_id
        self.program_path = program_path

        # Load the compiled program
        lm = dspy.LM(f"anthropic/{model_id}")
        dspy.configure(lm=lm)

        self.program = LiteratureReviewProgram()
        self.program.load(program_path)

    def search_semantic_scholar(self, query: str) -> list[PaperResult]:
        """Search Semantic Scholar API.

        In production, this calls the real API with rate limiting,
        caching, and error handling.
        """
        import requests

        url = "https://api.semanticscholar.org/graph/v1/paper/search"
        params = {
            "query": query,
            "limit": 10,
            "fields": "title,abstract,doi,year,authors",
        }
        resp = requests.get(url, params=params, timeout=10)
        resp.raise_for_status()
        data = resp.json()

        results = []
        for paper in data.get("data", []):
            if paper.get("abstract") and paper.get("doi"):
                results.append(PaperResult(
                    title=paper["title"],
                    abstract=paper["abstract"],
                    doi=paper["doi"],
                    year=paper.get("year", 0),
                    authors=[
                        a["name"] for a in paper.get("authors", [])
                    ],
                ))
        return results

    def run(self, request: WorkbenchLitReviewRequest
            ) -> WorkbenchLitReviewResult:
        """Execute a literature review and return a typed result."""

        # Run the compiled DSPy program
        prediction = self.program(
            research_question=request.research_question,
            search_fn=self.search_semantic_scholar,
            max_papers=request.max_papers,
        )

        # Compute quality score using the same metric
        # (no ground truth, so we use heuristic components only)
        quality = _compute_heuristic_quality(prediction)

        return WorkbenchLitReviewResult(
            request=request,
            synthesis=prediction.synthesis,
            agreements=prediction.agreements,
            contradictions=prediction.contradictions,
            gaps=prediction.gaps,
            confidence=prediction.confidence,
            papers_reviewed=prediction.papers_reviewed,
            claims_by_paper=prediction.claims_by_paper,
            quality_score=quality,
            model_id=self.model_id,
            compiled_program_version=self.program_path,
            timestamp=datetime.now(),
        )


def _compute_heuristic_quality(prediction) -> float:
    """Heuristic quality score without ground truth."""
    scores = []
    # Synthesis length
    synth_words = len(prediction.synthesis.split())
    scores.append(min(synth_words / 100, 1.0))
    # Structure completeness
    scores.append(1.0 if prediction.agreements else 0.0)
    scores.append(1.0 if prediction.gaps else 0.0)
    scores.append(1.0 if prediction.confidence in {
        'high', 'medium', 'low'
    } else 0.0)
    return sum(scores) / len(scores)

7. Step 6: Deploy and Iterate

The deployment pattern for compiled DSPy programs has three components: the saved program state (a JSON file), the model configuration, and the quality monitoring loop.

import dspy

# Save the compiled program
compiled_program.save("lit_review_v1_claude_sonnet.json")

# --- In production ---

# Load and configure
lm = dspy.LM("anthropic/claude-sonnet-4-20250514")
dspy.configure(lm=lm)

production_program = LiteratureReviewProgram()
production_program.load("lit_review_v1_claude_sonnet.json")

# Run with real search
result = production_program(
    research_question="What are the latest advances in protein "
    "structure prediction beyond AlphaFold?",
    search_fn=search_semantic_scholar,
    max_papers=15,
)

# --- Model migration ---
# When switching models, recompile (do not reuse old prompts)
new_lm = dspy.LM("anthropic/claude-opus-4-20250514")
dspy.configure(lm=new_lm)

recompiled = optimizer.compile(
    LiteratureReviewProgram(),
    trainset=train,
    valset=val,
)
recompiled.save("lit_review_v2_claude_opus.json")

# Verify quality did not regress
new_score = evaluator(recompiled)
assert new_score >= compiled_score * 0.95, (
    f"Quality regression: {new_score:.3f} < {compiled_score * 0.95:.3f}"
)

8. The Refactoring Pattern

Converting a brittle prompt chain to a compiled DSPy program follows a repeatable pattern. This checklist captures each step:

1. Identify the stages: each distinct LLM call becomes a signature.
2. Type the interfaces: replace string inputs/outputs with typed fields (lists, enums, nested objects).
3. Extract the logic: move control flow (loops, conditionals, error handling) into the module's forward method.
4. Inject dependencies: external calls (APIs, databases, file reads) become parameters, not hardcoded calls.
5. Define the metric: decide what "correct" means, then code it as a function from (example, prediction) to [0, 1].
6. Collect training data: 15 to 25 input/output examples typically suffice for many tasks.
7. Compile: run an optimizer to generate optimal prompts and demonstrations.
8. Evaluate: compare uncompiled vs. compiled on a held-out test set.
9. Deploy: save the compiled state, load in production, monitor quality.

This pattern connects directly to the software engineering practices from Chapter 8 (testing, type safety, dependency injection) and scales to the multi-agent systems in Chapter 17, where each agent is itself a compiled DSPy module with a typed interface. Figure 10.3.1 illustrates DSPy compilation pipeline from brittle prompt chain to compiled program.

Exercises

1. Conceptual: The refactoring pattern in this section transforms a brittle prompt chain into a typed, compiled program. Identify two additional benefits that this transformation provides for scientific reproducibility beyond the seven problems listed in the "Starting Point" section. How does the compiled program's JSON state file contribute to the provenance requirements of the Discovery Workbench?
2. Coding: Extend the LiteratureReviewProgram with a fourth stage: GenerateHypotheses. This stage takes the synthesis, gaps, and contradictions from Stage 3 and proposes three testable hypotheses that could address the identified gaps. Define the signature, add it to the module, extend the quality metric to score hypothesis quality (specificity, testability, novelty), recompile, and compare the four-stage program against the three-stage version.
3. Analysis: Run the compiled literature review program on five different research questions from different scientific domains (e.g., protein folding, battery materials, climate modeling, drug resistance, social network dynamics). For each, measure the quality score and inspect the generated prompts. Does the compiled program generalize across domains, or does it overfit to the training distribution? What modifications to the training set or metric would improve cross-domain performance?