41.1 Claim Extraction and Evidence Mapping

1. Anatomy of a Scientific Claim

A paper reports "94.7% accuracy on MMLU," and within hours three automated systems cite it as the new state of the art, yet not one of them checks whether the evaluation used the official test split or a contaminated subset. Not every sentence in a paper is a claim. "We used PyTorch 2.1" is a methodological statement. "Related work includes Smith et al. (2023)" is a citation. "Our model achieves 94.7% accuracy on MMLU" is a claim: it asserts a specific, verifiable result. Extracting that claim precisely is the first step toward catching the gap between what is reported and what is real.

We define a scientific claim as an assertion that (1) attributes a measurable or qualifiable outcome to a specific method, system, or intervention, and (2) could in principle be verified or refuted by re-executing an experiment or re-analyzing data. This definition excludes background statements, definitions, and methodological descriptions, while including numerical results, comparative statements ("outperforms baseline X by 3.2 points"), and qualitative findings ("the model fails to generalize to out-of-domain inputs").

In 2015, the Open Science Collaboration attempted to reproduce 100 psychology studies and found that fewer than half held up; many of the failures traced back to claims that were never checked against their original data or code. Systematic claim extraction exists to close exactly that gap before results propagate further.

Claim extraction identifies verifiable assertions in unstructured scientific text and converts each one into a typed, machine-readable object with explicit fields for the metric, the result, the dataset, and the provenance location. It matters because a single paper can contain dozens of claims scattered across the abstract, results tables, figure captions, and supplementary material. Without systematic extraction, reviewers and automated agents miss claims or conflate distinct results. The mechanism is a two-stage filter: a sentence classifier first separates claim-bearing sentences from background prose, then a structured parser decomposes each flagged sentence into subject, predicate, value, and context fields. Use claim extraction when you need to audit, compare, or reproduce results at scale; for a one-off reading of a single paper, manual annotation with a structured template works equally well. In short: a claim you cannot decompose into subject, predicate, value, and context is a claim you cannot verify.

A structured claim consists of several components. The subject identifies what is being evaluated (a model, a method, a treatment). The predicate specifies the measured property (accuracy, F1, reduction in tumor volume). The value gives the result, either as a number with units or as a qualitative assessment. The context identifies the conditions: the dataset, the split, the evaluation protocol, the baseline. And the provenance points to where in the paper the claim appears: section, sentence, table cell, or figure caption.

