LLM response models for prompting and response validation
This commit is contained in:
@@ -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):
|
||||
|
||||
+106
-10
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user