API Reference¶

Auto-generated API documentation via mkdocstrings. The pages below pull docstrings directly from source, so they stay in sync automatically as the codebase evolves.

Core¶

`promptcanary.core.suite.CanarySuite(name, prompts, probes, *, description='', tags=None, default_system_prompt=None)` ¶

Holds a collection of prompts and probes, and drives the canary run.

Parameters:

Name	Type	Description	Default
`name`	`str`	Human-readable name for this suite.	required
`prompts`	`list[CanaryPrompt]`	List of :class:`CanaryPrompt` objects.	required
`probes`	`list[BaseProbe]`	List of :class:`BaseProbe` instances to run on every response.	required
`description`	`str`	Optional description shown in reports.	`''`
`tags`	`list[str] \| None`	Optional tags for categorisation.	`None`

Example::

from promptcanary import CanarySuite, LiteLLMProvider
from promptcanary.core.probes import JsonValidityProbe, StepByStepProbe

suite = CanarySuite(
    name="production-agent",
    prompts=[CanaryPrompt(text="Return JSON: {name: 'Alice', age: 30}")],
    probes=[JsonValidityProbe(), StepByStepProbe(expect_steps=False)],
)
provider = LiteLLMProvider("openai/gpt-4o-mini")
result = suite.run(provider)
print(result.overall_score)

Source code in promptcanary/core/suite.py

def __init__(
    self,
    name: str,
    prompts: list[CanaryPrompt],
    probes: list[BaseProbe],
    *,
    description: str = "",
    tags: list[str] | None = None,
    default_system_prompt: str | None = None,
) -> None:
    if not prompts:
        raise ValueError("CanarySuite requires at least one prompt.")
    if not probes:
        raise ValueError("CanarySuite requires at least one probe.")

    self.name = name
    self.prompts = prompts
    self.probes = probes
    self.description = description
    self.tags = tags or []
    self.default_system_prompt = default_system_prompt

`from_yaml(path)` `classmethod` ¶

Load a CanarySuite from a YAML configuration file.

Expected YAML structure::

name: my-suite
description: Tests production agent behaviour.
probes:
  - type: json_validity
  - type: step_by_step
    expect_steps: false
  - type: keyword_presence
    required_keywords: ["Paris"]
prompts:
  - text: "What is the capital of France?"
    expected_keywords: ["Paris"]
    description: "Basic geography canary"

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to the `canary.yaml` file.	required

Returns:

Type	Description
`CanarySuite`	A fully configured :class:`CanarySuite`.

Source code in promptcanary/core/suite.py

@classmethod
def from_yaml(cls, path: str | Path) -> CanarySuite:
    """Load a CanarySuite from a YAML configuration file.

    Expected YAML structure::

        name: my-suite
        description: Tests production agent behaviour.
        probes:
          - type: json_validity
          - type: step_by_step
            expect_steps: false
          - type: keyword_presence
            required_keywords: ["Paris"]
        prompts:
          - text: "What is the capital of France?"
            expected_keywords: ["Paris"]
            description: "Basic geography canary"

    Args:
        path: Path to the ``canary.yaml`` file.

    Returns:
        A fully configured :class:`CanarySuite`.
    """
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"Canary config not found: {p}")

    with p.open("r", encoding="utf-8") as f:
        config: dict[str, Any] = yaml.safe_load(f)

    return cls._from_dict(config)

`run(provider, *, temperature=None, max_tokens=None, seed=None, show_progress=True)` ¶

Run all prompts against the provider, apply all probes, return results.

Parameters:

Name	Type	Description	Default
`provider`	`BaseLLMProvider`	The LLM provider to query.	required
`temperature`	`float \| None`	Override default temperature (0.0 recommended).	`None`
`max_tokens`	`int \| None`	Override max_tokens.	`None`
`seed`	`int \| None`	Override seed for reproducibility.	`None`
`show_progress`	`bool`	Whether to show a Rich progress bar.	`True`

Returns:

