diff --git a/frigate/api/chat.py b/frigate/api/chat.py index 3e10714ca..c40bfa548 100644 --- a/frigate/api/chat.py +++ b/frigate/api/chat.py @@ -3,6 +3,7 @@ import base64 import json import logging +import math import time from datetime import datetime from typing import Any, Dict, Generator, List, Optional @@ -26,6 +27,7 @@ from frigate.api.defs.response.chat_response import ( ) from frigate.api.defs.tags import Tags from frigate.api.event import events +from frigate.embeddings.util import ZScoreNormalization from frigate.genai.utils import build_assistant_message_for_conversation from frigate.jobs.vlm_watch import ( get_vlm_watch_job, @@ -98,6 +100,52 @@ class VLMMonitorRequest(BaseModel): zones: List[str] = [] +# Similarity fusion weights for find_similar_objects. +# Visual dominates because the feature's primary use case is "same specific object." +# If these change, update the test in test_chat_find_similar_objects.py. +VISUAL_WEIGHT = 0.65 +DESCRIPTION_WEIGHT = 0.35 + +# Must match or stay <= the k used by EmbeddingsContext vec searches +# (see frigate/embeddings/__init__.py search_thumbnail/search_description). +# Pre-filtering a larger pool is wasted work — vec will only rank top-k anyway. +CANDIDATE_CAP = 100 + + +def _distance_to_score(distance: float, stats: ZScoreNormalization) -> float: + """Convert a cosine distance to a [0, 1] similarity score. + + Uses the existing ZScoreNormalization stats maintained by EmbeddingsContext + to normalize across deployments, then a bounded sigmoid. Lower distance -> + higher score. If stats are uninitialized (stddev == 0), returns a neutral + 0.5 so the fallback ordering by raw distance still dominates. + """ + if stats.stddev == 0: + return 0.5 + z = (distance - stats.mean) / stats.stddev + # Sigmoid on -z so that small distance (good) -> high score. + return 1.0 / (1.0 + math.exp(z)) + + +def _fuse_scores( + visual_score: Optional[float], + description_score: Optional[float], +) -> Optional[float]: + """Weighted fusion of visual and description similarity scores. + + If one side is missing (e.g., no description embedding for this event), + the other side's score is returned alone with no penalty. If both are + missing, returns None and the caller should drop the event. + """ + if visual_score is None and description_score is None: + return None + if visual_score is None: + return description_score + if description_score is None: + return visual_score + return VISUAL_WEIGHT * visual_score + DESCRIPTION_WEIGHT * description_score + + def get_tool_definitions() -> List[Dict[str, Any]]: """ Get OpenAI-compatible tool definitions for Frigate. diff --git a/frigate/test/test_chat_find_similar_objects.py b/frigate/test/test_chat_find_similar_objects.py new file mode 100644 index 000000000..ee4e91e2e --- /dev/null +++ b/frigate/test/test_chat_find_similar_objects.py @@ -0,0 +1,57 @@ +"""Tests for the find_similar_objects chat tool.""" + +import unittest +from unittest.mock import MagicMock + +from frigate.api.chat import ( + DESCRIPTION_WEIGHT, + VISUAL_WEIGHT, + _distance_to_score, + _fuse_scores, +) +from frigate.embeddings.util import ZScoreNormalization + + +class TestDistanceToScore(unittest.TestCase): + def test_lower_distance_gives_higher_score(self): + stats = ZScoreNormalization() + # Seed the stats with a small distribution so stddev > 0. + stats._update([0.1, 0.2, 0.3, 0.4, 0.5]) + + close_score = _distance_to_score(0.1, stats) + far_score = _distance_to_score(0.5, stats) + + self.assertGreater(close_score, far_score) + self.assertGreaterEqual(close_score, 0.0) + self.assertLessEqual(close_score, 1.0) + self.assertGreaterEqual(far_score, 0.0) + self.assertLessEqual(far_score, 1.0) + + def test_uninitialized_stats_returns_neutral_score(self): + stats = ZScoreNormalization() # n == 0, stddev == 0 + self.assertEqual(_distance_to_score(0.3, stats), 0.5) + + +class TestFuseScores(unittest.TestCase): + def test_weights_sum_to_one(self): + self.assertAlmostEqual(VISUAL_WEIGHT + DESCRIPTION_WEIGHT, 1.0) + + def test_fuses_both_sides(self): + fused = _fuse_scores(visual_score=0.8, description_score=0.4) + expected = VISUAL_WEIGHT * 0.8 + DESCRIPTION_WEIGHT * 0.4 + self.assertAlmostEqual(fused, expected) + + def test_missing_description_uses_visual_only(self): + fused = _fuse_scores(visual_score=0.7, description_score=None) + self.assertAlmostEqual(fused, 0.7) + + def test_missing_visual_uses_description_only(self): + fused = _fuse_scores(visual_score=None, description_score=0.6) + self.assertAlmostEqual(fused, 0.6) + + def test_both_missing_returns_none(self): + self.assertIsNone(_fuse_scores(visual_score=None, description_score=None)) + + +if __name__ == "__main__": + unittest.main()