Skip to content

Config Validation

The config module validates YAML pipeline files without loading real models. Use it in CI, examples, and local development before running expensive model downloads.

Minimal example:

from SafeLens.config import config_summary, validate_pipeline_config_file

config = validate_pipeline_config_file("examples/config.yaml")
print(config_summary(config))

Generate the JSON Schema:

from SafeLens.config import write_pipeline_config_json_schema

write_pipeline_config_json_schema("schemas/pipeline-config.schema.json")

Configuration schema generation and static validation.

ConfigValidationError

Bases: ValueError

Raised when a SafeLens YAML config fails static validation.

Source code in src/SafeLens/config.py 
24
25
class ConfigValidationError(ValueError):
    """Raised when a SafeLens YAML config fails static validation."""

config_summary(config)

Return a small serializable summary of a validated config.

Source code in src/SafeLens/config.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def config_summary(config: PipelineConfig) -> dict[str, Any]:
    """Return a small serializable summary of a validated config."""
    return {
        "model": {
            "source": config.model.source,
            "name": config.model.name,
        },
        "methods": {
            "probes": [spec.name for spec in config.pipeline.probes],
            "monitors": [spec.name for spec in config.pipeline.monitors],
            "attributors": [spec.name for spec in config.pipeline.attributors],
        },
        "dataset_size": len(config.dataset),
        "report_path": config.output.report_path,
    }

format_pydantic_errors(exc)

Format Pydantic errors for CLI users.

Source code in src/SafeLens/config.py 
142
143
144
145
146
147
148
def format_pydantic_errors(exc: ValidationError) -> str:
    """Format Pydantic errors for CLI users."""
    lines = ["Invalid SafeLens config:"]
    for error in exc.errors():
        loc = ".".join(str(part) for part in error.get("loc", ())) or "<root>"
        lines.append(f"- {loc}: {error.get('msg', 'invalid value')}")
    return "\n".join(lines)

iter_layer_refs(spec)

Yield layer or hook references from a method config.

Source code in src/SafeLens/config.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def iter_layer_refs(spec: MethodSpec) -> Iterable[tuple[str, LayerRef]]:
    """Yield layer or hook references from a method config."""
    config = spec.config
    if "layers" in config:
        layers = config["layers"]
        if isinstance(layers, list):
            for index, layer in enumerate(layers):
                if isinstance(layer, int | str):
                    yield f"layers[{index}]", layer
        elif isinstance(layers, int | str):
            yield "layers", layers
    if "layer" in config and isinstance(config["layer"], int | str):
        yield "layer", config["layer"]
    for key in ("hook", "hook_name", "activation_name"):
        value = config.get(key)
        if isinstance(value, str):
            yield key, value

load_yaml_config(path)

Load a YAML config file into a dictionary.

Source code in src/SafeLens/config.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
def load_yaml_config(path: str | Path) -> dict[str, Any]:
    """Load a YAML config file into a dictionary."""
    config_path = Path(path)
    try:
        with config_path.open("r", encoding="utf-8") as handle:
            raw = yaml.safe_load(handle) or {}
    except yaml.YAMLError as exc:
        raise ConfigValidationError(f"Could not parse YAML config {config_path}: {exc}") from exc
    if not isinstance(raw, dict):
        raise ConfigValidationError(
            f"Config {config_path} must be a YAML mapping at the top level."
        )
    return raw

pipeline_config_json_schema()

Return the JSON Schema for SafeLens YAML pipeline configs.

Source code in src/SafeLens/config.py 
28
29
30
31
32
def pipeline_config_json_schema() -> dict[str, Any]:
    """Return the JSON Schema for SafeLens YAML pipeline configs."""
    schema = PipelineConfig.model_json_schema()
    schema["$schema"] = "https://json-schema.org/draft/2020-12/schema"
    return schema

run_report_json_schema()

Return the JSON Schema for SafeLens run reports.

Source code in src/SafeLens/config.py 
35
36
37
38
39
def run_report_json_schema() -> dict[str, Any]:
    """Return the JSON Schema for SafeLens run reports."""
    schema = RunReport.model_json_schema()
    schema["$schema"] = "https://json-schema.org/draft/2020-12/schema"
    return schema

validate_pipeline_config_file(path)

Load and statically validate a SafeLens pipeline config file.

Source code in src/SafeLens/config.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def validate_pipeline_config_file(path: str | Path) -> PipelineConfig:
    """Load and statically validate a SafeLens pipeline config file."""
    raw = load_yaml_config(path)
    try:
        config = PipelineConfig.model_validate(raw)
    except ValidationError as exc:
        raise ConfigValidationError(format_pydantic_errors(exc)) from exc

    errors = [
        *validate_registered_methods(config),
        *validate_static_hook_names(config),
    ]
    if errors:
        joined = "\n".join(f"- {error}" for error in errors)
        raise ConfigValidationError(f"Invalid SafeLens config:\n{joined}")
    return config

validate_registered_methods(config)

Return registry errors for method names referenced by a config.

Source code in src/SafeLens/config.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def validate_registered_methods(config: PipelineConfig) -> list[str]:
    """Return registry errors for method names referenced by a config."""
    load_builtin_methods()
    errors: list[str] = []
    sections = (
        ("pipeline.probes", config.pipeline.probes, get_probe),
        ("pipeline.monitors", config.pipeline.monitors, get_monitor),
        ("pipeline.attributors", config.pipeline.attributors, get_attributor),
    )
    for section, specs, getter in sections:
        for index, spec in enumerate(specs):
            try:
                getter(spec.name)
            except RegistryError as exc:
                errors.append(f"{section}[{index}].name: {exc.args[0]}")
    return errors

validate_static_hook_names(config)

Return static hook-name errors that can be checked without loading a model.

Source code in src/SafeLens/config.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def validate_static_hook_names(config: PipelineConfig) -> list[str]:
    """Return static hook-name errors that can be checked without loading a model."""
    if config.model.source not in {"qwen3", "qwen3_dense", "qwen3-dense"}:
        return []

    errors: list[str] = []
    for section, specs in (
        ("pipeline.probes", config.pipeline.probes),
        ("pipeline.monitors", config.pipeline.monitors),
        ("pipeline.attributors", config.pipeline.attributors),
    ):
        for spec_index, spec in enumerate(specs):
            for key_path, layer_ref in iter_layer_refs(spec):
                try:
                    validate_qwen3_hook_ref(layer_ref)
                except ValueError as exc:
                    errors.append(f"{section}[{spec_index}].config.{key_path}: {exc}")
    return errors

write_pipeline_config_json_schema(path)

Write the pipeline config JSON Schema to disk.

Source code in src/SafeLens/config.py 
42
43
44
45
46
47
48
49
def write_pipeline_config_json_schema(path: str | Path) -> None:
    """Write the pipeline config JSON Schema to disk."""
    output_path = Path(path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(
        json.dumps(pipeline_config_json_schema(), indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )