From af60d2db488b5ba9074425fb5ee8dffac0b76b0e Mon Sep 17 00:00:00 2001 From: Nicolas Mowen Date: Thu, 3 Sep 2026 08:02:22 -0600 Subject: [PATCH] GenAI Chat Improvements (#24173) * Initial tool approval implementation * Cleanups and fixes * Improve robustness of loading case --- docs/static/frigate-api.yaml | 12 + frigate/api/chat.py | 655 ++++++++++++++----- frigate/api/chat_util.py | 11 +- frigate/api/defs/request/chat_body.py | 11 +- frigate/genai/prompts.py | 124 ++++ frigate/test/test_chat_tool_approval.py | 479 ++++++++++++++ web/public/locales/en/views/chat.json | 16 + web/src/components/chat/ChatComposer.tsx | 14 +- web/src/components/chat/ChatSettings.tsx | 39 ++ web/src/components/chat/ToolApprovalCard.tsx | 97 +++ web/src/components/chat/ToolCallsGroup.tsx | 11 +- web/src/pages/Chat.tsx | 164 ++++- web/src/types/chat.ts | 10 + web/src/utils/chatUtil.ts | 70 +- 14 files changed, 1506 insertions(+), 207 deletions(-) create mode 100644 frigate/test/test_chat_tool_approval.py create mode 100644 web/src/components/chat/ToolApprovalCard.tsx diff --git a/docs/static/frigate-api.yaml b/docs/static/frigate-api.yaml index 6a63bc98bf..3ed7a2c1f5 100644 --- a/docs/static/frigate-api.yaml +++ b/docs/static/frigate-api.yaml @@ -7762,6 +7762,18 @@ components: description: Per-request thinking toggle. None means use the provider default. Ignored by providers that do not expose a per-request thinking switch. + tool_decisions: + additionalProperties: + type: string + enum: + - approve + - reject + type: object + title: Tool Decisions + description: Decisions for tool calls that paused for approval, keyed + by tool call ID. Send these with the conversation chain returned + alongside an approval request; rejected calls are reported to the + model as declined instead of being executed. type: object required: - messages diff --git a/frigate/api/chat.py b/frigate/api/chat.py index ff0a91ebe0..7e1c9aa3da 100644 --- a/frigate/api/chat.py +++ b/frigate/api/chat.py @@ -10,6 +10,7 @@ from functools import reduce from typing import Any, Literal import cv2 +import numpy as np from fastapi import APIRouter, Body, Depends, HTTPException, Request from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel @@ -23,6 +24,7 @@ from frigate.api.chat_util import ( chunk_content, distance_to_score, format_events_with_local_time, + format_local_time, fuse_scores, hydrate_event, parse_iso_to_timestamp, @@ -33,29 +35,44 @@ from frigate.api.defs.response.chat_response import ( ChatCompletionResponse, ChatMessageResponse, ToolCall, + ToolCallInvocation, ) from frigate.api.defs.tags import Tags from frigate.api.event import _build_attribute_filter_clause, events +from frigate.api.export import _build_export_job, _validate_export_source from frigate.config import FrigateConfig from frigate.config.classification import SemanticSearchModelEnum from frigate.genai.prompts import ( build_chat_system_prompt, get_attribute_classifications, get_tool_definitions, + get_write_tool_names, + strip_tool_access, ) -from frigate.genai.utils import build_assistant_message_for_conversation +from frigate.genai.utils import ( + build_assistant_message_for_conversation, + parse_tool_calls_from_message, +) +from frigate.jobs.export import ExportQueueFullError, start_export_job from frigate.jobs.vlm_watch import ( get_vlm_watch_job, start_vlm_watch_job, stop_vlm_watch_job, ) -from frigate.models import Event +from frigate.models import Event, Export, ExportCase +from frigate.record.export import PlaybackSourceEnum +from frigate.util.file import get_event_thumbnail_bytes, load_event_snapshot_image from frigate.util.object_names import get_categorized_object_names logger = logging.getLogger(__name__) router = APIRouter(tags=[Tags.chat]) +# Tool result recorded for a rejected write tool call. Providers require a +# result for every requested call; the user's intent is conveyed in a +# follow-up user message built by _rejection_message. +TOOL_REJECTED_RESULT: dict[str, str] = {"error": "user_rejected"} + class ToolExecuteRequest(BaseModel): """Request model for tool execution.""" @@ -666,29 +683,39 @@ async def _get_live_frame_image_url( frame = frame_processor.get_current_frame(camera, {}) if frame is None: return None - height, width = frame.shape[:2] - target_height = 480 - if height > target_height: - scale = target_height / height - frame = cv2.resize( - frame, - (int(width * scale), target_height), - interpolation=cv2.INTER_AREA, - ) - _, img_encoded = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 85]) - b64 = base64.b64encode(img_encoded.tobytes()).decode("utf-8") - return f"data:image/jpeg;base64,{b64}" + return _encode_frame_data_url(frame) except Exception as e: logger.debug("Failed to get live frame for %s: %s", camera, e) return None +def _encode_frame_data_url(frame: np.ndarray, target_height: int = 480) -> str: + """Downscale a BGR frame and encode it as a JPEG data URL for the model.""" + height, width = frame.shape[:2] + if height > target_height: + scale = target_height / height + frame = cv2.resize( + frame, + (int(width * scale), target_height), + interpolation=cv2.INTER_AREA, + ) + _, img_encoded = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 85]) + b64 = base64.b64encode(img_encoded.tobytes()).decode("utf-8") + return f"data:image/jpeg;base64,{b64}" + + +def _request_roles(request: Request) -> list[str]: + """Roles from the auth proxy header, split on the configured separator.""" + separator = request.app.frigate_config.proxy.separator + header = request.headers.get("remote-role", "") + return [r.strip() for r in header.split(separator) if r.strip()] + + async def _execute_set_camera_state( request: Request, arguments: dict[str, Any], ) -> dict[str, Any]: - role = request.headers.get("remote-role", "") - if "admin" not in [r.strip() for r in role.split(",")]: + if "admin" not in _request_roles(request): return {"error": "Admin privileges required to change camera settings."} camera = arguments.get("camera", "").strip() @@ -738,6 +765,189 @@ def _execute_get_categorized_object_names( return {"names": names} +def _execute_get_export_cases(allowed_cameras: list[str]) -> dict[str, Any]: + """List export cases with how many accessible exports each one holds.""" + from peewee import fn + + count_rows = ( + Export.select(Export.export_case, fn.COUNT(Export.id)) + .where(Export.camera << allowed_cameras, Export.export_case.is_null(False)) + .group_by(Export.export_case) + .tuples() + ) + counts = {case_id: count for case_id, count in count_rows} + + cases: list[dict[str, Any]] = [] + for case in ExportCase.select().order_by(ExportCase.created_at.desc()): + created_at = case.created_at + cases.append( + { + "id": case.id, + "name": case.name, + "description": case.description, + "created_at_local": format_local_time(created_at.timestamp()) + if isinstance(created_at, datetime) + else str(created_at), + "export_count": counts.get(case.id, 0), + } + ) + + if not cases: + return {"cases": [], "message": "No export cases exist yet."} + + return {"cases": cases} + + +async def _execute_create_export( + request: Request, + arguments: dict[str, Any], + allowed_cameras: list[str], +) -> dict[str, Any]: + """Queue a recording export, optionally attached to an existing case.""" + config = request.app.frigate_config + camera = (arguments.get("camera") or "").strip() + start_time = parse_iso_to_timestamp(arguments.get("start_time")) + end_time = parse_iso_to_timestamp(arguments.get("end_time")) + name = (arguments.get("name") or "").strip() or None + + if not camera or start_time is None or end_time is None: + return {"error": "camera, start_time, and end_time are all required."} + + if camera not in config.cameras: + return {"error": f"Camera '{camera}' not found."} + + if camera not in allowed_cameras: + return {"error": f"Camera '{camera}' not found or access denied"} + + if end_time <= start_time: + return {"error": "end_time must be after start_time."} + + try: + playback_source = PlaybackSourceEnum(arguments.get("source") or "recordings") + except ValueError: + return {"error": "source must be 'recordings' or 'preview'."} + + # Mirror the export API: attaching to an existing case is admin-only + # until case-level ACLs exist. + export_case_id = (arguments.get("export_case_id") or "").strip() or None + if export_case_id is not None: + if "admin" not in _request_roles(request): + return {"error": "Only admins can attach exports to an existing case."} + try: + ExportCase.get(ExportCase.id == export_case_id) + except ExportCase.DoesNotExist: + return {"error": f"Export case '{export_case_id}' not found."} + + source_error = _validate_export_source( + camera, start_time, end_time, playback_source + ) + if source_error is not None: + return {"error": source_error} + + export_job = _build_export_job( + camera, + start_time, + end_time, + name, + None, + playback_source, + export_case_id, + chapters=config.cameras[camera].record.export.chapters, + ) + try: + start_export_job(config, export_job) + except ExportQueueFullError: + return {"error": "Export queue is full. Try again once current exports finish."} + + return { + "success": True, + "export_id": export_job.id, + "status": "queued", + "camera": camera, + "name": name, + "source": playback_source.value, + "start_time_local": format_local_time(start_time), + "end_time_local": format_local_time(end_time), + "export_case_id": export_case_id, + "message": "Export queued. It will appear on the Export page when finished.", + } + + +async def _execute_get_event_image( + request: Request, + arguments: dict[str, Any], + allowed_cameras: list[str], +) -> dict[str, Any]: + """Attach an event's thumbnail or snapshot for a vision model to view.""" + event_id = (arguments.get("event_id") or "").strip() + if not event_id: + return {"error": "event_id is required."} + + image_type = arguments.get("image") or "thumbnail" + if image_type not in ("thumbnail", "snapshot"): + return {"error": "image must be 'thumbnail' or 'snapshot'."} + + try: + event = Event.get(Event.id == event_id) + except Event.DoesNotExist: + return {"error": f"Could not find event {event_id}."} + + if event.camera not in allowed_cameras: + return {"error": f"Event {event_id} not found or access denied"} + + chat_client = request.app.genai_manager.chat_client + if chat_client is None or not chat_client.supports_vision: + return { + "error": ( + "The configured chat model does not support vision, so images " + "cannot be viewed." + ) + } + + note = None + frame = None + if image_type == "snapshot": + if event.has_snapshot: + frame, _ = load_event_snapshot_image(event) + if frame is None: + note = "Snapshot not available; returning the thumbnail instead." + image_type = "thumbnail" + + if frame is None: + thumbnail = get_event_thumbnail_bytes(event) + if thumbnail: + frame = cv2.imdecode( + np.frombuffer(thumbnail, dtype=np.uint8), cv2.IMREAD_COLOR + ) + + if frame is None: + return {"error": f"No image is available for event {event_id}."} + + result: dict[str, Any] = { + "id": event.id, + "camera": event.camera, + "label": event.label, + "sub_label": event.sub_label, + "zones": event.zones, + "start_time_local": format_local_time(event.start_time), + "image": image_type, + } + if event.end_time is not None: + result["end_time_local"] = format_local_time(event.end_time) + description = (event.data or {}).get("description") + if description: + result["description"] = description + if note: + result["note"] = note + + result["_image_url"] = _encode_frame_data_url(frame) + result["_image_text"] = ( + f"Here is the {image_type} for event {event.id} " + f"({event.sub_label or event.label} on {event.camera})." + ) + return result + + async def _execute_tool_internal( tool_name: str, arguments: dict[str, Any], @@ -793,11 +1003,18 @@ async def _execute_tool_internal( return _execute_get_profile_status(request) elif tool_name == "get_recap": return _execute_get_recap(arguments, allowed_cameras) + elif tool_name == "get_export_cases": + return _execute_get_export_cases(allowed_cameras) + elif tool_name == "create_export": + return await _execute_create_export(request, arguments, allowed_cameras) + elif tool_name == "get_event_image": + return await _execute_get_event_image(request, arguments, allowed_cameras) else: logger.error( "Tool call failed: unknown tool %r. Expected one of: search_objects, find_similar_objects, " "get_categorized_object_names, get_live_context, start_camera_watch, stop_camera_watch, " - "get_profile_status, get_recap. Arguments received: %s", + "get_profile_status, get_recap, get_export_cases, create_export, get_event_image. " + "Arguments received: %s", tool_name, json.dumps(arguments), ) @@ -1026,14 +1243,74 @@ def _execute_get_recap( return {"error": "Failed to fetch recap data."} +def _pending_tool_calls_from_tail( + conversation: list[dict[str, Any]], +) -> list[dict[str, Any]] | None: + """Return the tool calls of a trailing assistant message, if any. + + A conversation that ends with an assistant message requesting tools is a + resume after an approval pause: the client sends the chain back with its + decisions and the loop runs those calls before asking the model again. + """ + if not conversation: + return None + tail = conversation[-1] + if tail.get("role") != "assistant" or not tail.get("tool_calls"): + return None + return parse_tool_calls_from_message(tail) + + +def _tool_calls_awaiting_approval( + pending_tool_calls: list[dict[str, Any]], + body: ChatCompletionRequest, + write_tools: set[str], +) -> list[dict[str, Any]]: + """Return the write tool calls the user still has to decide on.""" + return [ + { + "id": tc["id"], + "name": tc["name"], + "arguments": tc.get("arguments") or {}, + } + for tc in pending_tool_calls + if tc["name"] in write_tools and tc["id"] not in body.tool_decisions + ] + + +def _rejection_message(tool_names: list[str]) -> dict[str, Any]: + """User message telling the model a rejected call should not proceed. + + Uses list-form content so the UI, which only renders string user + content, does not show it as something the user typed. + """ + names = ", ".join(name.replace("_", " ") for name in tool_names) + return { + "role": "user", + "content": [ + { + "type": "text", + "text": ( + f"I do not want to proceed with the {names} call. Ask me for " + "clarification or suggest adjustments instead of running it." + ), + } + ], + } + + async def _execute_pending_tools( pending_tool_calls: list[dict[str, Any]], request: Request, allowed_cameras: list[str], + decisions: dict[str, str] | None = None, ) -> tuple[list[ToolCall], list[dict[str, Any]], list[dict[str, Any]]]: """ Execute a list of tool calls. + Calls the user rejected (per `decisions`) are not executed; they get a + placeholder result and a user message saying not to proceed is appended + after the tool results. + Returns: (ToolCall list for API response, tool result dicts for conversation, @@ -1042,10 +1319,28 @@ async def _execute_pending_tools( tool_calls_out: list[ToolCall] = [] tool_results: list[dict[str, Any]] = [] extra_messages: list[dict[str, Any]] = [] + rejected_tools: list[str] = [] for tool_call in pending_tool_calls: tool_name = tool_call["name"] tool_args = tool_call.get("arguments") or {} tool_call_id = tool_call["id"] + if decisions and decisions.get(tool_call_id) == "reject": + logger.debug( + "Tool %s (id: %s) was rejected by the user", tool_name, tool_call_id + ) + rejected_tools.append(tool_name) + rejected_content = json.dumps(TOOL_REJECTED_RESULT) + tool_calls_out.append( + ToolCall(name=tool_name, arguments=tool_args, response=rejected_content) + ) + tool_results.append( + { + "role": "tool", + "tool_call_id": tool_call_id, + "content": rejected_content, + } + ) + continue logger.debug( f"Executing tool: {tool_name} (id: {tool_call_id}) with arguments: {json.dumps(tool_args, indent=2)}" ) @@ -1079,17 +1374,21 @@ async def _execute_pending_tools( if isinstance(evt, dict) ] - # Extract _image_url from get_live_context results — images can - # only be sent in user messages, not tool results + # Extract _image_url from tool results — images can only be sent + # in user messages, not tool results if isinstance(tool_result, dict) and "_image_url" in tool_result: image_url = tool_result.pop("_image_url") + image_text = tool_result.pop("_image_text", None) or ( + "Here is the current live image from camera " + f"'{tool_result.get('camera', 'unknown')}'." + ) extra_messages.append( { "role": "user", "content": [ { "type": "text", - "text": f"Here is the current live image from camera '{tool_result.get('camera', 'unknown')}'.", + "text": image_text, }, { "type": "image_url", @@ -1133,6 +1432,8 @@ async def _execute_pending_tools( "content": error_content, } ) + if rejected_tools: + extra_messages.append(_rejection_message(rejected_tools)) return (tool_calls_out, tool_results, extra_messages) @@ -1179,6 +1480,8 @@ async def chat_completion( attribute_classifications=attribute_classifications, embeddings_language=_embeddings_language(config), ) + write_tools = get_write_tool_names(tools) + llm_tools = strip_tool_access(tools) conversation = [] # Build the system message only when the client hasn't already pinned one. @@ -1217,6 +1520,10 @@ async def chat_completion( tool_calls: list[ToolCall] = [] max_iterations = body.max_tool_iterations + # Resume after an approval pause: run the tail's tool calls (honoring the + # client's decisions) before asking the model for anything new. + resume_pending = _pending_tool_calls_from_tail(conversation) + logger.debug( f"Starting chat completion with {len(conversation)} message(s), " f"{len(tools)} tool(s) available, max_iterations={max_iterations}" @@ -1228,93 +1535,64 @@ async def chat_completion( async def stream_body_llm(): nonlocal conversation, stream_iterations + pending: list[dict[str, Any]] | None = resume_pending - def _emit_chain(extra: list[dict[str, Any]] | None = None): + def _emit(payload: dict[str, Any]) -> bytes: + return json.dumps(payload).encode("utf-8") + b"\n" + + def _emit_chain(extra: list[dict[str, Any]] | None = None) -> bytes: # Return the full conversation (including the system message) so # the client persists and replays it verbatim next turn. - chain = conversation + (extra or []) - return ( - json.dumps({"type": "messages", "messages": chain}).encode("utf-8") - + b"\n" + return _emit( + {"type": "messages", "messages": conversation + (extra or [])} ) while stream_iterations < max_iterations: if await request.is_disconnected(): logger.debug("Client disconnected, stopping chat stream") return - logger.debug( - f"Streaming LLM (iteration {stream_iterations + 1}/{max_iterations}) " - f"with {len(conversation)} message(s)" - ) - async for event in genai_client.chat_with_tools_stream( - messages=conversation, - tools=tools if tools else None, - tool_choice="auto", - enable_thinking=body.enable_thinking, - ): - if await request.is_disconnected(): - logger.debug("Client disconnected, stopping chat stream") - return - kind, value = event - if kind == "content_delta": - yield ( - json.dumps({"type": "content", "delta": value}).encode( - "utf-8" - ) - + b"\n" - ) - elif kind == "reasoning_delta": - yield ( - json.dumps({"type": "reasoning", "delta": value}).encode( - "utf-8" - ) - + b"\n" - ) - elif kind == "stats": - yield ( - json.dumps({"type": "stats", **value}).encode("utf-8") - + b"\n" - ) - elif kind == "message": - msg = value - if msg.get("finish_reason") == "error": - yield ( - json.dumps( + + if pending is None: + logger.debug( + f"Streaming LLM (iteration {stream_iterations + 1}/{max_iterations}) " + f"with {len(conversation)} message(s)" + ) + async for event in genai_client.chat_with_tools_stream( + messages=conversation, + tools=llm_tools if llm_tools else None, + tool_choice="auto", + enable_thinking=body.enable_thinking, + ): + if await request.is_disconnected(): + logger.debug("Client disconnected, stopping chat stream") + return + kind, value = event + if kind == "content_delta": + yield _emit({"type": "content", "delta": value}) + elif kind == "reasoning_delta": + yield _emit({"type": "reasoning", "delta": value}) + elif kind == "stats": + yield _emit({"type": "stats", **value}) + elif kind == "message": + msg = value + if msg.get("finish_reason") == "error": + yield _emit( { "type": "error", "error": "An error occurred while processing your request.", } - ).encode("utf-8") - + b"\n" - ) - return - pending = msg.get("tool_calls") - if pending: - stream_iterations += 1 - conversation.append( - build_assistant_message_for_conversation( - msg.get("content"), pending - ) - ) - if await request.is_disconnected(): - logger.debug( - "Client disconnected before tool execution" ) return - ( - _executed_calls, - tool_results, - extra_msgs, - ) = await _execute_pending_tools( - pending, request, allowed_cameras - ) - conversation.extend(tool_results) - conversation.extend(extra_msgs) - # Emit the running chain so the client can render tool - # calls live and replay them verbatim next turn. - yield _emit_chain() - break - else: + requested = msg.get("tool_calls") + if requested: + stream_iterations += 1 + conversation.append( + build_assistant_message_for_conversation( + msg.get("content"), requested + ) + ) + pending = requested + break # Streaming never appends the final assistant message # to the conversation, so add it to the chain. yield _emit_chain( @@ -1325,11 +1603,41 @@ async def chat_completion( } ] ) - yield (json.dumps({"type": "done"}).encode("utf-8") + b"\n") + yield _emit({"type": "done"}) return - else: + if pending is None: + # The stream ended without a final message; nothing + # more to run. + break + + awaiting = _tool_calls_awaiting_approval(pending, body, write_tools) + if awaiting: + # Pause before running write tools. The client shows the + # calls, collects decisions, and resends the chain. + yield _emit_chain() + yield _emit({"type": "approval_required", "tool_calls": awaiting}) + yield _emit({"type": "done"}) + return + + if await request.is_disconnected(): + logger.debug("Client disconnected before tool execution") + return + ( + _executed_calls, + tool_results, + extra_msgs, + ) = await _execute_pending_tools( + pending, request, allowed_cameras, decisions=body.tool_decisions + ) + conversation.extend(tool_results) + conversation.extend(extra_msgs) + pending = None + # Emit the running chain so the client can render tool + # calls live and replay them verbatim next turn. yield _emit_chain() - yield json.dumps({"type": "done"}).encode("utf-8") + b"\n" + + yield _emit_chain() + yield _emit({"type": "done"}) return StreamingResponse( stream_body_llm(), @@ -1338,102 +1646,129 @@ async def chat_completion( ) try: + pending_tool_calls = resume_pending while tool_iterations < max_iterations: - logger.debug( - f"Calling LLM (iteration {tool_iterations + 1}/{max_iterations}) " - f"with {len(conversation)} message(s) in conversation" - ) - response = genai_client.chat_with_tools( - messages=conversation, - tools=tools if tools else None, - tool_choice="auto", - enable_thinking=body.enable_thinking, - ) - - if response.get("finish_reason") == "error": - logger.error("GenAI client returned an error") - return JSONResponse( - content={ - "error": "An error occurred while processing your request.", - }, - status_code=500, - ) - - conversation.append( - build_assistant_message_for_conversation( - response.get("content"), response.get("tool_calls") - ) - ) - - pending_tool_calls = response.get("tool_calls") - if not pending_tool_calls: + if pending_tool_calls is None: logger.debug( - f"Chat completion finished with final answer (iterations: {tool_iterations})" + f"Calling LLM (iteration {tool_iterations + 1}/{max_iterations}) " + f"with {len(conversation)} message(s) in conversation" + ) + response = genai_client.chat_with_tools( + messages=conversation, + tools=llm_tools if llm_tools else None, + tool_choice="auto", + enable_thinking=body.enable_thinking, ) - final_content = response.get("content") or "" - if body.stream: - final_reasoning = response.get("reasoning") + if response.get("finish_reason") == "error": + logger.error("GenAI client returned an error") + return JSONResponse( + content={ + "error": "An error occurred while processing your request.", + }, + status_code=500, + ) - chain = list(conversation) + conversation.append( + build_assistant_message_for_conversation( + response.get("content"), response.get("tool_calls") + ) + ) - async def stream_body() -> Any: - yield ( - json.dumps({"type": "messages", "messages": chain}).encode( - "utf-8" - ) - + b"\n" - ) - # Emit the full reasoning trace up front when the - # underlying client did not stream it - if final_reasoning: + pending_tool_calls = response.get("tool_calls") + if not pending_tool_calls: + logger.debug( + f"Chat completion finished with final answer (iterations: {tool_iterations})" + ) + final_content = response.get("content") or "" + + if body.stream: + final_reasoning = response.get("reasoning") + + chain = list(conversation) + + async def stream_body() -> Any: yield ( json.dumps( - {"type": "reasoning", "delta": final_reasoning} + {"type": "messages", "messages": chain} ).encode("utf-8") + b"\n" ) - # Stream content in word-sized chunks for smooth UX - for part in chunk_content(final_content): - yield ( - json.dumps({"type": "content", "delta": part}).encode( - "utf-8" + # Emit the full reasoning trace up front when the + # underlying client did not stream it + if final_reasoning: + yield ( + json.dumps( + {"type": "reasoning", "delta": final_reasoning} + ).encode("utf-8") + + b"\n" ) - + b"\n" - ) - yield json.dumps({"type": "done"}).encode("utf-8") + b"\n" + # Stream content in word-sized chunks for smooth UX + for part in chunk_content(final_content): + yield ( + json.dumps( + {"type": "content", "delta": part} + ).encode("utf-8") + + b"\n" + ) + yield json.dumps({"type": "done"}).encode("utf-8") + b"\n" - return StreamingResponse( - stream_body(), - media_type="application/x-ndjson", + return StreamingResponse( + stream_body(), + media_type="application/x-ndjson", + ) + + return JSONResponse( + content=ChatCompletionResponse( + message=ChatMessageResponse( + role="assistant", + content=final_content, + reasoning=response.get("reasoning"), + tool_calls=None, + ), + finish_reason=response.get("finish_reason", "stop"), + tool_iterations=tool_iterations, + tool_calls=tool_calls, + messages=list(conversation), + ).model_dump(), ) + tool_iterations += 1 + logger.debug( + f"Tool calls detected (iteration {tool_iterations}/{max_iterations}): " + f"{len(pending_tool_calls)} tool(s) to execute" + ) + + awaiting = _tool_calls_awaiting_approval( + pending_tool_calls, body, write_tools + ) + if awaiting: + # Pause before running write tools; the client resends the + # returned chain with its decisions to continue. return JSONResponse( content=ChatCompletionResponse( message=ChatMessageResponse( role="assistant", - content=final_content, - reasoning=response.get("reasoning"), - tool_calls=None, + content=None, + tool_calls=[ToolCallInvocation(**tc) for tc in awaiting], ), - finish_reason=response.get("finish_reason", "stop"), + finish_reason="approval_required", tool_iterations=tool_iterations, tool_calls=tool_calls, messages=list(conversation), ).model_dump(), ) - tool_iterations += 1 - logger.debug( - f"Tool calls detected (iteration {tool_iterations}/{max_iterations}): " - f"{len(pending_tool_calls)} tool(s) to execute" - ) executed_calls, tool_results, extra_msgs = await _execute_pending_tools( - pending_tool_calls, request, allowed_cameras + pending_tool_calls, + request, + allowed_cameras, + decisions=body.tool_decisions, ) tool_calls.extend(executed_calls) conversation.extend(tool_results) conversation.extend(extra_msgs) + pending_tool_calls = None logger.debug( f"Added {len(tool_results)} tool result(s) to conversation. " f"Continuing with next LLM call..." diff --git a/frigate/api/chat_util.py b/frigate/api/chat_util.py index a2f29c75b2..79e4b1f1d4 100644 --- a/frigate/api/chat_util.py +++ b/frigate/api/chat_util.py @@ -44,6 +44,11 @@ def chunk_content(content: str, chunk_size: int = 80) -> Generator[str, None, No yield " ".join(current) +def format_local_time(timestamp: float) -> str: + """Format a unix timestamp as the server-local string quoted to users.""" + return datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %I:%M:%S %p") + + def format_events_with_local_time( events_list: list[dict[str, Any]], ) -> list[dict[str, Any]]: @@ -58,11 +63,9 @@ def format_events_with_local_time( 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") + copy_evt["start_time_local"] = format_local_time(start_ts) 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") + copy_evt["end_time_local"] = format_local_time(end_ts) except (TypeError, ValueError, OSError): pass result.append(copy_evt) diff --git a/frigate/api/defs/request/chat_body.py b/frigate/api/defs/request/chat_body.py index 5ca674fbc8..4a74c39259 100644 --- a/frigate/api/defs/request/chat_body.py +++ b/frigate/api/defs/request/chat_body.py @@ -1,6 +1,6 @@ """Chat API request models.""" -from typing import Any +from typing import Any, Literal from pydantic import BaseModel, Field @@ -59,3 +59,12 @@ class ChatCompletionRequest(BaseModel): "Ignored by providers that do not expose a per-request thinking switch." ), ) + tool_decisions: dict[str, Literal["approve", "reject"]] = Field( + default_factory=dict, + description=( + "Decisions for tool calls that paused for approval, keyed by tool " + "call ID. Send these with the conversation chain returned alongside " + "an approval request; rejected calls are reported to the model as " + "declined instead of being executed." + ), + ) diff --git a/frigate/genai/prompts.py b/frigate/genai/prompts.py index b6df26d5fc..90374e39c1 100644 --- a/frigate/genai/prompts.py +++ b/frigate/genai/prompts.py @@ -311,6 +311,10 @@ def get_tool_definitions( Descriptions here stay mechanical: which tool to reach for, and how the filters relate to each other, is stated once in the system prompt so the guidance is not paid for twice on every request. + + Each definition carries a Frigate-only `access` field ("read" or "write"); + write tools pause for user approval in the chat loop. Strip it with + `strip_tool_access` before sending the list to a provider. """ search_objects_properties: dict[str, Any] = { "camera": { @@ -382,6 +386,7 @@ def get_tool_definitions( return [ { "type": "function", + "access": "read", "function": { "name": "search_objects", "description": search_objects_description, @@ -394,6 +399,7 @@ def get_tool_definitions( }, { "type": "function", + "access": "read", "function": { "name": "get_categorized_object_names", "description": ( @@ -411,6 +417,7 @@ def get_tool_definitions( }, { "type": "function", + "access": "read", "function": { "name": "find_similar_objects", "description": ( @@ -474,6 +481,7 @@ def get_tool_definitions( }, { "type": "function", + "access": "write", "function": { "name": "set_camera_state", "description": ( @@ -529,6 +537,7 @@ def get_tool_definitions( }, { "type": "function", + "access": "read", "function": { "name": "get_live_context", "description": ( @@ -553,6 +562,7 @@ def get_tool_definitions( }, { "type": "function", + "access": "write", "function": { "name": "start_camera_watch", "description": ( @@ -596,6 +606,7 @@ def get_tool_definitions( }, { "type": "function", + "access": "write", "function": { "name": "stop_camera_watch", "description": "Cancel the currently running watch job.", @@ -608,6 +619,7 @@ def get_tool_definitions( }, { "type": "function", + "access": "read", "function": { "name": "get_profile_status", "description": ( @@ -624,6 +636,7 @@ def get_tool_definitions( }, { "type": "function", + "access": "read", "function": { "name": "get_recap", "description": ( @@ -656,9 +669,120 @@ def get_tool_definitions( }, }, }, + { + "type": "function", + "access": "read", + "function": { + "name": "get_export_cases", + "description": ( + "List the export cases (named groups of exported clips) with " + "their IDs, descriptions, and how many exports each holds. " + "Call this before create_export when the user wants a clip " + "added to an existing case." + ), + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + }, + { + "type": "function", + "access": "write", + "function": { + "name": "create_export", + "description": ( + "Export a camera's recording for a time range to a " + "downloadable file, optionally attached to an existing export " + "case. Only call this when the user explicitly asks to export " + "or save a clip." + ), + "parameters": { + "type": "object", + "properties": { + "camera": { + "type": "string", + "description": "Camera ID to export from.", + }, + "start_time": { + "type": "string", + "description": "Start of the clip in ISO 8601 format (e.g. '2025-03-15T08:00:00').", + }, + "end_time": { + "type": "string", + "description": "End of the clip in ISO 8601 format (e.g. '2025-03-15T08:05:00').", + }, + "name": { + "type": "string", + "description": "Friendly name for the export (optional).", + }, + "source": { + "type": "string", + "enum": ["recordings", "preview"], + "description": ( + "'recordings' (default) exports full-quality footage; " + "'preview' builds a low-resolution timelapse." + ), + "default": "recordings", + }, + "export_case_id": { + "type": "string", + "description": ( + "ID of an existing export case to attach the export " + "to. Use get_export_cases to find it." + ), + }, + }, + "required": ["camera", "start_time", "end_time"], + }, + }, + }, + { + "type": "function", + "access": "read", + "function": { + "name": "get_event_image", + "description": ( + "View the thumbnail or snapshot image of a specific tracked " + "object so you can describe what it shows. Use the event id " + "from search_objects, find_similar_objects, or an attached " + "event." + ), + "parameters": { + "type": "object", + "properties": { + "event_id": { + "type": "string", + "description": "ID of the tracked object to view.", + }, + "image": { + "type": "string", + "enum": ["thumbnail", "snapshot"], + "description": ( + "'thumbnail' (default) is a small crop of the object; " + "'snapshot' is the full camera frame." + ), + "default": "thumbnail", + }, + }, + "required": ["event_id"], + }, + }, + }, ] +def get_write_tool_names(tools: list[dict[str, Any]]) -> set[str]: + """Names of the tools whose `access` is "write" (they change state).""" + return {tool["function"]["name"] for tool in tools if tool.get("access") == "write"} + + +def strip_tool_access(tools: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Drop the Frigate-only `access` field before handing tools to a provider.""" + return [{k: v for k, v in tool.items() if k != "access"} for tool in tools] + + def build_chat_system_prompt( config: FrigateConfig, allowed_cameras: list[str], diff --git a/frigate/test/test_chat_tool_approval.py b/frigate/test/test_chat_tool_approval.py new file mode 100644 index 0000000000..46873bebb3 --- /dev/null +++ b/frigate/test/test_chat_tool_approval.py @@ -0,0 +1,479 @@ +"""Tests for chat tool approval and the export and event image tools.""" + +import asyncio +import base64 +import json +import os +import tempfile +import unittest +from datetime import datetime +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import cv2 +import numpy as np +from playhouse.sqlite_ext import SqliteExtDatabase + +from frigate.api.chat import ( + TOOL_REJECTED_RESULT, + _execute_create_export, + _execute_get_event_image, + _execute_get_export_cases, + _execute_pending_tools, + _pending_tool_calls_from_tail, + _tool_calls_awaiting_approval, +) +from frigate.api.defs.request.chat_body import ChatCompletionRequest +from frigate.genai.prompts import ( + get_tool_definitions, + get_write_tool_names, + strip_tool_access, +) +from frigate.genai.utils import build_assistant_message_for_conversation +from frigate.jobs.export import ExportQueueFullError +from frigate.models import Event, Export, ExportCase, Previews, Recordings + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +def _request(role: str = "admin", supports_vision: bool = True): + camera = SimpleNamespace( + record=SimpleNamespace(export=SimpleNamespace(chapters=None)), + ) + app = SimpleNamespace( + frigate_config=SimpleNamespace( + cameras={"driveway": camera, "garage": camera}, + proxy=SimpleNamespace(separator=","), + ), + genai_manager=SimpleNamespace( + chat_client=SimpleNamespace(supports_vision=supports_vision), + ), + ) + return SimpleNamespace(app=app, headers={"remote-role": role}) + + +def _body(**kwargs) -> ChatCompletionRequest: + return ChatCompletionRequest(messages=[], **kwargs) + + +WRITE_TOOLS = get_write_tool_names(get_tool_definitions()) + + +class TestToolRegistry(unittest.TestCase): + def test_every_tool_declares_access(self): + for tool in get_tool_definitions(): + self.assertIn(tool.get("access"), ("read", "write"), tool) + + def test_write_tools(self): + self.assertEqual( + WRITE_TOOLS, + { + "set_camera_state", + "start_camera_watch", + "stop_camera_watch", + "create_export", + }, + ) + + def test_strip_tool_access_removes_frigate_field(self): + for tool in strip_tool_access(get_tool_definitions()): + self.assertNotIn("access", tool) + self.assertEqual(set(tool), {"type", "function"}) + + def test_new_tools_are_registered(self): + names = {t["function"]["name"] for t in get_tool_definitions()} + self.assertIn("get_export_cases", names) + self.assertIn("create_export", names) + self.assertIn("get_event_image", names) + + def test_create_export_requires_time_range(self): + tool = next( + t + for t in get_tool_definitions() + if t["function"]["name"] == "create_export" + ) + params = tool["function"]["parameters"] + self.assertEqual(params["required"], ["camera", "start_time", "end_time"]) + self.assertNotIn("event_id", params["properties"]) + self.assertNotIn("new_case_name", params["properties"]) + + def test_get_event_image_requires_event_id(self): + tool = next( + t + for t in get_tool_definitions() + if t["function"]["name"] == "get_event_image" + ) + self.assertEqual(tool["function"]["parameters"]["required"], ["event_id"]) + + +class TestApprovalHelpers(unittest.TestCase): + def test_tail_without_tool_calls_is_not_pending(self): + self.assertIsNone(_pending_tool_calls_from_tail([])) + self.assertIsNone( + _pending_tool_calls_from_tail([{"role": "user", "content": "hi"}]) + ) + self.assertIsNone( + _pending_tool_calls_from_tail([{"role": "assistant", "content": "ok"}]) + ) + + def test_tail_with_tool_calls_is_parsed(self): + tail = build_assistant_message_for_conversation( + None, [{"id": "call_1", "name": "create_export", "arguments": {"a": 1}}] + ) + pending = _pending_tool_calls_from_tail([{"role": "user"}, tail]) + self.assertEqual( + pending, [{"id": "call_1", "name": "create_export", "arguments": {"a": 1}}] + ) + + def test_read_tools_never_await_approval(self): + pending = [{"id": "c1", "name": "search_objects", "arguments": {}}] + self.assertEqual( + _tool_calls_awaiting_approval(pending, _body(), WRITE_TOOLS), [] + ) + + def test_write_tools_await_approval(self): + pending = [ + {"id": "c1", "name": "search_objects", "arguments": {}}, + {"id": "c2", "name": "create_export", "arguments": {"camera": "x"}}, + ] + awaiting = _tool_calls_awaiting_approval(pending, _body(), WRITE_TOOLS) + self.assertEqual( + awaiting, + [{"id": "c2", "name": "create_export", "arguments": {"camera": "x"}}], + ) + + def test_decided_calls_skip_approval(self): + pending = [ + {"id": "c2", "name": "create_export", "arguments": {}}, + {"id": "c3", "name": "set_camera_state", "arguments": {}}, + ] + body = _body(tool_decisions={"c2": "approve", "c3": "reject"}) + self.assertEqual(_tool_calls_awaiting_approval(pending, body, WRITE_TOOLS), []) + + +class TestExecutePendingTools(unittest.TestCase): + def test_rejected_call_is_not_executed(self): + execute = AsyncMock(return_value={"success": True}) + pending = [{"id": "c1", "name": "create_export", "arguments": {}}] + with patch("frigate.api.chat._execute_tool_internal", execute): + calls, results, extra = _run( + _execute_pending_tools( + pending, _request(), ["driveway"], decisions={"c1": "reject"} + ) + ) + execute.assert_not_called() + self.assertEqual(json.loads(results[0]["content"]), TOOL_REJECTED_RESULT) + self.assertEqual(results[0]["tool_call_id"], "c1") + self.assertEqual(calls[0].name, "create_export") + # The user's intent goes to the model as a follow-up user message. + self.assertEqual(len(extra), 1) + self.assertEqual(extra[0]["role"], "user") + text = extra[0]["content"][0]["text"] + self.assertIn("do not want to proceed", text) + self.assertIn("create export", text) + self.assertIn("clarification", text) + + def test_approved_call_is_executed(self): + execute = AsyncMock(return_value={"success": True}) + pending = [{"id": "c1", "name": "create_export", "arguments": {}}] + with patch("frigate.api.chat._execute_tool_internal", execute): + _calls, results, _extra = _run( + _execute_pending_tools( + pending, _request(), ["driveway"], decisions={"c1": "approve"} + ) + ) + execute.assert_awaited_once() + self.assertEqual(json.loads(results[0]["content"]), {"success": True}) + + def test_image_text_becomes_user_message(self): + execute = AsyncMock( + return_value={ + "id": "evt", + "_image_url": "data:image/jpeg;base64,xx", + "_image_text": "Here is the thumbnail.", + } + ) + pending = [{"id": "c1", "name": "get_event_image", "arguments": {}}] + with patch("frigate.api.chat._execute_tool_internal", execute): + _calls, results, extra = _run( + _execute_pending_tools(pending, _request(), ["driveway"]) + ) + self.assertEqual(json.loads(results[0]["content"]), {"id": "evt"}) + self.assertEqual(extra[0]["role"], "user") + self.assertEqual(extra[0]["content"][0]["text"], "Here is the thumbnail.") + self.assertEqual( + extra[0]["content"][1]["image_url"]["url"], "data:image/jpeg;base64,xx" + ) + + +class DatabaseTestCase(unittest.TestCase): + models = [Event, Export, ExportCase, Recordings, Previews] + + def setUp(self): + self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + self.tmp.close() + self.db = SqliteExtDatabase(self.tmp.name) + for model in self.models: + model.bind(self.db, bind_refs=False, bind_backrefs=False) + self.db.connect() + self.db.create_tables(self.models) + + def tearDown(self): + self.db.close() + os.unlink(self.tmp.name) + + def make_event(self, event_id, camera="driveway", thumbnail="", **overrides): + fields = dict( + id=event_id, + label="car", + sub_label=None, + camera=camera, + start_time=1_700_000_100, + end_time=1_700_000_110, + top_score=0.9, + score=0.9, + false_positive=False, + zones=[], + thumbnail=thumbnail, + has_clip=True, + has_snapshot=False, + region=[0, 0, 1, 1], + box=[0, 0, 1, 1], + area=1, + retain_indefinitely=False, + ratio=1.0, + plus_id="", + model_hash="", + detector_type="", + model_type="", + data={}, + ) + fields.update(overrides) + return Event.create(**fields) + + def make_case(self, case_id, name="Case"): + now = datetime.fromtimestamp(1_700_000_000) + return ExportCase.create( + id=case_id, name=name, description=None, created_at=now, updated_at=now + ) + + +def _jpeg_base64() -> str: + frame = np.zeros((8, 8, 3), dtype=np.uint8) + _, encoded = cv2.imencode(".jpg", frame) + return base64.b64encode(encoded.tobytes()).decode("utf-8") + + +class TestGetEventImage(DatabaseTestCase): + def test_requires_vision(self): + self.make_event("evt", thumbnail=_jpeg_base64()) + result = _run( + _execute_get_event_image( + _request(supports_vision=False), {"event_id": "evt"}, ["driveway"] + ) + ) + self.assertIn("vision", result["error"]) + + def test_unknown_event(self): + result = _run( + _execute_get_event_image(_request(), {"event_id": "nope"}, ["driveway"]) + ) + self.assertIn("nope", result["error"]) + + def test_camera_access_denied(self): + self.make_event("evt", camera="garage", thumbnail=_jpeg_base64()) + result = _run( + _execute_get_event_image(_request(), {"event_id": "evt"}, ["driveway"]) + ) + self.assertIn("access denied", result["error"]) + + def test_thumbnail_is_attached(self): + self.make_event( + "evt", thumbnail=_jpeg_base64(), data={"description": "a red car"} + ) + result = _run( + _execute_get_event_image(_request(), {"event_id": "evt"}, ["driveway"]) + ) + self.assertEqual(result["id"], "evt") + self.assertEqual(result["image"], "thumbnail") + self.assertEqual(result["description"], "a red car") + self.assertIn("start_time_local", result) + self.assertTrue(result["_image_url"].startswith("data:image/jpeg;base64,")) + self.assertIn("thumbnail", result["_image_text"]) + + def test_snapshot_falls_back_to_thumbnail(self): + self.make_event("evt", thumbnail=_jpeg_base64(), has_snapshot=False) + result = _run( + _execute_get_event_image( + _request(), {"event_id": "evt", "image": "snapshot"}, ["driveway"] + ) + ) + self.assertEqual(result["image"], "thumbnail") + self.assertIn("note", result) + + def test_no_image_available(self): + self.make_event("evt", thumbnail="") + with patch("frigate.api.chat.get_event_thumbnail_bytes", return_value=None): + result = _run( + _execute_get_event_image(_request(), {"event_id": "evt"}, ["driveway"]) + ) + self.assertIn("error", result) + + +class TestGetExportCases(DatabaseTestCase): + def test_no_cases(self): + result = _execute_get_export_cases(["driveway"]) + self.assertEqual(result["cases"], []) + self.assertIn("message", result) + + def test_counts_only_accessible_exports(self): + self.make_case("case_a", name="Break-in") + self.make_case("case_b", name="Empty") + for idx, camera in enumerate(["driveway", "driveway", "garage"]): + Export.create( + id=f"exp_{idx}", + camera=camera, + name=f"Export {idx}", + date=datetime.fromtimestamp(1_700_000_000 + idx), + video_path=f"/exports/{idx}.mp4", + thumb_path=f"/exports/{idx}.jpg", + in_progress=False, + export_case="case_a", + ) + + result = _execute_get_export_cases(["driveway"]) + by_id = {c["id"]: c for c in result["cases"]} + self.assertEqual(by_id["case_a"]["export_count"], 2) + self.assertEqual(by_id["case_b"]["export_count"], 0) + self.assertEqual(by_id["case_a"]["name"], "Break-in") + self.assertIn("created_at_local", by_id["case_a"]) + + +class TestCreateExport(DatabaseTestCase): + def setUp(self): + super().setUp() + Recordings.create( + id="rec_1", + camera="driveway", + path="/recordings/rec_1.mp4", + start_time=1_700_000_000, + end_time=1_700_001_000, + duration=1000, + ) + + @staticmethod + def _iso(timestamp: int) -> str: + return datetime.fromtimestamp(timestamp).strftime("%Y-%m-%dT%H:%M:%S") + + def _args(self, **overrides): + args = { + "camera": "driveway", + "start_time": self._iso(1_700_000_100), + "end_time": self._iso(1_700_000_200), + } + args.update(overrides) + return args + + def test_requires_camera_and_range(self): + result = _run(_execute_create_export(_request(), {}, ["driveway"])) + self.assertIn("required", result["error"]) + + def test_camera_access_denied(self): + result = _run( + _execute_create_export( + _request(), self._args(camera="garage"), ["driveway"] + ) + ) + self.assertIn("access denied", result["error"]) + + def test_end_before_start(self): + result = _run( + _execute_create_export( + _request(), + self._args(end_time=self._iso(1_700_000_050)), + ["driveway"], + ) + ) + self.assertIn("after start_time", result["error"]) + + def test_existing_case_requires_admin(self): + self.make_case("case_a") + result = _run( + _execute_create_export( + _request(role="viewer"), + self._args(export_case_id="case_a"), + ["driveway"], + ) + ) + self.assertIn("admins", result["error"]) + + def test_unknown_case(self): + result = _run( + _execute_create_export( + _request(), self._args(export_case_id="nope"), ["driveway"] + ) + ) + self.assertIn("nope", result["error"]) + + def test_export_is_queued(self): + self.make_case("case_a") + with patch("frigate.api.chat.start_export_job") as start: + result = _run( + _execute_create_export( + _request(), + self._args(name="Delivery", export_case_id="case_a"), + ["driveway"], + ) + ) + self.assertTrue(result["success"]) + self.assertEqual(result["status"], "queued") + self.assertEqual(result["camera"], "driveway") + self.assertEqual(result["export_case_id"], "case_a") + job = start.call_args.args[1] + self.assertEqual(job.camera, "driveway") + self.assertEqual(job.request_start_time, 1_700_000_100) + self.assertEqual(job.request_end_time, 1_700_000_200) + self.assertEqual(job.name, "Delivery") + self.assertEqual(job.export_case_id, "case_a") + + def test_no_recordings_in_range(self): + result = _run( + _execute_create_export( + _request(), + { + "camera": "driveway", + "start_time": "2030-01-01T00:00:00", + "end_time": "2030-01-01T01:00:00", + }, + ["driveway"], + ) + ) + self.assertIn("No recordings", result["error"]) + + def test_queue_full(self): + with patch( + "frigate.api.chat.start_export_job", side_effect=ExportQueueFullError() + ): + result = _run( + _execute_create_export(_request(), self._args(), ["driveway"]) + ) + self.assertIn("queue is full", result["error"]) + + +class TestSetCameraStateRoles(unittest.TestCase): + def test_non_admin_is_rejected(self): + from frigate.api.chat import _execute_set_camera_state + + result = _run( + _execute_set_camera_state( + _request(role="viewer"), + {"camera": "driveway", "feature": "detect", "value": "OFF"}, + ) + ) + self.assertIn("Admin", result["error"]) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/web/public/locales/en/views/chat.json b/web/public/locales/en/views/chat.json index 363b0e68e4..0a4e028741 100644 --- a/web/public/locales/en/views/chat.json +++ b/web/public/locales/en/views/chat.json @@ -55,6 +55,12 @@ "auto_scroll": { "title": "Auto-scroll", "desc": "Follow new messages as they arrive." + }, + "always_allow": { + "title": "Always allowed actions", + "desc": "Actions the assistant may run without asking first.", + "none": "None", + "reset": "Reset" } }, "stats": { @@ -68,5 +74,15 @@ }, "thinking": { "toggle": "Toggle thinking" + }, + "approval": { + "title": "Approve {{tool}}?", + "desc": "This action changes something in Frigate. Review the details before approving.", + "approve": "Approve", + "always_allow": "Always allow", + "reject": "Reject", + "approved": "Approved", + "rejected": "Rejected", + "placeholder": "Approve or reject the pending action to continue" } } diff --git a/web/src/components/chat/ChatComposer.tsx b/web/src/components/chat/ChatComposer.tsx index 5ccfe93786..b165c4eb81 100644 --- a/web/src/components/chat/ChatComposer.tsx +++ b/web/src/components/chat/ChatComposer.tsx @@ -26,6 +26,9 @@ type ChatComposerProps = { isLoading?: boolean; onStop?: () => void; + /** Blocks input without showing the stop button, e.g. while a tool call + * is waiting for the user's approval. */ + disabled?: boolean; attachedEventId?: string | null; onClearAttachment?: () => void; @@ -45,6 +48,7 @@ export function ChatComposer({ setThinkingEnabled, isLoading = false, onStop, + disabled = false, attachedEventId, onClearAttachment, onAttach, @@ -62,6 +66,7 @@ export function ChatComposer({ const showPaperclip = !!onAttach; const showStop = isLoading && !!onStop; + const inputBlocked = isLoading || disabled; return (
@@ -77,7 +82,7 @@ export function ChatComposer({ {attachedEventId && ( sendMessage(text)} - disabled={isLoading} + disabled={inputBlocked} /> )}
@@ -85,7 +90,7 @@ export function ChatComposer({ )} {supportsThinking && ( @@ -103,7 +108,7 @@ export function ChatComposer({ !thinkingEnabled && "text-secondary-foreground", )} onClick={() => setThinkingEnabled(!thinkingEnabled)} - disabled={isLoading} + disabled={inputBlocked} > @@ -122,6 +127,7 @@ export function ChatComposer({ onChange={(e) => setInput(e.target.value)} onKeyDown={handleKeyDown} aria-busy={isLoading} + disabled={disabled} /> {showStop ? (
+ +
+
+
{t("settings.always_allow.title")}
+
+ {t("settings.always_allow.desc")} +
+
+ {alwaysAllowTools.length > 0 ? ( +
+ {alwaysAllowTools.map((name) => ( + + {formatToolName(name)} + + ))} +
+ ) : ( +
+ {t("settings.always_allow.none")} +
+ )} + +
); diff --git a/web/src/components/chat/ToolApprovalCard.tsx b/web/src/components/chat/ToolApprovalCard.tsx new file mode 100644 index 0000000000..4ca17694af --- /dev/null +++ b/web/src/components/chat/ToolApprovalCard.tsx @@ -0,0 +1,97 @@ +import { useTranslation } from "react-i18next"; +import { LuShieldAlert, LuCheck, LuX } from "react-icons/lu"; +import { Button } from "@/components/ui/button"; +import { cn } from "@/lib/utils"; +import { formatToolName } from "@/utils/chatUtil"; +import type { PendingToolCall, ToolDecision } from "@/types/chat"; + +type ToolApprovalCardProps = { + toolCall: PendingToolCall; + decision?: ToolDecision; + onApprove: (id: string) => void; + onAlwaysAllow: (id: string, name: string) => void; + onReject: (id: string) => void; +}; + +/** + * Prompt shown when the assistant wants to run a state-changing tool. + * Renders the call's arguments and approve / always allow / reject actions; + * once decided it collapses into a status line. + */ +export function ToolApprovalCard({ + toolCall, + decision, + onApprove, + onAlwaysAllow, + onReject, +}: ToolApprovalCardProps) { + const { t } = useTranslation(["views/chat"]); + const displayName = formatToolName(toolCall.name); + const hasArguments = Object.keys(toolCall.arguments ?? {}).length > 0; + + return ( +
+
+ +
+ + {t("approval.title", { tool: displayName })} + + + {t("approval.desc")} + +
+
+ {hasArguments && ( + 
+          {JSON.stringify(toolCall.arguments, null, 2)}
+
+ )} + {decision ? ( +
+ {decision === "approve" ? ( + + ) : ( + + )} + {decision === "approve" + ? t("approval.approved") + : t("approval.rejected")} +
+ ) : ( +
+ + + +
+ )} +
+ ); +} diff --git a/web/src/components/chat/ToolCallsGroup.tsx b/web/src/components/chat/ToolCallsGroup.tsx index 7cb20a1adb..731548e46b 100644 --- a/web/src/components/chat/ToolCallsGroup.tsx +++ b/web/src/components/chat/ToolCallsGroup.tsx @@ -7,19 +7,12 @@ import { } from "@/components/ui/collapsible"; import { LuChevronsUpDown } from "react-icons/lu"; import type { ToolCall } from "@/types/chat"; +import { formatToolName } from "@/utils/chatUtil"; type ToolCallsGroupProps = { toolCalls: ToolCall[]; }; -function normalizeName(name: string): string { - return name - .replace(/_/g, " ") - .split(" ") - .map((word) => word.charAt(0).toUpperCase() + word.slice(1).toLowerCase()) - .join(" "); -} - export function ToolCallsGroup({ toolCalls }: ToolCallsGroupProps) { const grouped = useMemo(() => { const map = new Map(); @@ -53,7 +46,7 @@ type ToolCallRowProps = { function ToolCallRow({ name, calls }: ToolCallRowProps) { const { t } = useTranslation(["views/chat"]); const [open, setOpen] = useState(false); - const displayName = normalizeName(name); + const displayName = formatToolName(name); const label = calls.length > 1 ? `${displayName} (\u00d7${calls.length})` : displayName; diff --git a/web/src/pages/Chat.tsx b/web/src/pages/Chat.tsx index bd92097873..6242dd88dc 100644 --- a/web/src/pages/Chat.tsx +++ b/web/src/pages/Chat.tsx @@ -8,6 +8,7 @@ import { ChatEventThumbnailsRow } from "@/components/chat/ChatEventThumbnailsRow import { MessageBubble } from "@/components/chat/ChatMessage"; import { ReasoningBubble } from "@/components/chat/ReasoningBubble"; import { ToolCallsGroup } from "@/components/chat/ToolCallsGroup"; +import { ToolApprovalCard } from "@/components/chat/ToolApprovalCard"; import { ChatStartingState } from "@/components/chat/ChatStartingState"; import { ChatComposer } from "@/components/chat/ChatComposer"; import ChatSettings from "@/components/chat/ChatSettings"; @@ -15,11 +16,13 @@ import type { ChatMessage, ChatStats, GenAIModelsResponse, + PendingToolCall, ShowStatsMode, + ToolDecision, } from "@/types/chat"; import { usePersistence } from "@/hooks/use-persistence"; import { - getEventIdsFromSearchObjectsToolCalls, + getEventIdsFromToolCalls, getFindSimilarObjectsFromToolCalls, prependAttachment, streamChatCompletion, @@ -40,6 +43,13 @@ const hasText = (content: unknown): content is string => const toWire = (messages: ChatMessage[]): ChatMessage[] => messages.map(({ reasoning: _r, stats: _s, ...rest }) => rest); +// Stable default so usePersistence does not reload on every render. +const NO_TOOLS: string[] = []; + +type ResumeOptions = { + toolDecisions: Record; +}; + export default function ChatPage() { const { t } = useTranslation(["views/chat"]); const [input, setInput] = useState(""); @@ -48,6 +58,21 @@ export default function ChatPage() { const [isLoading, setIsLoading] = useState(false); const [error, setError] = useState(null); const [attachedEventId, setAttachedEventId] = useState(null); + // Write tool calls the backend paused on, plus the user's decisions so far. + const [pendingApprovals, setPendingApprovals] = useState< + PendingToolCall[] | null + >(null); + const [approvalDecisions, setApprovalDecisions] = useState< + Record + >({}); + // Tools the user chose to always allow. Kept only in this browser; the + // backend never sees the list, the client just answers for them. + const [alwaysAllowTools, setAlwaysAllowTools] = usePersistence( + "chat-always-allow-tools", + NO_TOOLS, + ); + const alwaysAllowRef = useRef(NO_TOOLS); + alwaysAllowRef.current = alwaysAllowTools ?? NO_TOOLS; const [showStats, setShowStats] = usePersistence( "chat-show-stats", "while_generating", @@ -62,6 +87,7 @@ export default function ChatPage() { ); const scrollRef = useRef(null); const abortRef = useRef(null); + const loadingRef = useRef(false); const { data: genaiInfo } = useSWR("genai/models", { revalidateOnFocus: false, @@ -92,14 +118,27 @@ export default function ChatPage() { }, [messages, streaming, autoScroll]); const submitConversation = useCallback( - async (messagesToSend: ChatMessage[]) => { - if (isLoading) return; + async function submit( + messagesToSend: ChatMessage[], + resume?: ResumeOptions, + ) { + if (loadingRef.current) return; const last = messagesToSend[messagesToSend.length - 1]; - if (!last || last.role !== "user" || !hasText(last.content)) return; + if (!last) return; + // A normal turn ends with the user's message; a resume after an + // approval pause ends with the assistant's pending tool calls. + if (resume) { + if (last.role !== "assistant" || !last.tool_calls?.length) return; + } else if (last.role !== "user" || !hasText(last.content)) { + return; + } setError(null); + setPendingApprovals(null); + setApprovalDecisions({}); setMessages(messagesToSend); setStreaming({ content: "", reasoning: "", chain: [] }); + loadingRef.current = true; setIsLoading(true); const baseURL = axios.defaults.baseURL ?? ""; @@ -116,6 +155,7 @@ export default function ChatPage() { let stats: ChatStats | undefined; let reasoning = ""; let hadError = false; + let approvals: PendingToolCall[] | null = null; await streamChatCompletion( url, @@ -138,32 +178,99 @@ export default function ChatPage() { stats = s; setStreaming((cur) => (cur ? { ...cur, stats: s } : cur)); }, + onApprovalRequired: (toolCalls) => { + approvals = toolCalls; + }, onError: (message) => { hadError = true; setError(message); }, onDone: () => { abortRef.current = null; + loadingRef.current = false; setIsLoading(false); setStreaming(null); const lastMsg = chain[chain.length - 1]; if (!hadError && lastMsg?.role === "assistant") { - setMessages( - chain.map((m, i) => - i === chain.length - 1 - ? { ...m, reasoning: reasoning || undefined, stats } - : m, - ), + const committed = chain.map((m, i) => + i === chain.length - 1 + ? { ...m, reasoning: reasoning || undefined, stats } + : m, ); + setMessages(committed); + if (approvals?.length) { + // Calls to always-allowed tools are answered here without + // prompting; anything else waits for the user. + const allowed = alwaysAllowRef.current; + const auto: Record = {}; + for (const tc of approvals) { + if (allowed.includes(tc.name)) auto[tc.id] = "approve"; + } + if (Object.keys(auto).length === approvals.length) { + submit(committed, { toolDecisions: auto }); + } else { + setApprovalDecisions(auto); + setPendingApprovals(approvals); + } + } } }, defaultErrorMessage: t("error"), }, controller.signal, - supportsThinking ? { enableThinking: !!thinkingEnabled } : {}, + { + ...(supportsThinking ? { enableThinking: !!thinkingEnabled } : {}), + toolDecisions: resume?.toolDecisions, + }, ); }, - [isLoading, supportsThinking, t, thinkingEnabled], + [supportsThinking, t, thinkingEnabled], + ); + + // Resume the paused turn once every pending call has a decision. + const applyDecisions = useCallback( + (next: Record) => { + setApprovalDecisions(next); + if (!pendingApprovals) return; + if (!pendingApprovals.every((tc) => next[tc.id] !== undefined)) return; + submitConversation(messages, { toolDecisions: next }); + }, + [messages, pendingApprovals, submitConversation], + ); + + const handleApprove = useCallback( + (id: string) => applyDecisions({ ...approvalDecisions, [id]: "approve" }), + [applyDecisions, approvalDecisions], + ); + + const handleReject = useCallback( + (id: string) => applyDecisions({ ...approvalDecisions, [id]: "reject" }), + [applyDecisions, approvalDecisions], + ); + + const handleAlwaysAllow = useCallback( + (id: string, name: string) => { + const current = alwaysAllowTools ?? NO_TOOLS; + const allowed = current.includes(name) ? current : [...current, name]; + setAlwaysAllowTools(allowed); + const next = { ...approvalDecisions, [id]: "approve" as const }; + for (const tc of pendingApprovals ?? []) { + if (tc.name === name) next[tc.id] = "approve"; + } + applyDecisions(next); + }, + [ + alwaysAllowTools, + applyDecisions, + approvalDecisions, + pendingApprovals, + setAlwaysAllowTools, + ], + ); + + const clearAlwaysAllowTools = useCallback( + () => setAlwaysAllowTools(NO_TOOLS), + [setAlwaysAllowTools], ); const recentEventIds = useMemo(() => { @@ -174,7 +281,7 @@ export default function ChatPage() { const calls = toolCallsForMessage(msg, responses); const similar = getFindSimilarObjectsFromToolCalls(calls); if (similar) return similar.results.map((e) => e.id); - const events = getEventIdsFromSearchObjectsToolCalls(calls); + const events = getEventIdsFromToolCalls(calls); if (events.length > 0) return events.map((e) => e.id); } return []; @@ -197,19 +304,25 @@ export default function ChatPage() { const stopGeneration = useCallback(() => { abortRef.current?.abort(); abortRef.current = null; + loadingRef.current = false; setIsLoading(false); setStreaming(null); + setPendingApprovals(null); + setApprovalDecisions({}); }, []); const startNewChat = useCallback(() => { abortRef.current?.abort(); abortRef.current = null; + loadingRef.current = false; setIsLoading(false); setStreaming(null); setMessages([]); setInput(""); setAttachedEventId(null); setError(null); + setPendingApprovals(null); + setApprovalDecisions({}); }, []); const handleEditSubmit = useCallback( @@ -260,7 +373,7 @@ export default function ChatPage() { const calls = toolCallsForMessage(msg, responses); const contentText = hasText(msg.content) ? msg.content : ""; const similar = getFindSimilarObjectsFromToolCalls(calls); - const events = similar ? [] : getEventIdsFromSearchObjectsToolCalls(calls); + const events = similar ? [] : getEventIdsFromToolCalls(calls); return (
@@ -324,6 +437,8 @@ export default function ChatPage() { setShowStats={setShowStats} autoScroll={autoScroll ?? true} setAutoScroll={setAutoScroll} + alwaysAllowTools={alwaysAllowTools ?? NO_TOOLS} + clearAlwaysAllowTools={clearAlwaysAllowTools} />
{renderList.map((msg, i) => renderMessage(msg, i))} + {pendingApprovals && !streaming && ( +
+ {pendingApprovals.map((tc) => ( + + ))} +
+ )} {streaming && !finalShown && (streaming.content || streaming.reasoning ? ( @@ -391,7 +520,12 @@ export default function ChatPage() { setInput={setInput} sendMessage={sendMessage} isLoading={isLoading} - placeholder={t("placeholder")} + disabled={pendingApprovals != null} + placeholder={ + pendingApprovals != null + ? t("approval.placeholder") + : t("placeholder") + } attachedEventId={attachedEventId} onClearAttachment={handleClearAttachment} onAttach={setAttachedEventId} diff --git a/web/src/types/chat.ts b/web/src/types/chat.ts index e31b71c86f..fe3fdd4d18 100644 --- a/web/src/types/chat.ts +++ b/web/src/types/chat.ts @@ -20,11 +20,21 @@ export type ChatMessage = { }; export type ToolCall = { + id?: string; name: string; arguments?: Record; response?: string; }; +export type ToolDecision = "approve" | "reject"; + +/** A state-changing tool call the backend paused on, awaiting the user. */ +export type PendingToolCall = { + id: string; + name: string; + arguments: Record; +}; + export type StartingRequest = { label: string; prompt: string; diff --git a/web/src/utils/chatUtil.ts b/web/src/utils/chatUtil.ts index b7aeb8088d..a2bc7a929f 100644 --- a/web/src/utils/chatUtil.ts +++ b/web/src/utils/chatUtil.ts @@ -1,4 +1,10 @@ -import type { ChatMessage, ChatStats, ToolCall } from "@/types/chat"; +import type { + ChatMessage, + ChatStats, + PendingToolCall, + ToolCall, + ToolDecision, +} from "@/types/chat"; export type StreamChatCallbacks = { /** Streamed delta of the assistant's final answer text. */ @@ -11,6 +17,10 @@ export type StreamChatCallbacks = { onChain: (chain: ChatMessage[]) => void; /** Token/timing stats for the turn. */ onStats: (stats: ChatStats) => void; + /** The backend paused before running state-changing tools; the chain + * emitted just before this ends with the assistant message requesting + * them. Resend that chain with `toolDecisions` to continue. */ + onApprovalRequired?: (toolCalls: PendingToolCall[]) => void; /** Called when the stream sends an error or fetch fails. */ onError: (message: string) => void; /** Called when the stream finishes (success or error). */ @@ -30,6 +40,7 @@ type StatsChunk = { type StreamChunk = | { type: "error"; error: string } | { type: "messages"; messages: ChatMessage[] } + | { type: "approval_required"; tool_calls: PendingToolCall[] } | { type: "content"; delta: string } | { type: "reasoning"; delta: string } | StatsChunk; @@ -40,6 +51,8 @@ type StreamChunk = */ export type StreamChatOptions = { enableThinking?: boolean; + /** Decisions for tool calls that paused for approval, keyed by call id. */ + toolDecisions?: Record; }; export async function streamChatCompletion( @@ -55,6 +68,7 @@ export async function streamChatCompletion( onReasoningDelta, onChain, onStats, + onApprovalRequired, onError, onDone, defaultErrorMessage = "Something went wrong. Please try again.", @@ -68,6 +82,9 @@ export async function streamChatCompletion( if (options.enableThinking !== undefined) { body.enable_thinking = options.enableThinking; } + if (options.toolDecisions && Object.keys(options.toolDecisions).length) { + body.tool_decisions = options.toolDecisions; + } const res = await fetch(url, { method: "POST", headers, @@ -103,6 +120,10 @@ export async function streamChatCompletion( onChain(data.messages ?? []); return "continue"; } + if (data.type === "approval_required") { + onApprovalRequired?.(data.tool_calls ?? []); + return "continue"; + } if (data.type === "content" && data.delta !== undefined) { onContentDelta(data.delta); return "continue"; @@ -198,6 +219,7 @@ export function toolCallsForMessage( } } return { + id: tc.id, name: tc.function?.name ?? "", arguments: args, response: responses.get(tc.id), @@ -205,28 +227,48 @@ export function toolCallsForMessage( }); } +/** Human-readable tool name: "search_objects" -> "Search Objects". */ +export function formatToolName(name: string): string { + return name + .replace(/_/g, " ") + .split(" ") + .map((word) => word.charAt(0).toUpperCase() + word.slice(1).toLowerCase()) + .join(" "); +} + +const hasStringId = (item: unknown): item is { id: string } => + !!item && + typeof item === "object" && + "id" in item && + typeof (item as { id: unknown }).id === "string"; + /** - * Parse search_objects tool call response(s) into event ids for thumbnails. + * Collect event ids from tool responses that reference tracked objects: + * search_objects returns a list of events and get_event_image a single one. */ -export function getEventIdsFromSearchObjectsToolCalls( +export function getEventIdsFromToolCalls( toolCalls: ToolCall[] | undefined, ): { id: string }[] { if (!toolCalls?.length) return []; const results: { id: string }[] = []; + const seen = new Set(); + const push = (item: unknown) => { + if (hasStringId(item) && !seen.has(item.id)) { + seen.add(item.id); + results.push({ id: item.id }); + } + }; for (const tc of toolCalls) { - if (tc.name !== "search_objects" || !tc.response?.trim()) continue; + if (!tc.response?.trim()) continue; + if (tc.name !== "search_objects" && tc.name !== "get_event_image") { + continue; + } try { const parsed = JSON.parse(tc.response) as unknown; - if (!Array.isArray(parsed)) continue; - for (const item of parsed) { - if ( - item && - typeof item === "object" && - "id" in item && - typeof (item as { id: unknown }).id === "string" - ) { - results.push({ id: (item as { id: string }).id }); - } + if (Array.isArray(parsed)) { + parsed.forEach(push); + } else { + push(parsed); } } catch { // ignore parse errors