API Reference

This page provides exhaustive, automatically generated technical API reference documentation for the mutant-ai core modules.

Engine

mutant.core.engine

mutant/core/engine.py — V0.5

MutationEngine orchestrates the 6-stage pipeline. Python coordinates; LLMs generate.

V0.5 Changes: - BehaviorProfile is built during analysis and cached on PipelineContext. - Coverage gap detection feeds the planner with structured gap data. - Selective quality review: only ~40% of cases are judged (configurable). - Batch generation: dimensions generate all mutations in one prompt.

Pipeline: analyze_behavior (+ profile cache) → plan_mutations (gap-aware) → generate_mutations (batched) → quality_review (selective) → deduplicate → output

MutationEngine

Orchestrates the V0.4 mutation pipeline.

All generation is delegated to the LLM. Python only coordinates.

Example

engine = MutationEngine(provider=OpenAIProvider()) result = await engine.run(scenario, count=20) print(result.stats) print(result.mutation_plan)

Source code in mutant/core/engine.py

class MutationEngine:
    """Orchestrates the V0.4 mutation pipeline.

    All generation is delegated to the LLM. Python only coordinates.

    Example
    -------
    >>> engine = MutationEngine(provider=OpenAIProvider())
    >>> result = await engine.run(scenario, count=20)
    >>> print(result.stats)
    >>> print(result.mutation_plan)
    """

    def __init__(
        self,
        provider: BaseLLMProvider,
        registry: MutationRegistry | None = None,
        cache: BaseCache | None = None,
    ) -> None:
        self._provider: Any = _CachedProvider(provider, cache) if cache else provider
        self._registry = registry or _default_registry

    async def run(
        self,
        scenario: Scenario,
        config: MutationConfig | None = None,
        **kwargs: Any,
    ) -> MutationResult:
        """Run the full 6-stage mutation pipeline.

        Parameters
        ----------
        scenario : Scenario
        config : MutationConfig | None
            Configuration object. If provided, overrides kwargs.
        **kwargs : Any
            Config parameters if config is not provided.
        """
        if config is None:
            config = MutationConfig(**kwargs)

        if config.verbose:
            logger.info("Analyzing scenario...")

        active_dims = self._resolve_dimensions(
            dimension_ids=config.dimension_ids,
            exclude_dimension_ids=config.exclude_dimension_ids,
            categories=config.categories,
            severities=config.severities,
        )
        [d.id for d in self._registry.all()]

        pipeline_config = PipelineConfig(
            count=config.count,
            concurrency=config.concurrency,
            temperature=config.temperature,
            max_tokens=config.max_tokens,
            quality_review=config.quality_review,
            quality_batch_size=config.quality_batch_size,
            deduplicate=config.deduplicate,
            generate_rationale=config.generate_rationale,
            generate_tags=config.generate_tags,
            dimension_ids=[d.id for d in active_dims],
            prompts=config.prompts,
        )
        ctx = PipelineContext(scenario=scenario, config=pipeline_config)
        dims_by_id = {d.id: d for d in active_dims}

        # Stage 1
        ctx = await analyze_behavior(ctx, self._provider)

        if config.verbose:
            logger.info("Planning mutations...")

        # Stage 2
        ctx = await plan_mutations(ctx, self._provider, active_dims)

        if config.verbose:
            logger.info("Generating mutations...")

        # Stage 3
        ctx = await generate_mutations(ctx, self._provider, dims_by_id)

        # Stage 4 — Quality Review
        if config.quality_review:
            if config.verbose:
                logger.info("Reviewing quality...")
            from mutant.pipeline.stages import quality_review as _qr

            ctx = await _qr(ctx, self._provider)
        else:
            ctx.reviewed_cases = ctx.raw_cases

        # Stage 5 — Deduplication
        if config.deduplicate and len(ctx.reviewed_cases) > 1:
            if config.verbose:
                logger.info("Deduplicating...")
            ctx = await deduplicate(ctx, self._provider)
        else:
            ctx.deduplicated_cases = ctx.reviewed_cases

        if config.verbose:
            logger.info("Completed.")

        ctx.final_cases = ctx.output_cases

        return MutationResult(
            cases=ctx.final_cases,
            behavior_analysis=ctx.behavior_analysis,
            mutation_plan=ctx.mutation_plan,
        )

    def _resolve_dimensions(
        self,
        *,
        dimension_ids: list[str] | None,
        exclude_dimension_ids: list[str] | None,
        categories: list[str] | None,
        severities: list[str] | None,
    ) -> list[MutationDimension]:
        pool = self._registry.all()
        if dimension_ids:
            id_set = set(dimension_ids)
            pool = [d for d in pool if d.id in id_set]
        if exclude_dimension_ids:
            exclude_set = set(exclude_dimension_ids)
            pool = [d for d in pool if d.id not in exclude_set]
        if categories:
            cat_set = {c.lower() for c in categories}
            pool = [d for d in pool if d.category.value.lower() in cat_set]
        if severities:
            sev_set = {s.lower() for s in severities}
            pool = [d for d in pool if d.severity.value.lower() in sev_set]
        if not pool:
            raise ValueError(
                "No dimensions match the provided filters. "
                "Check your dimension ids, categories, and severities."
            )
        return pool

run(scenario, config=None, **kwargs) async

Run the full 6-stage mutation pipeline.

Parameters:

Name	Type	Description	Default
scenario	Scenario		required
config	MutationConfig | None	Configuration object. If provided, overrides kwargs.	None
**kwargs	Any	Config parameters if config is not provided.	{}

Source code in mutant/core/engine.py

async def run(
    self,
    scenario: Scenario,
    config: MutationConfig | None = None,
    **kwargs: Any,
) -> MutationResult:
    """Run the full 6-stage mutation pipeline.

    Parameters
    ----------
    scenario : Scenario
    config : MutationConfig | None
        Configuration object. If provided, overrides kwargs.
    **kwargs : Any
        Config parameters if config is not provided.
    """
    if config is None:
        config = MutationConfig(**kwargs)

    if config.verbose:
        logger.info("Analyzing scenario...")

    active_dims = self._resolve_dimensions(
        dimension_ids=config.dimension_ids,
        exclude_dimension_ids=config.exclude_dimension_ids,
        categories=config.categories,
        severities=config.severities,
    )
    [d.id for d in self._registry.all()]

    pipeline_config = PipelineConfig(
        count=config.count,
        concurrency=config.concurrency,
        temperature=config.temperature,
        max_tokens=config.max_tokens,
        quality_review=config.quality_review,
        quality_batch_size=config.quality_batch_size,
        deduplicate=config.deduplicate,
        generate_rationale=config.generate_rationale,
        generate_tags=config.generate_tags,
        dimension_ids=[d.id for d in active_dims],
        prompts=config.prompts,
    )
    ctx = PipelineContext(scenario=scenario, config=pipeline_config)
    dims_by_id = {d.id: d for d in active_dims}

    # Stage 1
    ctx = await analyze_behavior(ctx, self._provider)

    if config.verbose:
        logger.info("Planning mutations...")

    # Stage 2
    ctx = await plan_mutations(ctx, self._provider, active_dims)

    if config.verbose:
        logger.info("Generating mutations...")

    # Stage 3
    ctx = await generate_mutations(ctx, self._provider, dims_by_id)

    # Stage 4 — Quality Review
    if config.quality_review:
        if config.verbose:
            logger.info("Reviewing quality...")
        from mutant.pipeline.stages import quality_review as _qr

        ctx = await _qr(ctx, self._provider)
    else:
        ctx.reviewed_cases = ctx.raw_cases

    # Stage 5 — Deduplication
    if config.deduplicate and len(ctx.reviewed_cases) > 1:
        if config.verbose:
            logger.info("Deduplicating...")
        ctx = await deduplicate(ctx, self._provider)
    else:
        ctx.deduplicated_cases = ctx.reviewed_cases

    if config.verbose:
        logger.info("Completed.")

    ctx.final_cases = ctx.output_cases

    return MutationResult(
        cases=ctx.final_cases,
        behavior_analysis=ctx.behavior_analysis,
        mutation_plan=ctx.mutation_plan,
    )

augment(dataset, provider, mutations_per_case=None, quality_review=None, dimensions=None, verbose=None, generate_rationale=None, generate_tags=None, concurrency=3, config=None, **kwargs) async

Augment an existing dataset by mutating each scenario.

Source code in mutant/core/engine.py

async def augment(
    dataset: Sequence[Scenario],
    provider: BaseLLMProvider,
    mutations_per_case: int | None = None,
    quality_review: bool | None = None,
    dimensions: list[str] | None = None,
    verbose: bool | None = None,
    generate_rationale: bool | None = None,
    generate_tags: bool | None = None,
    concurrency: int = 3,
    config: MutationConfig | None = None,
    **kwargs: Any,
) -> AugmentedDataset:
    """Augment an existing dataset by mutating each scenario."""
    import anyio

    if config is None:
        config = MutationConfig(
            count=mutations_per_case if mutations_per_case is not None else 5,
            quality_review=quality_review if quality_review is not None else True,
            dimension_ids=dimensions,
            verbose=verbose if verbose is not None else False,
            generate_rationale=generate_rationale
            if generate_rationale is not None
            else True,
            generate_tags=generate_tags if generate_tags is not None else True,
            **kwargs,
        )
    else:
        updates: dict[str, Any] = {}
        if mutations_per_case is not None:
            updates["count"] = mutations_per_case
        if quality_review is not None:
            updates["quality_review"] = quality_review
        if verbose is not None:
            updates["verbose"] = verbose
        if dimensions is not None:
            updates["dimension_ids"] = dimensions
        if generate_rationale is not None:
            updates["generate_rationale"] = generate_rationale
        if generate_tags is not None:
            updates["generate_tags"] = generate_tags
        if updates:
            config = config.model_copy(update=updates)

    all_results: list[MutationResult | None] = [None for _ in dataset]
    semaphore = anyio.Semaphore(concurrency)

    async def _process_one(
        idx: int, scenario: Scenario, task_id: Any = None, progress: Any = None
    ) -> None:
        async with semaphore:
            try:
                # Disable child verbosity so the progress bar isn't interrupted by logs
                child_config = config.model_copy(update={"verbose": False})
                result = await mutate(scenario, provider=provider, config=child_config)
                all_results[idx] = result
            except Exception as e:
                logger.error(f"Failed to process scenario {idx}: {e}")
            finally:
                if progress is not None and task_id is not None:
                    progress.advance(task_id)

    if config.verbose:
        from rich.progress import (
            BarColumn,
            Progress,
            SpinnerColumn,
            TaskProgressColumn,
            TextColumn,
        )

        with Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            BarColumn(),
            TaskProgressColumn(),
        ) as progress:
            task_id = progress.add_task(
                "[cyan]Augmenting dataset...", total=len(dataset)
            )
            async with anyio.create_task_group() as tg:
                for i, scenario in enumerate(dataset):
                    tg.start_soon(_process_one, i, scenario, task_id, progress)
    else:
        async with anyio.create_task_group() as tg:
            for i, scenario in enumerate(dataset):
                tg.start_soon(_process_one, i, scenario, None, None)

    final_cases = []
    for res in all_results:
        if res:
            final_cases.extend(res.cases)

    return AugmentedDataset(cases=final_cases)

augment_sync(dataset, provider, mutations_per_case=None, quality_review=None, dimensions=None, verbose=None, generate_rationale=None, generate_tags=None, **kwargs)

Synchronous wrapper for dataset augmentation.

Source code in mutant/core/engine.py

def augment_sync(
    dataset: Sequence[Scenario],
    provider: BaseLLMProvider,
    mutations_per_case: int | None = None,
    quality_review: bool | None = None,
    dimensions: list[str] | None = None,
    verbose: bool | None = None,
    generate_rationale: bool | None = None,
    generate_tags: bool | None = None,
    **kwargs: Any,
) -> AugmentedDataset:
    """Synchronous wrapper for dataset augmentation."""
    return asyncio.run(
        augment(
            dataset,
            provider,
            mutations_per_case=mutations_per_case,
            quality_review=quality_review,
            dimensions=dimensions,
            verbose=verbose,
            **kwargs,
        )
    )

mutate(scenario, provider, count=None, quality_review=None, dimensions=None, verbose=None, generate_rationale=None, generate_tags=None, config=None, cache=None, registry=None, **kwargs) async

Generate LLM-powered behavioral mutations. Primary public API.

Source code in mutant/core/engine.py

async def mutate(
    scenario: Scenario,
    provider: BaseLLMProvider,
    count: int | None = None,
    quality_review: bool | None = None,
    dimensions: list[str] | None = None,
    verbose: bool | None = None,
    generate_rationale: bool | None = None,
    generate_tags: bool | None = None,
    config: MutationConfig | None = None,
    cache: BaseCache | None = None,
    registry: MutationRegistry | None = None,
    **kwargs: Any,
) -> MutationResult:
    """Generate LLM-powered behavioral mutations. Primary public API."""
    engine = MutationEngine(provider=provider, registry=registry, cache=cache)
    if config is None:
        config = MutationConfig(
            count=count if count is not None else 50,
            quality_review=quality_review if quality_review is not None else True,
            dimension_ids=dimensions,
            verbose=verbose if verbose is not None else False,
            generate_rationale=generate_rationale
            if generate_rationale is not None
            else True,
            generate_tags=generate_tags if generate_tags is not None else True,
            **kwargs,
        )
    else:
        # User explicitly passed a config, but we can override with common kwargs if provided
        updates: dict[str, Any] = {}
        if count is not None:
            updates["count"] = count
        if quality_review is not None:
            updates["quality_review"] = quality_review
        if verbose is not None:
            updates["verbose"] = verbose
        if dimensions is not None:
            updates["dimension_ids"] = dimensions
        if generate_rationale is not None:
            updates["generate_rationale"] = generate_rationale
        if generate_tags is not None:
            updates["generate_tags"] = generate_tags
        if updates:
            config = config.model_copy(update=updates)

    return await engine.run(scenario, config=config, **kwargs)

mutate_sync(scenario, provider, count=None, quality_review=None, dimensions=None, verbose=None, generate_rationale=None, generate_tags=None, config=None, **kwargs)

Synchronous wrapper for mutate.

Source code in mutant/core/engine.py

def mutate_sync(
    scenario: Scenario,
    provider: BaseLLMProvider,
    count: int | None = None,
    quality_review: bool | None = None,
    dimensions: list[str] | None = None,
    verbose: bool | None = None,
    generate_rationale: bool | None = None,
    generate_tags: bool | None = None,
    config: MutationConfig | None = None,
    **kwargs: Any,
) -> MutationResult:
    """Synchronous wrapper for mutate."""
    return asyncio.run(
        mutate(
            scenario,
            provider,
            count=count,
            quality_review=quality_review,
            dimensions=dimensions,
            verbose=verbose,
            config=config,
            **kwargs,
        )
    )

Scenarios

mutant.core.scenario

Scenario — the original behavioral situation under test.

Scenario

Bases: BaseModel

Represents a single, original behavioral scenario to be mutated.

A scenario is the atomic unit of input to Mutant. Developers write one realistic scenario; Mutant generates hundreds of behavioral mutations from it.

Attributes:

Name	Type	Description
title	str	Short human-readable label for the scenario.
description	str	Full description of the scenario. This is the text that mutations will be applied to.
context	dict[str, Any]	Optional extra metadata (agent name, domain, tags, etc.).
tags	list[str]	Free-form labels for filtering / grouping during reporting.

Examples:

>>> s = Scenario(
...     title="Refund Request",
...     description="Customer bought a laptop. Requests a refund after 10 days.",
...     tags=["customer-support", "refund"],
... )
>>> print(s.title)
'Refund Request'

Source code in mutant/core/scenario.py

class Scenario(BaseModel):
    """Represents a single, original behavioral scenario to be mutated.

    A scenario is the atomic unit of input to Mutant. Developers write one
    realistic scenario; Mutant generates hundreds of behavioral mutations from it.

    Attributes
    ----------
    title:
        Short human-readable label for the scenario.
    description:
        Full description of the scenario. This is the text that mutations
        will be applied to.
    context:
        Optional extra metadata (agent name, domain, tags, etc.).
    tags:
        Free-form labels for filtering / grouping during reporting.

    Examples
    --------
    >>> s = Scenario(
    ...     title="Refund Request",
    ...     description="Customer bought a laptop. Requests a refund after 10 days.",
    ...     tags=["customer-support", "refund"],
    ... )
    >>> print(s.title)
    'Refund Request'
    """

    title: str = Field(..., min_length=1, description="Short human-readable label.")
    description: str = Field(
        ..., min_length=5, description="Full description of the scenario."
    )
    domain: str | None = Field(
        None, description="Optional explicit domain (e.g., healthcare, finance)."
    )
    context: dict[str, Any] = Field(
        default_factory=dict,
        description="Optional structured context (organization, jurisdiction, risk_level, compliance, agent_type, tools, etc.).",
    )
    tags: list[str] = Field(
        default_factory=list,
        description="Free-form labels for filtering / grouping.",
    )

    model_config = {"frozen": False, "extra": "forbid"}

    @model_validator(mode="after")
    def _normalise_tags(self) -> Scenario:
        self.tags = [t.lower().strip() for t in self.tags if t.strip()]
        return self

    # ── Convenience ───────────────────────────────────────────────────────────

    def with_description(self, description: str) -> Scenario:
        """Return a shallow copy with a new description (used internally by mutations)."""
        return self.model_copy(update={"description": description})

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Scenario:
        """Create a Scenario from a dictionary, safely inferring missing fields."""
        title = data.pop("title", data.get("id", "Imported Scenario"))
        description = data.pop("description", data.pop("text", None))
        if not description:
            raise ValueError("Dictionary must contain a 'description' or 'text' key.")
        domain = data.pop("domain", None)
        tags = data.pop("tags", [])
        return cls(
            title=title, description=description, domain=domain, tags=tags, context=data
        )

    @classmethod
    def from_messages(
        cls, messages: list[dict[str, str]], title: str = "Chat Scenario"
    ) -> Scenario:
        """Create a Scenario from a list of chat messages."""
        lines = []
        for msg in messages:
            role = msg.get("role", "unknown").capitalize()
            content = msg.get("content", "")
            lines.append(f"{role}: {content}")
        return cls(
            title=title,
            description="\n\n".join(lines),
            context={"format": "chat_messages"},
        )

    @classmethod
    def from_chat(cls, chat_text: str, title: str = "Chat Scenario") -> Scenario:
        """Create a Scenario directly from raw chat text."""
        return cls(title=title, description=chat_text, context={"format": "raw_chat"})

    @classmethod
    def from_dataframe_row(
        cls, row: Any, text_column: str = "text", title_column: str | None = None
    ) -> Scenario:
        """Create a Scenario from a Pandas or Polars DataFrame row."""
        # Standard dict conversion handles pandas Series natively if using .to_dict() beforehand,
        # but if this is a raw Series or namedtuple from iterrows/itertuples:
        if hasattr(row, "to_dict"):
            data = row.to_dict()
        elif hasattr(row, "_asdict"):
            data = row._asdict()
        elif isinstance(row, dict):
            data = dict(row)
        else:
            raise TypeError(
                "Row must be convertible to a dictionary (e.g. Pandas Series)."
            )

        title = (
            data.pop(title_column)
            if title_column and title_column in data
            else "Scenario from row"
        )
        description = data.pop(text_column, None)
        if not description:
            raise ValueError(f"Row missing required text_column: '{text_column}'")

        return cls(title=str(title), description=str(description), context=data)

    def __repr__(self) -> str:  # pragma: no cover
        return (
            f"Scenario(title={self.title!r}, description={self.description[:40]!r}...)"
        )

