From 973c06c5f2f7b94e25b48101eb33103ee89bd022 Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Wed, 8 Apr 2026 16:48:15 -0500 Subject: [PATCH] extract pure chat helpers to chat_util module --- frigate/api/chat.py | 149 +++--------------- frigate/api/chat_util.py | 135 ++++++++++++++++ .../test/test_chat_find_similar_objects.py | 24 +-- 3 files changed, 167 insertions(+), 141 deletions(-) create mode 100644 frigate/api/chat_util.py diff --git a/frigate/api/chat.py b/frigate/api/chat.py index 05b87538e0..0ceb688193 100644 --- a/frigate/api/chat.py +++ b/frigate/api/chat.py @@ -3,12 +3,11 @@ import base64 import json import logging -import math import operator import time from datetime import datetime from functools import reduce -from typing import Any, Dict, Generator, List, Optional +from typing import Any, Dict, List, Optional import cv2 from fastapi import APIRouter, Body, Depends, Request @@ -20,6 +19,14 @@ from frigate.api.auth import ( get_allowed_cameras_for_filter, require_camera_access, ) +from frigate.api.chat_util import ( + chunk_content, + distance_to_score, + format_events_with_local_time, + fuse_scores, + hydrate_event, + parse_iso_to_timestamp, +) from frigate.api.defs.query.events_query_parameters import EventsQueryParams from frigate.api.defs.request.chat_body import ChatCompletionRequest from frigate.api.defs.response.chat_response import ( @@ -29,7 +36,6 @@ 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, @@ -43,49 +49,6 @@ logger = logging.getLogger(__name__) router = APIRouter(tags=[Tags.chat]) -def _chunk_content(content: str, chunk_size: int = 80) -> Generator[str, None, None]: - """Yield content in word-aware chunks for streaming.""" - if not content: - return - words = content.split(" ") - current: List[str] = [] - current_len = 0 - for w in words: - current.append(w) - current_len += len(w) + 1 - if current_len >= chunk_size: - yield " ".join(current) + " " - current = [] - current_len = 0 - if current: - yield " ".join(current) - - -def _format_events_with_local_time( - events_list: List[Dict[str, Any]], -) -> List[Dict[str, Any]]: - """Add human-readable local start/end times to each event for the LLM.""" - result = [] - for evt in events_list: - if not isinstance(evt, dict): - result.append(evt) - continue - copy_evt = dict(evt) - try: - start_ts = evt.get("start_time") - end_ts = evt.get("end_time") - if start_ts is not None: - dt_start = datetime.fromtimestamp(start_ts) - copy_evt["start_time_local"] = dt_start.strftime("%Y-%m-%d %I:%M:%S %p") - if end_ts is not None: - dt_end = datetime.fromtimestamp(end_ts) - copy_evt["end_time_local"] = dt_end.strftime("%Y-%m-%d %I:%M:%S %p") - except (TypeError, ValueError, OSError): - pass - result.append(copy_evt) - return result - - class ToolExecuteRequest(BaseModel): """Request model for tool execution.""" @@ -103,47 +66,6 @@ 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 - - -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. @@ -550,39 +472,6 @@ async def _execute_search_objects( ) -def _parse_iso_to_timestamp(value: Optional[str]) -> Optional[float]: - """Parse an ISO-8601 string as server-local time -> unix timestamp. - - Mirrors the parsing _execute_search_objects uses so both tools accept the - same format from the LLM. - """ - if value is None: - return None - try: - s = value.replace("Z", "").strip()[:19] - dt = datetime.strptime(s, "%Y-%m-%dT%H:%M:%S") - return time.mktime(dt.timetuple()) - except (ValueError, AttributeError, TypeError): - logger.warning("Invalid timestamp format: %s", value) - return None - - -def _hydrate_event(event: Event, score: Optional[float] = None) -> Dict[str, Any]: - """Convert an Event row into the dict shape returned by find_similar_objects.""" - data: Dict[str, Any] = { - "id": event.id, - "camera": event.camera, - "label": event.label, - "sub_label": event.sub_label, - "start_time": event.start_time, - "end_time": event.end_time, - "zones": event.zones, - } - if score is not None: - data["score"] = score - return data - - async def _execute_find_similar_objects( request: Request, arguments: Dict[str, Any], @@ -625,8 +514,8 @@ async def _execute_find_similar_objects( } # 3. Parse params. - after = _parse_iso_to_timestamp(arguments.get("after")) - before = _parse_iso_to_timestamp(arguments.get("before")) + after = parse_iso_to_timestamp(arguments.get("after")) + before = parse_iso_to_timestamp(arguments.get("before")) cameras = arguments.get("cameras") if cameras: @@ -685,7 +574,7 @@ async def _execute_find_similar_objects( if not vec_ids: return { - "anchor": _hydrate_event(anchor), + "anchor": hydrate_event(anchor), "results": [], "similarity_mode": similarity_mode, "candidate_truncated": candidate_truncated, @@ -714,16 +603,16 @@ async def _execute_find_similar_objects( scored: List[tuple[str, float]] = [] for eid in eligible: v_score = ( - _distance_to_score(visual_distances[eid], context.thumb_stats) + distance_to_score(visual_distances[eid], context.thumb_stats) if eid in visual_distances else None ) d_score = ( - _distance_to_score(description_distances[eid], context.desc_stats) + distance_to_score(description_distances[eid], context.desc_stats) if eid in description_distances else None ) - fused = _fuse_scores(v_score, d_score) + fused = fuse_scores(v_score, d_score) if fused is None: continue if min_score is not None and fused < min_score: @@ -733,10 +622,10 @@ async def _execute_find_similar_objects( scored.sort(key=lambda pair: pair[1], reverse=True) scored = scored[:limit] - results = [_hydrate_event(eligible[eid], score=score) for eid, score in scored] + results = [hydrate_event(eligible[eid], score=score) for eid, score in scored] return { - "anchor": _hydrate_event(anchor), + "anchor": hydrate_event(anchor), "results": results, "similarity_mode": similarity_mode, "candidate_truncated": candidate_truncated, @@ -1246,7 +1135,7 @@ async def _execute_pending_tools( json.dumps(tool_args), ) if tool_name == "search_objects" and isinstance(tool_result, list): - tool_result = _format_events_with_local_time(tool_result) + tool_result = format_events_with_local_time(tool_result) _keys = { "id", "camera", @@ -1561,7 +1450,7 @@ When a user refers to a specific object they have seen or describe with identify + b"\n" ) # Stream content in word-sized chunks for smooth UX - for part in _chunk_content(final_content): + for part in chunk_content(final_content): yield ( json.dumps({"type": "content", "delta": part}).encode( "utf-8" diff --git a/frigate/api/chat_util.py b/frigate/api/chat_util.py new file mode 100644 index 0000000000..743c38e57c --- /dev/null +++ b/frigate/api/chat_util.py @@ -0,0 +1,135 @@ +"""Pure, stateless helpers used by the chat tool dispatchers. + +These were extracted from frigate/api/chat.py to keep that module focused on +route handlers, tool dispatchers, and streaming loop internals. Nothing in +this file touches the FastAPI request, the embeddings context, or the chat +loop state — all inputs and outputs are plain data. +""" + +import logging +import math +import time +from datetime import datetime +from typing import Any, Dict, Generator, List, Optional + +from frigate.embeddings.util import ZScoreNormalization +from frigate.models import Event + +logger = logging.getLogger(__name__) + + +# 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 + + +def chunk_content(content: str, chunk_size: int = 80) -> Generator[str, None, None]: + """Yield content in word-aware chunks for streaming.""" + if not content: + return + words = content.split(" ") + current: List[str] = [] + current_len = 0 + for w in words: + current.append(w) + current_len += len(w) + 1 + if current_len >= chunk_size: + yield " ".join(current) + " " + current = [] + current_len = 0 + if current: + yield " ".join(current) + + +def format_events_with_local_time( + events_list: List[Dict[str, Any]], +) -> List[Dict[str, Any]]: + """Add human-readable local start/end times to each event for the LLM.""" + result = [] + for evt in events_list: + if not isinstance(evt, dict): + result.append(evt) + continue + copy_evt = dict(evt) + try: + start_ts = evt.get("start_time") + end_ts = evt.get("end_time") + if start_ts is not None: + dt_start = datetime.fromtimestamp(start_ts) + copy_evt["start_time_local"] = dt_start.strftime("%Y-%m-%d %I:%M:%S %p") + if end_ts is not None: + dt_end = datetime.fromtimestamp(end_ts) + copy_evt["end_time_local"] = dt_end.strftime("%Y-%m-%d %I:%M:%S %p") + except (TypeError, ValueError, OSError): + pass + result.append(copy_evt) + return result + + +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 parse_iso_to_timestamp(value: Optional[str]) -> Optional[float]: + """Parse an ISO-8601 string as server-local time -> unix timestamp. + + Mirrors the parsing _execute_search_objects uses so both tools accept the + same format from the LLM. + """ + if value is None: + return None + try: + s = value.replace("Z", "").strip()[:19] + dt = datetime.strptime(s, "%Y-%m-%dT%H:%M:%S") + return time.mktime(dt.timetuple()) + except (ValueError, AttributeError, TypeError): + logger.warning("Invalid timestamp format: %s", value) + return None + + +def hydrate_event(event: Event, score: Optional[float] = None) -> Dict[str, Any]: + """Convert an Event row into the dict shape returned by find_similar_objects.""" + data: Dict[str, Any] = { + "id": event.id, + "camera": event.camera, + "label": event.label, + "sub_label": event.sub_label, + "start_time": event.start_time, + "end_time": event.end_time, + "zones": event.zones, + } + if score is not None: + data["score"] = score + return data diff --git a/frigate/test/test_chat_find_similar_objects.py b/frigate/test/test_chat_find_similar_objects.py index f6c2bc4f6b..38055658e1 100644 --- a/frigate/test/test_chat_find_similar_objects.py +++ b/frigate/test/test_chat_find_similar_objects.py @@ -10,12 +10,14 @@ from unittest.mock import MagicMock from playhouse.sqlite_ext import SqliteExtDatabase from frigate.api.chat import ( + _execute_find_similar_objects, + get_tool_definitions, +) +from frigate.api.chat_util import ( DESCRIPTION_WEIGHT, VISUAL_WEIGHT, - _distance_to_score, - _execute_find_similar_objects, - _fuse_scores, - get_tool_definitions, + distance_to_score, + fuse_scores, ) from frigate.embeddings.util import ZScoreNormalization from frigate.models import Event @@ -31,8 +33,8 @@ class TestDistanceToScore(unittest.TestCase): # 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) + 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) @@ -42,7 +44,7 @@ class TestDistanceToScore(unittest.TestCase): def test_uninitialized_stats_returns_neutral_score(self): stats = ZScoreNormalization() # n == 0, stddev == 0 - self.assertEqual(_distance_to_score(0.3, stats), 0.5) + self.assertEqual(distance_to_score(0.3, stats), 0.5) class TestFuseScores(unittest.TestCase): @@ -50,20 +52,20 @@ class TestFuseScores(unittest.TestCase): 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) + 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) + 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) + 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)) + self.assertIsNone(fuse_scores(visual_score=None, description_score=None)) class TestToolDefinition(unittest.TestCase):