Transformers

All extractors inherit from ExtractorInterface[T, R]. Each implements a forward(input) method (synchronous) and gets an aforward(input) method (async) for free.

Base interface

ExtractorInterface

Bases: Module, Generic[T, R]

Generic interface for an extractor that takes an input of type T and returns an output of type R.

Material extraction

DspyTextExtractor(signature, lm, retry_temperatures=None)

Bases: MaterialExtractorInterface

A text extractor that uses dspy to extract any arbitrary text from the publication text.

Implements temperature escalation retry on failures to improve robustness against transient validation errors.

Initialize the extractor with a dspy signature and language model.

Parameters:

Name	Type	Description	Default
signature	Signature	The dspy signature specifying input/output fields.	required
lm	LM	The language model to use for prediction.	required
retry_temperatures	list[float] | None	Temperatures to try on failures. Defaults to [0.0, 0.3, 0.5].	None

Source code in src/llm_synthesis/transformers/material_extraction/dspy_extraction.py

def __init__(
    self,
    signature: type[dspy.Signature],
    lm: dspy.LM,
    retry_temperatures: list[float] | None = None,
):
    """
    Initialize the extractor with a dspy signature and language model.

    Args:
        signature (dspy.Signature): The dspy signature specifying
                                    input/output fields.
        lm (dspy.LM): The language model to use for prediction.
        retry_temperatures: Temperatures to try on failures.
                             Defaults to [0.0, 0.3, 0.5].
    """
    self._validate_signature(signature)
    self.signature = signature
    self.retry_temperatures = retry_temperatures or [0.0, 0.3, 0.5]

    # Strip any pre-existing response_format / extra_body so retries
    # have full control over LM kwargs.
    _clean_keys = {"response_format", "extra_body"}
    if _clean_keys & set(lm.kwargs):
        self.lm = copy.copy(lm)
        self.lm.kwargs = {
            k: v for k, v in lm.kwargs.items() if k not in _clean_keys
        }
    else:
        self.lm = lm

Methods:

forward(input)

Extract text from the given str using the language model and signature.

Retries at escalating temperatures on failure.

Parameters:

Name	Type	Description	Default
input	str	The str from which to extract text.	required

Returns:

Name	Type	Description
str	str	The extracted text from the str.

Source code in src/llm_synthesis/transformers/material_extraction/dspy_extraction.py

def forward(self, input: str) -> str:
    """
    Extract text from the given str using the language model and signature.

    Retries at escalating temperatures on failure.

    Args:
        input (str): The str from which to extract text.

    Returns:
        str: The extracted text from the str.
    """
    predict_kwargs = {"publication_text": input}
    output_field = next(iter(self.signature.output_fields.keys()))
    last_exc: Exception | None = None

    for t_idx, temp in enumerate(self.retry_temperatures):
        lm = self._lm_with_overrides({"temperature": temp})
        try:
            with dspy.settings.context(
                lm=lm,
                adapter=dspy.adapters.JSONAdapter(),
            ):
                return dspy.ChainOfThought(self.signature)(
                    **predict_kwargs
                ).__getattr__(output_field)
        except Exception as e:
            last_exc = e
            if t_idx < len(self.retry_temperatures) - 1:
                log.warning(
                    "Material extractor: failure at temp=%.1f: %r"
                    " — retrying at temp=%.1f",
                    temp,
                    e,
                    self.retry_temperatures[t_idx + 1],
                )
            else:
                log.warning(
                    "Material extractor: all temperatures exhausted: %r",
                    e,
                )
        finally:
            # lm is a throwaway copy.copy(self.lm) (see
            # _lm_with_overrides) -- cost accumulates on the copy, not
            # self.lm, so propagate it back or callers reading
            # self.lm.get_cost() always see 0.
            copy_cost = getattr(lm, "_cumulative_cost_usd", None)
            if copy_cost and hasattr(self.lm, "_cumulative_cost_usd"):
                self.lm._cumulative_cost_usd += copy_cost

    if last_exc is None:
        # Defensive: the loop body always sets last_exc on failure and
        # returns on success, so this branch is unreachable in practice.
        raise RuntimeError(
            "material extraction failed without recording an exception"
        )
    raise last_exc

make_dspy_text_extractor_signature(signature_name='DspyTextExtractorSignature', instructions='Extract the synthesis paragraph from the publication text.', input_description='The publication text to extract the synthesis paragraph from.', output_name='synthesis_paragraph', output_description='The extracted synthesis paragraph.')

Create a dspy signature for extracting text from publication text.

Parameters:

Name	Type	Description	Default
signature_name	str	Name of the signature.	'DspyTextExtractorSignature'
instructions	str	Instructions for the signature.	'Extract the synthesis paragraph from the publication text.'
input_description	str	Description for the publication text input.	'The publication text to extract the synthesis paragraph from.'
output_name	str	Name of the output field.	'synthesis_paragraph'
output_description	str	Description for the output field.	'The extracted synthesis paragraph.'

Returns:

Type	Description
type[Signature]	dspy.Signature: The constructed dspy signature for text extraction.

Source code in src/llm_synthesis/transformers/material_extraction/dspy_extraction.py

def make_dspy_text_extractor_signature(
    signature_name: str = "DspyTextExtractorSignature",
    instructions: str = "Extract the synthesis paragraph from the publication"
    " text.",
    input_description: str = "The publication text to extract the synthesis"
    " paragraph from.",
    output_name: str = "synthesis_paragraph",
    output_description: str = "The extracted synthesis paragraph.",
) -> type[dspy.Signature]:
    """
    Create a dspy signature for extracting text from publication text.

    Args:
        signature_name (str): Name of the signature.
        instructions (str): Instructions for the signature.
        input_description (str): Description for the publication text input.
        output_name (str): Name of the output field.
        output_description (str): Description for the output field.

    Returns:
        dspy.Signature: The constructed dspy signature for text extraction.
    """
    signature = {
        "publication_text": (
            str,
            dspy.InputField(description=input_description),
        ),
        output_name: (str, dspy.OutputField(description=output_description)),
    }
    return dspy.make_signature(
        signature_name=signature_name,
        instructions=instructions,
        signature=signature,
    )

Synthesis extraction

DspySynthesisExtractor(signature, lm, retry_temperatures=None)

Bases: SynthesisExtractorInterface

Extractor that uses dspy to extract a structured synthesis ontology for a specific material from the entire paper text.

Implements temperature escalation retry and bare-JSON recovery on failures to improve robustness without changing the happy-path behavior.

Initialize the extractor with a dspy signature and language model.

Parameters:

Name	Type	Description	Default
signature	Signature	The dspy signature specifying input/output fields.	required
lm	LM	The language model to use for prediction.	required
retry_temperatures	list[float] | None	Temperatures to try on failures. Defaults to [0.0, 0.3, 0.5].	None

Source code in src/llm_synthesis/transformers/synthesis_extraction/dspy_synthesis_extraction.py

def __init__(
    self,
    signature: type[dspy.Signature],
    lm: dspy.LM,
    retry_temperatures: list[float] | None = None,
):
    """
    Initialize the extractor with a dspy signature and language model.

    Args:
        signature (dspy.Signature): The dspy signature specifying
                                    input/output fields.
        lm (dspy.LM): The language model to use for prediction.
        retry_temperatures: Temperatures to try on failures.
                             Defaults to [0.0, 0.3, 0.5].
    """
    self._validate_signature(signature)
    self.signature = signature
    self.retry_temperatures = retry_temperatures or [0.0, 0.3, 0.5]

    # Strip any pre-existing response_format / extra_body so retries
    # have full control over LM kwargs.
    _clean_keys = {"response_format", "extra_body"}
    if _clean_keys & set(lm.kwargs):
        self.lm = copy.copy(lm)
        self.lm.kwargs = {
            k: v for k, v in lm.kwargs.items() if k not in _clean_keys
        }
    else:
        self.lm = lm

Methods:

forward(input)

Extract a structured synthesis ontology for a specific material from the given paper text.

Retries at escalating temperatures on failure. Attempts bare-JSON recovery before escalating. Falls back to a minimal ontology if all attempts are exhausted.

Parameters:

Name	Type	Description	Default
input	tuple[str, str]	Tuple of (paper_text, material_name).	required

Returns:

Name	Type	Description
GeneralSynthesisOntology	GeneralSynthesisOntology	The structured synthesis ontology for the specific material.

Source code in src/llm_synthesis/transformers/synthesis_extraction/dspy_synthesis_extraction.py

