latent.workspace¶
Workspace path resolution utilities.
This module provides centralized workspace path resolution that can be configured via environment variables or a TOML configuration file for flexibility across different project structures.
Configuration Priority: 1. Environment variables (highest priority) 2. TOML configuration file (config/latent.toml) 3. Default values (lowest priority)
Functions¶
get_agent_studio_api_key¶
Resolve Agent Studio API key for Authorization: Bearer header.
Priority: AGENT_STUDIO_API_KEY env var > [agent_studio].api_key in latent.toml > None.
get_agent_studio_url¶
Resolve Agent Studio API base URL.
Priority: AGENT_STUDIO_URL env var > [agent_studio].url in latent.toml
http://localhost:8000
get_cache_directory¶
Get the cache directory for storing Prefect task results.
Configuration priority: 1. LATENT_CACHE_DIR environment variable 2. workspace.cache_dir from config/latent.toml 3. .cache/ directory relative to workspace root
Returns: Path to cache directory
Example: >>> cache_dir = get_cache_directory() >>> # Returns: {workspace}/.cache/
get_data_directory¶
Get the data directory for storing input datasets.
This function handles input data. See get_output_directory() for
output data.
Configuration priority: 1. LATENT_DATA_DIR environment variable 2. workspace.data_dir from config/latent.toml 3. data/ directory relative to workspace root
Args: flow_name: Optional flow name to get flow-specific input data directory
Returns: Path to data directory (or flow-specific input subdirectory)
Example: >>> input_dir = get_data_directory("my_flow") >>> # Returns: data/my_flow/input/
get_fixtures_directory¶
Get the fixtures directory for committed eval inputs.
Configuration priority: 1. LATENT_FIXTURES_DIR environment variable 2. workspace.fixtures_dir from config/latent.toml 3. data/fixtures/ directory relative to workspace root
Distinct from get_stable_directory(): stable holds curated artifacts a
deployment consumes (glossaries, taxonomies), fixtures hold the eval inputs
a flow is scored against. Both are version-controlled, so neither is
auto-created — a missing directory is a checkout problem, and creating it
empty would turn that into a confusing "dataset not found" instead.
Reach for this over get_data_directory() whenever the input must be
reviewable in a diff: that one resolves under data/<flow>/input/, which
is generated and gitignored.
Returns: Path to fixtures directory
Example: >>> fixtures_dir = get_fixtures_directory() >>> cases = fixtures_dir / "guardrails" / "cases_en.yaml"
get_flows_directory¶
Get the flows directory where flow configurations are stored.
Configuration priority: 1. LATENT_FLOWS_DIR environment variable 2. workspace.flows_dir from config/latent.toml 3. flows/ directory relative to workspace root
Returns: Path to flows directory
Example: >>> flows_dir = get_flows_directory() >>> flow_config = flows_dir / "my_flow" / "parameters.yaml"
get_logs_directory¶
Get the logs directory for storing flow logs.
Configuration priority: 1. LATENT_LOGS_DIR environment variable 2. workspace.logs_dir from config/latent.toml 3. logs/ directory relative to workspace root
Args: flow_name: Optional flow name to get flow-specific logs directory
Returns: Path to logs directory (or flow-specific subdirectory)
Example: >>> logs_dir = get_logs_directory() >>> flow_logs = get_logs_directory("my_flow")
get_mlflow_artifacts_directory¶
Get the MLFlow artifacts directory for storing experiment artifacts.
This is separate from the mlruns directory (which stores metadata in SQLite). Artifacts are stored here and organized by flow/experiment.
Configuration priority: 1. LATENT_MLARTIFACTS_DIR environment variable 2. workspace.mlartifacts_dir from config/latent.toml 3. mlartifacts/ directory relative to workspace root
Args: flow_name: Optional flow name to get flow-specific artifacts directory
Returns: Path to mlartifacts directory (or flow-specific subdirectory)
Example: >>> artifacts_dir = get_mlflow_artifacts_directory("my_flow") >>> # Returns: {workspace}/mlartifacts/my_flow/
get_mlruns_directory¶
Get the MLFlow runs directory for storing experiment data.
Configuration priority: 1. LATENT_MLRUNS_DIR environment variable 2. workspace.mlruns_dir from config/latent.toml 3. mlruns/ directory relative to workspace root
Returns: Path to mlruns directory
Example: >>> mlruns_dir = get_mlruns_directory()
get_output_directory¶
Get the output directory for storing output datasets on the filesystem.
Outputs are always persisted here, regardless of MLflow state. MLflow provides lineage tracking on top of filesystem persistence.
Configuration priority: 1. LATENT_DATA_DIR environment variable (base) 2. workspace.data_dir from config/latent.toml 3. data/ directory relative to workspace root
Args: flow_name: Flow name to get flow-specific output directory
Returns:
Path to data/
Example: >>> output_dir = get_output_directory("my_flow") >>> # Returns: data/my_flow/output/
get_stable_directory¶
Get the stable directory for curated, version-controlled deployment artifacts.
Configuration priority: 1. LATENT_STABLE_DIR environment variable 2. workspace.stable_dir from config/latent.toml 3. data/stable/ directory relative to workspace root
Unlike other directory functions, this does NOT auto-create the directory since stable artifacts are version-controlled.
Returns: Path to stable directory
Example: >>> stable_dir = get_stable_directory() >>> glossary = stable_dir / "glossary.json"
get_workspace_root¶
Get the workspace root directory.
Configuration priority: 1. LATENT_WORKSPACE_ROOT environment variable 2. workspace.root from config/latent.toml 3. Current working directory (fallback)
When latent is installed as a package, you should set LATENT_WORKSPACE_ROOT or configure workspace.root in latent.toml (or use the cwd fallback).
Returns: Path to workspace root directory
Example: >>> root = get_workspace_root() >>> flows_dir = root / "flows" >>> data_dir = root / "data"
validate_workspace_paths¶
Validate that critical workspace paths are accessible.
This function checks that the required directories exist and provides helpful error messages if they don't.
Args: flow_name: Name of the flow to validate paths for
Returns: Dictionary mapping path type to Path object
Raises: FileNotFoundError: If critical paths don't exist with helpful error message
Example: >>> paths = validate_workspace_paths("my_flow") >>> print(paths['flows_dir'])