Skip to content

Pipeline

The SynthesisPerformancePipeline is the main orchestrator for end-to-end extraction. It chains material extraction, synthesis extraction, judge evaluation, and optional figure/performance linking.

SynthesisPerformancePipeline

SynthesisPerformancePipeline(material_extractor, synthesis_extractor, judge=None, linking_judge=None, plot_extractor=None, series_linker=None, plot_filter_config=None, figure_segmenter='dino', florence_repo_id='amayuelas/plot-visualization-florence-2-lora-32')

End-to-end pipeline: Paper → Materials → Synthesis → Performance Linking.

This pipeline processes scientific papers to extract: 1. Materials synthesized in the paper 2. Detailed synthesis procedures for each material 3. Performance data from plots, linked to specific materials

The pipeline is modular - each component can be customized or replaced.

Initialize the pipeline.

Parameters:

Name Type Description Default
material_extractor

Extractor for identifying materials in paper

required
synthesis_extractor

Extractor for synthesis procedures

required
judge

Optional judge for evaluating synthesis quality

None
linking_judge

Optional judge for evaluating linking quality

None
plot_extractor

Optional plot extractor (e.g. ClaudeLinePlotDataExtractor).

None
series_linker SeriesMaterialLinker | None

Optional linker for matching series to materials

None
plot_filter_config PlotFilterConfig | None

Optional config for filtering plots

None
figure_segmenter str

Backend for figure segmentation, "dino" (default) or "florence".

'dino'
florence_repo_id str

HuggingFace LoRA repo used when figure_segmenter="florence".

'amayuelas/plot-visualization-florence-2-lora-32'
Source code in src/llm_synthesis/services/pipelines/synthesis_performance_pipeline.py
def __init__(
    self,
    material_extractor,
    synthesis_extractor,
    judge=None,
    linking_judge=None,
    plot_extractor=None,
    series_linker: SeriesMaterialLinker | None = None,
    plot_filter_config: PlotFilterConfig | None = None,
    figure_segmenter: str = "dino",
    florence_repo_id: str = (
        "amayuelas/plot-visualization-florence-2-lora-32"
    ),
):
    """Initialize the pipeline.

    Args:
        material_extractor: Extractor for identifying materials in paper
        synthesis_extractor: Extractor for synthesis procedures
        judge: Optional judge for evaluating synthesis quality
        linking_judge: Optional judge for evaluating linking quality
        plot_extractor: Optional plot extractor
            (e.g. ClaudeLinePlotDataExtractor).
        series_linker: Optional linker for matching series to materials
        plot_filter_config: Optional config for filtering plots
        figure_segmenter: Backend for figure segmentation,
            ``"dino"`` (default) or ``"florence"``.
        florence_repo_id: HuggingFace LoRA repo used when
            ``figure_segmenter="florence"``.
    """
    self.material_extractor = material_extractor
    self.synthesis_extractor = synthesis_extractor
    self.judge = judge
    self.linking_judge = linking_judge
    self.plot_extractor = plot_extractor
    self.series_linker = series_linker
    self.plot_filter = (
        PlotFilter(plot_filter_config)
        if plot_filter_config
        else PlotFilter()
    )
    self.figure_segmenter = figure_segmenter
    self.florence_repo_id = florence_repo_id

Methods:

extract_materials(paper_text)

Step 1: Extract list of materials from paper text.

Parameters:

Name Type Description Default
paper_text str

Full paper text

required

Returns:

Type Description
list[str]

List of material names

Source code in src/llm_synthesis/services/pipelines/synthesis_performance_pipeline.py
def extract_materials(self, paper_text: str) -> list[str]:
    """Step 1: Extract list of materials from paper text.

    Args:
        paper_text: Full paper text

    Returns:
        List of material names
    """
    logger.info("Step 1: Extracting materials...")
    materials_text = self.material_extractor.forward(
        input=clean_text(paper_text)
    )

    if not materials_text:
        logger.warning("  No materials found")
        return []

    materials = [
        m.strip()
        for m in materials_text.replace("\n", ",").split(",")
        if m.strip()
    ]
    logger.info(f"  Found {len(materials)} materials: {materials}")
    return materials

extract_synthesis(paper_text, material)

Step 2: Extract synthesis procedure for a single material.

Parameters:

Name Type Description Default
paper_text str

Full paper text

required
material str

Material name to extract synthesis for

required

Returns:

Type Description
tuple[GeneralSynthesisOntology, Any]

Tuple of (synthesis ontology, evaluation result or None)

Source code in src/llm_synthesis/services/pipelines/synthesis_performance_pipeline.py
def extract_synthesis(
    self,
    paper_text: str,
    material: str,
) -> tuple[GeneralSynthesisOntology, Any]:
    """Step 2: Extract synthesis procedure for a single material.

    Args:
        paper_text: Full paper text
        material: Material name to extract synthesis for

    Returns:
        Tuple of (synthesis ontology, evaluation result or None)
    """
    logger.info(f"Step 2: Extracting synthesis for '{material}'...")

    try:
        synthesis = self.synthesis_extractor.forward(
            input=(clean_text(paper_text), material)
        )

        # Evaluate if judge is available
        evaluation = None
        if self.judge:
            try:
                evaluation = self.judge.forward(
                    (
                        clean_text(paper_text),
                        json.dumps(synthesis.model_dump()),
                        material,
                    )
                )
                logger.info(
                    f"  Evaluation score: "
                    f"{evaluation.scores.overall_score}/5.0"
                )
            except Exception as e:
                logger.warning(f"  Judge evaluation failed: {e}")

        return synthesis, evaluation

    except Exception as e:
        logger.error(f"  Synthesis extraction failed: {e}")
        return (
            GeneralSynthesisOntology(
                target_compound=material,
                target_compound_type="other",
                synthesis_method="other",
                notes=f"Extraction failed: {e}",
            ),
            None,
        )

extract_figures(markdown_text)

Step 3: Extract and classify figures from markdown.

Parameters:

Name Type Description Default
markdown_text str

Markdown text with embedded base64 images

required

Returns:

Type Description
list[FigureInfo]

List of quantitative figure info objects

Source code in src/llm_synthesis/services/pipelines/synthesis_performance_pipeline.py
def extract_figures(self, markdown_text: str) -> list[FigureInfo]:
    """Step 3: Extract and classify figures from markdown.

    Args:
        markdown_text: Markdown text with embedded base64 images

    Returns:
        List of quantitative figure info objects
    """
    logger.info("Step 3: Extracting figures...")

    try:
        from llm_synthesis.transformers.figure_extraction.regex_figure_extractor import (  # noqa: E501
            FigureExtractorMarkdown,
        )

        extractor = FigureExtractorMarkdown(
            segmenter=self.figure_segmenter,
            florence_repo_id=self.florence_repo_id,
        )
        all_figures = extractor.forward(markdown_text)

        # Filter to only quantitative figures (classified by ResNet).
        # This avoids sending non-quantitative figures (schematics,
        # microscopy, etc.) to the Claude VLM, saving compute.
        quantitative_figures = [f for f in all_figures if f.quantitative]
        logger.info(
            f"  Found {len(all_figures)} figures, "
            f"{len(quantitative_figures)} quantitative"
        )
        return quantitative_figures

    except Exception as e:
        import traceback

        logger.warning(
            f"  Figure extraction failed: {e}\n{traceback.format_exc()}"
        )
        return []

extract_plot_data(figures, paper_text, si_text='')

Step 4: Extract data from quantitative plots.

Parameters:

Name Type Description Default
figures list[FigureInfo]

List of FigureInfo for quantitative figures

required
paper_text str

Full paper text for context

required
si_text str

Supplementary information text

''

Returns:

Type Description
tuple[list[ExtractedLinePlotData], list[FigureInfo]]

Tuple of (list of plot data, list of corresponding figures)

Source code in src/llm_synthesis/services/pipelines/synthesis_performance_pipeline.py
def extract_plot_data(
    self,
    figures: list[FigureInfo],
    paper_text: str,
    si_text: str = "",
) -> tuple[list[ExtractedLinePlotData], list[FigureInfo]]:
    """Step 4: Extract data from quantitative plots.

    Args:
        figures: List of FigureInfo for quantitative figures
        paper_text: Full paper text for context
        si_text: Supplementary information text

    Returns:
        Tuple of (list of plot data, list of corresponding figures)
    """
    if not self.plot_extractor:
        logger.info(
            "Step 4: Skipping plot extraction (no extractor configured)"
        )
        return [], []

    logger.info(f"Step 4: Extracting data from {len(figures)} plots...")
    plots = []
    plot_figures = []

    for fig in figures:
        plot_data, _ = self._extract_one_plot(fig, paper_text, si_text)
        if plot_data is not None:
            plots.append(plot_data)
            plot_figures.append(fig)
            logger.info(
                f"    {fig.figure_reference}: "
                f"{len(plot_data.name_to_coordinates)} series extracted"
            )

    logger.info(f"  Extracted data from {len(plots)} plots")
    return plots, plot_figures

