From 3754b3e4a9ae9589a5cabb9bedfeac1594eebf9b Mon Sep 17 00:00:00 2001 From: Mathias Scheider Date: Fri, 2 Oct 2026 22:36:19 +0200 Subject: [PATCH] LLM response models for prompting and response validation --- src/photools3/llm_session.py | 47 +++++------ src/photools3/models.py | 116 ++++++++++++++++++++++++--- test/test_llm_session_integration.py | 10 +-- test/test_llm_session_mocked.py | 16 ++-- 4 files changed, 136 insertions(+), 53 deletions(-) diff --git a/src/photools3/llm_session.py b/src/photools3/llm_session.py index 3dde285..95bd3ff 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,11 @@ from photools3.models import ( LlmMessageType, LlmRequest, LlmRequestMessage, + LlmResponseCaptionsAndKeywords, + LlmResponseGeneral, ) 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 +23,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 @@ -64,18 +67,16 @@ class LlmSession: 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: - # try: - # 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 - return result_raw + # Clean the response (strip markdown code blocks if present) + json_str = re.sub( + r"^```json(.*)```$", r"\1", result_raw, flags=re.DOTALL + ).strip() + + try: + # Parse and validate against the class-level response_model + return self.response_model.model_validate_json(json_str) + except (json.JSONDecodeError, ValidationError) as e: + raise ValueError(f"Failed to parse or validate LLM response: {e}") def _build_request( self, @@ -117,24 +118,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, ) @@ -146,7 +138,6 @@ class LlmSessionCaptionsAndKeywords(LlmSession): model_id=model_id, ) - class LlmSessionQuality(LlmSession): def __init__(self, temperature: float = settings.llm_default_temperature): diff --git a/src/photools3/models.py b/src/photools3/models.py index 3df8e8d..ff158a5 100644 --- a/src/photools3/models.py +++ b/src/photools3/models.py @@ -1,7 +1,42 @@ -from enum import Enum, StrEnum +from enum import StrEnum +from functools import lru_cache from typing import Annotated, Any, Literal -from pydantic import BaseModel, BeforeValidator, Field, model_validator +from pydantic import ( + BaseModel, + BeforeValidator, + Field, + ValidationError, + 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): @@ -47,14 +82,75 @@ class LlmMessageContent(BaseModel): raise ValueError("image_url must be set when type is 'image_url'") return self -class LlmResponseKeywords(BaseModel): - caption: str = Field(description="Description of the image") - tags:list[str] = Field(description="keywords to tag the image") -class Rating(Enum): - ok = Annotated[str, Field( description="bla" )] - review = Annotated[str, Field(description="")] +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 = Literal["ok", "review", "not_ok"] - reason:str + 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 paersonal taste).' + ) + ), + ] + ) + + @model_validator(mode="after") + def _not_ok_needs_reason(self): + if self.rating != "ok" and self.reason is None: + raise ValidationError("lower-than-ok rating needs a reason") + elif self.rating == "ok" and self.reason is not None: + raise ValidationError("ok rating needs no reason") + return self 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..b37a58b 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,7 @@ 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 +77,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