def forward(self, input: tuple[str, str]) -> GeneralSynthesisOntology:
    """
    Extract a structured synthesis ontology for a specific material
    from the given paper text.

    Retries at escalating temperatures on failure. Attempts bare-JSON
    recovery before escalating. Falls back to a minimal ontology if all
    attempts are exhausted.

    Args:
        input (tuple[str, str]): Tuple of (paper_text, material_name).

    Returns:
        GeneralSynthesisOntology: The structured synthesis ontology
                                  for the specific material.
    """
    paper_text, material_name = input
    predict_kwargs = {
        "paper_text": paper_text,
        "material_name": material_name,
    }

    last_exc: Exception | None = None

    for t_idx, temp in enumerate(self.retry_temperatures):
        lm = self._lm_with_overrides({"temperature": temp})
        try:
            with dspy.settings.context(
                lm=lm, adapter=SynthesisJSONAdapter()
            ):
                result = dspy.Predict(self.signature)(**predict_kwargs)
                synthesis_data = result.__getattr__(
                    next(iter(self.signature.output_fields.keys()))
                )

                # Ensure required fields are present
                if (
                    not hasattr(synthesis_data, "target_compound_type")
                    or synthesis_data.target_compound_type is None
                ):
                    synthesis_data.target_compound_type = "other"
                if (
                    not hasattr(synthesis_data, "synthesis_method")
                    or synthesis_data.synthesis_method is None
                ):
                    synthesis_data.synthesis_method = "other"

                return synthesis_data

        except Exception as e:
            # Attempt bare-JSON recovery before escalating temperature.
            recovered = self._try_recover_bare_json(e, material_name)
            if recovered is not None:
                logging.info(
                    "Synthesis extractor: recovered bare JSON for %s"
                    " (temp=%.1f)",
                    material_name,
                    temp,
                )
                return recovered

            last_exc = e
            if t_idx < len(self.retry_temperatures) - 1:
                logging.warning(
                    "Synthesis extractor: failure at temp=%.1f for %s:"
                    " %r — retrying at temp=%.1f",
                    temp,
                    material_name,
                    e,
                    self.retry_temperatures[t_idx + 1],
                )
            else:
                logging.warning(
                    "Synthesis extractor: all temperatures exhausted"
                    " for %s: %r",
                    material_name,
                    e,
                )

    # Try to parse raw response as JSON and extract structured_synthesis
    try:
        # Get the raw response from the LM
        raw_response = (
            self.lm.history[-1]["response"]
            if hasattr(self.lm, "history") and self.lm.history
            else None
        )
        if raw_response:
            # Try to parse as JSON
            parsed = json.loads(raw_response)
            if "structured_synthesis" in parsed:
                synthesis_data = parsed["structured_synthesis"]
                # Ensure required fields are present
                if "target_compound_type" not in synthesis_data:
                    synthesis_data["target_compound_type"] = "other"
                if (
                    "synthesis_method" not in synthesis_data
                    or synthesis_data["synthesis_method"] is None
                ):
                    synthesis_data["synthesis_method"] = "other"
                return GeneralSynthesisOntology(**synthesis_data)
    except Exception as json_error:
        logging.debug(f"Failed to parse JSON response: {json_error}")

    # Fallback: create a minimal synthesis ontology if extraction fails
    logging.warning(
        f"Failed to extract synthesis for {material_name}: {last_exc}"
    )
    return GeneralSynthesisOntology(
        target_compound=material_name,
        target_compound_type="other",
        synthesis_method="other",
        starting_materials=[],
        steps=[],
        equipment=[],
        notes=f"Extraction failed: {last_exc!s}",
    )

make_dspy_synthesis_extractor_signature(signature_name='DspySynthesisExtractorSignature', instructions='Extract structured synthesis for a specific material from the paper. Output only a valid JSON with the structured_synthesis field.', paper_text_description='Complete paper text to search for the material synthesis procedure.', material_name_description='The name of the specific material to extract synthesis for.', output_name='structured_synthesis', output_description='The extracted structured synthesis for specific material as a JSON.')

Create signature for extracting a materials-specific synthesis ontology.

Parameters:

Name	Type	Description	Default
signature_name	str	Name of the signature.	'DspySynthesisExtractorSignature'
instructions	str	Instructions for the signature.	'Extract structured synthesis for a specific material from the paper. Output only a valid JSON with the structured_synthesis field.'
paper_text_description	str	Description for the paper text input.	'Complete paper text to search for the material synthesis procedure.'
material_name_description	str	Description for material name input.	'The name of the specific material to extract synthesis for.'
output_name	str	Name of the output field.	'structured_synthesis'
output_description	str	Description for the output field.	'The extracted structured synthesis for specific material as a JSON.'

Returns:

Type	Description
type[Signature]	dspy.Signature: The dspy signature for synthesis extraction.

Source code in src/llm_synthesis/transformers/synthesis_extraction/dspy_synthesis_extraction.py

def make_dspy_synthesis_extractor_signature(
    signature_name: str = "DspySynthesisExtractorSignature",
    instructions: str = (
        "Extract structured synthesis for a specific material from the paper. "
        "Output only a valid JSON with the structured_synthesis field."
    ),
    paper_text_description: str = (
        "Complete paper text to search for the material synthesis procedure."
    ),
    material_name_description: str = (
        "The name of the specific material to extract synthesis for."
    ),
    output_name: str = "structured_synthesis",
    output_description: str = (
        "The extracted structured synthesis for specific material as a JSON."
    ),
) -> type[dspy.Signature]:
    """
    Create signature for extracting a materials-specific synthesis ontology.

    Args:
        signature_name (str): Name of the signature.
        instructions (str): Instructions for the signature.
        paper_text_description (str): Description for the paper text input.
        material_name_description (str): Description for material name input.
        output_name (str): Name of the output field.
        output_description (str): Description for the output field.

    Returns:
        dspy.Signature: The dspy signature for synthesis extraction.
    """
    signature = {
        "paper_text": (
            str,
            dspy.InputField(description=paper_text_description),
        ),
        "material_name": (
            str,
            dspy.InputField(description=material_name_description),
        ),
        output_name: (
            GeneralSynthesisOntology,
            dspy.OutputField(description=output_description),
        ),
    }
    return dspy.make_signature(
        signature_name=signature_name,
        instructions=instructions,
        signature=signature,
    )

PDF extraction

DoclingPDFExtractor(pipeline='standard', table_mode='accurate', add_page_images=False, use_gpu=True, scale=2.0, format='markdown')

Bases: PdfExtractorInterface

An extractor for extracting content from PDF files using the Docling library

This class provides functionality to convert PDF files into various formats such as Markdown, doctags, JSON, or tokens. It supports different modes for handling images within the PDF, including embedding, referencing, or using placeholders. The extractor can be configured with various options such as pipeline type, table extraction mode, GPU usage, and scaling.

Attributes:

Name	Type	Description
pipeline	str	The pipeline to use for PDF processing
(default	str	"standard").
table_mode	str	The mode for table extraction (default: "accurate").
add_page_images	bool	Whether to include page images in the output
(default	bool	False).
use_gpu	bool	Whether to use GPU for processing (default: True).
scale	float	The scaling factor for images (default: 2.0).
format	str	The output format. Options are "markdown", "doctags",
"json",	or "tokens" (default	"markdown").

Methods:

extract_to_markdown(pdf_data: bytes) -> str:
    Converts a PDF file to Markdown format. Supports different image
    modes and raises a ValueError if an invalid image mode is provided.

Source code in src/llm_synthesis/transformers/pdf_extraction/docling_pdf_extractor.py

def __init__(
    self,
    pipeline: str = "standard",
    table_mode: str = "accurate",
    add_page_images: bool = False,
    use_gpu: bool = True,
    scale: float = 2.0,
    format: str = "markdown",
):
    self.pipeline = pipeline
    self.table_mode = table_mode
    self.add_page_images = add_page_images
    self.use_gpu = use_gpu
    self.scale = scale
    self.format = format

Methods:

forward(input)

Extracts text and figures from a PDF and returns them as markdown with embedded figures.

Parameters:

Name	Type	Description	Default
pdf_data		The PDF data as bytes.	required

Returns:

Type	Description
str	The extracted text as markdown with embedded figures.

Source code in src/llm_synthesis/transformers/pdf_extraction/docling_pdf_extractor.py

def forward(self, input: bytes) -> str:
    """
    Extracts text and figures from a PDF and returns them as markdown with
    embedded figures.

    Args:
        pdf_data: The PDF data as bytes.

    Returns:
        The extracted text as markdown with embedded figures.
    """
    opts = PdfPipelineOptions(
        pipeline=self.pipeline,
        table_mode=self.table_mode,
        generate_picture_images=True,
        generate_page_images=self.add_page_images,
        images_scale=self.scale,
        ocr=True,
        batch_size=4 if self.use_gpu else 1,
    )
    conv = DocumentConverter(
        format_options={
            InputFormat.PDF: PdfFormatOption(pipeline_options=opts)
        }
    )
    result = conv.convert(
        DocumentStream(name="pdf", stream=io.BytesIO(input))
    )
    doc = result.document

    return doc.export_to_markdown(image_mode="embedded")

