mllm_shap.observability package#

Submodules#

mllm_shap.observability.bridge module#

Bridge between TelemetryProbe and structured observability sinks.

class mllm_shap.observability.bridge.TelemetryBridge(run_id: str, sink: ObservabilitySink)[source]#

Bases: object

Dual-writes telemetry events to structured sink.

custom_metric(key: str, value: float | int | str | bool) → None[source]#: Emit a metric update as event.

static from_probe(probe: TelemetryProbe | None, run_id: str, sink: ObservabilitySink) → TelemetryBridge[source]#: Create bridge and backfill known probe metrics if available.

run_id: str#: Unique identifier for the SHAP run, used to correlate events and spans in the observability sink.

sink: ObservabilitySink#: The observability sink to which telemetry events will be emitted. This should be an instance of a class that implements the ObservabilitySink interface, such as InMemoryObservabilitySink or a custom implementation that sends events to an external monitoring system.

stage_span(stage: str, elapsed_ms: float) → None[source]#: Emit explicit stage duration span.

mllm_shap.observability.events module#

Structured event/span models for explainability observability.

class mllm_shap.observability.events.StageSpan(run_id: str, stage: str, elapsed_ms: float, attrs: dict[str, ~typing.Any]=<factory>)[source]#

Bases: object

Structured span for stage timing.

attrs: dict[str, Any]#: Additional attributes for the span, represented as a dictionary of key-value pairs. This can be used to include any relevant information about the stage execution that may be useful for analysis in the observability sink.

elapsed_ms: float#: The elapsed time for the stage, measured in milliseconds. This should represent the total time taken to execute the stage, and can be used for performance analysis in the observability sink.

run_id: str#: The unique identifier for the SHAP run that this span is associated with. This should match the run_id used in the TelemetryBridge to allow correlation of events and spans in the observability sink.

stage: str#: The name of the stage that this span is measuring. This should correspond to the name of the stage in the pipeline, such as ‘SamplingStage’, ‘AttributionStage’, or ‘PostprocessingStage’.

class mllm_shap.observability.events.TraceEvent(name: str, run_id: str, ts: float = <factory>, attrs: dict[str, ~typing.Any]=<factory>)[source]#

Bases: object

Single structured event in explainability run.

attrs: dict[str, Any]#: Additional attributes for the event, represented as a dictionary of key-value pairs. This can be used to include any relevant information about the event that may be useful for analysis in the observability sink, such as stage names, metric keys and values, or probe snapshot details

name: str#: The name of the event, used to identify the type of event in the observability sink. This should be a descriptive name that indicates the purpose of the event, such as ‘stage_start’, ‘stage_end’, ‘metric’, or ‘probe_snapshot’.

run_id: str#: The unique identifier for the SHAP run that this event is associated with. This should match the run_id used in the TelemetryBridge to allow correlation of events and spans in the observability sink.

ts: float#: Timestamp of the event, represented as seconds since the epoch. This will be automatically set to the current time when the event is created, but can be overridden if needed.

mllm_shap.observability.sink module#

Observability sink interfaces and in-memory implementation.

class mllm_shap.observability.sink.InMemoryObservabilitySink(events: list[TraceEvent] = <factory>, spans: list[StageSpan] = <factory>)[source]#

Bases: ObservabilitySink

Testing/debug sink storing all events in memory.

emit_event(event: TraceEvent) → None[source]#: Emit structured event.

emit_span(span: StageSpan) → None[source]#: Emit structured stage span.

events: list[TraceEvent]#

spans: list[StageSpan]#

class mllm_shap.observability.sink.ObservabilitySink[source]#

Bases: ABC

Abstract event sink for explainability traces.

abstractmethod emit_event(event: TraceEvent) → None[source]#: Emit structured event.

abstractmethod emit_span(span: StageSpan) → None[source]#: Emit structured stage span.

Module contents#

Observability primitives for SHAP pipelines.

class mllm_shap.observability.InMemoryObservabilitySink(events: list[TraceEvent] = <factory>, spans: list[StageSpan] = <factory>)[source]#

Bases: ObservabilitySink

Testing/debug sink storing all events in memory.

emit_event(event: TraceEvent) → None[source]#: Emit structured event.

emit_span(span: StageSpan) → None[source]#: Emit structured stage span.

events: list[TraceEvent]#

spans: list[StageSpan]#

class mllm_shap.observability.ObservabilitySink[source]#

Bases: ABC

Abstract event sink for explainability traces.

abstractmethod emit_event(event: TraceEvent) → None[source]#: Emit structured event.

abstractmethod emit_span(span: StageSpan) → None[source]#: Emit structured stage span.

class mllm_shap.observability.StageSpan(run_id: str, stage: str, elapsed_ms: float, attrs: dict[str, ~typing.Any]=<factory>)[source]#

Bases: object

Structured span for stage timing.

attrs: dict[str, Any]#: Additional attributes for the span, represented as a dictionary of key-value pairs. This can be used to include any relevant information about the stage execution that may be useful for analysis in the observability sink.

elapsed_ms: float#: The elapsed time for the stage, measured in milliseconds. This should represent the total time taken to execute the stage, and can be used for performance analysis in the observability sink.

run_id: str#: The unique identifier for the SHAP run that this span is associated with. This should match the run_id used in the TelemetryBridge to allow correlation of events and spans in the observability sink.

stage: str#: The name of the stage that this span is measuring. This should correspond to the name of the stage in the pipeline, such as ‘SamplingStage’, ‘AttributionStage’, or ‘PostprocessingStage’.

class mllm_shap.observability.TraceEvent(name: str, run_id: str, ts: float = <factory>, attrs: dict[str, ~typing.Any]=<factory>)[source]#

Bases: object

Single structured event in explainability run.

attrs: dict[str, Any]#: Additional attributes for the event, represented as a dictionary of key-value pairs. This can be used to include any relevant information about the event that may be useful for analysis in the observability sink, such as stage names, metric keys and values, or probe snapshot details

name: str#: The name of the event, used to identify the type of event in the observability sink. This should be a descriptive name that indicates the purpose of the event, such as ‘stage_start’, ‘stage_end’, ‘metric’, or ‘probe_snapshot’.

run_id: str#: The unique identifier for the SHAP run that this event is associated with. This should match the run_id used in the TelemetryBridge to allow correlation of events and spans in the observability sink.

ts: float#: Timestamp of the event, represented as seconds since the epoch. This will be automatically set to the current time when the event is created, but can be overridden if needed.