"""
Structured representation of a scientific claim.
Each claim is a typed object with components that map
directly to the evidence needed for verification.
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class ClaimType(Enum):
    NUMERICAL = "numerical"        # "achieves 94.7% accuracy"
    COMPARATIVE = "comparative"    # "outperforms X by 3.2 points"
    QUALITATIVE = "qualitative"    # "generalizes to unseen domains"
    STATISTICAL = "statistical"    # "significant at p < 0.01"
    CAUSAL = "causal"              # "treatment reduces tumor volume"


class EvidenceStrength(Enum):
    STRONG = "strong"              # Artifact exists, result verified
    MODERATE = "moderate"          # Artifact exists, not yet verified
    WEAK = "weak"                  # No artifact, citation only
    UNSUPPORTED = "unsupported"    # No evidence found


@dataclass
class ClaimValue:
    """The measured result of a claim."""
    magnitude: Optional[float] = None       # 94.7
    unit: Optional[str] = None              # "percent"
    qualifier: Optional[str] = None         # "approximately", "at least"
    confidence_interval: Optional[tuple] = None  # (93.1, 96.3)
    p_value: Optional[float] = None         # 0.01
    text: str = ""                          # Original text: "94.7%"


@dataclass
class Claim:
    """A single scientific claim extracted from text."""
    claim_id: str                  # Unique identifier
    claim_type: ClaimType
    subject: str                   # "our proposed model"
    predicate: str                 # "accuracy"
    value: ClaimValue
    context: dict = field(default_factory=dict)  # dataset, split, etc.
    provenance: dict = field(default_factory=dict)  # section, sentence_idx
    evidence_strength: EvidenceStrength = EvidenceStrength.UNSUPPORTED
    linked_artifacts: list = field(default_factory=list)
    source_sentence: str = ""      # Full original sentence

    def is_verifiable(self) -> bool:
        """A claim is verifiable if it has a numerical value or
        references a specific, reproducible experiment."""
        return (
            self.claim_type in (ClaimType.NUMERICAL, ClaimType.STATISTICAL)
            or (self.claim_type == ClaimType.COMPARATIVE
                and self.value.magnitude is not None)
        )

    def summary(self) -> str:
        """Human-readable one-line summary of the claim."""
        val = self.value.text or str(self.value.magnitude)
        ctx = self.context.get("dataset", "unspecified context")
        return f"{self.subject} | {self.predicate} = {val} | {ctx}"

The Claim dataclass in Listing 41.1 is the atomic unit of our validation pipeline. Every downstream operation (evidence mapping, artifact verification, reproducibility scoring) consumes and annotates Claim objects. The is_verifiable method implements our definition: only claims with measurable outcomes qualify for automated verification.

2. Extracting Claims from Scientific Text

Claim extraction operates at two levels: sentence classification (which sentences contain claims?) and claim parsing (what are the structured components of each claim?). We build a hybrid system that combines rule-based pattern matching for numerical claims with LLM-assisted extraction for complex or qualitative claims.

2.1 Rule-Based Numerical Claim Extraction

Numerical claims follow predictable patterns. A sentence like "Our model achieves 94.7% accuracy on the MMLU benchmark" contains a metric name ("accuracy"), a value ("94.7%"), and a context ("MMLU benchmark"). Regular expressions can capture these patterns with high precision, though at the cost of recall for unusual phrasings.

"""
Rule-based extraction of numerical claims from scientific text.
Patterns target common reporting formats: percentages, counts,
p-values, confidence intervals, and comparative deltas.
"""
import re
from typing import Generator


# Pattern components (composed into full claim patterns)
FLOAT = r"(\d+\.?\d*)"
PERCENT = FLOAT + r"\s*[%％]"
PVALUE = r"[pP]\s*[<>=≤≥]\s*" + FLOAT
CI_PATTERN = (
    r"\[?\(?" + FLOAT + r"\s*[,;–-]\s*" + FLOAT + r"\)?\]?"
)
METRIC_NAMES = (
    r"(?:accuracy|precision|recall|F1(?:\s*score)?|AUC(?:\s*-?\s*ROC)?"
    r"|BLEU|ROUGE(?:-[12L])?"
    r"|perplexity|loss|error\s*rate|IoU|mAP|R²|RMSE|MAE"
    r"|sensitivity|specificity|PPV|NPV)"
)
COMPARISON_VERBS = (
    r"(?:outperforms?|surpass(?:es)?|exceeds?|improves?\s+(?:upon|over)"
    r"|beats?|achieves?\s+higher|reduces?)"
)


class NumericalClaimExtractor:
    """Extract numerical claims using regex pattern matching."""

    def __init__(self):
        self.patterns = self._compile_patterns()

    def _compile_patterns(self) -> list[tuple[str, re.Pattern]]:
        """Build ordered list of (claim_type, pattern) pairs."""
        return [
            # "achieves 94.7% accuracy on MMLU"
            ("metric_report", re.compile(
                r"(?:achieves?|obtains?|reaches?|reports?|yields?)"
                r"\s+(?:an?\s+)?" + PERCENT
                + r"\s+" + METRIC_NAMES,
                re.IGNORECASE
            )),
            # "accuracy of 94.7% on MMLU"
            ("metric_of", re.compile(
                METRIC_NAMES + r"\s+of\s+" + PERCENT,
                re.IGNORECASE
            )),
            # "outperforms the baseline by 3.2 points"
            ("comparative", re.compile(
                COMPARISON_VERBS + r".*?by\s+" + FLOAT
                + r"\s*(?:points?|pp|percentage\s+points?|%)",
                re.IGNORECASE
            )),
            # "p < 0.01" or "p = 0.03"
            ("significance", re.compile(PVALUE, re.IGNORECASE)),
            # "95% confidence interval [92.1, 97.3]"
            ("confidence_interval", re.compile(
                r"(?:\d+%?\s+)?(?:confidence\s+interval|CI)\s*"
                + CI_PATTERN,
                re.IGNORECASE
            )),
        ]

    def extract(
        self, text: str, source_id: str = ""
    ) -> Generator[Claim, None, None]:
        """Extract claims from a block of scientific text.

        Args:
            text: The input text (abstract, section, or full paper).
            source_id: Identifier for provenance tracking.

        Yields:
            Claim objects for each detected numerical assertion.
        """
        sentences = self._split_sentences(text)
        claim_counter = 0

        for sent_idx, sentence in enumerate(sentences):
            for claim_type_name, pattern in self.patterns:
                for match in pattern.finditer(sentence):
                    claim_counter += 1
                    claim = self._match_to_claim(
                        match, claim_type_name, sentence,
                        sent_idx, source_id, claim_counter
                    )
                    if claim is not None:
                        yield claim

    def _match_to_claim(
        self, match: re.Match, claim_type_name: str,
        sentence: str, sent_idx: int, source_id: str, counter: int
    ) -> Claim | None:
        """Convert a regex match into a structured Claim object."""
        groups = match.groups()
        if not groups:
            return None

        # Parse the primary numerical value
        try:
            magnitude = float(groups[0])
        except (ValueError, IndexError):
            return None

        # Determine claim type
        type_map = {
            "metric_report": ClaimType.NUMERICAL,
            "metric_of": ClaimType.NUMERICAL,
            "comparative": ClaimType.COMPARATIVE,
            "significance": ClaimType.STATISTICAL,
            "confidence_interval": ClaimType.NUMERICAL,
        }

        value = ClaimValue(
            magnitude=magnitude,
            unit="percent" if "%" in match.group() else None,
            text=match.group().strip()
        )

        # Extract dataset context using simple heuristics
        context = self._extract_context(sentence)

        return Claim(
            claim_id=f"{source_id}_claim_{counter:04d}",
            claim_type=type_map.get(
                claim_type_name, ClaimType.NUMERICAL
            ),
            subject=self._extract_subject(sentence),
            predicate=self._extract_metric(sentence),
            value=value,
            context=context,
            provenance={
                "source_id": source_id,
                "sentence_index": sent_idx,
                "match_span": (match.start(), match.end()),
            },
            source_sentence=sentence.strip()
        )

    def _split_sentences(self, text: str) -> list[str]:
        """Split text into sentences, handling abbreviations."""
        # Simple sentence splitter; production systems use spaCy
        parts = re.split(r'(?<=[.!?])\s+(?=[A-Z])', text)
        return [s.strip() for s in parts if s.strip()]

    def _extract_subject(self, sentence: str) -> str:
        """Extract the subject (what is being evaluated)."""
        patterns = [
            r"(?:our|the\s+proposed)\s+([\w\s]+?)(?:\s+(?:achieves?|obtains?))",
            r"([\w\s]+?)\s+(?:outperforms?|surpass)",
        ]
        for pat in patterns:
            m = re.search(pat, sentence, re.IGNORECASE)
            if m:
                return m.group(1).strip()
        return "unspecified"

    def _extract_metric(self, sentence: str) -> str:
        """Extract the metric name from the sentence."""
        m = re.search(METRIC_NAMES, sentence, re.IGNORECASE)
        return m.group().lower() if m else "unspecified"

    def _extract_context(self, sentence: str) -> dict:
        """Extract dataset and evaluation context."""
        context = {}
        # Look for "on X" patterns for dataset names
        m = re.search(
            r"\bon\s+(?:the\s+)?([\w\-]+(?:\s+[\w\-]+)?)"
            r"\s+(?:benchmark|dataset|test\s+set|corpus)",
            sentence, re.IGNORECASE
        )
        if m:
            context["dataset"] = m.group(1).strip()
        return context


# --- Demonstration ---
extractor = NumericalClaimExtractor()
sample_abstract = (
    "We present DiscoveryNet, a foundation model for scientific claim "
    "verification. DiscoveryNet achieves 94.7% accuracy on the SciFact "
    "benchmark, outperforming the previous best by 3.2 percentage points. "
    "On the FEVER dataset, our model obtains an F1 score of 89.3%. "
    "All improvements are statistically significant (p < 0.001). "
    "The 95% confidence interval for SciFact accuracy is [93.1, 96.3]."
)

for claim in extractor.extract(sample_abstract, source_id="paper_001"):
    print(claim.summary())
    # paper_001 | accuracy = 94.7% | SciFact
    # paper_001 | unspecified = 3.2 percentage points | {}
    # paper_001 | f1 score = 89.3% | {}
    # paper_001 | unspecified = 0.001 | {}

The extractor in Listing 41.2 achieves high precision on standard reporting formats but misses claims that use non-standard phrasing. For example, "the error rate dropped to single digits" is a numerical claim that no regex will catch. For these cases, we turn to LLM-assisted extraction.

2.2 LLM-Assisted Claim Extraction

Large language models (LLMs) excel at understanding the semantics of scientific text, including implicit claims and complex phrasing. We use structured output generation to produce claim objects directly from text, with the LLM serving as a flexible parser that handles the long tail of formats our regex patterns miss.

"""
LLM-assisted claim extraction using structured output.
The LLM acts as a semantic parser, converting free-text
scientific claims into structured Claim objects.
"""
import json
from typing import Any


CLAIM_EXTRACTION_PROMPT = """\
You are a scientific claim extractor. Given a passage from a
scientific paper, identify all verifiable claims and return them
as a JSON array.