MistralPDFExtractor(structured=False, mistral_api_key=None, max_retries=3, retry_base_delay=2.0)

Bases: PdfExtractorInterface

A PDF extractor that uses the Mistral OCR API to extract content from PDF files and optionally convert it to Markdown format. This extractor supports embedding images as data URIs and can return structured JSON output if required.

Attributes:

Name	Type	Description
structured	bool	Determines whether the output should be structured JSON.
embed_images	bool	Indicates whether images should be embedded as data URIs in the Markdown output.
mistral_api_key	str	The API key for authenticating with the Mistral OCR API. If not provided, it will be fetched from the environment variable MISTRAL_API_KEY.
mistral_api_client	Mistral	The client instance for interacting with the Mistral OCR API.

Methods:

Name	Description
extract_to_markdown	bytes) -> str: Extracts content from a PDF file and converts it to Markdown format. Optionally embeds images as data URIs and supports structured JSON output.

Source code in src/llm_synthesis/transformers/pdf_extraction/mistral_pdf_extractor.py

def __init__(
    self,
    structured: bool = False,
    mistral_api_key: str | None = None,
    max_retries: int = 3,
    retry_base_delay: float = 2.0,
):
    self.structured = structured
    self.max_retries = max_retries
    self.retry_base_delay = retry_base_delay
    self.mistral_api_key = mistral_api_key or os.environ.get(
        "MISTRAL_API_KEY"
    )
    if self.mistral_api_key is None:
        LOGGER.error(
            "MISTRAL_API_KEY is not set. Please provide it as an argument "
            "or set it in the environment."
        )
        raise ValueError(
            "MISTRAL_API_KEY must be set either as an argument"
            " or in the environment."
        )
    self.mistral_api_client = Mistral(api_key=self.mistral_api_key)

Methods:

forward(input)

Extracts text and figures from a PDF and returns them as markdown with embedded figures.

Parameters:

Name	Type	Description	Default
pdf_data		The PDF data as bytes.	required

Returns:

Type	Description
str	The extracted text as markdown with embedded figures.

Source code in src/llm_synthesis/transformers/pdf_extraction/mistral_pdf_extractor.py

def forward(self, input: bytes) -> str:
    """
    Extracts text and figures from a PDF and returns them as markdown
    with embedded figures.

    Args:
        pdf_data: The PDF data as bytes.

    Returns:
        The extracted text as markdown with embedded figures.
    """
    data_uri = self._get_data_uri_from_bytes(input)

    resp = self._call_with_retry(
        self.mistral_api_client.ocr.process,
        document={"type": "document_url", "document_url": data_uri},
        model="mistral-ocr-latest",
        include_image_base64=True,
    )

    if self.structured:  # <-- optional JSON dump
        return json.dumps(resp.to_dict(), indent=2)

    return self._process_pages(resp)

Plot data extraction

ClaudeLinePlotDataExtractor(model_name, prompt=resources.LINE_CHART_PROMPT_WITH_CONTEXT, max_tokens=1024, temperature=0.0, use_figure_context=True)

Bases: LinePlotDataExtractorInterface

Source code in src/llm_synthesis/transformers/plot_extraction/claude_extraction/plot_data_extraction.py

def __init__(
    self,
    model_name: str,
    prompt: str = resources.LINE_CHART_PROMPT_WITH_CONTEXT,
    max_tokens: int = 1024,
    temperature: float = 0.0,
    use_figure_context: bool = True,
):
    super().__init__()
    self.claude_client = ClaudeAPIClient(model_name)
    self.prompt = prompt
    self.max_tokens = max_tokens
    self.temperature = temperature
    self.use_figure_context = use_figure_context

Methods:

get_cost()

Get cumulative cost from Claude client.

Source code in src/llm_synthesis/transformers/plot_extraction/claude_extraction/plot_data_extraction.py

def get_cost(self) -> float:
    """Get cumulative cost from Claude client."""
    return self.claude_client.get_cost()

reset_cost()

Reset costs in Claude client.

Source code in src/llm_synthesis/transformers/plot_extraction/claude_extraction/plot_data_extraction.py

