Llm client quality (#14)

Reviewed-on: #14
Co-authored-by: Mathias Scheider <mathias2784@web.de>

src/photools3/cli.py

@@ -1,9 +1,12 @@
 from pathlib import Path
-from typing import Annotated, Any
+from typing import Annotated
 
 import typer
 
-from photools3.llm_session import LlmSessionCaptionsAndKeywords
+from photools3.llm_session import (
+    LlmSessionCaptionsAndKeywords,
+    LlmSessionQuality,
+)
 from photools3.photo import Photo
 
 app = typer.Typer(help="PhotoTools3: A CLI for image processing.")
@@ -30,39 +33,33 @@ def main(
 def keywords(
     file_path: Annotated[Path, typer.Argument(..., help="Path to the image file")],
 ):
-    """Extract keywords from a single image."""
-    if not file_path.exists():
-        typer.echo(f"Error: File '{file_path}' does not exist.", err=True)
-        raise typer.Exit(code=1)
+    session = LlmSessionCaptionsAndKeywords()
+    _run_llm_task(file_path, session)
 
+
+@app.command("quality")
+def quality(
+    file_path: Annotated[Path, typer.Argument(..., help="Path to the image file")],
+):
+    session = LlmSessionQuality()
+    _run_llm_task(file_path, session)
+
+
+def _run_llm_task(
+    file_path: Path, session: LlmSessionCaptionsAndKeywords | LlmSessionQuality
+):
     try:
         typer.echo(f"Processing: {file_path}")
-        result = _run_keywords_extraction(file_path)
-        _print_results(result)
+        photo = Photo(file_path)
+        result = session.call([photo])
+        print(result)
+    except FileNotFoundError:
+        typer.echo(f"Error: File '{file_path}' does not exist.", err=True)
+        raise typer.Exit(code=1)
     except Exception as e:  # noqa: BLE001
         typer.echo(f"Error processing image: {e}", err=True)
         raise typer.Exit(code=1)
 
 
-def _run_keywords_extraction(file_path: Path) -> Any:
-    """Handles the core logic of loading the photo and calling the LLM."""
-    session = LlmSessionCaptionsAndKeywords()
-    photo = Photo(file_path)
-    return session.call([photo])
-
-
-def _print_results(result: Any) -> None:
-    """Formats and prints the extraction results to the console."""
-    typer.echo("\n--- Results ---")
-    if isinstance(result, dict):
-        caption = result.get("caption", "No caption found.")
-        tags = result.get("tags", [])
-        typer.echo(f"Caption: {caption}")
-        typer.echo(f"Tags: {', '.join(tags)}")
-    else:
-        typer.echo(str(result))
-    typer.echo("---------------")
-
-
 if __name__ == "__main__":
     app()

src/photools3/llm_session.py

@@ -3,7 +3,7 @@ import re
 
 import openai
 from openai import OpenAI
-from utils.misc import readlines
+from pydantic import ValidationError
 
 from photools3.models import (
     LlmMessageContent,
@@ -11,9 +11,12 @@ from photools3.models import (
     LlmMessageType,
     LlmRequest,
     LlmRequestMessage,
+    LlmResponseCaptionsAndKeywords,
+    LlmResponseGeneral,
+    LlmResponseQuality,
 )
 from photools3.photo import Photo
-from photools3.settings import KEYWORDS_LIST_PATH, Settings
+from photools3.settings import Settings
 
 settings = Settings()  # type: ignore[reportCallIssue]
 
@@ -21,6 +24,7 @@ settings = Settings()  # type: ignore[reportCallIssue]
 class LlmSession:
     base_url: str = str(settings.llm_api_base_url)
     api_key: str = settings.llm_api_key
+    response_model = LlmResponseGeneral
 
     def __init__(
         self, system_prompt: str, temperature: float = settings.llm_default_temperature
@@ -57,24 +61,47 @@ class LlmSession:
         prompt: str,
         photos: list[Photo],
         model_id=None,
+        max_retries: int = 2,
     ):
         if not model_id:
             model_id = self.models[0].id
 
         request = self._build_request(prompt, photos, model_id)
-        result_raw = self.client_chat_completion_create(request)
 
-        try:
-            result = json.loads(result_raw)
-        except json.JSONDecodeError as e:
+        last_error = None
+        for attempt in range(max_retries + 1):
             try:
+                result_raw = self.client_chat_completion_create(request)
+                # Clean the response (strip markdown code blocks if present)
                 json_str = re.sub(
                     r"^```json(.*)```$", r"\1", result_raw, flags=re.DOTALL
                 ).strip()
-                result = json.loads(json_str)
-            except json.JSONDecodeError:
-                raise e
-        return result
+
+                # Parse and validate against the class-level response_model
+                return self.response_model.model_validate_json(json_str)
+
+            except (json.JSONDecodeError, ValidationError) as e:
+                last_error = e
+                if attempt < max_retries:
+                    # Prepare a correction prompt
+                    correction_prompt = (
+                        f"The previous response was invalid due to: {e}. "
+                        f"Please provide the corrected JSON based on this content: {json_str if 'json_str' in locals() else 'the original input'}"
+                    )
+                    # Append correction prompt as a new user message for the next attempt
+                    request.messages.append(
+                        LlmRequestMessage(
+                            role=LlmMessageRole.user,
+                            content=correction_prompt,
+                        )
+                    )
+                    continue
+                else:
+                    raise ValueError(
+                        f"Failed to parse or validate LLM response after {max_retries} retries: {e}"
+                    )
+            except Exception:
+                raise
 
     def _build_request(
         self,
@@ -116,24 +143,15 @@ class LlmSession:
 
 
 class LlmSessionCaptionsAndKeywords(LlmSession):
-    keywords_list = ",".join([s.strip() for s in readlines(str(KEYWORDS_LIST_PATH))])
+    response_model = LlmResponseCaptionsAndKeywords
 
     def __init__(self, temperature: float = settings.llm_default_temperature):
         super().__init__(
             system_prompt=(
-                "# System prompt\n"
-                "\n"
                 "You are a helpful photographer's assistant that helps to write **captions** and **keywords** to photos:\n"
-                "- Answer in JSON format so it can easily be integrated in automatic workflows.\n"
-                "  - JSON format is `{'caption': '...', 'tags': ['tag1', 'tag2', ...]}`.\n"
-                "  - Do not add any other text outside of this JSON format!\n"
-                "- When writing captions, ONLY describe what is visible on the photo. Do not interpret!\n"
-                "- Only use keywords from the keyword list, appended below.\n"
-                "- Answer in German.\n"
-                "\n"
-                "## Keyword list"
-                "\n"
-                f"{self.keywords_list}"
+                "Answer in German.\n"
+                "Output must be a JSON of the following schema:\n"
+                f"{self.response_model.model_json_schema()}"
             ),
             temperature=temperature,
         )
@@ -144,3 +162,25 @@ class LlmSessionCaptionsAndKeywords(LlmSession):
             photos=photos,
             model_id=model_id,
         )
+
+
+class LlmSessionQuality(LlmSession):
+    response_model = LlmResponseQuality
+
+    def __init__(self, temperature: float = settings.llm_default_temperature):
+        super().__init__(
+            system_prompt=(
+                "You are a helpful photographer's assistant that helps to identify "
+                "photos that can be sorted out due to qulity issues:\n"
+                "Output must be a JSON of the following schema:\n"
+                f"{self.response_model.model_json_schema()}"
+            ),
+            temperature=temperature,
+        )
+
+    def call(self, photos: list[Photo], model_id=None):
+        return super().call(
+            prompt="Assess the quality of the following photos to help me decide which ones to sort out",
+            photos=photos,
+            model_id=model_id,
+        )

src/photools3/models.py

@@ -1,7 +1,41 @@
 from enum import StrEnum
-from typing import Annotated, Any
+from functools import lru_cache
+from typing import Annotated, Any, Literal
 
-from pydantic import BaseModel, BeforeValidator, model_validator
+from pydantic import (
+    BaseModel,
+    BeforeValidator,
+    Field,
+    field_validator,
+    model_validator,
+)
+
+from .settings import KEYWORDS_LIST_PATH
+
+
+@lru_cache
+def load_valid_keywords() -> set[str]:
+    """Loads all valid keywords from the configuration file."""
+    if not KEYWORDS_LIST_PATH.exists():
+        return set()
+
+    keywords = set()
+    with open(KEYWORDS_LIST_PATH, "r", encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            # Each line can contain hierarchical tags separated by '|'
+            # e.g., 'architektur|gebäude' means both 'architektur' and 'gebäude' are valid.
+            parts = line.split("|")
+            current_path = ""
+            for part in parts:
+                if current_path:
+                    current_path += "|" + part
+                else:
+                    current_path = part
+                keywords.add(current_path)
+    return keywords
 
 
 class LlmRequest(BaseModel):
@@ -46,3 +80,76 @@ class LlmMessageContent(BaseModel):
         if self.type == "image_url" and self.image_url is None:
             raise ValueError("image_url must be set when type is 'image_url'")
         return self
+
+
+class LlmResponseGeneral(BaseModel):
+    text: str
+
+
+class LlmResponseCaptionsAndKeywords(BaseModel):
+    caption: Annotated[
+        str,
+        Field(
+            description=(
+                "A brief but precise description of what the picture shows. "
+                "No interpretation; just what is visible"
+            )
+        ),
+    ]
+    tags: list[str] = Field(
+        description=(
+            "A list of keywords that properly tag the photo. "
+            "Allowed keywords: "
+            + str(load_valid_keywords())
+            .replace("'", '"')
+            .replace("{", "[")
+            .replace("}", "]")
+        )
+    )
+
+    @field_validator("tags")
+    @classmethod
+    def validate_tags(cls, v: list[str]) -> list[str]:
+        valid_keywords = load_valid_keywords()
+        for tag in v:
+            if tag not in valid_keywords:
+                raise ValueError(
+                    f"'{tag}' is not a valid keyword. Must be one from the configuration."
+                )
+        return v
+
+
+class LlmResponseQuality(BaseModel):
+    rating: Annotated[
+        Literal["ok", "review", "not_ok"],
+        Field(
+            description=(
+                "General quality assessment of the photo. "
+                'Helping to decide whether to keep ("ok"), '
+                'to consider sorting out ("review"), '
+                'or to almost certainly sort out ("not_ok")'
+            )
+        ),
+    ]
+    reason: (
+        None
+        | Annotated[
+            Literal["blur", "exposure", "style"],
+            Field(
+                description=(
+                    'Possible reasons for a lower-than-"ok" assessment. '
+                    '"blur": Obviously mis-focussed or motion-blurred. '
+                    '"exposure": severly under- or over-exposed. '
+                    '"style": photo has severe style issues (beyond personal taste).'
+                )
+            ),
+        ]
+    )
+
+    @model_validator(mode="after")
+    def _not_ok_needs_reason(self):
+        if self.rating != "ok" and self.reason is None:
+            raise ValueError("lower-than-ok rating needs a reason")
+        elif self.rating == "ok" and self.reason is not None:
+            raise ValueError("ok rating needs no reason")
+        return self

src/photools3/photo.py

@@ -7,6 +7,10 @@ from PIL import Image
 from photools3.models import transform_to_image_dict
 
 
+def _exist_guard(path: Path):
+    pass
+
+
 class Photo:
 
     def __init__(self, path: Path):