Type	Description
`CanaryRunResult`	A fully-populated :class:`CanaryRunResult`.

Source code in promptcanary/core/suite.py

def run(
    self,
    provider: BaseLLMProvider,
    *,
    temperature: float | None = None,
    max_tokens: int | None = None,
    seed: int | None = None,
    show_progress: bool = True,
) -> CanaryRunResult:
    """Run all prompts against the provider, apply all probes, return results.

    Args:
        provider:      The LLM provider to query.
        temperature:   Override default temperature (0.0 recommended).
        max_tokens:    Override max_tokens.
        seed:          Override seed for reproducibility.
        show_progress: Whether to show a Rich progress bar.

    Returns:
        A fully-populated :class:`CanaryRunResult`.
    """
    from rich.progress import Progress, SpinnerColumn, TextColumn, TimeElapsedColumn

    # Build a ProviderConfig for this run
    provider_cfg = provider.config
    if temperature is not None or max_tokens is not None or seed is not None:
        provider_cfg = ProviderConfig(
            model_id=provider_cfg.model_id,
            temperature=temperature if temperature is not None else provider_cfg.temperature,
            max_tokens=max_tokens if max_tokens is not None else provider_cfg.max_tokens,
            seed=seed if seed is not None else provider_cfg.seed,
            extra_params=provider_cfg.extra_params,
        )

    result = CanaryRunResult(
        suite_name=self.name,
        provider=provider_cfg,
    )

    total = len(self.prompts)
    all_responses = []
    all_probe_results = []

    progress_ctx = (
        Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            TimeElapsedColumn(),
        )
        if show_progress
        else _NoopContext()
    )

    with progress_ctx as progress:
        task = progress.add_task(
            f"[cyan]Running {total} prompt(s) x {len(self.probes)} probe(s)…",
            total=total,
        )

        for i, prompt in enumerate(self.prompts, 1):
            # ── Call provider ─────────────────────────────────────────────
            system = prompt.system_prompt or self.default_system_prompt
            t0 = time.perf_counter()
            llm_response = provider.complete(prompt, system_prompt=system)
            latency_ms = (time.perf_counter() - t0) * 1000

            # Patch latency if provider didn't set it
            if llm_response.latency_ms is None:
                # Rebuild with latency (frozen model)
                llm_response = llm_response.model_copy(
                    update={"latency_ms": round(latency_ms, 2)}
                )

            all_responses.append(llm_response)

            # ── Apply probes ──────────────────────────────────────────────
            for probe in self.probes:
                try:
                    probe_result = probe.evaluate(prompt, llm_response)
                except Exception as exc:
                    # Probe errors become scored failures, never crashes
                    from promptcanary.core.models import ProbeResult

                    probe_result = ProbeResult(
                        probe_id=probe.probe_id,
                        probe_name=probe.name,
                        category=probe.category,
                        prompt_id=prompt.id,
                        passed=False,
                        score=0.0,
                        details=f"Probe raised an exception: {type(exc).__name__}: {exc}",
                    )
                all_probe_results.append(probe_result)

            progress.update(task, advance=1, description=f"[cyan]Prompt {i}/{total}…")

    # Populate result (Pydantic frozen workaround: reassign)
    result.probe_results.extend(all_probe_results)
    result.llm_responses.extend(all_responses)
    result.finished_at = datetime.now(timezone.utc)

    return result

`to_yaml_template()` ¶

Render this suite back to a YAML config string (useful for init).

Source code in promptcanary/core/suite.py

def to_yaml_template(self) -> str:
    """Render this suite back to a YAML config string (useful for init)."""
    import yaml as _yaml

    data: dict[str, Any] = {
        "name": self.name,
        "description": self.description or "",
        "tags": self.tags,
        "probes": [{"type": p.probe_id} for p in self.probes],
        "prompts": [
            {
                "text": cp.text,
                "description": cp.description,
                "tags": cp.tags,
            }
            for cp in self.prompts
        ],
    }
    return _yaml.dump(data, default_flow_style=False, allow_unicode=True)

`promptcanary.core.comparator.compare(baseline, current, *, regression_threshold=0.05, improvement_threshold=0.05)` ¶

Compare a current run against a saved baseline snapshot.

Parameters:

Name	Type	Description	Default
`baseline`	`BaselineSnapshot`	Saved :class:`BaselineSnapshot`.	required
`current`	`CanaryRunResult`	Fresh :class:`CanaryRunResult` from the same suite.	required
`regression_threshold`	`float`	Minimum score drop (Δ) to declare a regression.	`0.05`
`improvement_threshold`	`float`	Minimum score gain (Δ) to declare an improvement.	`0.05`

Returns:

Name	Type	Description
`A`	`DriftReport`	class:`DriftReport` with full comparison data.

Raises:

Type	Description
`ValueError`	If suite names do not match (guards against comparing incompatible suites).

Source code in promptcanary/core/comparator.py

def compare(
    baseline: BaselineSnapshot,
    current: CanaryRunResult,
    *,
    regression_threshold: float = 0.05,
    improvement_threshold: float = 0.05,
) -> DriftReport:
    """Compare a current run against a saved baseline snapshot.

    Args:
        baseline:               Saved :class:`BaselineSnapshot`.
        current:                Fresh :class:`CanaryRunResult` from the same suite.
        regression_threshold:   Minimum score drop (Δ) to declare a regression.
        improvement_threshold:  Minimum score gain (Δ) to declare an improvement.

    Returns:
        A :class:`DriftReport` with full comparison data.

    Raises:
        ValueError: If suite names do not match (guards against comparing incompatible suites).
    """
    if baseline.suite_name != current.suite_name:
        raise ValueError(
            f"Suite name mismatch: baseline is for '{baseline.suite_name}' "
            f"but current run is for '{current.suite_name}'. "
            "Cannot compare results from different suites."
        )

    # Index baseline probe results by (probe_id, prompt_id)
    baseline_index: dict[tuple[str, str], ProbeResult] = {
        (r.probe_id, r.prompt_id): r for r in baseline.run_result.probe_results
    }
    current_index: dict[tuple[str, str], ProbeResult] = {
        (r.probe_id, r.prompt_id): r for r in current.probe_results
    }

    # All keys in both baseline and current
    all_keys = set(baseline_index) | set(current_index)
    comparisons: list[ProbeComparison] = []

    for key in sorted(all_keys, key=lambda k: (k[0], k[1])):
        probe_id, prompt_id = key
        b_result = baseline_index.get(key)
        c_result = current_index.get(key)

        if b_result is None or c_result is None:
            # One side is missing — treat missing as score 0.0 / failed
            b_score = b_result.score if b_result else 0.0
            b_passed = b_result.passed if b_result else False
            b_details = b_result.details if b_result else "Not present in baseline."
            c_score = c_result.score if c_result else 0.0
            c_passed = c_result.passed if c_result else False
            c_details = c_result.details if c_result else "Not present in current run."
            category = (b_result or c_result).category  # type: ignore[union-attr]
            probe_name = (b_result or c_result).probe_name  # type: ignore[union-attr]
        else:
            b_score, b_passed, b_details = b_result.score, b_result.passed, b_result.details
            c_score, c_passed, c_details = c_result.score, c_result.passed, c_result.details
            category = c_result.category
            probe_name = c_result.probe_name

        delta = c_score - b_score
        regression = (b_passed and not c_passed and delta <= -regression_threshold) or (
            not b_passed and not c_passed and delta <= -regression_threshold
        )
        improvement = (not b_passed and c_passed and delta >= improvement_threshold) or (
            b_passed and c_passed and delta >= improvement_threshold
        )

        comparisons.append(
            ProbeComparison(
                probe_id=probe_id,
                probe_name=probe_name,
                category=category,
                prompt_id=prompt_id,
                baseline_score=b_score,
                current_score=c_score,
                score_delta=delta,
                baseline_passed=b_passed,
                current_passed=c_passed,
                regression=regression,
                improvement=improvement,
                baseline_details=b_details,
                current_details=c_details,
            )
        )

    return DriftReport(
        suite_name=current.suite_name,
        provider=current.provider,
        baseline_snapshot_id=baseline.snapshot_id,
        baseline_created_at=baseline.created_at,
        current_run_id=current.run_id,
        comparisons=comparisons,
    )

`promptcanary.core.models.CanaryPrompt` ¶

Bases: BaseModel

A single prompt entry in a CanarySuite, with optional metadata.

`promptcanary.core.models.CanaryRunResult` ¶

Bases: BaseModel

Aggregated results for one full run of a CanarySuite against one provider.

`by_category` `property` ¶

Group probe results by category.

`duration_ms` `property` ¶

Total wall-clock duration of the run in milliseconds.

`failed_probes` `property` ¶

Convenience: all probe results that did not pass.

`overall_score` `property` ¶

Mean score across all probe results. Returns 1.0 if no probes ran.

`pass_rate` `property` ¶

Fraction of probes that passed. Returns 1.0 if no probes ran.

`promptcanary.core.models.BaselineSnapshot` ¶

Bases: BaseModel

A saved baseline — the "known-good" state a future run is compared against.

`promptcanary.core.models.DriftReport` ¶

Bases: BaseModel

The authoritative, fully-structured output of a drift comparison.

`severity` `property` ¶

Heuristic severity rating based on regression count and magnitude.

`summary` `property` ¶

Single-sentence human-readable summary suitable for notifications.

`promptcanary.core.models.ProbeComparison` ¶

Bases: BaseModel

Side-by-side comparison of a single probe between baseline and current run.

`promptcanary.core.models.ProbeResult` ¶

Bases: BaseModel

Result of running a single Probe against a single LLMResponse.

`promptcanary.core.models.ProviderConfig` ¶

Bases: BaseModel

Identifies and configures an LLM provider + model endpoint.

Probes¶

`promptcanary.core.probes.base.BaseProbe` ¶

Bases: ABC

Abstract base class for all PromptCanary probes.

Subclasses must implement :meth:evaluate.

Attributes:

Name	Type	Description
`probe_id`	`str`	Stable machine-readable identifier (snake_case).
`name`	`str`	Human-readable display name.
`category`	`ProbeCategory`	High-level :class:`ProbeCategory`.
`description`	`str`	One-sentence description shown in reports.

`call(prompt, response)` ¶

Make probes callable directly: probe(prompt, response).

Source code in promptcanary/core/probes/base.py

def __call__(self, prompt: CanaryPrompt, response: LLMResponse) -> ProbeResult:
    """Make probes callable directly: ``probe(prompt, response)``."""
    return self.evaluate(prompt, response)

`evaluate(prompt, response)` `abstractmethod` ¶

Run this probe and return a structured result.

Parameters:

Name	Type	Description	Default
`prompt`	`CanaryPrompt`	The :class:`CanaryPrompt` that was sent.	required
`response`	`LLMResponse`	The :class:`LLMResponse` received.	required

Returns:

Name	Type	Description
`A`	`ProbeResult`	class:`ProbeResult` with score, pass/fail, and details.

Source code in promptcanary/core/probes/base.py

@abc.abstractmethod
def evaluate(self, prompt: CanaryPrompt, response: LLMResponse) -> ProbeResult:
    """Run this probe and return a structured result.

    Args:
        prompt:   The :class:`CanaryPrompt` that was sent.
        response: The :class:`LLMResponse` received.

    Returns:
        A :class:`ProbeResult` with score, pass/fail, and details.
    """
    ...

`promptcanary.core.probes.base.probe(probe_id, *, name='', category=ProbeCategory.CUSTOM, description='')` ¶

Decorator that turns a plain function into a registered BaseProbe.

Usage::

@probe("my_probe", name="My Custom Probe", category=ProbeCategory.CUSTOM)
def evaluate(prompt: CanaryPrompt, response: LLMResponse) -> ProbeResult:
    passed = "hello" in response.content.lower()
    return ProbeResult(
        probe_id="my_probe",
        probe_name="My Custom Probe",
        category=ProbeCategory.CUSTOM,
        prompt_id=prompt.id,
        passed=passed,
        score=1.0 if passed else 0.0,
        details="Checked for greeting.",
    )

Source code in promptcanary/core/probes/base.py

def probe(
    probe_id: str,
    *,
    name: str = "",
    category: ProbeCategory = ProbeCategory.CUSTOM,
    description: str = "",
) -> Callable[[Callable[[CanaryPrompt, LLMResponse], ProbeResult]], type[BaseProbe]]:
    """Decorator that turns a plain function into a registered ``BaseProbe``.

    Usage::

        @probe("my_probe", name="My Custom Probe", category=ProbeCategory.CUSTOM)
        def evaluate(prompt: CanaryPrompt, response: LLMResponse) -> ProbeResult:
            passed = "hello" in response.content.lower()
            return ProbeResult(
                probe_id="my_probe",
                probe_name="My Custom Probe",
                category=ProbeCategory.CUSTOM,
                prompt_id=prompt.id,
                passed=passed,
                score=1.0 if passed else 0.0,
                details="Checked for greeting.",
            )
    """

    def decorator(fn: Callable[[CanaryPrompt, LLMResponse], ProbeResult]) -> type[BaseProbe]:
        _name = name or fn.__name__.replace("_", " ").title()
        _captured_fn = fn

        # Build the class dynamically with evaluate already defined so ABC is satisfied
        def _evaluate(self: BaseProbe, p: CanaryPrompt, r: LLMResponse) -> ProbeResult:
            return _captured_fn(p, r)

        _FunctionalProbe = type(  # noqa: N806  (dynamic class creation — uppercase is intentional)
            _name,
            (BaseProbe,),
            {
                "probe_id": probe_id,
                "name": _name,
                "category": category,
                "description": description,
                "evaluate": _evaluate,
            },
        )

        _PROBE_REGISTRY[probe_id] = _FunctionalProbe
        return _FunctionalProbe

    return decorator

Providers¶

`promptcanary.providers.base.BaseLLMProvider(config)` ¶

Bases: ABC

Abstract base class for LLM provider adapters.

Subclasses must implement :meth:complete.

Parameters:

Name	Type	Description	Default
`config`	`ProviderConfig`	A :class:`ProviderConfig` describing the model and parameters.	required

Source code in promptcanary/providers/base.py

def __init__(self, config: ProviderConfig) -> None:
    self._config = config

`complete(prompt, *, system_prompt=None)` `abstractmethod` ¶

Send a prompt to the provider and return a structured response.

Parameters:

Name	Type	Description	Default
`prompt`	`CanaryPrompt`	The :class:`CanaryPrompt` to send.	required
`system_prompt`	`str \| None`	Optional system-level instruction override.	`None`

Returns:

Name	Type	Description
`A`	`LLMResponse`	class:`LLMResponse` with content and metadata.

Raises:

Type	Description
`ProviderError`	On any API or network error.

Source code in promptcanary/providers/base.py

@abc.abstractmethod
def complete(
    self,
    prompt: CanaryPrompt,
    *,
    system_prompt: str | None = None,
) -> LLMResponse:
    """Send a prompt to the provider and return a structured response.

    Args:
        prompt:        The :class:`CanaryPrompt` to send.
        system_prompt: Optional system-level instruction override.

    Returns:
        A :class:`LLMResponse` with content and metadata.

    Raises:
        ProviderError: On any API or network error.
    """
    ...

`promptcanary.providers.litellm.LiteLLMProvider(model_id_or_config, *, temperature=0.0, max_tokens=1024, seed=42, **extra_params)` ¶

Bases: BaseLLMProvider

LLM provider backed by the LiteLLM library.

This is the recommended provider for PromptCanary. It supports all major cloud providers and local models via a single unified interface.

Parameters:

Name	Type	Description	Default
`model_id_or_config`	`str \| ProviderConfig`	Either a LiteLLM model string (e.g. `"openai/gpt-4o"`) or a :class:`ProviderConfig`.	required
`temperature`	`float`	Sampling temperature (default: 0.0 for reproducibility).	`0.0`
`max_tokens`	`int`	Max tokens in the response (default: 1024).	`1024`
`seed`	`int \| None`	Determinism seed (default: 42).	`42`
`**extra_params`	`object`	Additional kwargs forwarded to `litellm.completion()`.	`{}`

Examples::

# Quickstart
provider = LiteLLMProvider("openai/gpt-4o-mini")

# Anthropic
provider = LiteLLMProvider("anthropic/claude-3-5-sonnet-20241022")

# Local Ollama
provider = LiteLLMProvider("ollama/llama3", temperature=0.0)

# Full config
provider = LiteLLMProvider(
    "openai/gpt-4o",
    temperature=0.0,
    max_tokens=2048,
    extra_params={"response_format": {"type": "json_object"}},
)

Source code in promptcanary/providers/litellm.py

def __init__(
    self,
    model_id_or_config: str | ProviderConfig,
    *,
    temperature: float = 0.0,
    max_tokens: int = 1024,
    seed: int | None = 42,
    **extra_params: object,
) -> None:
    if isinstance(model_id_or_config, ProviderConfig):
        config = model_id_or_config
    else:
        config = ProviderConfig(
            model_id=model_id_or_config,
            temperature=temperature,
            max_tokens=max_tokens,
            seed=seed,
            extra_params=dict(extra_params),
        )
    super().__init__(config)

`complete(prompt, *, system_prompt=None)` ¶

Call the LLM via LiteLLM and return a structured :class:LLMResponse.

Parameters:

Name	Type	Description	Default
`prompt`	`CanaryPrompt`	The :class:`CanaryPrompt` to send.	required
`system_prompt`	`str \| None`	Optional system instruction.	`None`

Returns:

Name	Type	Description
`A`	`LLMResponse`	class:`LLMResponse` with content, token counts, and metadata.

Raises:

Type	Description
`ProviderError`	On API errors, auth failures, or network issues.

Source code in promptcanary/providers/litellm.py

def complete(
    self,
    prompt: CanaryPrompt,
    *,
    system_prompt: str | None = None,
) -> LLMResponse:
    """Call the LLM via LiteLLM and return a structured :class:`LLMResponse`.

    Args:
        prompt:        The :class:`CanaryPrompt` to send.
        system_prompt: Optional system instruction.

    Returns:
        A :class:`LLMResponse` with content, token counts, and metadata.

    Raises:
        ProviderError: On API errors, auth failures, or network issues.
    """
    try:
        import litellm
    except ImportError as e:
        raise ProviderError(
            "LiteLLM is not installed. Run: pip install litellm",
            model_id=self.config.model_id,
            raw_error=e,
        ) from e

    messages: list[dict[str, str]] = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": prompt.text})

    # Build kwargs from config
    kwargs: dict[str, object] = {
        "model": self.config.model_id,
        "messages": messages,
        "temperature": self.config.temperature,
        "max_tokens": self.config.max_tokens,
        **self.config.extra_params,
    }

    # Seed is provider-dependent; pass it if set
    if self.config.seed is not None:
        kwargs["seed"] = self.config.seed

    try:
        response = litellm.completion(**kwargs)
    except Exception as exc:
        # Normalise all litellm/provider errors into ProviderError
        status = getattr(exc, "status_code", None)
        raise ProviderError(
            f"Provider call failed: {exc}",
            model_id=self.config.model_id,
            status_code=status,
            raw_error=exc,
        ) from exc

    # Extract content safely
    try:
        content = response.choices[0].message.content or ""
        finish_reason = response.choices[0].finish_reason
    except (AttributeError, IndexError) as exc:
        raise ProviderError(
            f"Unexpected response structure from {self.config.model_id}: {exc}",
            model_id=self.config.model_id,
            raw_error=exc,
        ) from exc

    # Token usage (optional fields)
    usage = getattr(response, "usage", None)
    prompt_tokens = getattr(usage, "prompt_tokens", None)
    completion_tokens = getattr(usage, "completion_tokens", None)
    total_tokens = getattr(usage, "total_tokens", None)

    return LLMResponse(
        prompt_id=prompt.id,
        provider_model_id=self.config.model_id,
        content=content,
        finish_reason=finish_reason,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        total_tokens=total_tokens,
        raw_response=response.model_dump() if hasattr(response, "model_dump") else {},
    )

`from_config(config)` `classmethod` ¶

Create a provider directly from a :class:ProviderConfig.

Source code in promptcanary/providers/litellm.py

@classmethod
def from_config(cls, config: ProviderConfig) -> LiteLLMProvider:
    """Create a provider directly from a :class:`ProviderConfig`."""
    return cls(config)

Storage¶

`promptcanary.storage.file.FileBaselineStore(directory='baselines')` ¶

Stores and retrieves :class:BaselineSnapshot objects as JSON files.

Parameters:

Name	Type	Description	Default
`directory`	`str \| Path`	Directory where baseline JSON files are stored. Created automatically if it doesn't exist.	`'baselines'`

File naming: {suite_name}__{model_slug}__{timestamp}.json (URL-safe, avoiding characters that cause shell issues)

Example::

store = FileBaselineStore("baselines/")
snapshot = store.save(run_result)
loaded = store.load(snapshot.snapshot_id)
latest = store.load_latest("my-suite", "openai/gpt-4o")

Source code in promptcanary/storage/file.py

def __init__(self, directory: str | Path = "baselines") -> None:
    self.directory = Path(directory)
    self.directory.mkdir(parents=True, exist_ok=True)

`delete(snapshot_id)` ¶

Delete a baseline by snapshot ID. Returns True if deleted.

Source code in promptcanary/storage/file.py

def delete(self, snapshot_id: str) -> bool:
    """Delete a baseline by snapshot ID. Returns True if deleted."""
    matches = list(self.directory.glob(f"*{snapshot_id[:8]}*.json"))
    if not matches:
        return False
    for m in matches:
        m.unlink(missing_ok=True)
    return True

`list_baselines(suite_name=None, model_id=None)` ¶

List stored baselines with lightweight metadata (no full deserialization).

Returns a list of dicts with keys: path, snapshot_id, suite_name, model_id, created_at.

Source code in promptcanary/storage/file.py

def list_baselines(
    self,
    suite_name: str | None = None,
    model_id: str | None = None,
) -> list[dict[str, Any]]:
    """List stored baselines with lightweight metadata (no full deserialization).

    Returns a list of dicts with keys: path, snapshot_id, suite_name,
    model_id, created_at.
    """
    files = self._list_files(suite_name=suite_name, model_id=model_id)
    results = []
    for f in sorted(files, reverse=True):
        try:
            data = json.loads(f.read_text(encoding="utf-8"))
            results.append(
                {
                    "path": str(f),
                    "snapshot_id": data.get("snapshot_id", "?"),
                    "suite_name": data.get("suite_name", "?"),
                    "model_id": data.get("provider", {}).get("model_id", "?"),
                    "created_at": data.get("created_at", "?"),
                    "note": data.get("_note", ""),
                }
            )
        except Exception:
            continue
    return results

`load(snapshot_id)` ¶

Load a snapshot by its ID.

Parameters:

Name	Type	Description	Default
`snapshot_id`	`str`	The UUID of the snapshot to load.	required

Raises:

Type	Description
`FileNotFoundError`	If no snapshot with that ID exists.

Source code in promptcanary/storage/file.py

def load(self, snapshot_id: str) -> BaselineSnapshot:
    """Load a snapshot by its ID.

    Args:
        snapshot_id: The UUID of the snapshot to load.

    Raises:
        FileNotFoundError: If no snapshot with that ID exists.
    """
    matches = list(self.directory.glob(f"*_{snapshot_id[:8]}*.json"))
    if not matches:
        # Try full scan
        matches = [
            f
            for f in self.directory.glob("*.json")
            if snapshot_id in f.read_text(encoding="utf-8")
        ]
    if not matches:
        raise FileNotFoundError(
            f"No baseline snapshot found with ID starting with '{snapshot_id[:8]}'. "
            f"Available baselines: {[f.name for f in self.directory.glob('*.json')]}"
        )
    return self._load_file(matches[0])

`load_from_path(path)` ¶

Load a snapshot directly from a JSON file path.

Source code in promptcanary/storage/file.py

def load_from_path(self, path: str | Path) -> BaselineSnapshot:
    """Load a snapshot directly from a JSON file path."""
    return self._load_file(Path(path))

`load_latest(suite_name, model_id=None)` ¶

Load the most recently saved baseline for a given suite and provider.

Parameters:

Name	Type	Description	Default
`suite_name`	`str`	Name of the canary suite.	required
`model_id`	`str \| None`	Optional model ID filter (e.g. `"openai/gpt-4o"`).	`None`

Raises:

Type	Description
`FileNotFoundError`	If no matching baseline exists.

Source code in promptcanary/storage/file.py

def load_latest(
    self,
    suite_name: str,
    model_id: str | None = None,
) -> BaselineSnapshot:
    """Load the most recently saved baseline for a given suite and provider.

    Args:
        suite_name: Name of the canary suite.
        model_id:   Optional model ID filter (e.g. ``"openai/gpt-4o"``).

    Raises:
        FileNotFoundError: If no matching baseline exists.
    """
    candidates = self._list_files(suite_name=suite_name, model_id=model_id)
    if not candidates:
        raise FileNotFoundError(
            f"No baseline found for suite={suite_name!r}"
            + (f", model={model_id!r}" if model_id else "")
            + f". Store directory: {self.directory}"
        )
    # Most recent first (filename has ISO timestamp)
    candidates.sort(reverse=True)
    return self._load_file(candidates[0])

`save(run_result, *, note='', snapshot_id=None)` ¶

Save a :class:CanaryRunResult as a baseline snapshot.

Parameters:

Name	Type	Description	Default
`run_result`	`CanaryRunResult`	The run to persist.	required
`note`	`str`	Optional human note (stored in the JSON).	`''`
`snapshot_id`	`str \| None`	Override the auto-generated ID (useful for testing).	`None`

Returns:

Type	Description
`BaselineSnapshot`	The saved :class:`BaselineSnapshot`.

Source code in promptcanary/storage/file.py

def save(
    self,
    run_result: CanaryRunResult,
    *,
    note: str = "",
    snapshot_id: str | None = None,
) -> BaselineSnapshot:
    """Save a :class:`CanaryRunResult` as a baseline snapshot.

    Args:
        run_result:  The run to persist.
        note:        Optional human note (stored in the JSON).
        snapshot_id: Override the auto-generated ID (useful for testing).

    Returns:
        The saved :class:`BaselineSnapshot`.
    """
    snapshot = BaselineSnapshot(
        suite_name=run_result.suite_name,
        provider=run_result.provider,
        run_result=run_result,
    )
    if snapshot_id:
        # Pydantic frozen model — rebuild with custom ID
        snapshot = snapshot.model_copy(update={"snapshot_id": snapshot_id})

    filename = self._filename(
        suite_name=run_result.suite_name,
        model_id=run_result.provider.model_id,
        ts=snapshot.created_at,
        snap_id=snapshot.snapshot_id,
    )

    data: dict[str, Any] = snapshot.model_dump(mode="json")
    if note:
        data["_note"] = note

    path = self.directory / filename
    path.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8")

    return snapshot