A verifiable claim must:
1. Assert a specific, measurable outcome (numerical or qualitative)
2. Attribute the outcome to a method, system, or intervention
3. Be falsifiable through experiment or data analysis

For each claim, extract:
- subject: what is being evaluated
- predicate: what property is measured
- value: the result (number, comparison, or qualitative finding)
- unit: the unit of measurement (if applicable)
- dataset: the evaluation dataset or context (if mentioned)
- claim_type: one of [numerical, comparative, qualitative,
  statistical, causal]

Do NOT include:
- Background statements or definitions
- Methodological descriptions without outcomes
- Hedged speculation ("may", "could potentially")

Return ONLY a JSON array. Example:
[{
  "subject": "DiscoveryNet",
  "predicate": "accuracy",
  "value": 94.7,
  "unit": "percent",
  "dataset": "SciFact",
  "claim_type": "numerical"
}]

Passage:
"""


def extract_claims_with_llm(
    text: str,
    llm_client: Any,
    model: str = "claude-sonnet-4-20250514"
) -> list[Claim]:
    """Extract claims using an LLM with structured output.

    Args:
        text: Scientific text to analyze.
        llm_client: An Anthropic or OpenAI client instance.
        model: Model identifier for the API call.

    Returns:
        List of Claim objects parsed from LLM output.
    """
    response = llm_client.messages.create(
        model=model,
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": CLAIM_EXTRACTION_PROMPT + text
        }]
    )

    raw_text = response.content[0].text
    # Strip markdown code fences if present
    if raw_text.startswith("```"):
        raw_text = raw_text.split("\n", 1)[1].rsplit("```", 1)[0]

    claims_data = json.loads(raw_text)
    claims = []

    for i, cd in enumerate(claims_data):
        claim = Claim(
            claim_id=f"llm_claim_{i:04d}",
            claim_type=ClaimType(
                cd.get("claim_type", "numerical")
            ),
            subject=cd.get("subject", "unspecified"),
            predicate=cd.get("predicate", "unspecified"),
            value=ClaimValue(
                magnitude=cd.get("value") if isinstance(
                    cd.get("value"), (int, float)
                ) else None,
                unit=cd.get("unit"),
                text=str(cd.get("value", "")),
            ),
            context={"dataset": cd.get("dataset", "")},
            source_sentence=text[:200],  # Truncated for reference
        )
        claims.append(claim)

    return claims

2.3 Merging and Deduplicating Claims

When both extractors run on the same text, they produce overlapping results. A merge step deduplicates claims by comparing their value, predicate, and context fields. Two claims match if they reference the same metric on the same dataset with values within a small tolerance (to handle rounding differences between abstract and table).

"""
Merge and deduplicate claims from multiple extractors.
Uses value proximity and context overlap to identify duplicates.
"""
from itertools import combinations


def claims_match(a: Claim, b: Claim, tol: float = 0.5) -> bool:
    """Determine if two claims refer to the same result.

    Two claims match if they share the same predicate (metric),
    similar context (dataset), and values within tolerance.
    """
    # Same metric?
    if a.predicate != b.predicate and a.predicate != "unspecified":
        if b.predicate != "unspecified":
            return False

    # Same dataset context?
    ds_a = a.context.get("dataset", "").lower()
    ds_b = b.context.get("dataset", "").lower()
    if ds_a and ds_b and ds_a != ds_b:
        return False

    # Values within tolerance?
    if (a.value.magnitude is not None
            and b.value.magnitude is not None):
        return abs(a.value.magnitude - b.value.magnitude) <= tol

    # If no numerical comparison possible, check text overlap
    return a.value.text.strip() == b.value.text.strip()


def merge_claims(
    regex_claims: list[Claim],
    llm_claims: list[Claim]
) -> list[Claim]:
    """Merge claims from regex and LLM extractors.

    Priority: regex claims are preferred (more precise provenance),
    LLM-only claims are added if they have no regex match.
    """
    merged = list(regex_claims)  # Start with regex claims
    used_llm = set()

    # Match LLM claims to regex claims
    for rc in regex_claims:
        for j, lc in enumerate(llm_claims):
            if j not in used_llm and claims_match(rc, lc):
                # Enrich regex claim with LLM-extracted fields
                if rc.subject == "unspecified":
                    rc.subject = lc.subject
                if not rc.context.get("dataset"):
                    rc.context.update(lc.context)
                used_llm.add(j)

    # Add unmatched LLM claims
    for j, lc in enumerate(llm_claims):
        if j not in used_llm:
            merged.append(lc)

    return merged

3. Evidence Mapping Through Scholarly APIs

Extracting and merging claims gives us a structured inventory of what a paper asserts; the natural next question is whether any external evidence actually backs those assertions up.

Once claims are extracted, the next step is finding the evidence that supports or refutes them. Evidence comes in layers: the paper itself, the papers it cites, the datasets it references, the code repositories it links, and the experiment logs it (hopefully) provides. Scholarly metadata APIs let us traverse these layers programmatically.

3.1 Crossref: DOI Resolution and Citation Metadata

Crossref is the authoritative registry for DOIs (Digital Object Identifiers). Every DOI resolves to a metadata record containing the title, authors, publication date, journal, and (crucially) a list of cited references. This reference list is the first layer of our evidence map: it tells us what prior work the authors claim to build upon.

"""
Evidence mapping through the Crossref REST API.
Retrieves bibliographic metadata, citation lists,
and cross-reference integrity checks.
"""
import requests
import time
from dataclasses import dataclass


@dataclass
class CrossrefRecord:
    """Bibliographic record from Crossref."""
    doi: str
    title: str
    authors: list[str]
    year: int
    journal: str
    reference_count: int
    is_referenced_by_count: int
    references: list[str]  # List of cited DOIs


class CrossrefClient:
    """Client for the Crossref REST API with polite pooling."""

    BASE_URL = "https://api.crossref.org/works"

    def __init__(self, email: str):
        """Initialize with a contact email for the polite pool.

        Crossref gives faster responses to requests that include
        a mailto parameter, placing them in the 'polite' pool.
        """
        self.session = requests.Session()
        self.session.params = {"mailto": email}
        self.session.headers.update({
            "User-Agent": f"DiscoveryAI/1.0 (mailto:{email})"
        })

    def get_work(self, doi: str) -> CrossrefRecord | None:
        """Fetch metadata for a single DOI.

        Args:
            doi: A DOI string (e.g., "10.1038/s41586-023-06221-2").

        Returns:
            CrossrefRecord or None if not found.
        """
        url = f"{self.BASE_URL}/{doi}"
        try:
            resp = self.session.get(url, timeout=30)
            resp.raise_for_status()
        except requests.RequestException:
            return None

        msg = resp.json().get("message", {})
        authors = [
            f"{a.get('given', '')} {a.get('family', '')}".strip()
            for a in msg.get("author", [])
        ]
        refs = [
            r.get("DOI", "")
            for r in msg.get("reference", [])
            if r.get("DOI")
        ]
        title_parts = msg.get("title", [""])
        year_parts = msg.get("published-print", {}).get(
            "date-parts", [[None]]
        )

        return CrossrefRecord(
            doi=doi,
            title=title_parts[0] if title_parts else "",
            authors=authors,
            year=year_parts[0][0] if year_parts[0][0] else 0,
            journal=msg.get("container-title", [""])[0],
            reference_count=msg.get("reference-count", 0),
            is_referenced_by_count=msg.get(
                "is-referenced-by-count", 0
            ),
            references=refs,
        )

    def verify_citations(
        self, paper_doi: str, claimed_refs: list[str]
    ) -> dict:
        """Check which claimed references actually appear
        in the Crossref metadata for a paper.

        Returns:
            Dictionary with verified, missing, and extra refs.
        """
        record = self.get_work(paper_doi)
        if record is None:
            return {"error": f"Could not fetch {paper_doi}"}

        actual_refs = set(record.references)
        claimed_set = set(claimed_refs)

        return {
            "verified": list(claimed_set & actual_refs),
            "missing_from_metadata": list(
                claimed_set - actual_refs
            ),
            "in_metadata_not_claimed": list(
                actual_refs - claimed_set
            ),
            "total_actual": len(actual_refs),
            "total_claimed": len(claimed_set),
        }

3.2 OpenAlex: Citation Graph Traversal

While Crossref provides the authoritative DOI registry, OpenAlex, an open catalog of scholarly metadata maintained by OurResearch, offers a richer knowledge graph of scholarly works, authors, institutions, and concepts. For evidence mapping, OpenAlex is particularly valuable because it provides bidirectional citation links (both "cites" and "cited by"), concept tagging, and open-access status indicators. As of 2024, OpenAlex has deprecated its concepts field in favor of a richer topics taxonomy; the code below uses concepts for clarity, but production pipelines should query topics and its nested subfield and field attributes instead.

"""
OpenAlex client for citation graph traversal and
author/institution-level evidence mapping.
"""


class OpenAlexClient:
    """Client for the OpenAlex API."""

    BASE_URL = "https://api.openalex.org"

    def __init__(self, email: str):
        self.session = requests.Session()
        self.session.params = {"mailto": email}

    def get_work_by_doi(self, doi: str) -> dict | None:
        """Fetch an OpenAlex work record by DOI."""
        url = f"{self.BASE_URL}/works/doi:{doi}"
        try:
            resp = self.session.get(url, timeout=30)
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException:
            return None

    def get_citing_works(
        self, openalex_id: str, limit: int = 50
    ) -> list[dict]:
        """Find works that cite the given work.

        Useful for checking whether a claim has been
        independently reproduced by other groups.
        """
        url = f"{self.BASE_URL}/works"
        params = {
            "filter": f"cites:{openalex_id}",
            "per_page": min(limit, 200),
            "sort": "cited_by_count:desc"
        }
        try:
            resp = self.session.get(url, params=params, timeout=30)
            resp.raise_for_status()
            return resp.json().get("results", [])
        except requests.RequestException:
            return []

    def build_evidence_map(self, doi: str) -> dict:
        """Build a multi-layer evidence map for a paper.

        Returns:
            A dictionary with citation context, citing works,
            author history, and independent reproduction signals.
        """
        work = self.get_work_by_doi(doi)
        if work is None:
            return {"error": f"Work not found: {doi}"}

        oa_id = work.get("id", "")

        # Layer 1: Direct citation context
        cited_works = work.get("referenced_works", [])

        # Layer 2: Who cites this paper?
        citing = self.get_citing_works(oa_id, limit=20)

        # Layer 3: Author track record
        authors = work.get("authorships", [])
        author_ids = [
            a["author"]["id"] for a in authors
            if a.get("author", {}).get("id")
        ]

        # Layer 4: Concept overlap with citing papers
        paper_concepts = {
            c["display_name"]
            for c in work.get("concepts", [])
            if c.get("score", 0) > 0.5
        }

        independent_reproductions = []
        for cw in citing:
            # A citing work is an "independent reproduction" if
            # it shares concepts but has no author overlap
            cw_authors = {
                a["author"]["id"]
                for a in cw.get("authorships", [])
                if a.get("author", {}).get("id")
            }
            cw_concepts = {
                c["display_name"]
                for c in cw.get("concepts", [])
                if c.get("score", 0) > 0.5
            }
            if (not cw_authors & set(author_ids)
                    and len(cw_concepts & paper_concepts) >= 2):
                independent_reproductions.append({
                    "id": cw.get("id"),
                    "title": cw.get("title"),
                    "year": cw.get("publication_year"),
                    "concept_overlap": list(
                        cw_concepts & paper_concepts
                    ),
                })

        return {
            "paper_id": oa_id,
            "title": work.get("title"),
            "cited_works_count": len(cited_works),
            "cited_by_count": work.get("cited_by_count", 0),
            "independent_reproductions": independent_reproductions,
            "author_count": len(authors),
            "open_access": work.get("open_access", {}).get(
                "is_oa", False
            ),
        }

4. Citation Integrity and Anomaly Detection

The scholarly APIs in the previous section let us retrieve metadata and traverse citation graphs, but they also reveal patterns that call the integrity of those citations into question.

Citation checking goes beyond verifying that references exist. Research documents systematic citation anomalies: phantom references (citations to nonexistent papers), citation cartels (where a group of authors systematically cite each other's work to inflate their metrics), self-citation inflation, and reference list manipulation. Automated anomaly detection of these patterns matters because corrupted citations undermine the evidence chain for every claim a paper makes.

"""
Citation anomaly detector.
Identifies phantom references, excessive self-citation,
citation cartel patterns, and temporal anomalies.
"""
from collections import Counter


@dataclass
class CitationAnomaly:
    """A detected anomaly in citation patterns."""
    anomaly_type: str      # phantom, self_citation, cartel, temporal
    severity: str          # low, medium, high
    description: str
    evidence: dict


class CitationAnomalyDetector:
    """Detect anomalous citation patterns in a paper."""

    def __init__(
        self,
        crossref: CrossrefClient,
        openalex: OpenAlexClient
    ):
        self.crossref = crossref
        self.openalex = openalex

    def detect_anomalies(
        self, doi: str
    ) -> list[CitationAnomaly]:
        """Run all anomaly checks on a paper.

        Args:
            doi: The paper's DOI.

        Returns:
            List of detected anomalies, sorted by severity.
        """
        anomalies = []
        record = self.crossref.get_work(doi)
        if record is None:
            return anomalies

        work = self.openalex.get_work_by_doi(doi)

        # Check 1: Phantom references
        anomalies.extend(
            self._check_phantom_refs(record)
        )

        # Check 2: Self-citation rate
        if work:
            anomalies.extend(
                self._check_self_citation(record, work)
            )

        # Check 3: Temporal anomalies
        anomalies.extend(
            self._check_temporal_anomalies(record)
        )

        # Sort by severity
        severity_order = {"high": 0, "medium": 1, "low": 2}
        anomalies.sort(
            key=lambda a: severity_order.get(a.severity, 3)
        )
        return anomalies

    def _check_phantom_refs(
        self, record: CrossrefRecord
    ) -> list[CitationAnomaly]:
        """Check for references that don't resolve to real papers."""
        anomalies = []
        unresolvable = []

        for ref_doi in record.references[:50]:  # Limit API calls
            ref_record = self.crossref.get_work(ref_doi)
            if ref_record is None:
                unresolvable.append(ref_doi)
            time.sleep(0.1)  # Rate limiting

        if unresolvable:
            ratio = len(unresolvable) / max(
                len(record.references), 1
            )
            severity = (
                "high" if ratio > 0.1
                else "medium" if ratio > 0.05
                else "low"
            )
            anomalies.append(CitationAnomaly(
                anomaly_type="phantom",
                severity=severity,
                description=(
                    f"{len(unresolvable)} of "
                    f"{len(record.references)} references "
                    f"could not be resolved via Crossref "
                    f"({ratio:.1%} unresolvable rate)"
                ),
                evidence={
                    "unresolvable_dois": unresolvable,
                    "total_refs": len(record.references),
                }
            ))
        return anomalies

    def _check_self_citation(
        self, record: CrossrefRecord, work: dict
    ) -> list[CitationAnomaly]:
        """Detect excessive self-citation rates."""
        anomalies = []
        paper_authors = set(record.authors)

        self_cited = 0
        checked = 0
        for ref_doi in record.references[:30]:
            ref_record = self.crossref.get_work(ref_doi)
            if ref_record is not None:
                checked += 1
                ref_authors = set(ref_record.authors)
                if paper_authors & ref_authors:
                    self_cited += 1
            time.sleep(0.1)

        if checked > 0:
            rate = self_cited / checked
            # Self-citation rates above 25% are unusual;
            # above 40% warrant investigation
            if rate > 0.25:
                severity = "high" if rate > 0.40 else "medium"
                anomalies.append(CitationAnomaly(
                    anomaly_type="self_citation",
                    severity=severity,
                    description=(
                        f"Self-citation rate of {rate:.1%} "
                        f"({self_cited}/{checked} checked refs). "
                        f"Typical range is roughly 5-15% in most fields."
                    ),
                    evidence={
                        "self_cited": self_cited,
                        "checked": checked,
                        "rate": rate,
                    }
                ))
        return anomalies

    def _check_temporal_anomalies(
        self, record: CrossrefRecord
    ) -> list[CitationAnomaly]:
        """Detect references to future papers or suspicious
        clustering of citation years."""
        anomalies = []
        ref_years = []

        for ref_doi in record.references[:30]:
            ref_record = self.crossref.get_work(ref_doi)
            if ref_record and ref_record.year:
                ref_years.append(ref_record.year)
                # Future citation check
                if ref_record.year > record.year + 1:
                    anomalies.append(CitationAnomaly(
                        anomaly_type="temporal",
                        severity="high",
                        description=(
                            f"Paper from {record.year} cites "
                            f"a work from {ref_record.year}: "
                            f"'{ref_record.title[:60]}...'"
                        ),
                        evidence={
                            "paper_year": record.year,
                            "ref_year": ref_record.year,
                            "ref_doi": ref_doi,
                        }
                    ))
            time.sleep(0.1)

        # Check for suspicious year clustering
        if ref_years:
            year_counts = Counter(ref_years)
            most_common_year, count = year_counts.most_common(1)[0]
            ratio = count / len(ref_years)
            if ratio > 0.5 and len(ref_years) > 5:
                anomalies.append(CitationAnomaly(
                    anomaly_type="temporal",
                    severity="medium",
                    description=(
                        f"{ratio:.0%} of references are from "
                        f"{most_common_year}. Unusual clustering "
                        f"may indicate narrow literature review."
                    ),
                    evidence={
                        "year_distribution": dict(year_counts),
                        "dominant_year": most_common_year,
                        "concentration": ratio,
                    }
                ))
        return anomalies

5. Building the Claim-to-Artifact Linkage Graph

Citation integrity checks tell us whether the bibliographic chain is sound, but a clean reference list alone cannot confirm that a reported result is reproducible; for that, we need to trace each claim to the actual computational artifacts behind it.

The final step in evidence mapping is connecting claims not just to other papers but to concrete computational artifacts: datasets, code repositories, model checkpoints, and experiment logs. This is where claim validation transitions from bibliometric analysis to computational verification.

A claim-to-artifact linkage graph is a directed graph where claim nodes connect to artifact nodes through typed edges: "measured_on" (claim to dataset), "produced_by" (claim to code), "logged_in" (claim to experiment tracker), and "derived_from" (claim to upstream claim). Building this graph requires extracting artifact references from the paper text (GitHub URLs, dataset identifiers, MLflow experiment IDs, where MLflow is an open-source platform for tracking machine learning experiments) and verifying that each artifact actually exists and is accessible. Figure 41.1.1 illustrates claim-to-artifact linkage graph.

"""
Claim-to-artifact linkage graph construction.
Connects extracted claims to their supporting computational
artifacts: datasets, code, model checkpoints, experiment logs.
"""
import re
from enum import Enum


class ArtifactType(Enum):
    DATASET = "dataset"
    CODE_REPO = "code_repository"
    MODEL_CHECKPOINT = "model_checkpoint"
    EXPERIMENT_LOG = "experiment_log"
    SUPPLEMENTARY = "supplementary_material"


