Skip to content

latent.registry.pipelines

Pipeline discovery and CLI generation.

This module provides utilities for auto-discovering @flow decorated functions and generating Typer CLI applications with dynamic options based on flow signatures.

Most discovery and registration logic is now in latent.flow_registry. This module provides the user-facing create_pipelines_cli() convenience for building project-specific CLI apps.

Functions

create_pipelines_cli

create_pipelines_cli(flows: dict[str, Callable], app_name: str) -> typer.Typer

Create Typer app with run/list commands and dynamic options.

Args: flows: Dict mapping flow_name -> flow_function app_name: Name of the app (e.g., "support", "risk")

Returns: Configured Typer app

discover_flows

discover_flows(package_name: str, package_path: Path) -> dict[str, Callable]

Scan package for flow.py files and collect @flow decorated functions.

Uses FlowRegistry (populated by @flow decorator) as primary source, with fallback to scanning module attributes for _latent_flow_name.

Args: package_name: Python package name (e.g., "pipelines") package_path: Path to package directory

Returns: Dict mapping flow_name -> flow_function

get_flow_options

get_flow_options(flow_fn: Callable) -> list[tuple[str, type, Any, str]]

Extract CLI options from flow function signature.

Returns list of (param_name, type, default, description) tuples.

Attributes

console