Save genai response to metadata model

frigate/data_processing/post/review_descriptions.py

@@ -92,7 +92,7 @@ class ReviewDescriptionProcessor(PostProcessorApi):
 
             # kickoff analysis
             threading.Thread(
-                target=self.run_analysis,
+                target=run_analysis,
                 args=(
                     self.genai_client,
                     camera,
@@ -102,23 +102,27 @@ class ReviewDescriptionProcessor(PostProcessorApi):
             ).start()
             self.tracked_review_items.pop(id)
 
-    def run_analysis(
-        self,
-        genai_client: GenAIClient,
-        camera: str,
-        final_data: dict[str, str],
-        thumbs: list[bytes],
-    ) -> None:
-        genai_client.generate_review_description(
-            {
-                "camera": camera,
-                "objects": final_data["data"]["objects"],
-                "recognized_objects": final_data["data"]["sub_labels"],
-                "zones": final_data["data"]["zones"],
-                "timestamp": datetime.datetime.fromtimestamp(final_data["end_time"]),
-            },
-            thumbs,
-        )
-
     def handle_request(self, request_data):
         pass
+
+
+@staticmethod
+def run_analysis(
+    genai_client: GenAIClient,
+    camera: str,
+    final_data: dict[str, str],
+    thumbs: list[bytes],
+) -> None:
+    metadata = genai_client.generate_review_description(
+        {
+            "camera": camera,
+            "objects": final_data["data"]["objects"],
+            "recognized_objects": final_data["data"]["sub_labels"],
+            "zones": final_data["data"]["zones"],
+            "timestamp": datetime.datetime.fromtimestamp(final_data["end_time"]),
+        },
+        thumbs,
+    )
+
+    if not metadata:
+        return None

frigate/genai/__init__.py

@@ -3,6 +3,7 @@
 import importlib
 import logging
 import os
+import re
 from typing import Any, Optional
 
 from playhouse.shortcuts import model_to_dict
@@ -36,7 +37,7 @@ class GenAIClient:
 
     def generate_review_description(
         self, review_data: dict[str, Any], thumbnails: list[bytes]
-    ) -> None:
+    ) -> ReviewMetadata | None:
         """Generate a description for the review item activity."""
         context_prompt = f"""
         Please analyze the image(s), which are in chronological order, strictly from the perspective of the {review_data["camera"].replace("_", " ")} security camera.
@@ -68,8 +69,20 @@ class GenAIClient:
         - The JSON must strictly match this structure:
         {ReviewMetadata.model_json_schema()["properties"]}
         """
-        logger.info(f"processing {review_data}")
-        logger.info(f"Got GenAI review: {self._send(context_prompt, thumbnails)}")
+        response = self._send(context_prompt, thumbnails)
+
+        if response:
+            clean_json = re.sub(
+                r"\n?```$", "", re.sub(r"^```[a-zA-Z0-9]*\n?", "", response)
+            )
+
+            try:
+                return ReviewMetadata.model_validate_json(clean_json)
+            except Exception:
+                # rarely LLMs can fail to follow directions on output format
+                return None
+        else:
+            return None
 
     def generate_object_description(
         self,