Step 5: Link plot series to materials.

Parameters:

Name Type Description Default
materials list[str]

List of material names

required
plots list[ExtractedLinePlotData]

List of extracted plot data

required
figures list[FigureInfo]

List of corresponding figure info

required

Returns:

Type Description
tuple[list[PlotMaterialMapping], LinkingStats]

Tuple of (list of mappings, linking statistics)

Source code in src/llm_synthesis/services/pipelines/synthesis_performance_pipeline.py
def link_performance(
    self,
    materials: list[str],
    plots: list[ExtractedLinePlotData],
    figures: list[FigureInfo],
) -> tuple[list[PlotMaterialMapping], LinkingStats]:
    """Step 5: Link plot series to materials.

    Args:
        materials: List of material names
        plots: List of extracted plot data
        figures: List of corresponding figure info

    Returns:
        Tuple of (list of mappings, linking statistics)
    """
    if not self.series_linker:
        logger.info(
            "Step 5: Skipping performance linking (no linker configured)"
        )
        return [], LinkingStats(total_plots_extracted=len(plots))

    logger.info(
        f"Step 5: Linking {len(plots)} plots to {len(materials)} "
        "materials..."
    )

    # Filter plots
    relevant_plots, skip_counts = self.plot_filter.filter_plots(plots)
    skipped_plots = []  # Could be enhanced to track details

    all_mappings = []
    for idx, plot in relevant_plots:
        fig = figures[idx]
        mapping = self._link_one_plot(idx, plot, fig, materials)
        if mapping is not None:
            all_mappings.append(mapping)
            logger.info(
                f"    Linking plot {idx} '{plot.title or 'N/A'}' "
                f"({len(plot.name_to_coordinates)} series)"
            )
            logger.info(
                f"      Matched: {len(mapping.mappings)}, "
                f"Unmatched: {mapping.unmatched_series}"
            )

    # Compute stats
    stats = compute_linking_stats(
        total_plots=len(plots),
        mappings=all_mappings,
        skip_counts=skip_counts,
        skipped_plots=skipped_plots,
    )

    return all_mappings, stats

process_paper(paper, skip_figures=False)

Process a single paper through the full pipeline.

Parameters:

Name Type Description Default
paper Paper

Paper object with text content

required
skip_figures bool

If True, skip figures and performance linking

False

Returns:

Type Description
PipelineResult | None

PipelineResult or None if processing failed