@dataclass
class Artifact:
    """A computational artifact supporting a claim."""
    artifact_id: str
    artifact_type: ArtifactType
    uri: str                     # URL, DOI, or local path
    accessible: bool = False     # Verified accessible?
    integrity_hash: str = ""     # SHA-256 if downloadable
    metadata: dict = field(default_factory=dict)


@dataclass
class EvidenceLink:
    """A typed edge between a claim and an artifact."""
    claim_id: str
    artifact_id: str
    link_type: str     # measured_on, produced_by, logged_in
    confidence: float  # 0 to 1


class ArtifactLinker:
    """Extract artifact references and link them to claims."""

    # Patterns for common artifact references
    GITHUB_PATTERN = re.compile(
        r"https?://github\.com/[\w\-]+/[\w\-.]+"
    )
    HUGGINGFACE_PATTERN = re.compile(
        r"https?://huggingface\.co/[\w\-]+/[\w\-.]+"
    )
    ARXIV_PATTERN = re.compile(
        r"(?:arxiv[:\s]+)?(\d{4}\.\d{4,5})"
    )
    DATASET_NAMES = {
        "imagenet", "cifar-10", "cifar-100", "mnist",
        "glue", "superglue", "squad", "mmlu", "hellaswag",
        "humaneval", "mbpp", "scifact", "fever", "coco",
        "wmt", "penn treebank", "wikitext"
    }

    def extract_artifacts(self, text: str) -> list[Artifact]:
        """Extract artifact references from paper text."""
        artifacts = []

        # GitHub repositories
        for match in self.GITHUB_PATTERN.finditer(text):
            url = match.group()
            artifacts.append(Artifact(
                artifact_id=f"github_{len(artifacts)}",
                artifact_type=ArtifactType.CODE_REPO,
                uri=url,
            ))

        # Hugging Face models/datasets
        for match in self.HUGGINGFACE_PATTERN.finditer(text):
            url = match.group()
            # Heuristic: /datasets/ in URL means dataset
            atype = (
                ArtifactType.DATASET if "/datasets/" in url
                else ArtifactType.MODEL_CHECKPOINT
            )
            artifacts.append(Artifact(
                artifact_id=f"hf_{len(artifacts)}",
                artifact_type=atype,
                uri=url,
            ))

        # Named datasets mentioned in text
        text_lower = text.lower()
        for ds_name in self.DATASET_NAMES:
            if ds_name in text_lower:
                artifacts.append(Artifact(
                    artifact_id=f"dataset_{ds_name}",
                    artifact_type=ArtifactType.DATASET,
                    uri=f"named:{ds_name}",
                    metadata={"name": ds_name}
                ))

        return artifacts

    def verify_accessibility(
        self, artifact: Artifact
    ) -> Artifact:
        """Check whether an artifact URI is accessible."""
        if artifact.uri.startswith("http"):
            try:
                resp = requests.head(
                    artifact.uri, timeout=10,
                    allow_redirects=True
                )
                artifact.accessible = resp.status_code < 400
            except requests.RequestException:
                artifact.accessible = False
        elif artifact.uri.startswith("named:"):
            # Named datasets: assume accessible if well-known
            artifact.accessible = True
        return artifact

    def link_claims_to_artifacts(
        self,
        claims: list[Claim],
        artifacts: list[Artifact]
    ) -> list[EvidenceLink]:
        """Create evidence links between claims and artifacts.

        Uses heuristic matching: a claim links to a dataset
        artifact if the dataset name appears in the claim's
        context, and to a code artifact if the code repo
        is mentioned in the same section.
        """
        links = []

        for claim in claims:
            claim_dataset = claim.context.get(
                "dataset", ""
            ).lower()

            for artifact in artifacts:
                link = self._try_link(
                    claim, artifact, claim_dataset
                )
                if link is not None:
                    links.append(link)

        return links

    def _try_link(
        self, claim: Claim, artifact: Artifact,
        claim_dataset: str
    ) -> EvidenceLink | None:
        """Attempt to create a link between a claim and artifact."""

        # Dataset match: claim mentions the dataset by name
        if (artifact.artifact_type == ArtifactType.DATASET
                and claim_dataset):
            ds_name = artifact.metadata.get("name", "").lower()
            if ds_name and ds_name in claim_dataset:
                return EvidenceLink(
                    claim_id=claim.claim_id,
                    artifact_id=artifact.artifact_id,
                    link_type="measured_on",
                    confidence=0.9,
                )

        # Code repo: link all numerical claims to available code
        if (artifact.artifact_type == ArtifactType.CODE_REPO
                and claim.is_verifiable()
                and artifact.accessible):
            return EvidenceLink(
                claim_id=claim.claim_id,
                artifact_id=artifact.artifact_id,
                link_type="produced_by",
                confidence=0.6,
            )

        return None