from_chat(chat_text, title='Chat Scenario') classmethod

Create a Scenario directly from raw chat text.

Source code in mutant/core/scenario.py

@classmethod
def from_chat(cls, chat_text: str, title: str = "Chat Scenario") -> Scenario:
    """Create a Scenario directly from raw chat text."""
    return cls(title=title, description=chat_text, context={"format": "raw_chat"})

from_dataframe_row(row, text_column='text', title_column=None) classmethod

Create a Scenario from a Pandas or Polars DataFrame row.

Source code in mutant/core/scenario.py

@classmethod
def from_dataframe_row(
    cls, row: Any, text_column: str = "text", title_column: str | None = None
) -> Scenario:
    """Create a Scenario from a Pandas or Polars DataFrame row."""
    # Standard dict conversion handles pandas Series natively if using .to_dict() beforehand,
    # but if this is a raw Series or namedtuple from iterrows/itertuples:
    if hasattr(row, "to_dict"):
        data = row.to_dict()
    elif hasattr(row, "_asdict"):
        data = row._asdict()
    elif isinstance(row, dict):
        data = dict(row)
    else:
        raise TypeError(
            "Row must be convertible to a dictionary (e.g. Pandas Series)."
        )

    title = (
        data.pop(title_column)
        if title_column and title_column in data
        else "Scenario from row"
    )
    description = data.pop(text_column, None)
    if not description:
        raise ValueError(f"Row missing required text_column: '{text_column}'")

    return cls(title=str(title), description=str(description), context=data)

from_dict(data) classmethod

Create a Scenario from a dictionary, safely inferring missing fields.

Source code in mutant/core/scenario.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Scenario:
    """Create a Scenario from a dictionary, safely inferring missing fields."""
    title = data.pop("title", data.get("id", "Imported Scenario"))
    description = data.pop("description", data.pop("text", None))
    if not description:
        raise ValueError("Dictionary must contain a 'description' or 'text' key.")
    domain = data.pop("domain", None)
    tags = data.pop("tags", [])
    return cls(
        title=title, description=description, domain=domain, tags=tags, context=data
    )

from_messages(messages, title='Chat Scenario') classmethod

Create a Scenario from a list of chat messages.

Source code in mutant/core/scenario.py

@classmethod
def from_messages(
    cls, messages: list[dict[str, str]], title: str = "Chat Scenario"
) -> Scenario:
    """Create a Scenario from a list of chat messages."""
    lines = []
    for msg in messages:
        role = msg.get("role", "unknown").capitalize()
        content = msg.get("content", "")
        lines.append(f"{role}: {content}")
    return cls(
        title=title,
        description="\n\n".join(lines),
        context={"format": "chat_messages"},
    )

with_description(description)

Return a shallow copy with a new description (used internally by mutations).

Source code in mutant/core/scenario.py

def with_description(self, description: str) -> Scenario:
    """Return a shallow copy with a new description (used internally by mutations)."""
    return self.model_copy(update={"description": description})

Mutations

mutant.core.mutation

mutant/core/mutation.py — V0.4

All data models for the mutation pipeline. No generation logic lives here.

AugmentedDataset

Bases: BaseModel

Result of augmenting an entire dataset.

Source code in mutant/core/mutation.py

class AugmentedDataset(BaseModel):
    """Result of augmenting an entire dataset."""

    cases: list[EvaluationCase]

    @property
    def summary(self) -> str:
        return f"AugmentedDataset: {len(self.cases)} total cases generated."

    def _to_records(self, **kwargs: Any) -> list[dict[str, Any]]:
        # Default exclusions for cleaner output
        default_excludes = {
            "dimension_name",
            "category",
            "severity",
            "rationale",
            "behavioral_tags",
            "expected_failure_modes",
        }
        exclude = {k for k in default_excludes if kwargs.get(k) is not True}

        for k, v in kwargs.items():
            if v is False and k != "mutated_description":
                exclude.add(k)

        records = []
        for case in self.cases:
            record = case.model_dump(exclude=exclude if exclude else None)
            if "category" in record and isinstance(record["category"], Enum):
                record["category"] = record["category"].value
            if "severity" in record and isinstance(record["severity"], Enum):
                record["severity"] = record["severity"].value
            import json

            for key in ["behavioral_tags", "expected_failure_modes"]:
                if record.get(key):
                    record[key] = json.dumps(record[key])
            records.append(record)
        return records

    def to_dataframe(self, **kwargs: Any) -> Any:
        try:
            import pandas as pd
        except ImportError:
            raise ImportError("pandas is required. pip install pandas")
        return pd.DataFrame(self._to_records(**kwargs))

    def to_csv(self, path: str, **kwargs: Any) -> None:
        import csv

        records = self._to_records(**kwargs)
        if not records:
            return
        with open(path, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=records[0].keys())
            writer.writeheader()
            writer.writerows(records)

    def to_json(self, path: str, **kwargs: Any) -> None:
        exclude = {
            k for k, v in kwargs.items() if v is False and k != "mutated_description"
        }
        exclude_dict = {"cases": {"__all__": exclude}} if exclude else None
        with open(path, "w", encoding="utf-8") as f:
            f.write(self.model_dump_json(indent=2, exclude=exclude_dict))

    def to_jsonl(self, path: str, **kwargs: Any) -> None:
        exclude = {
            k for k, v in kwargs.items() if v is False and k != "mutated_description"
        }
        with open(path, "w", encoding="utf-8") as f:
            for case in self.cases:
                f.write(
                    case.model_dump_json(exclude=exclude if exclude else None) + "\n"
                )

    def to_parquet(self, path: str, **kwargs: Any) -> None:
        df = self.to_dataframe(**kwargs)
        df.to_parquet(path)

    def sort_by(self, field: str, descending: bool = False) -> AugmentedDataset:
        if not self.cases:
            return self

        def _key(c: EvaluationCase) -> Any:
            val = getattr(c, field)
            if isinstance(val, Enum):
                return list(type(val)).index(val)
            return val

        sorted_cases = sorted(self.cases, key=_key, reverse=descending)
        return self.model_copy(update={"cases": sorted_cases})

    def filter(self, **kwargs: Any) -> AugmentedDataset:
        filtered = self.cases
        for key, value in kwargs.items():
            if key == "dimension":
                filtered = [
                    c
                    for c in filtered
                    if value.lower() in c.dimension_id.lower()
                    or value.lower() in c.dimension_name.lower()
                ]
            elif key == "severity":
                filtered = [
                    c for c in filtered if c.severity.value.lower() == value.lower()
                ]
            elif key == "keyword":
                keyword = value.lower()
                filtered = [
                    c
                    for c in filtered
                    if keyword in c.mutated_description.lower()
                    or keyword in c.rationale.lower()
                ]
            else:
                filtered = [c for c in filtered if getattr(c, key, None) == value]

        return self.model_copy(update={"cases": filtered})

    def __len__(self) -> int:
        return len(self.cases)

    def __iter__(self) -> Any:
        return iter(self.cases)

    def __getitem__(self, index: int | slice) -> Any:
        return self.cases[index]

BehaviorAnalysis

Bases: BaseModel

Rich structural analysis of a scenario produced by the LLM.

Source code in mutant/core/mutation.py

class BehaviorAnalysis(BaseModel):
    """Rich structural analysis of a scenario produced by the LLM."""

    detected_domain: str = Field(
        default="", description="Inferred domain (e.g. e-commerce, healthcare)."
    )
    confidence: float = Field(
        default=1.0, ge=0.0, le=1.0, description="Domain detection confidence."
    )
    actors: list[str] = Field(
        default_factory=list, description="People or systems involved."
    )
    entities: list[str] = Field(
        default_factory=list, description="Key objects, values, identifiers."
    )
    goals: list[str] = Field(
        default_factory=list, description="What actors are trying to achieve."
    )
    constraints: list[str] = Field(
        default_factory=list, description="Rules or limits that apply."
    )
    assumptions: list[str] = Field(
        default_factory=list, description="Things the agent implicitly assumes."
    )
    policies: list[str] = Field(
        default_factory=list, description="Business or domain policies in play."
    )
    tools: list[str] = Field(
        default_factory=list, description="Tools or APIs the agent would use."
    )
    risks: list[str] = Field(
        default_factory=list, description="Areas where failure is likely or costly."
    )
    likely_failure_modes: list[str] = Field(
        default_factory=list, description="How a weak agent would fail."
    )
    ambiguities: list[str] = Field(
        default_factory=list, description="Underspecified or unclear elements."
    )

DimensionAllocation

Bases: BaseModel

Planner decision for one mutation dimension.

Source code in mutant/core/mutation.py

class DimensionAllocation(BaseModel):
    """Planner decision for one mutation dimension."""

    dimension_id: str
    dimension_name: str
    count: int
    priority: int
    rationale: str
    why_selected: str = Field(
        default="",
        description="Explicit explanation of why this dimension matters here.",
    )
    focus_areas: list[str] = Field(default_factory=list)
    difficulty: str = Field(default="medium", description="low | medium | high")
    mutation_type: str = Field(default="single", description="single | composed")

EvaluationCase

Bases: BaseModel

A single mutation case — rich enough to be used directly in evaluation.

Includes the mutated input, rationale, expected behaviors, and failure modes so it can be plugged directly into any evaluation framework.

Source code in mutant/core/mutation.py

class EvaluationCase(BaseModel):
    """A single mutation case — rich enough to be used directly in evaluation.

    Includes the mutated input, rationale, expected behaviors, and failure
    modes so it can be plugged directly into any evaluation framework.
    """

    id: str = Field(exclude=True)
    parent_id: str | None = Field(default=None, exclude=True)
    dimension_id: str
    dimension_name: str
    category: MutationCategory
    severity: MutationSeverity
    original_description: str
    mutated_description: str
    rationale: str = ""
    behavioral_tags: list[str] = Field(default_factory=list)
    # Scores
    quality_approved: bool = True

    model_config = {"frozen": True}

    def __repr__(self) -> str:  # pragma: no cover
        return (
            f"EvaluationCase(dim={self.dimension_name!r}, "
            f"sev={self.severity.value}, "
            f"approved={self.quality_approved})"
        )

GeneratedMutation

Bases: BaseModel

Raw output from the mutation generation LLM call.

Source code in mutant/core/mutation.py

class GeneratedMutation(BaseModel):
    """Raw output from the mutation generation LLM call."""

    mutated_description: str
    rationale: str = ""
    behavioral_tags: list[str] = Field(default_factory=list)
    realism_notes: str = ""

MutationPlan

Bases: BaseModel

Planner output — exposed in MutationResult for debugging.

Source code in mutant/core/mutation.py

class MutationPlan(BaseModel):
    """Planner output — exposed in MutationResult for debugging."""

    dimension_allocations: list[DimensionAllocation]
    relevant_dimensions: list[str] = Field(
        default_factory=list,
        description="IDs of dimensions applicable to this scenario",
    )
    irrelevant_dimensions: list[str] = Field(
        default_factory=list, description="IDs of dimensions not applicable"
    )
    coverage_strategy: str
    diversity_strategy: str = ""
    total_planned: int = 0
    expected_failure_modes: list[str] = Field(default_factory=list)

MutationResult

Bases: BaseModel

Complete output of a mutation run.

Source code in mutant/core/mutation.py

class MutationResult(BaseModel):
    """Complete output of a mutation run."""

    cases: list[EvaluationCase]
    behavior_analysis: BehaviorAnalysis | None = None
    mutation_plan: MutationPlan | None = None

    @property
    def count(self) -> int:
        return len(self.cases)

    @property
    def coverage_score(self) -> float:
        if not self.mutation_plan or not self.mutation_plan.relevant_dimensions:
            return 0.0
        explored = {c.dimension_id for c in self.cases}
        return len(explored) / max(len(self.mutation_plan.relevant_dimensions), 1)

    @property
    def summary(self) -> str:
        return f"MutationResult: {len(self.cases)} cases generated. Coverage: {self.coverage_score:.0%}."

    def filter(self, **kwargs: Any) -> MutationResult:
        """Filter cases based on attributes."""
        filtered = self.cases
        for key, value in kwargs.items():
            if key == "dimension":
                filtered = [
                    c
                    for c in filtered
                    if value.lower() in c.dimension_id.lower()
                    or value.lower() in c.dimension_name.lower()
                ]
            elif key == "severity":
                filtered = [
                    c for c in filtered if c.severity.value.lower() == value.lower()
                ]
            elif key == "keyword":
                keyword = value.lower()
                filtered = [
                    c
                    for c in filtered
                    if keyword in c.mutated_description.lower()
                    or keyword in c.rationale.lower()
                ]
            else:
                filtered = [c for c in filtered if getattr(c, key, None) == value]
        return self.model_copy(update={"cases": filtered})

    def explain(self, print_output: bool = True) -> None:
        """Print a structured explanation of the generation process and coverage."""
        if not print_output:
            return

        from rich.console import Console
        from rich.panel import Panel
        from rich.text import Text
        from rich import box

        console = Console()
        console.print()

        header_text = Text.from_markup(
            "[bold cyan]MUTATION GENERATION EXPLANATION[/bold cyan]\n"
            "[dim]Planner Strategy and Coverage Summary[/dim]"
        )
        console.print(
            Panel(header_text, box=box.DOUBLE, border_style="cyan", padding=(1, 4)),
            justify="center"
        )
        console.print()

        if self.mutation_plan:
            console.rule("[bold cyan]PLANNER STRATEGY[/bold cyan]", style="cyan")
            console.print()
            console.print(f"  [dim]Strategy:[/dim]              [bold white]{self.mutation_plan.coverage_strategy}[/bold white]")

            rel = ', '.join(self.mutation_plan.relevant_dimensions) if self.mutation_plan.relevant_dimensions else 'None'
            console.print(f"  [dim]Relevant Dimensions:[/dim]   [white]{rel}[/white]")

            skip = ', '.join(self.mutation_plan.irrelevant_dimensions) if self.mutation_plan.irrelevant_dimensions else 'None'
            console.print(f"  [dim]Skipped Dimensions:[/dim]    [white]{skip}[/white]")
            console.print()

            explored = {c.dimension_id for c in self.cases}
            rel_list = self.mutation_plan.relevant_dimensions or []
            unexplored = [d for d in rel_list if d not in explored]

            console.rule("[bold cyan]COVERAGE SUMMARY[/bold cyan]", style="cyan")
            console.print()

            score_color = "green" if self.coverage_score == 1.0 else ("yellow" if self.coverage_score > 0.5 else "red")
            console.print(f"  [dim]Coverage Score:[/dim]        [{score_color} bold]{self.coverage_score:.0%}[/{score_color} bold]")
            console.print(f"  [dim]Explored:[/dim]              [white]{len(explored)} dimensions out of {len(rel_list)}[/white]")

            if unexplored:
                console.print()
                console.print("  [bold red]Gaps remain in:[/bold red]")
                for gap in unexplored:
                    console.print(f"    [red]•[/red] {gap}")

            console.print()

    # ── Export Methods ────────────────────────────────────────────────────────

    def _to_records(self, **kwargs: Any) -> list[dict[str, Any]]:
        """Convert cases to a flat list of dicts suitable for tabular export."""
        # Default exclusions for cleaner output
        default_excludes = {
            "dimension_name",
            "category",
            "severity",
            "rationale",
            "behavioral_tags",
        }
        exclude = {k for k in default_excludes if kwargs.get(k) is not True}

        # Add explicit exclusions from kwargs
        for k, v in kwargs.items():
            if v is False and k != "mutated_description":
                exclude.add(k)

        records = []
        for case in self.cases:
            record = case.model_dump(exclude=exclude if exclude else None)
            if "category" in record and isinstance(record["category"], Enum):
                record["category"] = record["category"].value
            if "severity" in record and isinstance(record["severity"], Enum):
                record["severity"] = record["severity"].value
            # JSON-ify complex fields for flat tabular formats
            import json

            for key in ["behavioral_tags", "generation_metadata", "metadata"]:
                if record.get(key):
                    record[key] = json.dumps(record[key])
            records.append(record)
        return records

    def to_dataframe(self, **kwargs: Any) -> Any:
        """Convert results to a pandas DataFrame."""
        try:
            import pandas as pd
        except ImportError:
            raise ImportError(
                "pandas is required for to_dataframe(). Install it with `pip install pandas`"
            )
        return pd.DataFrame(self._to_records(**kwargs))

    def to_csv(self, path: str, **kwargs: Any) -> None:
        """Export results to CSV."""
        import csv

        records = self._to_records(**kwargs)
        if not records:
            return
        with open(path, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=records[0].keys())
            writer.writeheader()
            writer.writerows(records)

    def sort_by(self, field: str, descending: bool = False) -> MutationResult:
        """Sort the cases in this result by a specific field."""
        if not self.cases:
            return self

        def _key(c: EvaluationCase) -> Any:
            val = getattr(c, field)
            if isinstance(val, Enum):
                # Simple heuristic for enums (like Severity) to sort by their order
                return list(type(val)).index(val)
            return val

        sorted_cases = sorted(self.cases, key=_key, reverse=descending)
        return self.model_copy(update={"cases": sorted_cases})

    def __len__(self) -> int:
        return len(self.cases)

    def __iter__(self) -> Any:
        return iter(self.cases)

    def __getitem__(self, index: int | slice) -> Any:
        return self.cases[index]

    def to_json(self, path: str, **kwargs: Any) -> None:
        """Export full result object to JSON."""
        exclude = {
            k for k, v in kwargs.items() if v is False and k != "mutated_description"
        }
        exclude_dict = {"cases": {"__all__": exclude}} if exclude else None
        with open(path, "w", encoding="utf-8") as f:
            f.write(self.model_dump_json(indent=2, exclude=exclude_dict))

    def to_jsonl(self, path: str, **kwargs: Any) -> None:
        """Export cases to a JSON Lines file."""
        exclude = {
            k for k, v in kwargs.items() if v is False and k != "mutated_description"
        }
        with open(path, "w", encoding="utf-8") as f:
            for case in self.cases:
                f.write(
                    case.model_dump_json(exclude=exclude if exclude else None) + "\n"
                )

    def to_parquet(self, path: str, **kwargs: Any) -> None:
        """Export results to Parquet."""
        df = self.to_dataframe(**kwargs)
        df.to_parquet(path)

    def to_huggingface(self, **kwargs: Any) -> Any:
        """Convert results to a HuggingFace Dataset."""
        try:
            from datasets import Dataset
        except ImportError:
            raise ImportError(
                "datasets is required for to_huggingface(). Install it with `pip install datasets`"
            )
        return Dataset.from_list(self._to_records(**kwargs))

explain(print_output=True)

Print a structured explanation of the generation process and coverage.

Source code in mutant/core/mutation.py

def explain(self, print_output: bool = True) -> None:
    """Print a structured explanation of the generation process and coverage."""
    if not print_output:
        return

    from rich.console import Console
    from rich.panel import Panel
    from rich.text import Text
    from rich import box

    console = Console()
    console.print()

    header_text = Text.from_markup(
        "[bold cyan]MUTATION GENERATION EXPLANATION[/bold cyan]\n"
        "[dim]Planner Strategy and Coverage Summary[/dim]"
    )
    console.print(
        Panel(header_text, box=box.DOUBLE, border_style="cyan", padding=(1, 4)),
        justify="center"
    )
    console.print()

    if self.mutation_plan:
        console.rule("[bold cyan]PLANNER STRATEGY[/bold cyan]", style="cyan")
        console.print()
        console.print(f"  [dim]Strategy:[/dim]              [bold white]{self.mutation_plan.coverage_strategy}[/bold white]")

        rel = ', '.join(self.mutation_plan.relevant_dimensions) if self.mutation_plan.relevant_dimensions else 'None'
        console.print(f"  [dim]Relevant Dimensions:[/dim]   [white]{rel}[/white]")

        skip = ', '.join(self.mutation_plan.irrelevant_dimensions) if self.mutation_plan.irrelevant_dimensions else 'None'
        console.print(f"  [dim]Skipped Dimensions:[/dim]    [white]{skip}[/white]")
        console.print()

        explored = {c.dimension_id for c in self.cases}
        rel_list = self.mutation_plan.relevant_dimensions or []
        unexplored = [d for d in rel_list if d not in explored]

        console.rule("[bold cyan]COVERAGE SUMMARY[/bold cyan]", style="cyan")
        console.print()

        score_color = "green" if self.coverage_score == 1.0 else ("yellow" if self.coverage_score > 0.5 else "red")
        console.print(f"  [dim]Coverage Score:[/dim]        [{score_color} bold]{self.coverage_score:.0%}[/{score_color} bold]")
        console.print(f"  [dim]Explored:[/dim]              [white]{len(explored)} dimensions out of {len(rel_list)}[/white]")

        if unexplored:
            console.print()
            console.print("  [bold red]Gaps remain in:[/bold red]")
            for gap in unexplored:
                console.print(f"    [red]•[/red] {gap}")

        console.print()

filter(**kwargs)

Filter cases based on attributes.

Source code in mutant/core/mutation.py

def filter(self, **kwargs: Any) -> MutationResult:
    """Filter cases based on attributes."""
    filtered = self.cases
    for key, value in kwargs.items():
        if key == "dimension":
            filtered = [
                c
                for c in filtered
                if value.lower() in c.dimension_id.lower()
                or value.lower() in c.dimension_name.lower()
            ]
        elif key == "severity":
            filtered = [
                c for c in filtered if c.severity.value.lower() == value.lower()
            ]
        elif key == "keyword":
            keyword = value.lower()
            filtered = [
                c
                for c in filtered
                if keyword in c.mutated_description.lower()
                or keyword in c.rationale.lower()
            ]
        else:
            filtered = [c for c in filtered if getattr(c, key, None) == value]
    return self.model_copy(update={"cases": filtered})

sort_by(field, descending=False)

Sort the cases in this result by a specific field.

Source code in mutant/core/mutation.py

def sort_by(self, field: str, descending: bool = False) -> MutationResult:
    """Sort the cases in this result by a specific field."""
    if not self.cases:
        return self

    def _key(c: EvaluationCase) -> Any:
        val = getattr(c, field)
        if isinstance(val, Enum):
            # Simple heuristic for enums (like Severity) to sort by their order
            return list(type(val)).index(val)
        return val

    sorted_cases = sorted(self.cases, key=_key, reverse=descending)
    return self.model_copy(update={"cases": sorted_cases})

to_csv(path, **kwargs)

Export results to CSV.

Source code in mutant/core/mutation.py

def to_csv(self, path: str, **kwargs: Any) -> None:
    """Export results to CSV."""
    import csv

    records = self._to_records(**kwargs)
    if not records:
        return
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=records[0].keys())
        writer.writeheader()
        writer.writerows(records)

to_dataframe(**kwargs)

Convert results to a pandas DataFrame.

Source code in mutant/core/mutation.py

def to_dataframe(self, **kwargs: Any) -> Any:
    """Convert results to a pandas DataFrame."""
    try:
        import pandas as pd
    except ImportError:
        raise ImportError(
            "pandas is required for to_dataframe(). Install it with `pip install pandas`"
        )
    return pd.DataFrame(self._to_records(**kwargs))

to_huggingface(**kwargs)

Convert results to a HuggingFace Dataset.

Source code in mutant/core/mutation.py

def to_huggingface(self, **kwargs: Any) -> Any:
    """Convert results to a HuggingFace Dataset."""
    try:
        from datasets import Dataset
    except ImportError:
        raise ImportError(
            "datasets is required for to_huggingface(). Install it with `pip install datasets`"
        )
    return Dataset.from_list(self._to_records(**kwargs))

to_json(path, **kwargs)

Export full result object to JSON.

Source code in mutant/core/mutation.py

def to_json(self, path: str, **kwargs: Any) -> None:
    """Export full result object to JSON."""
    exclude = {
        k for k, v in kwargs.items() if v is False and k != "mutated_description"
    }
    exclude_dict = {"cases": {"__all__": exclude}} if exclude else None
    with open(path, "w", encoding="utf-8") as f:
        f.write(self.model_dump_json(indent=2, exclude=exclude_dict))

to_jsonl(path, **kwargs)

Export cases to a JSON Lines file.

Source code in mutant/core/mutation.py

def to_jsonl(self, path: str, **kwargs: Any) -> None:
    """Export cases to a JSON Lines file."""
    exclude = {
        k for k, v in kwargs.items() if v is False and k != "mutated_description"
    }
    with open(path, "w", encoding="utf-8") as f:
        for case in self.cases:
            f.write(
                case.model_dump_json(exclude=exclude if exclude else None) + "\n"
            )

to_parquet(path, **kwargs)

Export results to Parquet.

Source code in mutant/core/mutation.py

def to_parquet(self, path: str, **kwargs: Any) -> None:
    """Export results to Parquet."""
    df = self.to_dataframe(**kwargs)
    df.to_parquet(path)

QualityReviewResult

Bases: BaseModel

Output of the quality review stage.

Source code in mutant/core/mutation.py

class QualityReviewResult(BaseModel):
    """Output of the quality review stage."""

    scores: list[QualityScore] = Field(default_factory=list)
    approved_ids: list[str] = Field(default_factory=list)
    rejected_ids: list[str] = Field(default_factory=list)

QualityScore

Bases: BaseModel

LLM quality verdict for a single mutation case.

Source code in mutant/core/mutation.py

class QualityScore(BaseModel):
    """LLM quality verdict for a single mutation case."""

    case_id: str
    preserves_task: bool
    preserves_domain: bool
    meaningful_mutation: bool
    realistic: bool
    sufficiently_different: bool
    approved: bool
    rejection_reason: str | None = None

    @property
    def overall_score(self) -> float:
        return 1.0 if self.approved else 0.0

Providers

mutant.providers.base

mutant/providers/base.py

Provider-agnostic LLM abstraction.

No provider-specific code leaks beyond this boundary. The rest of Mutant only ever sees BaseLLMProvider, LLMMessage, and LLMResponse.