def reset_cost(self) -> float:
    """Reset costs in Claude client."""
    return self.claude_client.reset_cost()

LiteLLMPlotDataExtractor(model, prompt=resources.LINE_CHART_PROMPT_WITH_CONTEXT, max_tokens=8192, temperature=0.0, api_key=None, api_base=None, extra_kwargs=None, retry_temperatures=None)

Bases: LinePlotDataExtractorInterface

Plot data extractor using litellm — works with any vision model.

Uses the same prompt and parsing logic as ClaudeLinePlotDataExtractor, but routes API calls through litellm for multi-provider support.

Source code in src/llm_synthesis/transformers/plot_extraction/litellm_plot_data_extraction.py

def __init__(
    self,
    model: str,
    prompt: str = resources.LINE_CHART_PROMPT_WITH_CONTEXT,
    max_tokens: int = 8192,
    temperature: float = 0.0,
    api_key: str | None = None,
    api_base: str | None = None,
    extra_kwargs: dict | None = None,
    retry_temperatures: list[float] | None = None,
):
    super().__init__()
    self.model = model
    self.prompt = prompt
    self.max_tokens = max_tokens
    self.temperature = temperature
    self.api_key = api_key
    self.api_base = api_base
    self.extra_kwargs = extra_kwargs or {}
    self.retry_temperatures = retry_temperatures or [temperature, 0.3, 0.5]
    self._cumulative_cost_usd = 0.0
    # This extractor instance is shared across all worker threads in
    # run_from_hf.py's ThreadPoolExecutor (built once in
    # build_pipeline()) -- without a lock, concurrent += on a shared
    # float silently drops updates and per-paper before/after cost
    # snapshots pick up other papers' concurrent cost, both of which
    # were observed inflating per-paper cost readings ~10x under
    # --workers 12.
    self._cost_lock = threading.Lock()

Performance linking

SeriesMaterialLinker(lm, prompt_template=DEFAULT_MATCHING_PROMPT)

Bases: PerformanceLinkingInterface

LLM-based transformer for matching plot series names to material names.

This transformer uses an LLM to semantically match series names from plots (e.g., "575", "Ni/Al2O3", "Sample A") to the actual material names extracted from the paper (e.g., "Mo2(C,N)Tx-575", "10%Ni/Al2O3").

Attributes:

Name	Type	Description
lm		DSPy language model for making predictions
prompt_template		Template for the matching prompt

Initialize the linker.

Parameters:

Name	Type	Description	Default
lm	LM	DSPy language model instance	required
prompt_template	str	Prompt template with placeholders for materials, series_names, context, plot_title, x_axis_label, x_axis_unit, y_axis_label, y_axis_unit	DEFAULT_MATCHING_PROMPT

Source code in src/llm_synthesis/transformers/performance_linking/series_material_linker.py

def __init__(
    self,
    lm: dspy.LM,
    prompt_template: str = DEFAULT_MATCHING_PROMPT,
):
    """Initialize the linker.

    Args:
        lm: DSPy language model instance
        prompt_template: Prompt template with placeholders for materials,
            series_names, context, plot_title, x_axis_label, x_axis_unit,
            y_axis_label, y_axis_unit
    """
    super().__init__()
    self.lm = lm
    self.prompt_template = prompt_template

Methods:

forward(input)

Match plot series names to material names using LLM.

Parameters:

Name	Type	Description	Default
input	LinkingInput	LinkingInput containing materials, series names, context, and plot metadata	required

Returns:

Type	Description
list[SeriesMapping]	List of validated SeriesMapping objects

Source code in src/llm_synthesis/transformers/performance_linking/series_material_linker.py

def forward(self, input: LinkingInput) -> list[SeriesMapping]:
    """Match plot series names to material names using LLM.

    Args:
        input: LinkingInput containing materials, series names, context,
            and plot metadata

    Returns:
        List of validated SeriesMapping objects
    """
    prompt = self._build_prompt(input)

    # Call LLM
    response = self.lm(prompt)
    first = response[0] if response else None
    if isinstance(first, dict):
        response_text = first.get("text", "")
    else:
        response_text = first or ""

    if not response_text:
        logger.warning(
            "LLM returned empty/None response (possibly truncated)"
        )
        return []

    # Parse response
    raw_mappings = self._parse_response(response_text)

    # Validate mappings
    validated = self._validate_mappings(
        raw_mappings,
        input.series_names,
        input.materials,
    )

    return validated