From 401488fa7eba5c99de5c8e354a74ba635deb7972 Mon Sep 17 00:00:00 2001 From: Mathias Scheider Date: Fri, 2 Oct 2026 21:31:21 +0000 Subject: [PATCH] Llm client quality (#14) Reviewed-on: https://gitea.schneiderserver.site/mathias2784/photools3/pulls/14 Co-authored-by: Mathias Scheider --- .env.template | 2 +- src/photools3/cli.py | 53 ++++++------- src/photools3/llm_session.py | 86 +++++++++++++++------ src/photools3/models.py | 111 ++++++++++++++++++++++++++- src/photools3/photo.py | 4 + test/test_cli.py | 7 +- test/test_llm_session_integration.py | 10 +-- test/test_llm_session_mocked.py | 18 ++--- 8 files changed, 220 insertions(+), 71 deletions(-) diff --git a/.env.template b/.env.template index 07b3ea4..69a0aea 100644 --- a/.env.template +++ b/.env.template @@ -1,3 +1,3 @@ LLM_API_BASE_URL = "http://localhost:1234/v1" LLM_API_KEY = "not-relevant" -LLM_DEFAULT_TEMPERATURE = 0.8 +LLM_DEFAULT_TEMPERATURE = 1.0 diff --git a/src/photools3/cli.py b/src/photools3/cli.py index 2071050..bff0dd0 100644 --- a/src/photools3/cli.py +++ b/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() diff --git a/src/photools3/llm_session.py b/src/photools3/llm_session.py index 492cfee..08fc89b 100644 --- a/src/photools3/llm_session.py +++ b/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, + ) diff --git a/src/photools3/models.py b/src/photools3/models.py index 260c114..f3b27da 100644 --- a/src/photools3/models.py +++ b/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 diff --git a/src/photools3/photo.py b/src/photools3/photo.py index ed5cf32..738254f 100644 --- a/src/photools3/photo.py +++ b/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): diff --git a/test/test_cli.py b/test/test_cli.py index af39888..c04e9e5 100644 --- a/test/test_cli.py +++ b/test/test_cli.py @@ -58,8 +58,11 @@ def test_keywords_command_success(mock_photo_logic, dummy_file): assert result.exit_code == 0 assert "Processing: " in result.stdout - assert "Caption: A beautiful test image" in result.stdout - assert "Tags: test, mock, image" in result.stdout + # The result is printed as a raw dictionary string + assert ( + "{'caption': 'A beautiful test image', 'tags': ['test', 'mock', 'image']}" + in result.stdout + ) def test_keywords_command_file_not_found(): diff --git a/test/test_llm_session_integration.py b/test/test_llm_session_integration.py index 29d1094..8de0694 100644 --- a/test/test_llm_session_integration.py +++ b/test/test_llm_session_integration.py @@ -7,7 +7,7 @@ from photools3.llm_session import LlmSession not LlmSession(system_prompt="").is_api_reachable(), reason="API not available" ) def test_llm_session(llm_session, test_photo): - llm_session.system_prompt = "Return only JSON format: {'keywords': ['...']}" + llm_session.system_prompt = "Return only JSON format: {'text': '...'}" result = llm_session.call( prompt="keywords for the attached picture", photos=[test_photo] ) @@ -21,7 +21,7 @@ def test_llm_session_captions_and_keywords( llm_session_captions_and_keywords, test_photo ): result = llm_session_captions_and_keywords.call(photos=[test_photo]) - assert "caption" in result - assert "tags" in result - assert isinstance(result["caption"], str) - assert isinstance(result["tags"], list) + assert result.caption + assert result.tags + assert isinstance(result.caption, str) + assert isinstance(result.tags, list) diff --git a/test/test_llm_session_mocked.py b/test/test_llm_session_mocked.py index b4e127c..c6b9bb0 100644 --- a/test/test_llm_session_mocked.py +++ b/test/test_llm_session_mocked.py @@ -21,9 +21,7 @@ def test_llm_session_mocked(test_photo): mock_choice = MagicMock() mock_message = MagicMock() - mock_message.content = ( - '{"caption": "A beautiful sunset", "tags": ["sunset", "nature"]}' - ) + mock_message.content = '{"text": "some text"}' mock_choice.message = mock_message mock_response.choices = [mock_choice] @@ -39,11 +37,9 @@ def test_llm_session_mocked(test_photo): ) # Verify the results - assert result["caption"] == "A beautiful sunset" - assert "sunset" in result["tags"] - assert "nature" in result["tags"] + assert result.text == "some text" - # Verify that the mock was called + # Verify that the mock was enough assert mock_client.chat.completions.create.called @@ -65,7 +61,9 @@ def test_llm_session_captions_and_keywords_mocked(test_photo): mock_choice = MagicMock() mock_message = MagicMock() - mock_message.content = '{"caption": "A mocked caption", "tags": ["mocked_tag"]}' + mock_message.content = ( + '{"caption": "A mocked caption", "tags": ["architektur"]}' + ) mock_choice.message = mock_message mock_response.choices = [mock_choice] @@ -81,8 +79,8 @@ def test_llm_session_captions_and_keywords_mocked(test_photo): result = mocked_session.call(photos=[test_photo]) # Verify the results - assert result["caption"] == "A mocked caption" - assert "mocked_tag" in result["tags"] + assert result.caption == "A mocked caption" + assert "architektur" in result.tags # Verify that the mock was called assert mock_client.chat.completions.create.called