gitea_admin/sientia-dataops-laborious-legacy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'release/SIENTIAPDE-1646' into feature/SIENTIAPDE-1986-simple-metrics-rewrite

Browse Source
This commit is contained in:
Vitor Pimentel dos Santos
2026-10-09 14:33:16 -03:00
committed by GitHub
parent 5d2bb1527f 0aa3b54827
commit 0a7392f0ba
9 changed files with 302 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
laborious/activities/gates.py
View File

@@ -78,6 +78,7 @@ input_path_confidence: Mapping[str, int] = {
'REPEAT': -1,
}
# MLFlow response filter function mappings
mlflow_response_filter_functions: dict[str, ResponseFilterFunc] = {
'API_ERROR': api_error_filter,
@@ -89,12 +90,34 @@ mlflow_response_path_confidence: Mapping[str, int] = {
'REPEAT': -1,
}
def _count_nan_values(data: DataFrame, _config: dict[str, Any]) -> int:
"""
Count rows for the NAN_VALUES filter.
`NAN_VALUES` only triggers when every value in the DataFrame is NaN, so every row in it
matched the condition — same convention as `_count_specific_variables_null_values`:
counting rows that satisfied the filter's own trigger condition.
Return:
int: Number of rows in the evaluated DataFrame.
"""
return len(data)
# MLFlow content filter function mappings
mlflow_content_filter_functions: dict[str, ContentFilterFunc] = {
'NAN_VALUES': nan_values_filter,
'EMPTY_DATA': filter_empty_data,
}
# `EMPTY_DATA` reuses `_count_empty_data` — same function (`filter_empty_data`) and same
# trigger condition (absence of rows) as `input_gate`'s own EMPTY_DATA.
mlflow_content_filter_counters: dict[str, InputCounterFunc] = {
'NAN_VALUES': _count_nan_values,
'EMPTY_DATA': _count_empty_data,
}
mlflow_content_path_confidence: Mapping[str, int] = {
'STOP': -1,
'CONTINUE': 18,
@@ -320,7 +343,9 @@ class Gates(SientiaMonitoring):
return None, 0, '', None
@activity.defn(name='mlflow_response_gate')
def mlflow_response_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
def mlflow_response_gate(
self, input_data: dict[str, Any]
) -> tuple[str | None, int, str, dict[str, Any] | None]:
"""
Validate MLFlow API response quality and integrity.
@@ -334,6 +359,9 @@ class Gates(SientiaMonitoring):
3. Determines path decisions based on response validation results
4. Provides confidence scores and detailed validation comments
5. Handles API errors and response validation failures
6. Builds a metadata document describing which filter(s) triggered, with their policy
and a counter, whenever one does — this activity only computes it, it never writes it
(see `PredictionMetadata.write_prediction_metadata` for the actual write)
Args:
input_data: Configuration and data for response validation
@@ -345,10 +373,14 @@ class Gates(SientiaMonitoring):
- path_priority (list[str]): Priority order for path decisions
Returns:
tuple: (path_flag, confidence, comment)
tuple: (path_flag, confidence, comment, document)
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
- confidence (int): Confidence score for the decision
- comment (str): Detailed explanation of the decision
- document (dict | None): Metadata describing the filter(s) that produced
`path_flag`, shaped `{'filters': [{'name', 'policy', 'counter'}, ...]}`;
`None` when nothing was filtered. Not written anywhere by this activity —
the caller decides whether and how to persist it.
Raises:
Exception: If response validation fails or configuration is invalid
@@ -369,10 +401,11 @@ class Gates(SientiaMonitoring):
gate_type = input_data['type']
path_priority = input_data['path_priority']
filter_output = []
comments = []
# (filter_name, policy, filter_config) per filter that triggered, in configured order
triggered_filters: list[tuple[str, str, dict[str, Any]]] = []
status = payload.status or {}
for fil, config in filters.items():
@@ -381,15 +414,21 @@ class Gates(SientiaMonitoring):
policy, filter_config = self._read_filter_entry(config)
try:
if mlflow_response_filter_functions[fil](status, filter_config):
filter_output.append(policy)
comments.append(status.get('message', 'Unknown MLFlow API error'))
triggered_filters.append((fil, policy, filter_config))
# The real error lives one level in, under `content` — `request_transform`/
# `request_predict` nest `message`/`traceback` there on failure (mlflow.py),
# reusing the same `content` key the success path uses for the DataFrame
# (review finding, 2026-10-08: reading `status['message']` directly always
# fell back to the generic default, in every real production occurrence).
error_content = status.get('content') or {}
comments.append(error_content.get('message', 'Unknown MLFlow API error'))
self.send_notification(
metadata=metadata,
notification_id=f'{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}',
message=status.get('message', 'Unknown MLFlow API error'),
message=error_content.get('message', 'Unknown MLFlow API error'),
block='mlflow_gate',
level=NotificationLevel.ERROR,
attachment_content=status.get('traceback'),
attachment_content=error_content.get('traceback'),
)
except Exception as e:
trace = traceback.format_exc()
@@ -402,19 +441,47 @@ class Gates(SientiaMonitoring):
attachment_content=trace,
)
triggered_policies = [policy for _, policy, _ in triggered_filters]
for path_flag in path_priority:
if path_flag in filter_output:
if path_flag in triggered_policies:
self.info(f'Mlflow response gate result: {path_flag}', metadata)
return path_flag, mlflow_response_path_confidence[path_flag], ', '.join(comments)
matched = [fil for fil, policy, _ in triggered_filters if policy == path_flag]
document = {
'filters': [
{
'name': fil,
'policy': path_flag,
# No per-row/per-item unit to count here — `api_error_filter`
# evaluates a single API response as a whole, not tabular data —
# so, unlike `input_gate`'s per-filter counters, this is a fixed
# value: 1 (one API error event). Inlined rather than a counter
# function/dict, since there's only one response filter today and
# this isn't an override of anything (review finding).
'counter': 1,
}
for fil in matched
],
}
return (
path_flag,
mlflow_response_path_confidence[path_flag],
', '.join(comments),
document,
)
self.info('Nothing was filtered by the mlflow response gate', metadata)
del data
return None, 0, ''
return None, 0, '', None
@activity.defn(name='mlflow_content_gate')
def mlflow_content_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
def mlflow_content_gate(
self, input_data: dict[str, Any]
) -> tuple[str | None, int, str, dict[str, Any] | None]:
"""
Validate MLFlow prediction content quality and integrity.
@@ -428,6 +495,9 @@ class Gates(SientiaMonitoring):
3. Determines path decisions based on content validation results
4. Provides confidence scores and detailed validation comments
5. Handles content validation failures and quality issues
6. Builds a metadata document describing which filter(s) triggered, with their policy
and a counter, whenever one does — this activity only computes it, it never writes it
(see `PredictionMetadata.write_prediction_metadata` for the actual write)
Args:
input_data: Configuration and data for content validation
@@ -439,10 +509,14 @@ class Gates(SientiaMonitoring):
- path_priority (list[str]): Priority order for path decisions
Returns:
tuple: (path_flag, confidence, comment)
tuple: (path_flag, confidence, comment, document)
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
- confidence (int): Confidence score for the decision
- comment (str): Detailed explanation of the decision
- document (dict | None): Metadata describing the filter(s) that produced
`path_flag`, shaped `{'filters': [{'name', 'policy', 'counter'}, ...]}`;
`None` when nothing was filtered. Not written anywhere by this activity —
the caller decides whether and how to persist it.
Raises:
Exception: If content validation fails or configuration is invalid
@@ -458,7 +532,8 @@ class Gates(SientiaMonitoring):
gate_type = input_data['type']
path_priority = input_data['path_priority']
filter_output = []
# (filter_name, policy, filter_config) per filter that triggered, in configured order
triggered_filters: list[tuple[str, str, dict[str, Any]]] = []
self._debug_dataframe('Input data:', data, metadata)
self.debug(f'Filters: \n {filters}', metadata)
@@ -469,7 +544,7 @@ class Gates(SientiaMonitoring):
policy, filter_config = self._read_filter_entry(config)
try:
if mlflow_content_filter_functions[fil](data, filter_config):
filter_output.append(policy)
triggered_filters.append((fil, policy, filter_config))
self.send_notification(
metadata=metadata,
notification_id=f'{gate_type.upper()}_GATE_CONTENT_FILTER__{fil}',
@@ -489,20 +564,46 @@ class Gates(SientiaMonitoring):
attachment_content=trace,
)
triggered_policies = [policy for _, policy, _ in triggered_filters]
for path_flag in path_priority:
if path_flag in filter_output:
if path_flag in triggered_policies:
self.info(f'Mlflow content gate result: {path_flag}', metadata)
matched = [
(fil, filter_config)
for fil, policy, filter_config in triggered_filters
if policy == path_flag
]
document = {
'filters': [
{
'name': fil,
'policy': path_flag,
# `.get(..., lambda *_: 0)`, not `[fil]`: same defensive fallback as
# `input_gate` (review finding, 2026-09-30) — a filter added to
# `mlflow_content_filter_functions` without a matching counter here
# degrades to counter=0 instead of a raw KeyError.
'counter': mlflow_content_filter_counters.get(fil, lambda *_: 0)(
data, filter_config
),
}
for fil, filter_config in matched
],
}
return (
path_flag,
mlflow_content_path_confidence[path_flag],
'Transformed data not passed the content filter',
document,
)
self.info('Nothing was filtered by the mlflow content gate', metadata)
del data
return None, 0, ''
return None, 0, '', None
def get_prediction_store_policy(
self, prediction_store_policy: str, metadata: dict[str, Any]

8
laborious/activities/mlflow.py
View File

@@ -664,7 +664,7 @@ class MLFlow(SientiaMonitoring):
"""
Pick the first available reference CSV artifact path from the MLflow run.
Candidates are checked in priority order: ``retrain_input.csv``, then ``train_data.csv``.
Candidates are checked in priority order: ``evaluation_data.csv``, then ``test_data.csv``.
A path matches when it equals the candidate or ends with ``/<candidate>`` for nested layouts.
Args:
@@ -703,9 +703,9 @@ class MLFlow(SientiaMonitoring):
"""
Download reference training CSV from the MLflow run linked to the production alias.
Resolves ``retrain_input.csv`` or ``train_data.csv`` via artifact listing before download.
``retrain_input.csv`` is preferred when both exist (most recent retrain snapshot). Used by
drift workflows to compare live data against the reference distribution logged with the model.
Resolves ``evaluation_data.csv`` or ``test_data.csv`` via artifact listing before download.
``evaluation_data.csv`` is preferred when both exist. Used by drift workflows to compare
live data against the reference distribution logged with the model.
Timestamps are normalized to ``DATETIME_FORMAT`` string columns before returning records.
Args:

23
laborious/workflows/sub_workflows/prediction_process.py
View File

@@ -156,7 +156,7 @@ class PredictionProcess:
)
# Validate MLFlow transform response
path_flag, confidence, comment = await workflow.execute_local_activity_method(
path_flag, confidence, comment, document = await workflow.execute_local_activity_method(
Activities.mlflow_response_gate,
{
**metadata,
@@ -171,11 +171,11 @@ class PredictionProcess:
# Handle path decision based on transform validation
if await self.path_flag_handler(
data, path_flag, input_data, confidence, last_timestamp, comment
data, path_flag, input_data, confidence, last_timestamp, comment, document
):
return
path_flag, confidence, comment = await workflow.execute_local_activity_method(
path_flag, confidence, comment, document = await workflow.execute_local_activity_method(
Activities.mlflow_content_gate,
{
**metadata,
@@ -189,7 +189,7 @@ class PredictionProcess:
)
if await self.path_flag_handler(
data, path_flag, input_data, confidence, last_timestamp, comment
data, path_flag, input_data, confidence, last_timestamp, comment, document
):
return
@@ -206,7 +206,7 @@ class PredictionProcess:
)
# Validate MLFlow prediction response
path_flag, confidence, comment = await workflow.execute_local_activity_method(
path_flag, confidence, comment, document = await workflow.execute_local_activity_method(
Activities.mlflow_response_gate,
{
**metadata,
@@ -221,7 +221,7 @@ class PredictionProcess:
# Handle path decision based on prediction validation
if await self.path_flag_handler(
data, path_flag, input_data, confidence, last_timestamp, comment
data, path_flag, input_data, confidence, last_timestamp, comment, document
):
return
@@ -262,8 +262,7 @@ class PredictionProcess:
Args:
metadata: Workflow metadata for contextual logging.
document: Metadata produced by the gate that resolved this `path_flag`; `None` when
nothing was filtered, or when the gate that produced this `path_flag` doesn't
build one yet.
nothing was filtered.
Return:
The generated id, or `None` when there's nothing to write, including when the
@@ -271,9 +270,11 @@ class PredictionProcess:
"""
if not document:
# Skip the remote activity call entirely rather than dispatching it just to have
# `write_prediction_metadata` decide there's nothing to write — `mlflow_response_gate`
# and `mlflow_content_gate` don't build a `document` yet, so this is the common case
# today, not a rare edge case (review finding, 2026-10-01).
# `write_prediction_metadata` decide there's nothing to write (review finding,
# 2026-10-01). All 3 gates now build a `document` whenever they return a non-`None`
# `path_flag`, so this branch is a defensive fallback today, not the common case it
# was when this was first written — kept in case a future caller passes a `path_flag`
# without one.
return None
try: