biolm.pipeline

Overview

The pipeline framework provides:

• Multi-stage orchestration with automatic dependency resolution

• DuckDB caching for predictions, embeddings, and structures

• Resumability — skip completed stages on re-run

• Streaming for large datasets

• Visualization — funnel plots, PCA/UMAP, distributions

Quick start

from biolm.pipeline import GenerativePipeline, SaturationMutagenesisConfig

config = SaturationMutagenesisConfig(
    parent_sequence="MKTAYIAKQRQ",
    scoring_model="esm2-650m",
    scoring_action="predict",
    score_field="logits",
    top_n=10,
)
pipeline = GenerativePipeline(configs=[config])
results = pipeline.run()

Config hierarchy

All pipeline configs inherit from ScoringProtocolConfig. Use isinstance to dispatch on config type:

from biolm.pipeline.generative import (
    ScoringProtocolConfig,
    GenerativeProtocolConfig,
    SaturationMutagenesisConfig,
    IterativeMaskingDMSConfig,
    DirectGenerationConfig,
)

if isinstance(config, SaturationMutagenesisConfig):
    pipeline = GenerativePipeline(configs=[config])
elif isinstance(config, DirectGenerationConfig):
    pipeline = GenerativePipeline(configs=[config])

ScoringProtocolConfig
├── SaturationMutagenesisConfig
├── IterativeMaskingDMSConfig
└── DirectGenerationConfig (extends GenerativeProtocolConfig)

Model quick-reference

Config classes

Field details are documented on each class below.

class biolm.pipeline.generative.ScoringProtocolConfig

Bases: object

Marker base class for stages that score sequences and return a ranked subset.

Subclasses call a BioLM prediction model to assign a numeric score to each variant sequence enumerated by the config (e.g. all single-point mutants), then sort and return the top-N results. They do not produce wholly new sequences — the output is a scored, ranked subset of the candidate library built internally by the config (no separate generation step is needed).

The scoring_action field on concrete subclasses selects the client method: 'predict' calls BioLMApiClient.predict() and 'score' calls BioLMApiClient.score(). Both are allowlisted ('predict', 'score') to prevent arbitrary method dispatch.

Use isinstance(config, ScoringProtocolConfig)() to branch on config type in pipeline dispatch functions.

Example:

default Copy

from biolm.pipeline.generative import (
    ScoringProtocolConfig,
    SaturationMutagenesisConfig,
)

def dispatch(config):
    if isinstance(config, ScoringProtocolConfig):
        return run_scoring_stage(config)  # returns ranked variant library
    return run_generative_stage(config)   # returns novel sequences

config = SaturationMutagenesisConfig(
    parent_sequence="MKTAYIAKQRQ",
    scoring_model="thermompnn-d",
    score_field="ddg",
)
assert isinstance(config, ScoringProtocolConfig)  # True

class biolm.pipeline.generative.GenerativeProtocolConfig

Bases: object

Marker base class for stages that produce new amino-acid sequences.

Subclasses drive generative or masked-language models to emit novel sequences — either via autoregressive sampling (ProteinMPNN, DSM, AntiFold) or by greedy-argmax masking over an MLM (ESM2, ESMC). Unlike ScoringProtocolConfig subclasses, these stages expand the candidate pool rather than filtering it; they are the source stage in a generative design pipeline.

Use isinstance(config, GenerativeProtocolConfig)() to branch on config type in pipeline dispatch functions.

Example:

default Copy

from biolm.pipeline.generative import (
    GenerativeProtocolConfig,
    DirectGenerationConfig,
    IterativeMaskingDMSConfig,
)

def dispatch(config):
    if isinstance(config, GenerativeProtocolConfig):
        return run_generative_stage(config)  # emits new sequences
    return run_scoring_stage(config)          # returns ranked variants

cfg_direct = DirectGenerationConfig(
    "dsm-150m-base", sequence="MKTAYIAKQRQ", num_sequences=50
)
cfg_dms = IterativeMaskingDMSConfig(
    parent_sequence="MKTAYIAKQRQ", model_name="esm2-650m"
)
assert isinstance(cfg_direct, GenerativeProtocolConfig)  # True
assert isinstance(cfg_dms, GenerativeProtocolConfig)     # True

class biolm.pipeline.generative.SaturationMutagenesisConfig(parent_sequence: str, scoring_model: str, positions: list[int] | None = None, alphabet: str = 'ACDEFGHIKLMNPQRSTVWY', scoring_action: str = 'predict', scoring_params: dict[str, typing.Any] = <factory>, score_field: str = 'ddg', top_n: int | None = 50, ascending: bool = True, exclude_synonymous: bool = True, batch_size: int = 8, label: str | None = None, pdb_str: str | None = None, chain: str = 'A')