6. Putting It Together: The Extraction Pipeline

The complete claim extraction and evidence mapping pipeline chains these components: sentence-level claim extraction (regex + LLM), claim merging and deduplication, scholarly metadata retrieval (Crossref + OpenAlex), citation anomaly detection, artifact extraction and linking, and evidence strength scoring. Figure 41.1 illustrates the six stages and the data flow between them.

The pipeline assigns each claim an evidence strength based on the layers of evidence available. A claim with a verified artifact, an accessible code repository, and independent reproductions in the citation graph receives "strong" evidence. A claim with only a citation to the authors' own prior work receives "weak" evidence. A claim with no supporting artifacts and phantom references receives "unsupported."

Mental Model

Think of evidence mapping like a home inspector verifying a real estate listing. The listing (the paper) claims "new roof, updated plumbing, no structural issues." The inspector does not take the seller's word; instead, she traces each claim to a concrete artifact: the roofing receipt from the contractor, the plumbing permit on file with the city, the structural engineer's signed report. If the receipt exists and the contractor confirms the work, evidence is strong. If the seller says "I have the receipt somewhere" but cannot produce it, evidence is weak. If the permit number on the listing does not match any city record, that is a phantom reference. The inspector's job is not to declare the house good or bad; it is to grade how verifiable each claim is, so the buyer knows where to dig deeper. Evidence mapping does exactly this for scientific claims: it checks whether the receipts (datasets, code, experiment logs) exist, are accessible, and match what the paper says.

Formally, we model evidence strength as a weighted sum of binary indicators:

where \(c\) is a claim and the weights \(w_i\) reflect the relative importance of each evidence layer. In practice, we use \(w_1 = 0.15\), \(w_2 = 0.20\), \(w_3 = 0.25\), \(w_4 = 0.30\), \(w_5 = 0.10\). Independent reproduction (\(w_4\)) carries twice the weight of mere artifact existence (\(w_1\)), because a result that another lab has replicated is far more trustworthy than one that simply ships a code repository nobody has re-run. A score above 0.7 maps to "strong," 0.4 to 0.7 maps to "moderate," and below 0.4 maps to "weak." A score at or near zero, where no artifact, code, or independent reproduction exists, corresponds to the "unsupported" level in the EvidenceStrength enum from Listing 41.1. These threshold values are illustrative defaults, informed by patterns in known-reproducible and known-irreproducible papers from the Open Science Collaboration replication studies (see the chapter bibliography); in practice, they should be tuned for the target domain and claim type.