Model Monitoring Inference Aggregator block #818
Conversation
robiscoding
requested review from
PawelPeczek-Roboflow,
grzegorz-roboflow,
yeldarby,
probicheaux and
hansent
as code owners
| description="Reference data to extract property from", | ||
| examples=["$steps.my_step.predictions"], | ||
| ) | ||
| frequency: Union[ |
There was a problem hiding this comment.
I considered using the RateLimiter block, but this export block would always require that as a dependency. I'm not sure if there's a pattern for making another block required to run, so I just added this to the params for now, open to doing it another way
There was a problem hiding this comment.
name is misleading, as it suggests compaction of predictions to be sent in batches, which does not take place
There was a problem hiding this comment.
Yeah maybe something like SamplePredictions, ReportPredictionsSample, ModelMonitoringPredictionsSampler
| "model_type": detections.data.get("prediction_type", [""])[i], | ||
| } | ||
| results.append(formatted_det) | ||
| elif isinstance(detections, dict): |
There was a problem hiding this comment.
What case would predictions be returned as a dict? I saw in other Blocks that predictions will either be of type sv.Detections or dict
There was a problem hiding this comment.
I have general concern with this PR - what is the quality of data acquired in that way? Is that meaningful monitoring at the end of the day? What do we want to display as the result of running monitoring about a week-long stream? How would people understand that something is wrong with the model?
I mean, we have stream running and we sub sample of predictions based on time, suggesting to update once per few seconds. We do it as this is basically unfeasible to push through the wire all predictions - but maybe - in this scenario it makes sense to compute aggregates of predictions on the client end and submit compacted results once per interval?
| def get_manifest(cls) -> Type[WorkflowBlockManifest]: | ||
| return BlockManifest | ||
|
|
||
| def is_in_reporting_range(self, frequency: int) -> bool: |
There was a problem hiding this comment.
please move into usage first, then declaration (run(...) method first
| def is_in_reporting_range(self, frequency: int) -> bool: | ||
| now = datetime.now() | ||
| last_report_time_str = self._cache.get(LAST_REPORT_TIME_CACHE_KEY) | ||
| print( |
There was a problem hiding this comment.
please remove print statements
| self, | ||
| fire_and_forget: bool, | ||
| predictions: Union[sv.Detections, dict], | ||
| frequency: int = 3, |
There was a problem hiding this comment.
defaults should not be set here, but in manifest
| BLOCK_NAME = "Roboflow Model Monitoring Exporter" | ||
|
|
||
|
|
||
| class BlockManifest(WorkflowBlockManifest): |
There was a problem hiding this comment.
please follow migration guide: https://inference.roboflow.com/workflows/execution_engine_changelog/#execution-engine-v130-inference-v0270
| ) | ||
| type: Literal[ | ||
| "roboflow_core/roboflow_model_monitoring_exporter@v1", | ||
| BLOCK_NAME, |
There was a problem hiding this comment.
please remove name aliasing, this is not needed for new blocks
| "error_status": False, | ||
| "message": "Not in reporting range, skipping report. (Ok)", | ||
| } | ||
| if self._api_key is None: |
There was a problem hiding this comment.
should be validated first - fail fast approach - clearer error handling
|
|
||
|
|
||
| # TODO: maybe make this a helper or decorator, it's used in multiple places | ||
| def get_workspace_name( |
There was a problem hiding this comment.
I will remember about this and if I see re-ocurring pattern of usage, I will move to common inference utils
| description="Reference data to extract property from", | ||
| examples=["$steps.my_step.predictions"], | ||
| ) | ||
| frequency: Union[ |
There was a problem hiding this comment.
name is misleading, as it suggests compaction of predictions to be sent in batches, which does not take place
|
|
||
| def is_in_reporting_range(self, frequency: int) -> bool: | ||
| now = datetime.now() | ||
| last_report_time_str = self._cache.get(LAST_REPORT_TIME_CACHE_KEY) |
There was a problem hiding this comment.
global key? how would that work on hosted platform?
This is particularly meant for inferences made by InferencePipeline in edge deployments. Model Monitoring currently does not get any data from InferencePipeline. This is because of the large volume of requests made by IP would overload our system. So the goal here is not to have comprehensive inference results data or an aggregation of it (that's coming later), but more of a health/status check at a regular interval that IP is running and making inferences. Is there a recommended way of doing this?
Aggregating is fine, but we were trying to defer doing aggregations until we had a better idea of what type of aggregation makes sense for this use case. I thought the Aggregator block could be used in conjunction with this but maybe not? |
|
is model monitoring in given form even acceptable to run on video? |
|
Today we need to close the topic, otherwise |
robiscoding
changed the title
| self._cache = cache | ||
| self._background_tasks = background_tasks | ||
| self._thread_pool_executor = thread_pool_executor | ||
| self._last_report_time_cache_key = "roboflow_model_monitoring_last_report_time" |
There was a problem hiding this comment.
that would have a side effect for all blocks of this type to have shared state of last report time - is that desired end?
There was a problem hiding this comment.
Oh no, the intent was for each instance of the block to have it's own last_report_time. Should I add a random string to the key name to make that happen?
There was a problem hiding this comment.
- if the expectation is that it works at hosted, I would use redis with the following structure of keys
workflows:steps_cache:roboflow_core/model_monitoring_inference_aggregator@v1:{uuid4()}:last_report_time
| last_report_time = now | ||
| else: | ||
| last_report_time = datetime.fromisoformat(last_report_time_str) | ||
| time_elapsed = int((now - last_report_time).total_seconds()) |
There was a problem hiding this comment.
type conversion to int seems not to be needed
| ) | ||
|
|
||
|
|
||
| def format_sv_detections_for_model_monitoring( |
There was a problem hiding this comment.
please change the name to reflect that sv.Detections and cls results are handled
| ) -> List[Prediction]: | ||
| results = [] | ||
| if isinstance(detections, sv.Detections): | ||
| num_detections = len(detections.data.get("detection_id", [])) |
There was a problem hiding this comment.
there is an iter in sv.Detections
for detection in detections: would be easier and less error-prone with getting the list of size 1, that seems to fail with strange error if other blocks break the contract
| if system_info: | ||
| for key, value in system_info.items(): | ||
| inference_data[key] = value | ||
| inference_data["inference_results"] = predictions |
There was a problem hiding this comment.
I have a strange feeling it may fail on serialisation - predictions is List of dataclasses which does not get automatically converted into json afaik
| LONG_DESCRIPTION = """ | ||
| This block periodically reports an aggregated sample of inference results to Roboflow Model Monitoring. | ||
|
|
||
| It aggregates predictions in memory between reports and then sends a representative sample of predictions at a regular interval specified by the `frequency` parameter. |
There was a problem hiding this comment.
I would elaborate on what we understand by representative - seems like "most confident for given class"
There was a problem hiding this comment.
Approving under following conditions (also reported in docs):
🚨 Limitations
-
The block is should not be relied on when running Workflow in
inferenceserver or via HTTP request to Roboflow
hosted platform, as the internal state is not persisted in a memory that would be accessible for all requests to
the server, causing aggregation to only have a scope of single request. We will solve that problem in future
releases if proven to be serious limitation for clients. -
This block do not have ability to separate aggregations for multiple videos processed by
InferencePipeline-
effectively aggregating data for all video feeds connected to single process runningInferencePipeline.
@robiscoding in the interest of having the feature I will push the PR, also fixing tests that were broken, but I will hold you responsible for any follow-up changes if there are bug / problem reports.
We can work with those current limitations as the block will still provide us with some observability of inferencepipeline running in edge environments. And yes we'll be iterating on this functionality so let me know if any issues come up in the interim, thanks for the thorough review! |
Description
Type of change
Please delete options that are not relevant.
How has this change been tested, please provide a testcase or example of how you tested the change?
Any specific deployment considerations
Docs