Source code in src/llm_synthesis/services/pipelines/synthesis_performance_pipeline.py
def process_paper(
    self,
    paper: Paper,
    skip_figures: bool = False,
) -> PipelineResult | None:
    """Process a single paper through the full pipeline.

    Args:
        paper: Paper object with text content
        skip_figures: If True, skip figures and performance linking

    Returns:
        PipelineResult or None if processing failed
    """
    logger.info(f"Processing: {paper.name}")

    # Step 1: Extract materials
    materials = self.extract_materials(paper.publication_text)
    if not materials:
        logger.warning("  No materials found, skipping paper")
        return None

    # Step 2: Extract synthesis for each material
    all_syntheses = []
    for material in materials:
        synthesis, evaluation = self.extract_synthesis(
            paper.publication_text, material
        )
        all_syntheses.append(
            SynthesisEntry(
                material=material,
                synthesis=synthesis,
                evaluation=evaluation,
            )
        )

    # Steps 3-5: Figure/plot extraction and linking (optional)
    performance_data = {}
    plot_mappings = []
    extracted_plots = []
    plot_figures: list[FigureInfo] = []
    linking_stats = None
    linking_evaluation = None

    if not skip_figures:
        # Step 3: Extract figures
        figures = self.extract_figures(paper.publication_text)

        if figures:
            # Step 4: Extract plot data
            plots, plot_figures = self.extract_plot_data(
                figures, paper.publication_text, paper.si_text
            )
            extracted_plots = plots

            if plots:
                # Step 5: Link performance (with graceful error handling)
                try:
                    plot_mappings, linking_stats = self.link_performance(
                        materials, plots, plot_figures
                    )

                    # Aggregate per material
                    performance_data = aggregate_all_materials_performance(
                        materials, plot_mappings, plots
                    )

                    # Step 6: Evaluate linking quality (optional)
                    if self.linking_judge and plot_mappings:
                        linking_evaluation = self._evaluate_linking(
                            paper_text=paper.publication_text,
                            all_syntheses=all_syntheses,
                            plots=plots,
                            plot_mappings=plot_mappings,
                            performance_data=performance_data,
                        )
                except Exception as e:
                    logger.warning(
                        f"  Performance linking failed: {e}. "
                        "Synthesis results saved without performance data."
                    )
                    # Keep empty defaults - synthesis still saved

    # Collect relevant plots for domain metric processors
    kept_relevant_plots: list[tuple[int, ExtractedLinePlotData]] = []
    if extracted_plots:
        kept_relevant_plots, _ = self.plot_filter.filter_plots(
            extracted_plots, log_skipped=False
        )

    # Build results
    results = []
    for entry in all_syntheses:
        results.append(
            SynthesisWithPerformanceEntry(
                material=entry.material,
                synthesis=entry.synthesis,
                evaluation=entry.evaluation,
                performance=performance_data.get(entry.material),
                linking_evaluation=linking_evaluation,
            )
        )

    # Summary
    materials_with_perf = [m for m in materials if m in performance_data]
    materials_without_perf = [
        m for m in materials if m not in performance_data
    ]

    return PipelineResult(
        paper_id=paper.id,
        paper_name=paper.name,
        materials=materials,
        results=results,
        plot_mappings=plot_mappings,
        num_plots=len(extracted_plots),
        linking_stats=linking_stats,
        materials_with_performance=materials_with_perf,
        materials_without_performance=materials_without_perf,
        relevant_plots=kept_relevant_plots,
        plot_figures=plot_figures,
    )

process_paper_async(paper, semaphore, skip_figures=False) async

Process one paper with concurrent LLM calls (asyncio + semaphore).

Same as process_paper but runs independent LLM calls in parallel: - Materials: one call, then synthesis+judge per material in parallel - Plot extraction: one call per figure in parallel - Linking: one call per plot in parallel

Parameters:

Name Type Description Default
paper Paper

Paper object with text content

required
semaphore Semaphore

Cap on concurrent LLM calls

required
skip_figures bool

If True, skip figures and performance linking

False

Returns:

Type Description
PipelineResult | None

PipelineResult or None if processing failed

