pluto
2df137ee2a
- OllamaClient: analyze_image, compare_images, is_available - Recognition pipeline: CLIP embedding -> cosine similarity -> threshold routing (known >0.85, unknown <0.60, ambiguous -> Ollama fallback) - Vision dispatcher: get_description() supports haiku/ollama/both backends - 9 embedding serialization tests Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
110 lines
3.6 KiB
Python
Executable File
110 lines
3.6 KiB
Python
Executable File
import base64
|
|
import logging
|
|
from typing import Optional
|
|
|
|
import anthropic
|
|
|
|
from ollama_client import OllamaClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PROMPT = (
|
|
"Briefly describe what's happening in this security camera image in 1-2 sentences. "
|
|
"Focus on the people, vehicles, or animals and their activity. Be concise."
|
|
)
|
|
|
|
|
|
def analyze_snapshot(client: anthropic.Anthropic, image_bytes: bytes) -> str | None:
|
|
"""Send a snapshot to Claude Haiku for description. Returns None on failure."""
|
|
try:
|
|
b64 = base64.standard_b64encode(image_bytes).decode("utf-8")
|
|
response = client.messages.create(
|
|
model="claude-haiku-4-5-20251001",
|
|
max_tokens=150,
|
|
messages=[
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/jpeg",
|
|
"data": b64,
|
|
},
|
|
},
|
|
{"type": "text", "text": PROMPT},
|
|
],
|
|
}
|
|
],
|
|
)
|
|
description = response.content[0].text.strip()
|
|
logger.info("Vision analysis complete: %d chars", len(description))
|
|
return description
|
|
except Exception:
|
|
logger.exception("Claude Vision API failed")
|
|
return None
|
|
|
|
|
|
async def analyze_snapshot_ollama(
|
|
ollama_client: OllamaClient, image_bytes: bytes
|
|
) -> str | None:
|
|
"""Send a snapshot to Ollama vision model for description.
|
|
|
|
Args:
|
|
ollama_client: Configured OllamaClient instance.
|
|
image_bytes: Raw image bytes (JPEG).
|
|
|
|
Returns:
|
|
Description string or None on failure.
|
|
"""
|
|
return await ollama_client.analyze_image(image_bytes, PROMPT)
|
|
|
|
|
|
async def get_description(
|
|
image_bytes: bytes,
|
|
vision_backend: str,
|
|
claude_client: Optional[anthropic.Anthropic] = None,
|
|
ollama_client: Optional[OllamaClient] = None,
|
|
) -> str | None:
|
|
"""Get a scene description using the configured vision backend.
|
|
|
|
Args:
|
|
image_bytes: Raw image bytes (JPEG).
|
|
vision_backend: One of "haiku", "ollama", or "both".
|
|
"both" tries Ollama first, falls back to Haiku.
|
|
claude_client: Anthropic client (required for "haiku" and "both").
|
|
ollama_client: OllamaClient instance (required for "ollama" and "both").
|
|
|
|
Returns:
|
|
Description string or None if all backends fail.
|
|
"""
|
|
if vision_backend == "ollama":
|
|
if ollama_client is None:
|
|
logger.error("Ollama backend selected but no client provided")
|
|
return None
|
|
return await analyze_snapshot_ollama(ollama_client, image_bytes)
|
|
|
|
if vision_backend == "haiku":
|
|
if claude_client is None:
|
|
logger.error("Haiku backend selected but no client provided")
|
|
return None
|
|
return analyze_snapshot(claude_client, image_bytes)
|
|
|
|
if vision_backend == "both":
|
|
# Try Ollama first, fall back to Haiku
|
|
if ollama_client is not None:
|
|
result = await analyze_snapshot_ollama(ollama_client, image_bytes)
|
|
if result is not None:
|
|
return result
|
|
logger.info("Ollama failed, falling back to Haiku")
|
|
|
|
if claude_client is not None:
|
|
return analyze_snapshot(claude_client, image_bytes)
|
|
|
|
logger.error("Both backend selected but no clients provided")
|
|
return None
|
|
|
|
logger.error("Unknown vision backend: %s", vision_backend)
|
|
return None
|