Bases: ScoringProtocolConfig

Source config that generates a single-mutant library and filters by a prediction model.

Enumerates every single amino-acid substitution at the specified positions, scores each variant with scoring_model using scoring_action, then returns the top-top_n variants ranked by score_field.

Typical use: ThermoMPNN-D or ESM2StabP-guided design — predict ΔΔG for all single-point mutants, keep the most stabilising ones.

Args:

parent_sequence: Wild-type sequence to mutate. scoring_model: BioLM model slug used to score variants (e.g. 'thermompnn-d'). positions: 0-indexed positions to enumerate. If None, all positions

in the sequence are enumerated.

alphabet: Amino acids to substitute. Defaults to the 20 canonical AAs. scoring_action: API action for the scoring model (default 'predict'). scoring_params: Extra params forwarded to the scoring model API. score_field: Key inside each model response that holds the numeric score

(default 'ddg'). Supports nested access with '.' separator, e.g. 'result.ddg'.

top_n: Number of top-scoring variants to retain (default 50). None

keeps all variants that receive a valid score.

ascending: If True, lower scores are better (e.g. negative ΔΔG means

stabilising). Defaults to True.

exclude_synonymous: If True (default), skip substitutions that are

identical to the wild-type residue.

batch_size: Sequences per API request when scoring (default 8). label: Optional label stored as source_label in results. pdb_str: Raw PDB file contents as a string (not a file path). When

provided, each scoring item is built as {"pdb": pdb_str, "mutations": [""], "chain": chain} instead of {"sequence": mutant_sequence}. Required by structure-aware models such as ThermoMPNN-D. Pass None (default) for sequence-only models like ESM2StabP.

chain: Chain identifier forwarded in structure-aware scoring items

(default 'A').

Example:

default Copy

from biolm.pipeline.generative import SaturationMutagenesisConfig, GenerativePipeline

# Score all single-mutant variants at three positions with ThermoMPNN-D
config = SaturationMutagenesisConfig(
    parent_sequence="MKTAYIAKQRQ",
    scoring_model="thermompnn-d",
    positions=[3, 7, 10],          # 0-indexed; None → all positions
    score_field="ddg",             # key in API response holding ΔΔG
    top_n=25,                      # keep top-25 most stabilising variants
    ascending=True,                # lower ΔΔG = more stabilising
    pdb_str=open("protein.pdb").read(),  # required by structure-aware models
)

pipeline = GenerativePipeline(configs=[config])
pipeline.add_prediction("esmfold", extractions="mean_plddt", columns="plddt")
results = pipeline.run()

alphabet: str = 'ACDEFGHIKLMNPQRSTVWY'

ascending: bool = True

batch_size: int = 8

chain: str = 'A'

exclude_synonymous: bool = True

label: str | None = None

parent_sequence: str

pdb_str: str | None = None

positions: list[int] | None = None

score_field: str = 'ddg'

scoring_action: str = 'predict'

scoring_model: str

scoring_params: dict[str, Any]

to_spec() → dict

top_n: int | None = 50

class biolm.pipeline.generative.IterativeMaskingDMSConfig(parent_sequence: str, model_name: str, positions: list[int] | None = None, rounds: int = 2, mask_token: str = '<mask>', alphabet: str = 'ACDEFGHIKLMNPQRSTVWY', exclude_synonymous: bool = True, batch_size: int = 32, label: str | None = None, action: str = 'predict')

Bases: GenerativeProtocolConfig

Source config that builds multi-point variants via sequential greedy masking.

Implements an iterative argmax masking procedure using a masked language model:

1. For each target position, mask it in the parent sequence and query the model for the highest-probability residue (greedy argmax — not sampled).

2. If rounds > 1, apply the round-1 preferred substitution at each position and repeat for round 2: mask each other target position in the round-1 sequence and collect the argmax residue.

3. Yield all resulting sequences as pipeline outputs.

This matches the ESM2 two-round DMS design pattern in the EGF generation notebook.

Args:

parent_sequence: Starting sequence. model_name: MLM model slug (e.g. 'esm2-650m', 'esmc-300m'). positions: 0-indexed positions to probe. Defaults to all positions. rounds: Number of sequential masking rounds (default 2). Round N uses

the variant produced by round N-1 as its starting sequence.

mask_token: Token inserted at masked positions (default ''). alphabet: Vocabulary used to identify valid AA positions in logits.

Defaults to the 20 canonical amino acids.

exclude_synonymous: Skip round-1 variants where the argmax matches WT

(default True).

batch_size: Sequences per API request (default 32). label: Optional label stored as source_label in results. action: API action for the model (default 'predict'; the model must

return logits). Only 'predict' is allowed because the greedy-argmax procedure requires per-token logit arrays.

Example:

default Copy