Source code in src/llm_synthesis/services/pipelines/synthesis_performance_pipeline.py
async def process_paper_async(
    self,
    paper: Paper,
    semaphore: asyncio.Semaphore,
    skip_figures: bool = False,
) -> PipelineResult | None:
    """Process one paper with concurrent LLM calls (asyncio + semaphore).

    Same as process_paper but runs independent LLM calls in parallel:
    - Materials: one call, then synthesis+judge per material in parallel
    - Plot extraction: one call per figure in parallel
    - Linking: one call per plot in parallel

    Args:
        paper: Paper object with text content
        semaphore: Cap on concurrent LLM calls
        skip_figures: If True, skip figures and performance linking

    Returns:
        PipelineResult or None if processing failed
    """
    logger.info(f"Processing: {paper.name}")

    # Step 1: Material extraction (one call)
    materials_text = await run_with_semaphore(
        semaphore,
        self.material_extractor.forward,
        input=clean_text(paper.publication_text),
    )
    if not materials_text:
        logger.warning("  No materials found")
        return None
    materials = [
        m.strip()
        for m in materials_text.replace("\n", ",").split(",")
        if m.strip()
    ]
    logger.info(f"  Found {len(materials)} materials: {materials}")

    # Step 2: Synthesis + judge per material (parallel)
    async def extract_synthesis_one(material: str) -> SynthesisEntry:
        try:
            synthesis = await run_with_semaphore(
                semaphore,
                self.synthesis_extractor.forward,
                input=(clean_text(paper.publication_text), material),
            )
            evaluation = None
            if self.judge:
                try:
                    evaluation = await run_with_semaphore(
                        semaphore,
                        self.judge.forward,
                        (
                            clean_text(paper.publication_text),
                            json.dumps(synthesis.model_dump()),
                            material,
                        ),
                    )
                    logger.info(
                        f"  [{material}] Evaluation score: "
                        f"{evaluation.scores.overall_score}/5.0"
                    )
                except Exception as e:
                    logger.warning(f"  Judge evaluation failed: {e}")
            return SynthesisEntry(
                material=material,
                synthesis=synthesis,
                evaluation=evaluation,
            )
        except Exception as e:
            logger.error(
                f"  Synthesis extraction failed for {material}: {e}"
            )
            return SynthesisEntry(
                material=material,
                synthesis=GeneralSynthesisOntology(
                    target_compound=material,
                    target_compound_type="other",
                    synthesis_method="other",
                    notes=f"Extraction failed: {e}",
                ),
                evaluation=None,
            )

    all_syntheses = await asyncio.gather(
        *[extract_synthesis_one(mat) for mat in materials]
    )
    all_syntheses = list(all_syntheses)

    # Steps 3-5: Figures, plot extraction, linking (optional)
    performance_data = {}
    plot_mappings = []
    extracted_plots = []
    plot_figures: list[FigureInfo] = []
    linking_stats = None
    linking_evaluation = None

    if not skip_figures:
        # Step 3:
        # Extract figures (CPU-bound, no LLM — run in thread directly)
        figures = await asyncio.to_thread(
            self.extract_figures, paper.publication_text
        )

        if figures:
            # Step 4: Extract plot data in parallel (one call per figure)
            paper_text = paper.publication_text
            si_text = paper.si_text or ""
            plot_results = await asyncio.gather(
                *[
                    run_with_semaphore(
                        semaphore,
                        self._extract_one_plot,
                        fig,
                        paper_text,
                        si_text,
                    )
                    for fig in figures
                ]
            )
            plots = []
            plot_figures = []
            for plot_data, fig in plot_results:
                if plot_data is not None:
                    plots.append(plot_data)
                    plot_figures.append(fig)
            extracted_plots = plots
            logger.info(f"  Extracted data from {len(plots)} plots")

            if plots:
                try:
                    # Step 5: Link each plot in parallel
                    relevant_plots, skip_counts = (
                        self.plot_filter.filter_plots(plots)
                    )
                    # filter_plots returns (idx, plot) where idx indexes
                    # `plots`; `plots` and `plot_figures` are built in
                    # lockstep just above, so they must match.
                    assert len(plots) == len(plot_figures), (
                        f"plots/plot_figures desynced: "
                        f"{len(plots)} vs {len(plot_figures)}"
                    )
                    skipped_plots = []
                    link_tasks = [
                        run_with_semaphore(
                            semaphore,
                            self._link_one_plot,
                            idx,
                            plot,
                            plot_figures[idx],
                            materials,
                        )
                        for idx, plot in relevant_plots
                    ]
                    mapping_results = await asyncio.gather(*link_tasks)
                    all_mappings = [
                        m for m in mapping_results if m is not None
                    ]
                    plot_mappings = all_mappings
                    linking_stats = compute_linking_stats(
                        total_plots=len(plots),
                        mappings=all_mappings,
                        skip_counts=skip_counts,
                        skipped_plots=skipped_plots,
                    )
                    performance_data = aggregate_all_materials_performance(
                        materials, plot_mappings, plots
                    )

                    # Step 6: Linking judge (one call)
                    if self.linking_judge and plot_mappings:
                        synthesis_json = json.dumps(
                            [
                                {
                                    "material": e.material,
                                    "synthesis": (
                                        e.synthesis.model_dump()
                                        if e.synthesis
                                        else None
                                    ),
                                }
                                for e in all_syntheses
                            ],
                            indent=2,
                        )
                        plot_data_json = json.dumps(
                            [p.model_dump() for p in plots], indent=2
                        )
                        linking_output_json = json.dumps(
                            {
                                "mappings": [
                                    m.model_dump() for m in plot_mappings
                                ],
                                "performance_per_material": {
                                    k: v.model_dump()
                                    for k, v in performance_data.items()
                                },
                            },
                            indent=2,
                        )
                        linking_evaluation = await run_with_semaphore(
                            semaphore,
                            self.linking_judge.forward,
                            (
                                clean_text(paper_text),
                                synthesis_json,
                                plot_data_json,
                                linking_output_json,
                            ),
                        )
                        if linking_evaluation:
                            logger.info(
                                f"  Linking evaluation score: "
                                f"{linking_evaluation.scores.overall_score}/5.0"
                            )
                except Exception as e:
                    logger.warning(
                        f"  Performance linking failed: {e}. "
                        "Synthesis saved without performance."
                    )

    # Collect relevant plots for domain metric processors
    kept_relevant_plots: list[tuple[int, ExtractedLinePlotData]] = []
    if extracted_plots:
        kept_relevant_plots, _ = self.plot_filter.filter_plots(
            extracted_plots, log_skipped=False
        )

    # Build results
    results = [
        SynthesisWithPerformanceEntry(
            material=entry.material,
            synthesis=entry.synthesis,
            evaluation=entry.evaluation,
            performance=performance_data.get(entry.material),
            linking_evaluation=linking_evaluation,
        )
        for entry in all_syntheses
    ]
    materials_with_perf = [m for m in materials if m in performance_data]
    materials_without_perf = [
        m for m in materials if m not in performance_data
    ]

    return PipelineResult(
        paper_id=paper.id,
        paper_name=paper.name,
        materials=materials,
        results=results,
        plot_mappings=plot_mappings,
        num_plots=len(extracted_plots),
        linking_stats=linking_stats,
        materials_with_performance=materials_with_perf,
        materials_without_performance=materials_without_perf,
        relevant_plots=kept_relevant_plots,
        plot_figures=plot_figures,
    )