BaseLLMProvider

Bases: ABC

Abstract base class for all LLM providers.

Implementors must only override complete(). JSON extraction and structured parsing are handled by this base class.

Example

provider = OpenAIProvider(api_key="sk-...") response = await provider.complete([ ... LLMMessage(role="user", content="Hello!") ... ]) print(response.content)

Source code in mutant/providers/base.py

class BaseLLMProvider(ABC):
    """Abstract base class for all LLM providers.

    Implementors must only override ``complete()``. JSON extraction and
    structured parsing are handled by this base class.

    Example
    -------
    >>> provider = OpenAIProvider(api_key="sk-...")
    >>> response = await provider.complete([
    ...     LLMMessage(role="user", content="Hello!")
    ... ])
    >>> print(response.content)
    """

    # Subclasses should set this.
    provider_name: str = "unknown"

    @abstractmethod
    async def complete(
        self,
        messages: list[LLMMessage],
        *,
        temperature: float = 0.8,
        max_tokens: int = 4096,
    ) -> LLMResponse:
        """Send messages to the provider and return its response.

        Parameters
        ----------
        messages:
            Conversation history. Use ``role="system"`` for system prompts.
        temperature:
            Sampling temperature (0.0 = deterministic, 1.0 = creative).
        max_tokens:
            Maximum tokens in the response.
        """

    async def complete_json(
        self,
        messages: list[LLMMessage],
        schema: type[T],
        *,
        temperature: float = 0.7,
        max_tokens: int = 4096,
        max_retries: int = 3,
    ) -> T:
        """Complete and parse the response as a Pydantic model.

        The default implementation appends a JSON reminder to the last user
        message, completes, then extracts and validates JSON. Providers that
        natively support structured output (e.g. OpenAI ``response_format``)
        can override this for reliability.

        Parameters
        ----------
        messages:
            Conversation messages.
        schema:
            The Pydantic ``BaseModel`` subclass to parse into.

        Raises
        ------
        ParseError
            If the response cannot be parsed as valid JSON or validated
            against ``schema``.
        """
        last_error = None
        for _attempt in range(max_retries):
            response = await self.complete(
                messages, temperature=temperature, max_tokens=max_tokens
            )
            try:
                return self._parse_json(response.content, schema)
            except ParseError as e:
                last_error = e
                # Adjust temperature slightly for retries to get a different result
                temperature = min(1.0, temperature + 0.1)

        raise last_error  # type: ignore

    # ── Internal helpers ───────────────────────────────────────────────────────

    @staticmethod
    def _parse_json(content: str, schema: type[T]) -> T:
        """Extract JSON from content and validate against schema."""
        # Strip markdown fences if present
        text = content.strip()
        if text.startswith("```"):
            lines = text.split("\n")
            # Drop first and last fence lines
            inner = (
                "\n".join(lines[1:-1])
                if lines[-1].strip() == "```"
                else "\n".join(lines[1:])
            )
            text = inner.strip()

        try:
            data = json.loads(text)
        except json.JSONDecodeError as exc:
            # Try to find JSON block within free text
            import re

            match = re.search(r"\{.*\}", text, re.DOTALL)
            if match:
                try:
                    data = json.loads(match.group())
                except json.JSONDecodeError:
                    raise ParseError(
                        f"Could not parse JSON from LLM response: {exc}",
                        raw_content=content,
                    ) from exc
            else:
                raise ParseError(
                    f"No JSON found in LLM response: {exc}",
                    raw_content=content,
                ) from exc

        try:
            return schema.model_validate(data)
        except Exception as exc:
            raise ParseError(
                f"LLM response did not match schema {schema.__name__}: {exc}",
                raw_content=content,
            ) from exc

    def __repr__(self) -> str:  # pragma: no cover
        return f"{self.__class__.__name__}(provider={self.provider_name!r})"

complete(messages, *, temperature=0.8, max_tokens=4096) abstractmethod async

Send messages to the provider and return its response.

Parameters:

Name	Type	Description	Default
messages	list[LLMMessage]	Conversation history. Use role="system" for system prompts.	required
temperature	float	Sampling temperature (0.0 = deterministic, 1.0 = creative).	0.8
max_tokens	int	Maximum tokens in the response.	4096

Source code in mutant/providers/base.py

@abstractmethod
async def complete(
    self,
    messages: list[LLMMessage],
    *,
    temperature: float = 0.8,
    max_tokens: int = 4096,
) -> LLMResponse:
    """Send messages to the provider and return its response.

    Parameters
    ----------
    messages:
        Conversation history. Use ``role="system"`` for system prompts.
    temperature:
        Sampling temperature (0.0 = deterministic, 1.0 = creative).
    max_tokens:
        Maximum tokens in the response.
    """

complete_json(messages, schema, *, temperature=0.7, max_tokens=4096, max_retries=3) async

Complete and parse the response as a Pydantic model.

The default implementation appends a JSON reminder to the last user message, completes, then extracts and validates JSON. Providers that natively support structured output (e.g. OpenAI response_format) can override this for reliability.

Parameters:

Name	Type	Description	Default
messages	list[LLMMessage]	Conversation messages.	required
schema	type[T]	The Pydantic BaseModel subclass to parse into.	required

Raises:

Type	Description
ParseError	If the response cannot be parsed as valid JSON or validated against schema.

Source code in mutant/providers/base.py

async def complete_json(
    self,
    messages: list[LLMMessage],
    schema: type[T],
    *,
    temperature: float = 0.7,
    max_tokens: int = 4096,
    max_retries: int = 3,
) -> T:
    """Complete and parse the response as a Pydantic model.

    The default implementation appends a JSON reminder to the last user
    message, completes, then extracts and validates JSON. Providers that
    natively support structured output (e.g. OpenAI ``response_format``)
    can override this for reliability.

    Parameters
    ----------
    messages:
        Conversation messages.
    schema:
        The Pydantic ``BaseModel`` subclass to parse into.

    Raises
    ------
    ParseError
        If the response cannot be parsed as valid JSON or validated
        against ``schema``.
    """
    last_error = None
    for _attempt in range(max_retries):
        response = await self.complete(
            messages, temperature=temperature, max_tokens=max_tokens
        )
        try:
            return self._parse_json(response.content, schema)
        except ParseError as e:
            last_error = e
            # Adjust temperature slightly for retries to get a different result
            temperature = min(1.0, temperature + 0.1)

    raise last_error  # type: ignore

LLMMessage

Bases: BaseModel

A single message in an LLM conversation.

Source code in mutant/providers/base.py

class LLMMessage(BaseModel):
    """A single message in an LLM conversation."""

    role: Literal["system", "user", "assistant"]
    content: str

    model_config = {"frozen": True}

LLMResponse

Bases: BaseModel

The response from an LLM provider.

Source code in mutant/providers/base.py

class LLMResponse(BaseModel):
    """The response from an LLM provider."""

    content: str
    model: str
    input_tokens: int | None = None
    output_tokens: int | None = None
    metadata: dict[str, Any] = {}

    model_config = {"frozen": True}

    @property
    def total_tokens(self) -> int | None:
        if self.input_tokens is not None and self.output_tokens is not None:
            return self.input_tokens + self.output_tokens
        return None

ParseError

Bases: Exception

Raised when structured output parsing fails.

Source code in mutant/providers/base.py

class ParseError(Exception):
    """Raised when structured output parsing fails."""

    def __init__(self, message: str, *, raw_content: str) -> None:
        super().__init__(message)
        self.raw_content = raw_content

ProviderError

Bases: Exception

Raised when an LLM provider returns an error.

Source code in mutant/providers/base.py

class ProviderError(Exception):
    """Raised when an LLM provider returns an error."""

    def __init__(
        self, message: str, *, provider: str, status_code: int | None = None
    ) -> None:
        super().__init__(message)
        self.provider = provider
        self.status_code = status_code

mutant.providers.gemini

Gemini provider implementation.

GeminiProvider

Bases: BaseLLMProvider

LLM provider for Google Gemini models.

Requires the gemini extra: pip install mutant-ai[gemini]

Parameters:

Name	Type	Description	Default
api_key	str | None	Google API key. Defaults to GOOGLE_API_KEY env var.	None
model	str	Model identifier. Default: "gemini-1.5-flash".	'gemini-1.5-flash'

Source code in mutant/providers/gemini.py