from biolm.pipeline.generative import IterativeMaskingDMSConfig, GenerativePipeline

# Two-round greedy DMS at three positions using ESM2-650M
config = IterativeMaskingDMSConfig(
    parent_sequence="MKTAYIAKQRQ",
    model_name="esm2-650m",
    positions=[2, 5, 8],   # 0-indexed; None → all positions
    rounds=2,              # round 1: per-position; round 2: 2-point combos
    exclude_synonymous=True,
)

pipeline = GenerativePipeline(configs=[config])
results = pipeline.run()
# results DataFrame has columns:
#   sequence, dms_round, dms_pos1, dms_aa1, dms_pos2, dms_aa2

action: str = 'predict'

alphabet: str = 'ACDEFGHIKLMNPQRSTVWY'

batch_size: int = 32

exclude_synonymous: bool = True

label: str | None = None

mask_token: str = '<mask>'

model_name: str

parent_sequence: str

positions: list[int] | None = None

rounds: int = 2

to_spec() → dict

class biolm.pipeline.generative.DirectGenerationConfig(model_name: str, structure_path: str | None = None, structure_column: str | None = None, sequence: str | None = None, item_field: str = 'pdb', params: dict[str, typing.Any] = <factory>, num_sequences: int = 100, temperature: float = 1.0, structure_from_stage: str | None = None, structure_from_model: str | None = None, n_runs: int = 1, label: str | None = None)

Bases: GenerativeProtocolConfig

Configuration for structure- or sequence-conditioned generation.

Use with models such as ProteinMPNN, AntiFold, HyperMPNN, LigandMPNN, DSM.

The caller is responsible for providing the correct item_field and params for the target model — these vary per model and are documented in the BioLM API schema (/schema//generate/). Common values:

Args:

model_name: BioLM model slug (e.g. 'protein-mpnn', 'antifold',

'dsm-150m-base').

structure_path: Path to a PDB or CIF file. structure_column: DataFrame column holding PDB strings (for chained

pipelines where structure was predicted upstream).

sequence: Parent sequence string for sequence-conditioned models (DSM). item_field: The item dict key expected by the model API — 'pdb' for

structure-conditioned models, 'sequence' for DSM. Defaults to 'pdb'.

params: Model-specific params dict (required for non-trivial calls).

Keys must exactly match the model’s API param names. When empty the stage sends {'num_sequences': num_sequences, 'temperature': temperature} as a simple fallback — useful only for models that accept these exact param names.

num_sequences: Fallback when params is empty (default 100). temperature: Fallback when params is empty (default 1.0).

Example:

default Copy

from biolm.pipeline.generative import DirectGenerationConfig, GenerativePipeline

# Structure-conditioned design with ProteinMPNN
cfg_mpnn = DirectGenerationConfig(
    model_name="protein-mpnn",
    structure_path="/path/to/protein.pdb",
    item_field="pdb",
    params={"num_sequences": 50, "temperature": 0.1},
)

# Sequence-conditioned design with DSM
cfg_dsm = DirectGenerationConfig(
    model_name="dsm-150m-base",
    sequence="MKTAYIAKQRQ",
    item_field="sequence",
    params={
        "num_sequences": 100,
        "temperature": 1.0,
        "remasking": "low_confidence",
        "step_divisor": 8,
    },
)

pipeline = GenerativePipeline(configs=[cfg_mpnn])
results = pipeline.run()

item_field: str = 'pdb'

label: str | None = None

model_name: str

n_runs: int = 1

num_sequences: int = 100

params: dict[str, Any]

sequence: str | None = None

structure_column: str | None = None

structure_from_model: str | None = None

structure_from_stage: str | None = None

structure_path: str | None = None

temperature: float = 1.0

to_spec() → dict

Return a serializable dict for pipeline definition persistence.

Pipeline examples

Saturation mutagenesis funnel:

from biolm.pipeline import GenerativePipeline, SaturationMutagenesisConfig

config = SaturationMutagenesisConfig(
    parent_sequence="MKTAYIAKQRQ",
    scoring_model="esm2-650m",
    scoring_action="predict",
    score_field="logits",
    top_n=10,
)
pipeline = GenerativePipeline(configs=[config])
results = pipeline.run()

Direct generation (ProteinMPNN):

from biolm.pipeline import GenerativePipeline, DirectGenerationConfig

config = DirectGenerationConfig(
    model_name="protein-mpnn",
    structure_path="design.pdb",
    num_sequences=10,
)
pipeline = GenerativePipeline(configs=[config])
results = pipeline.run()

Serialization and resumability

Pipeline definitions can be serialized with PipelineDef for round-trip storage. DuckDB caches predictions and embeddings so re-runs skip completed work.

Related

• Python SDK overview — SDK overview

• biolm.pipeline package — full pipeline module reference