mathias2784/photools3
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

Cap generated keywords and simplify LLM correction retries.

Browse Source
This commit is contained in:
mathias2784
2026-10-18 13:11:13 +02:00
parent 0ad504e4c9
commit 68bb2dc28c
9 changed files with 329 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.template
+3 -1
View File
@@ -4,5 +4,7 @@ LLM_DEFAULT_TEMPERATURE = 1.0
METADATA_PREFIX = "photools|"
LOG_LEVEL=INFO
LLM_FLUSH_EVERY_PHOTOS=50
LLM_REQUEST_TIMEOUT_SECONDS=10
LLM_MAX_COMPLETION_TOKENS=1024
LLM_MAX_RETRIES=5
LLM_MAX_KEYWORDS=15
AGENTS.md
+15
View File
@@ -1,27 +1,42 @@
# Agent Instructions: PhotoTools3
## Environment Setup
Requires a `.env` file with the following variables:
- `llm_api_key`: API key for the LLM provider.
- `llm_api_base_url`: Base URL for the LLM API.
- `llm_default_temperature`: Float value for temperature.
## System Dependencies
- `exiftool` binary: required by `pyexiftool` at runtime, not installable via pip/uv.
- Arch: `perl-image-exiftool`
- Debian/Ubuntu: `libimage-exiftool-perl`
## Development Toolchain
- **Dependency Management**: Use `uv`.
- **Python Version**: Requires Python `>=3.14`.
- **Testing**: Run `pytest` to execute the test suite in `test/`.
- **Linting & Formatting**: Use `ruff` and `black`.
## CLI Usage
The primary entry point is the `photools3` command:
- `photools3 keywords <image_path>`: Extracts keywords and captions from a single image.
## Architecture & Logic
- **`src/photools3/photo.py`**: Core image processing logic, including resizing, RGB conversion, and generating base64 JPEG encodings specifically for LLM vision compatibility.
- **`src/photools3/llm_session.py`**: Manages the workflow of sending processed images to the LLM and parsing results.
- **Configuration**: The list of keywords to extract is maintained in `config/keywords.txt`.
## Code style
- Avoid overly verbose docstrings or comments. Be precise!
## Comunication
- When done with code modifications, give a one-liner for the currently unstaged changes that can be used as a commit message.
README.md
+13 -8
View File
@@ -22,18 +22,23 @@ llm_api_key=your-api-key
llm_default_temperature=0.2
metadata_prefix=photools3:
LOG_LEVEL=DEBUG
LLM_FLUSH_EVERY_PHOTOS=50
LLM_REQUEST_TIMEOUT_SECONDS=10
LLM_MAX_COMPLETION_TOKENS=1024
LLM_MAX_RETRIES=5
LLM_MAX_KEYWORDS=15
```
Keyword suggestions are restricted to the vocabulary in `config/keywords.txt`.
`LOG_LEVEL=DEBUG` enables detailed per-request and per-photo timings. The flush
interval uses LM Studio's native model-management API to unload and reload the
model after every 50 attempted batch photos. This requires LM Studio's native
`/api/v1` model-management API. Set `LLM_FLUSH_EVERY_PHOTOS=0` to disable it;
use `0` for non-LM-Studio OpenAI-compatible providers.
`LLM_MAX_RETRIES` controls how many corrective retry requests are allowed after
the initial LLM response; `5` means up to 6 requests for a photo.
`LOG_LEVEL=DEBUG` enables detailed per-request and per-photo timings.
`LLM_MAX_RETRIES` controls retries for malformed or structurally invalid
responses; `5` means up to 6 requests for a photo. A valid response with
unsupported keywords gets at most one targeted correction retry, then remaining
invalid keywords are discarded.
Each request is limited by `LLM_REQUEST_TIMEOUT_SECONDS`; a timeout reloads the
LM Studio model before retrying. `LLM_MAX_COMPLETION_TOKENS` caps the generated
response length. When all retries are exhausted, the model is reloaded before
the failed photo is reported. `LLM_MAX_KEYWORDS` sets the maximum number of
distinct keywords requested and stored per photo; excess keywords are discarded.
## Generate metadata
src/photools3/llm_session.py
+119 -27
View File
@@ -19,6 +19,7 @@ from photools3.models import (
LlmResponseGeneral,
LlmResponseKeywords,
LlmResponseQuality,
load_valid_keywords,
)
from photools3.photo import Photo
from photools3.settings import settings
@@ -32,15 +33,23 @@ def _clean_llm_response(txt: str):
return txt
def _correction_prompt(exception: Exception, original_json: str) -> LlmRequestMessage:
correction_prompt = (
f"The previous response was invalid due to: {exception}. "
f"Please provide the corrected JSON based on this content: "
f"{original_json}"
)
def _correction_prompt(instruction: str) -> LlmRequestMessage:
return LlmRequestMessage(
role=LlmMessageRole.user,
content=correction_prompt,
content=instruction,
)
def _request_with_correction(base_request: LlmRequest, instruction: str) -> LlmRequest:
return base_request.model_copy(
update={"messages": [*base_request.messages, _correction_prompt(instruction)]}
)
def _validation_summary(exception: ValidationError) -> str:
return "; ".join(
f"{'.'.join(map(str, error['loc']))}: {error['msg']}"
for error in exception.errors(include_input=False)
)
@@ -62,6 +71,7 @@ class LlmSession:
self.client = OpenAI(
base_url=self.base_url,
api_key=self.api_key,
max_retries=0,
)
self._model_manager = LMStudioModelManager(self.base_url, self.api_key)
@@ -81,7 +91,10 @@ class LlmSession:
self._model_manager.reload_model(self._model_id)
def client_chat_completion_create(self, request: LlmRequest, choice_no: int = 0):
response = self.client.chat.completions.create(**request.model_dump())
response = self.client.chat.completions.create(
**request.model_dump(),
timeout=settings.llm_request_timeout_seconds,
)
return response.choices[choice_no].message.content
def is_api_reachable(self) -> bool:
@@ -117,11 +130,25 @@ class LlmSession:
base_request = self._build_request(prompt or self.prompt, photos, model_id)
request = base_request
tag_correction_used = False
valid_keywords = load_valid_keywords()
max_keywords = settings.llm_max_keywords
for attempt in range(retry_limit + 1):
request_started = time.perf_counter()
try:
result_raw = self.client_chat_completion_create(request)
except openai.APITimeoutError as exc:
logger.warning(
"LLM request timed out after %.2f seconds; reloading model",
time.perf_counter() - request_started,
)
self.flush_model()
if attempt < retry_limit:
continue
raise ValueError(
f"LLM request timed out after {retry_limit + 1} attempts"
) from exc
except Exception:
logger.debug(
"LLM request attempt %d/%d failed after %.2f seconds",
@@ -138,28 +165,88 @@ class LlmSession:
time.perf_counter() - request_started,
)
try:
json_str = _clean_llm_response(result_raw)
return self.response_model.model_validate_json(json_str)
except (json.JSONDecodeError, ValidationError) as exc:
response_data = json.loads(_clean_llm_response(result_raw))
except (json.JSONDecodeError, TypeError) as exc:
if attempt < retry_limit:
logger.info("Chatbot response is invalid; prompting to correct ...")
correction = _correction_prompt(
exc,
json_str if "json_str" in locals() else "unparseable response",
)
# Start each retry from the initial prompt and only the latest
# correction; do not grow the conversation with all past retries.
request = base_request.model_copy(
update={"messages": [*base_request.messages, correction]}
logger.info("Chatbot response was not valid JSON; retrying")
request = _request_with_correction(
base_request,
"Return one complete JSON object matching the schema. "
f"The previous response was invalid: {exc}",
)
continue
else:
raise ValueError(
f"Failed to parse or validate LLM response after {retry_limit} retries: {exc}"
logger.warning(
"LLM retry limit reached; reloading model before failing"
)
except Exception:
raise
self.flush_model()
raise ValueError(
f"Failed to parse LLM response after {retry_limit} retries: {exc}"
) from exc
if (
isinstance(response_data, dict)
and "tags" in self.response_model.model_fields
and isinstance(response_data.get("tags"), list)
):
tags = response_data["tags"]
considered_tags = tags[:max_keywords]
invalid_tags = [
tag
for tag in considered_tags
if not isinstance(tag, str) or tag not in valid_keywords
]
if invalid_tags and not tag_correction_used and attempt < retry_limit:
tag_correction_used = True
logger.info(
"LLM returned unsupported keywords; requesting one correction"
)
request = _request_with_correction(
base_request,
f"These keywords are not allowed: {invalid_tags!r}. "
"Replace them with exact allowed keywords when sensible; "
f"otherwise omit them. Return at most {max_keywords} "
"distinct keywords and do not list the vocabulary.",
)
continue
if len(tags) > max_keywords:
logger.info(
"Discarding keywords beyond configured limit of %d",
max_keywords,
)
if invalid_tags:
logger.warning("Discarding unsupported keywords: %s", invalid_tags)
response_data = dict(response_data)
valid_tags = [
tag
for tag in tags
if isinstance(tag, str) and tag in valid_keywords
]
response_data["tags"] = list(dict.fromkeys(valid_tags))[:max_keywords]
try:
return self.response_model.model_validate(response_data)
except ValidationError as exc:
if attempt < retry_limit:
logger.info("LLM response did not match the schema; retrying")
request = _request_with_correction(
base_request,
"Return one complete JSON object matching the schema. "
f"Validation issues: {_validation_summary(exc)}",
)
continue
logger.warning(
"LLM retry limit reached; reloading model before failing"
)
self.flush_model()
raise ValueError(
f"Failed to validate LLM response after {retry_limit} retries: "
f"{_validation_summary(exc)}"
) from exc
def _build_request(
self,
@@ -171,6 +258,7 @@ class LlmSession:
return LlmRequest(
model=model_id,
temperature=self.temperature,
max_tokens=settings.llm_max_completion_tokens,
messages=[
# System prompt
LlmRequestMessage(
@@ -202,13 +290,15 @@ class LlmSession:
class LlmSessionCaptionsAndKeywords(LlmSession):
response_model = LlmResponseCaptionsAndKeywords
prompt = "Give me a caption and keyword for the following photo(s)"
prompt = "Give me a caption and keywords for the following photo"
def __init__(self, temperature: float = settings.llm_default_temperature):
super().__init__(
system_prompt=(
"You are a helpful photographer's assistant that helps to write **captions** and **keywords** to photos:\n"
"Answer in German.\n"
f"Choose at most {settings.llm_max_keywords} distinct, relevant keywords. "
"Do not repeat or enumerate the allowed-keyword list.\n"
"Output must be a JSON of the following schema:\n"
f"{self.response_model.model_json_schema()}"
),
@@ -241,6 +331,8 @@ class LlmSessionKeywords(LlmSession):
system_prompt=(
"You are a helpful photographer's assistant that chooses keywords "
"for photos. Answer in German.\n"
f"Choose at most {settings.llm_max_keywords} distinct, relevant keywords. "
"Do not repeat or enumerate the allowed-keyword list.\n"
"Output must be a JSON of the following schema:\n"
f"{self.response_model.model_json_schema()}"
),
src/photools3/models.py
+22 -32
View File
@@ -1,3 +1,4 @@
import json
from enum import StrEnum
from functools import lru_cache
from typing import Annotated, Any, Literal
@@ -10,7 +11,7 @@ from pydantic import (
model_validator,
)
from .settings import KEYWORDS_LIST_PATH
from .settings import KEYWORDS_LIST_PATH, settings
@lru_cache
@@ -42,6 +43,7 @@ class LlmRequest(BaseModel):
model: str = "gpt-4o"
messages: list[LlmRequestMessage]
temperature: float | None = None
max_tokens: int
class LlmMessageRole(StrEnum):
@@ -98,18 +100,23 @@ class LlmResponseCaption(BaseModel):
]
class LlmResponseKeywords(BaseModel):
tags: list[str] = Field(
def _keyword_tags_field():
allowed_keywords = json.dumps(
sorted(load_valid_keywords()), ensure_ascii=False, separators=(",", ":")
)
return Field(
max_length=settings.llm_max_keywords,
description=(
"A list of keywords that properly tag the photo. "
"Allowed keywords: "
+ str(load_valid_keywords())
.replace("'", '"')
.replace("{", "[")
.replace("}", "]")
)
f"Choose at most {settings.llm_max_keywords} distinct, relevant keywords. "
"Do not enumerate the vocabulary. Allowed keywords: "
f"{allowed_keywords}"
),
)
class _LlmResponseWithTags(BaseModel):
tags: list[str] = _keyword_tags_field()
@field_validator("tags")
@classmethod
def validate_tags(cls, v: list[str]) -> list[str]:
@@ -122,7 +129,11 @@ class LlmResponseKeywords(BaseModel):
return v
class LlmResponseCaptionsAndKeywords(BaseModel):
class LlmResponseKeywords(_LlmResponseWithTags):
pass
class LlmResponseCaptionsAndKeywords(_LlmResponseWithTags):
caption: Annotated[
str,
Field(
@@ -132,27 +143,6 @@ class LlmResponseCaptionsAndKeywords(BaseModel):
)
),
]
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):
src/photools3/processing.py
+1 -10
View File
@@ -18,7 +18,6 @@ from photools3.models import (
LlmResponseKeywords,
)
from photools3.photo import Photo
from photools3.settings import settings
logger = logging.getLogger(__name__)
@@ -108,17 +107,9 @@ def process_batch(
mode = ProcessingMode(mode)
session = _create_session(mode)
outcomes: list[ProcessingOutcome] = []
flush_every = settings.llm_flush_every_photos
try:
for attempted_photos, item in enumerate(paths):
if flush_every and attempted_photos and attempted_photos % flush_every == 0:
logger.info(
"Flushing LM Studio model after %d attempted photos",
attempted_photos,
)
session.flush_model()
for item in paths:
path = Path(item)
photo_started = time.perf_counter()
try:
src/photools3/settings.py
+3 -1
View File
@@ -19,8 +19,10 @@ class Settings(BaseSettings):
llm_api_base_url: Url
llm_api_key: str
llm_default_temperature: float
llm_flush_every_photos: int = Field(default=0, ge=0)
llm_request_timeout_seconds: float = Field(default=10, gt=0)
llm_max_completion_tokens: int = Field(default=1024, gt=0)
llm_max_retries: int = Field(default=5, ge=0)
llm_max_keywords: int = Field(default=15, ge=1)
log_level: str = "INFO"
metadata_prefix: str
test/test_llm_session_mocked.py
+152 -4
View File
@@ -1,8 +1,10 @@
from unittest.mock import MagicMock, patch
import openai
import pytest
from photools3.llm_session import LlmSession
from photools3.llm_session import LlmSession, LlmSessionCaptionsAndKeywords
from photools3.models import LlmResponseCaptionsAndKeywords
def test_llm_session_mocked(test_photo):
@@ -119,8 +121,8 @@ def test_llm_session_retries_from_original_request_without_accumulating_messages
messages_per_attempt = [call.kwargs["messages"] for call in requests]
assert [len(messages) for messages in messages_per_attempt] == [2, 3, 3]
final_correction = messages_per_attempt[2][-1]["content"]
assert '"wrong": "second"' in final_correction
assert '"wrong": "first"' not in final_correction
assert "Field required" in final_correction
assert '"wrong"' not in final_correction
def test_llm_session_uses_configured_retry_count(monkeypatch):
@@ -149,7 +151,153 @@ def test_llm_session_uses_configured_retry_count(monkeypatch):
]
session = LlmSession(system_prompt="You are a helper.")
with pytest.raises(ValueError, match="after 1 retries"):
with (
patch.object(session, "flush_model") as flush_model,
pytest.raises(ValueError, match="after 1 retries"),
):
session.call(photos=[])
assert mock_client.chat.completions.create.call_count == 2
flush_model.assert_called_once()
def test_llm_session_reloads_after_timeout_and_retries(monkeypatch):
monkeypatch.setattr(
"photools3.llm_session.settings.llm_request_timeout_seconds", 10
)
monkeypatch.setattr("photools3.llm_session.settings.llm_max_completion_tokens", 768)
monkeypatch.setattr("photools3.llm_session.settings.llm_max_retries", 1)
with patch("photools3.llm_session.OpenAI") as mocked_openai:
mock_client = MagicMock()
mocked_openai.return_value = mock_client
mock_model = MagicMock()
mock_model.id = "gpt-4o"
mock_client.models.list.return_value.data = [mock_model]
response_message = MagicMock(content='{"text": "recovered"}')
response_choice = MagicMock(message=response_message)
success_response = MagicMock(choices=[response_choice])
mock_client.chat.completions.create.side_effect = [
openai.APITimeoutError(request=MagicMock()),
success_response,
]
session = LlmSession(system_prompt="You are a helper.")
with patch.object(session, "flush_model") as flush_model:
result = session.call(photos=[])
assert result.text == "recovered"
flush_model.assert_called_once()
calls = mock_client.chat.completions.create.call_args_list
assert len(calls) == 2
assert all(call.kwargs["timeout"] == 10 for call in calls)
assert all(call.kwargs["max_tokens"] == 768 for call in calls)
mocked_openai.assert_called_once_with(
base_url=session.base_url,
api_key=session.api_key,
max_retries=0,
)
@pytest.mark.parametrize("failure_kind", ["invalid_response", "timeout"])
def test_llm_session_reloads_when_retry_limit_is_exhausted(monkeypatch, failure_kind):
monkeypatch.setattr("photools3.llm_session.settings.llm_max_retries", 0)
with patch("photools3.llm_session.OpenAI") as mocked_openai:
mock_client = MagicMock()
mocked_openai.return_value = mock_client
mock_model = MagicMock()
mock_model.id = "gpt-4o"
mock_client.models.list.return_value.data = [mock_model]
if failure_kind == "timeout":
mock_client.chat.completions.create.side_effect = openai.APITimeoutError(
request=MagicMock()
)
else:
response_message = MagicMock(content='{"wrong": "shape"}')
response_choice = MagicMock(message=response_message)
mock_client.chat.completions.create.return_value = MagicMock(
choices=[response_choice]
)
session = LlmSession(system_prompt="You are a helper.")
with (
patch.object(session, "flush_model") as flush_model,
pytest.raises(ValueError),
):
session.call(photos=[])
flush_model.assert_called_once()
assert mock_client.chat.completions.create.call_count == 1
def test_invalid_keywords_get_one_correction_then_invalid_tags_are_dropped(monkeypatch):
monkeypatch.setattr("photools3.llm_session.settings.llm_max_keywords", 3)
monkeypatch.setattr("photools3.llm_session.settings.llm_max_retries", 5)
with patch("photools3.llm_session.OpenAI") as mocked_openai:
mock_client = MagicMock()
mocked_openai.return_value = mock_client
mock_model = MagicMock()
mock_model.id = "gpt-4o"
mock_client.models.list.return_value.data = [mock_model]
def response(content):
message = MagicMock(content=content)
choice = MagicMock(message=message)
return MagicMock(choices=[choice])
mock_client.chat.completions.create.side_effect = [
response(
'{"caption": "A caption", "tags": ["not_allowed", "architektur"]}'
),
response(
'{"caption": "A caption", '
'"tags": ["still_invalid", "architektur", "personen"]}'
),
]
session = LlmSessionCaptionsAndKeywords()
with patch.object(session, "flush_model") as flush_model:
result = session.call(photos=[])
assert result.tags == ["architektur", "personen"]
assert mock_client.chat.completions.create.call_count == 2
flush_model.assert_not_called()
correction = mock_client.chat.completions.create.call_args_list[1].kwargs[
"messages"
][-1]["content"]
assert "not_allowed" in correction
assert '"caption"' not in correction
def test_excess_keywords_are_truncated_to_configured_limit(monkeypatch):
monkeypatch.setattr("photools3.llm_session.settings.llm_max_keywords", 2)
with patch("photools3.llm_session.OpenAI") as mocked_openai:
mock_client = MagicMock()
mocked_openai.return_value = mock_client
mock_model = MagicMock()
mock_model.id = "gpt-4o"
mock_client.models.list.return_value.data = [mock_model]
message = MagicMock(
content=(
'{"caption": "A caption", '
'"tags": ["architektur", "personen", "landschaft"]}'
)
)
mock_client.chat.completions.create.return_value = MagicMock(
choices=[MagicMock(message=message)]
)
result = LlmSessionCaptionsAndKeywords().call(photos=[])
assert result.tags == ["architektur", "personen"]
assert mock_client.chat.completions.create.call_count == 1
def test_keyword_response_schema_includes_configured_max_items():
schema = LlmResponseCaptionsAndKeywords.model_json_schema()
assert schema["properties"]["tags"]["maxItems"] == 15
test/test_processing.py
-36
View File
@@ -127,39 +127,3 @@ def test_process_batch_is_sequential_reuses_session_and_continues_after_error():
assert all(len(call.args[0]) == 1 for call in session.call.call_args_list)
assert writer_class.return_value.write.call_count == 2
session.close.assert_called_once()
def test_process_batch_flushes_after_configured_attempt_count(monkeypatch):
monkeypatch.setattr("photools3.processing.settings.llm_flush_every_photos", 2)
class FakePhoto:
def __init__(self, path):
self.path = path
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
return None
response = LlmResponseCaptionsAndKeywords(caption="Ein Haus", tags=["architektur"])
session = MagicMock()
session.call.side_effect = [
response,
RuntimeError("temporary error"),
response,
response,
response,
]
paths = [Path(f"{i}.jpg") for i in range(5)]
with (
patch("photools3.processing._create_session", return_value=session),
patch("photools3.processing.Photo", FakePhoto),
patch("photools3.processing.MetadataWriter"),
):
outcomes = process_batch(paths, ProcessingMode.BOTH)
assert len(outcomes) == len(paths)
assert session.flush_model.call_count == 2
session.close.assert_called_once()