class GeminiProvider(BaseLLMProvider):
    """LLM provider for Google Gemini models.

    Requires the ``gemini`` extra: ``pip install mutant-ai[gemini]``

    Parameters
    ----------
    api_key:
        Google API key. Defaults to ``GOOGLE_API_KEY`` env var.
    model:
        Model identifier. Default: ``"gemini-1.5-flash"``.
    """

    provider_name = "gemini"

    # Global state for free-tier rate limiting
    _global_lock = None
    _last_request_time = 0.0

    def __init__(
        self,
        *,
        api_key: str | None = None,
        model: str = "gemini-1.5-flash",
        **kwargs: Any,
    ) -> None:
        try:
            from google import genai
        except ImportError as exc:
            raise ImportError(
                "Gemini provider requires the 'google-genai' package. "
                "Install it with: pip install google-genai"
            ) from exc

        self.client = genai.Client(api_key=api_key)
        self.model_name = model
        self._genai = genai

    async def complete(
        self,
        messages: list[LLMMessage],
        *,
        temperature: float = 0.8,
        max_tokens: int = 4096,
    ) -> LLMResponse:

        # Build prompt from messages
        parts: list[str] = []
        for m in messages:
            prefix = "" if m.role == "user" else f"[{m.role.upper()}] "
            parts.append(prefix + m.content)
        prompt = "\n\n".join(parts)

        import asyncio
        import time

        from tenacity import (
            retry,
            retry_if_exception,
            stop_after_attempt,
            wait_random_exponential,
        )

        # Initialize the global lock once per event loop
        if GeminiProvider._global_lock is None:
            GeminiProvider._global_lock = asyncio.Lock()

        def is_retryable(exc: BaseException) -> bool:
            exc_str = str(exc).lower()
            return (
                "429" in exc_str
                or "quota" in exc_str
                or "resourceexhausted" in exc_str
                or "too many requests" in exc_str
            )

        def before_sleep_print(retry_state: Any) -> None:
            exc = retry_state.outcome.exception()
            print(
                f"[WARNING] Gemini API Error ({type(exc).__name__}: {exc}). Sleeping {retry_state.next_action.sleep:.1f}s before retry "
                f"(Attempt {retry_state.attempt_number}/10)..."
            )

        @retry(
            wait=wait_random_exponential(multiplier=2, max=65),
            stop=stop_after_attempt(10),
            retry=retry_if_exception(is_retryable),
            before_sleep=before_sleep_print,
            reraise=True,
        )
        async def _generate() -> LLMResponse:
            try:
                # Enforce a strict global rate limit for free tier (approx 1 request per 4 seconds)
                async with GeminiProvider._global_lock:
                    now = time.monotonic()
                    time_since_last = now - GeminiProvider._last_request_time
                    if time_since_last < 4.5:
                        await asyncio.sleep(4.5 - time_since_last)

                    GeminiProvider._last_request_time = time.monotonic()

                response = await self.client.aio.models.generate_content(
                    model=self.model_name,
                    contents=prompt,
                    config=self._genai.types.GenerateContentConfig(
                        temperature=temperature,
                        max_output_tokens=max_tokens,
                    ),
                )
                return LLMResponse(
                    content=response.text,
                    model=self.model_name,
                )
            except Exception as e:
                if is_retryable(e):
                    raise
                raise ProviderError(
                    f"Gemini request failed: {e}",
                    provider=self.provider_name,
                ) from e

        return await _generate()

mutant.providers.openai

OpenAI provider implementation.

OpenAIProvider

Bases: BaseLLMProvider

LLM provider for OpenAI (GPT-4o, GPT-4-turbo, o1, etc.).

Requires the openai extra: pip install mutant-ai[openai]

Parameters:

Name	Type	Description	Default
api_key	str | None	OpenAI API key. Defaults to OPENAI_API_KEY environment variable.	None
model	str	Model identifier. Default: "gpt-4o-mini".	'gpt-4o-mini'
base_url	str | None	Override for OpenAI-compatible endpoints (e.g. Azure, local proxies).	None
default_headers	dict[str, str] | None	Extra headers to send with every request.	None

Example

provider = OpenAIProvider(model="gpt-4o") cases = await mutate(scenario, provider=provider, count=50)

Source code in mutant/providers/openai.py

class OpenAIProvider(BaseLLMProvider):
    """LLM provider for OpenAI (GPT-4o, GPT-4-turbo, o1, etc.).

    Requires the ``openai`` extra: ``pip install mutant-ai[openai]``

    Parameters
    ----------
    api_key:
        OpenAI API key. Defaults to ``OPENAI_API_KEY`` environment variable.
    model:
        Model identifier. Default: ``"gpt-4o-mini"``.
    base_url:
        Override for OpenAI-compatible endpoints (e.g. Azure, local proxies).
    default_headers:
        Extra headers to send with every request.

    Example
    -------
    >>> provider = OpenAIProvider(model="gpt-4o")
    >>> cases = await mutate(scenario, provider=provider, count=50)
    """

    provider_name = "openai"

    def __init__(
        self,
        *,
        api_key: str | None = None,
        model: str = "gpt-4o-mini",
        base_url: str | None = None,
        default_headers: dict[str, str] | None = None,
        **kwargs: Any,
    ) -> None:
        try:
            import openai  # noqa: F401
        except ImportError as exc:
            raise ImportError(
                "OpenAI provider requires the 'openai' package. "
                "Install it with: pip install mutant-ai[openai]"
            ) from exc

        import openai as _openai

        self.model = model
        self._client = _openai.AsyncOpenAI(
            api_key=api_key,
            base_url=base_url,
            default_headers=default_headers or {},
        )

    async def complete(
        self,
        messages: list[LLMMessage],
        *,
        temperature: float = 0.8,
        max_tokens: int = 4096,
    ) -> LLMResponse:
        try:
            response = await self._client.chat.completions.create(
                model=self.model,
                messages=[{"role": m.role, "content": m.content} for m in messages],
                temperature=temperature,
                max_tokens=max_tokens,
            )
            choice = response.choices[0]
            usage = response.usage
            return LLMResponse(
                content=choice.message.content or "",
                model=response.model,
                input_tokens=usage.prompt_tokens if usage else None,
                output_tokens=usage.completion_tokens if usage else None,
            )
        except Exception as exc:
            raise ProviderError(
                f"OpenAI request failed: {exc}",
                provider=self.provider_name,
            ) from exc

    async def complete_json(
        self,
        messages: list[LLMMessage],
        schema: type[T],
        *,
        temperature: float = 0.7,
        max_tokens: int = 4096,
    ) -> T:
        """Uses OpenAI JSON mode for reliable structured output."""
        try:
            response = await self._client.chat.completions.create(
                model=self.model,
                messages=[{"role": m.role, "content": m.content} for m in messages],
                temperature=temperature,
                max_tokens=max_tokens,
                response_format={"type": "json_object"},
            )
            content = response.choices[0].message.content or "{}"
            return self._parse_json(content, schema)
        except ParseError:
            raise
        except Exception as exc:
            raise ProviderError(
                f"OpenAI JSON request failed: {exc}",
                provider=self.provider_name,
            ) from exc

complete_json(messages, schema, *, temperature=0.7, max_tokens=4096) async

Uses OpenAI JSON mode for reliable structured output.

Source code in mutant/providers/openai.py

async def complete_json(
    self,
    messages: list[LLMMessage],
    schema: type[T],
    *,
    temperature: float = 0.7,
    max_tokens: int = 4096,
) -> T:
    """Uses OpenAI JSON mode for reliable structured output."""
    try:
        response = await self._client.chat.completions.create(
            model=self.model,
            messages=[{"role": m.role, "content": m.content} for m in messages],
            temperature=temperature,
            max_tokens=max_tokens,
            response_format={"type": "json_object"},
        )
        content = response.choices[0].message.content or "{}"
        return self._parse_json(content, schema)
    except ParseError:
        raise
    except Exception as exc:
        raise ProviderError(
            f"OpenAI JSON request failed: {exc}",
            provider=self.provider_name,
        ) from exc

mutant.providers.anthropic

Anthropic (Claude) provider implementation.

AnthropicProvider

Bases: BaseLLMProvider

LLM provider for Anthropic Claude models.

Requires the anthropic extra: pip install mutant-ai[anthropic]

Parameters:

Name	Type	Description	Default
api_key	str | None	Anthropic API key. Defaults to ANTHROPIC_API_KEY env var.	None
model	str	Model identifier. Default: "claude-3-5-haiku-20241022".	'claude-3-5-haiku-20241022'

Example

provider = AnthropicProvider(model="claude-3-5-sonnet-20241022") cases = await mutate(scenario, provider=provider, count=50)

Source code in mutant/providers/anthropic.py

class AnthropicProvider(BaseLLMProvider):
    """LLM provider for Anthropic Claude models.

    Requires the ``anthropic`` extra: ``pip install mutant-ai[anthropic]``

    Parameters
    ----------
    api_key:
        Anthropic API key. Defaults to ``ANTHROPIC_API_KEY`` env var.
    model:
        Model identifier. Default: ``"claude-3-5-haiku-20241022"``.

    Example
    -------
    >>> provider = AnthropicProvider(model="claude-3-5-sonnet-20241022")
    >>> cases = await mutate(scenario, provider=provider, count=50)
    """

    provider_name = "anthropic"

    def __init__(
        self,
        *,
        api_key: str | None = None,
        model: str = "claude-3-5-haiku-20241022",
        **kwargs: Any,
    ) -> None:
        try:
            import anthropic  # noqa: F401
        except ImportError as exc:
            raise ImportError(
                "Anthropic provider requires the 'anthropic' package. "
                "Install it with: pip install mutant-ai[anthropic]"
            ) from exc

        import anthropic as _anthropic

        self.model = model
        self._client = _anthropic.AsyncAnthropic(api_key=api_key)

    async def complete(
        self,
        messages: list[LLMMessage],
        *,
        temperature: float = 0.8,
        max_tokens: int = 4096,
    ) -> LLMResponse:
        # Separate system message from the rest
        system_content = ""
        user_messages = []
        for m in messages:
            if m.role == "system":
                system_content = m.content
            else:
                user_messages.append({"role": m.role, "content": m.content})

        try:
            kwargs: dict[str, Any] = {
                "model": self.model,
                "max_tokens": max_tokens,
                "temperature": temperature,
                "messages": user_messages,
            }
            if system_content:
                kwargs["system"] = system_content

            response = await self._client.messages.create(**kwargs)
            content = response.content[0].text if response.content else ""
            return LLMResponse(
                content=content,
                model=response.model,
                input_tokens=response.usage.input_tokens,
                output_tokens=response.usage.output_tokens,
            )
        except Exception as exc:
            raise ProviderError(
                f"Anthropic request failed: {exc}",
                provider=self.provider_name,
            ) from exc