save_results(result, output_dir) staticmethod

Save pipeline results to disk.

Parameters:

Name Type Description Default
result PipelineResult

PipelineResult to save

required
output_dir str

Base output directory

required
Source code in src/llm_synthesis/services/pipelines/synthesis_performance_pipeline.py
@staticmethod
def save_results(result: PipelineResult, output_dir: str) -> None:
    """Save pipeline results to disk.

    Args:
        result: PipelineResult to save
        output_dir: Base output directory
    """
    paper_dir = os.path.join(output_dir, result.paper_id)
    os.makedirs(paper_dir, exist_ok=True)

    # One file per material
    for entry in result.results:
        mat_name = sanitize_filename(entry.material)
        mat_path = os.path.join(paper_dir, f"{mat_name}.json")
        with open(mat_path, "w") as f:
            json.dump(entry.model_dump(), f, indent=2)

    # Plot mappings
    mappings_path = os.path.join(paper_dir, "performance_mappings.json")
    with open(mappings_path, "w") as f:
        json.dump(
            [m.model_dump() for m in result.plot_mappings], f, indent=2
        )

    # Summary
    summary = {
        "paper_id": result.paper_id,
        "paper_name": result.paper_name,
        "total_materials": len(result.materials),
        "materials_with_performance": len(
            result.materials_with_performance
        ),
        "materials_without_performance": len(
            result.materials_without_performance
        ),
        "materials_list": result.materials,
        "materials_with_performance_list": (
            result.materials_with_performance
        ),
        "materials_without_performance_list": (
            result.materials_without_performance
        ),
        "total_plots_extracted": result.num_plots,
    }

    if result.linking_stats:
        stats = result.linking_stats
        summary["plots_linked"] = stats.plots_linked
        summary["plots_skipped"] = {
            "not_relevant_x": stats.plots_skipped_not_relevant_x,
            "not_relevant_y": stats.plots_skipped_not_relevant_y,
            "no_series": stats.plots_skipped_no_series,
        }
        summary["confidence_breakdown"] = stats.confidence_counts
        summary["all_unmatched_series"] = stats.all_unmatched_series

    summary_path = os.path.join(paper_dir, "linking_summary.json")
    with open(summary_path, "w") as f:
        json.dump(summary, f, indent=2)

    logger.info(
        f"  Saved {len(result.results)} material files to {paper_dir}/"
    )

Result models

PipelineResult

Bases: BaseModel

Complete result from the synthesis + performance pipeline.

SynthesisWithPerformanceEntry

Bases: BaseModel

A material's synthesis procedure with linked performance data.