mutant.providers.litellm

LiteLLM provider — unified proxy for 100+ models.

LiteLLMProvider

Bases: BaseLLMProvider

LLM provider that wraps LiteLLM for access to 100+ models.

Requires the litellm extra: pip install mutant-ai[litellm]

LiteLLM supports: OpenAI, Anthropic, Gemini, Azure, Cohere, Mistral, Together, Replicate, and many more — using a unified OpenAI-compatible API.

Parameters:

Name	Type	Description	Default
model	str	LiteLLM model string e.g. "gpt-4o", "claude-3-5-sonnet-20241022", "gemini/gemini-1.5-pro", "ollama/llama3".	'gpt-4o-mini'
api_key	str | None	API key (if not set via environment).	None
kwargs	Any	Any additional kwargs passed to litellm.acompletion.	{}

Example

provider = LiteLLMProvider(model="claude-3-5-sonnet-20241022") cases = await mutate(scenario, provider=provider, count=50)

Source code in mutant/providers/litellm.py

class LiteLLMProvider(BaseLLMProvider):
    """LLM provider that wraps LiteLLM for access to 100+ models.

    Requires the ``litellm`` extra: ``pip install mutant-ai[litellm]``

    LiteLLM supports: OpenAI, Anthropic, Gemini, Azure, Cohere, Mistral,
    Together, Replicate, and many more — using a unified OpenAI-compatible API.

    Parameters
    ----------
    model:
        LiteLLM model string e.g. ``"gpt-4o"``, ``"claude-3-5-sonnet-20241022"``,
        ``"gemini/gemini-1.5-pro"``, ``"ollama/llama3"``.
    api_key:
        API key (if not set via environment).
    kwargs:
        Any additional kwargs passed to ``litellm.acompletion``.

    Example
    -------
    >>> provider = LiteLLMProvider(model="claude-3-5-sonnet-20241022")
    >>> cases = await mutate(scenario, provider=provider, count=50)
    """

    provider_name = "litellm"

    def __init__(
        self,
        *,
        model: str = "gpt-4o-mini",
        api_key: str | None = None,
        **kwargs: Any,
    ) -> None:
        try:
            import litellm  # noqa: F401
        except ImportError as exc:
            raise ImportError(
                "LiteLLM provider requires the 'litellm' package. "
                "Install it with: pip install mutant-ai[litellm]"
            ) from exc

        self.model = model
        self._api_key = api_key
        self._extra_kwargs = kwargs

    async def complete(
        self,
        messages: list[LLMMessage],
        *,
        temperature: float = 0.8,
        max_tokens: int = 4096,
    ) -> LLMResponse:
        try:
            import litellm

            kwargs: dict[str, Any] = {
                "model": self.model,
                "messages": [{"role": m.role, "content": m.content} for m in messages],
                "temperature": temperature,
                "max_tokens": max_tokens,
                **self._extra_kwargs,
            }
            if self._api_key:
                kwargs["api_key"] = self._api_key

            response = await litellm.acompletion(**kwargs)
            content = response.choices[0].message.content or ""
            usage = response.usage
            return LLMResponse(
                content=content,
                model=response.model or self.model,
                input_tokens=getattr(usage, "prompt_tokens", None),
                output_tokens=getattr(usage, "completion_tokens", None),
            )
        except Exception as exc:
            raise ProviderError(
                f"LiteLLM request failed: {exc}",
                provider=self.provider_name,
            ) from exc

Reports

mutant.reports.html

HTML report generator.

HtmlReport

Generates a self-contained HTML report.

Example

report = HtmlReport() report.save(scenario, cases, path="report.html")

Source code in mutant/reports/html.py

class HtmlReport:
    """Generates a self-contained HTML report.

    Example
    -------
    >>> report = HtmlReport()
    >>> report.save(scenario, cases, path="report.html")
    """

    def render(self, scenario: Scenario, cases: list[MutationCase]) -> str:
        data = _build_report_dict(scenario, cases)
        return render_prompt("mutant_report.html", data=data)

    def save(
        self, scenario: Scenario, cases: list[MutationCase], path: str | Path
    ) -> Path:
        output = Path(path)
        output.parent.mkdir(parents=True, exist_ok=True)
        output.write_text(self.render(scenario, cases), encoding="utf-8")
        return output

mutant.reports.json

JSON and Markdown report generators.

JsonReport

Serialises mutation results to JSON.

Example

report = JsonReport() report.save(scenario, cases, path="report.json")

Source code in mutant/reports/json.py

class JsonReport:
    """Serialises mutation results to JSON.

    Example
    -------
    >>> report = JsonReport()
    >>> report.save(scenario, cases, path="report.json")
    """

    def render(self, scenario: Scenario, cases: list[MutationCase]) -> str:
        """Return the report as a JSON string."""
        data = _build_report_dict(scenario, cases)
        return json.dumps(data, indent=2, ensure_ascii=False)

    def save(
        self, scenario: Scenario, cases: list[MutationCase], path: str | Path
    ) -> Path:
        """Render and save the report to ``path``. Returns the resolved path."""
        output = Path(path)
        output.parent.mkdir(parents=True, exist_ok=True)
        output.write_text(self.render(scenario, cases), encoding="utf-8")
        return output

render(scenario, cases)

Return the report as a JSON string.

Source code in mutant/reports/json.py

def render(self, scenario: Scenario, cases: list[MutationCase]) -> str:
    """Return the report as a JSON string."""
    data = _build_report_dict(scenario, cases)
    return json.dumps(data, indent=2, ensure_ascii=False)

save(scenario, cases, path)

Render and save the report to path. Returns the resolved path.

Source code in mutant/reports/json.py

def save(
    self, scenario: Scenario, cases: list[MutationCase], path: str | Path
) -> Path:
    """Render and save the report to ``path``. Returns the resolved path."""
    output = Path(path)
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(self.render(scenario, cases), encoding="utf-8")
    return output

MarkdownReport

Serialises mutation results to a Markdown document.

Example

report = MarkdownReport() report.save(scenario, cases, path="report.md")

Source code in mutant/reports/json.py

class MarkdownReport:
    """Serialises mutation results to a Markdown document.

    Example
    -------
    >>> report = MarkdownReport()
    >>> report.save(scenario, cases, path="report.md")
    """

    def render(self, scenario: Scenario, cases: list[MutationCase]) -> str:
        data = _build_report_dict(scenario, cases)
        lines: list[str] = [
            f"# Mutant Report — {data['scenario']['title']}",
            "",
            f"**Generated:** {data['generated_at']}  ",
            f"**Report ID:** `{data['report_id']}`",
            "",
            "## Original Scenario",
            "",
            f"> {data['scenario']['description']}",
            "",
            "## Summary",
            "",
            f"- **Total mutations:** {data['summary']['total']}",
            "",
            "### By Category",
            "",
        ]
        for cat, count in sorted(data["summary"]["by_category"].items()):
            lines.append(f"- `{cat}`: {count}")
        lines += [
            "",
            "### By Severity",
            "",
        ]
        for sev, count in sorted(data["summary"]["by_severity"].items()):
            lines.append(f"- `{sev}`: {count}")
        lines += [
            "",
            "---",
            "",
            "## Mutation Cases",
            "",
        ]
        for i, case in enumerate(data["cases"], 1):
            lines += [
                f"### {i}. {case['dimension_name']}",
                "",
                "| Field | Value |",
                "|-------|-------|",
                f"| **ID** | `{case['id']}` |",
                f"| **Dimension** | `{case['dimension_id']}` |",
                f"| **Category** | `{case['category']}` |",
                f"| **Severity** | `{case['severity']}` |",
                "",
                "**Rationale:**",
                "",
                f"> {case['rationale']}",
                "",
                "**Original:**",
                "",
                f"> {case['original']}",
                "",
                "**Mutated:**",
                "",
                f"> {case['mutated']}",
                "",
            ]
        return "\n".join(lines)

    def save(
        self, scenario: Scenario, cases: list[MutationCase], path: str | Path
    ) -> Path:
        """Render and save the report to ``path``. Returns the resolved path."""
        output = Path(path)
        output.parent.mkdir(parents=True, exist_ok=True)
        output.write_text(self.render(scenario, cases), encoding="utf-8")
        return output

save(scenario, cases, path)

Render and save the report to path. Returns the resolved path.

Source code in mutant/reports/json.py

def save(
    self, scenario: Scenario, cases: list[MutationCase], path: str | Path
) -> Path:
    """Render and save the report to ``path``. Returns the resolved path."""
    output = Path(path)
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(self.render(scenario, cases), encoding="utf-8")
    return output