Llm client quality #14
+1
-1
@@ -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 = 0.5
|
||||
|
||||
+18
-47
@@ -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, LlmSessionQuality
|
||||
from photools3.llm_session import (
|
||||
LlmSessionCaptionsAndKeywords,
|
||||
LlmSessionQuality,
|
||||
)
|
||||
from photools3.photo import Photo
|
||||
|
||||
app = typer.Typer(help="PhotoTools3: A CLI for image processing.")
|
||||
@@ -30,65 +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)
|
||||
|
||||
try:
|
||||
typer.echo(f"Processing: {file_path}")
|
||||
result = _run_keywords_extraction(file_path)
|
||||
_print_results(result)
|
||||
except Exception as e: # noqa: BLE001
|
||||
typer.echo(f"Error processing image: {e}", 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")],
|
||||
):
|
||||
"""Assess general quality of photos"""
|
||||
if not file_path.exists():
|
||||
typer.echo(f"Error: File '{file_path}' does not exist.", err=True)
|
||||
raise typer.Exit(code=1)
|
||||
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_qa(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 _run_qa(file_path: Path) -> Any:
|
||||
"""Handles the core logic of loading the photo and calling the LLM."""
|
||||
session = LlmSessionQuality()
|
||||
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()
|
||||
|
||||
@@ -13,6 +13,7 @@ from photools3.models import (
|
||||
LlmRequestMessage,
|
||||
LlmResponseCaptionsAndKeywords,
|
||||
LlmResponseGeneral,
|
||||
LlmResponseQuality,
|
||||
)
|
||||
from photools3.photo import Photo
|
||||
from photools3.settings import Settings
|
||||
@@ -60,23 +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)
|
||||
|
||||
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()
|
||||
|
||||
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}")
|
||||
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,
|
||||
@@ -140,29 +165,15 @@ class LlmSessionCaptionsAndKeywords(LlmSession):
|
||||
|
||||
|
||||
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"
|
||||
"- Your answer is a simple string that must strictly comply with this scheme: "
|
||||
"'{flag}|{reason}', "
|
||||
"where {flag} and {reason} are placeholders for possible values "
|
||||
"that are explained in the following points.\n"
|
||||
"- {flag} is the general image quality assessment."
|
||||
"It can have the following values:\n"
|
||||
" - `not_ok`: photo lacks basic quality and should almost certainly be deleted.\n"
|
||||
" - `review`: photo has some quality issues, not severe but might be deleted.\n"
|
||||
" - `ok`: no issues, photo can be kept.\n"
|
||||
"- {reason} is the reason for the photo being flagged as `not_ok` or `review`. "
|
||||
"Thus, for any photo that is not `ok`, one or more reasons have to be given. "
|
||||
"If there is more than one reason, give a comma-separated list: "
|
||||
"'{flag}|{reason1},{flag}|{reason2},...'. "
|
||||
"Possible reasons for a lower-than-`ok` assessment are:\n"
|
||||
" - `blur`: Obviously mis-focussed or motion-blurred.\n"
|
||||
" - `exposure`: severly under- or over-exposed.\n"
|
||||
" - `style`: photo has severe style issues (beyond paersonal taste)\n"
|
||||
"Output must be a JSON of the following schema:\n"
|
||||
f"{self.response_model.model_json_schema()}"
|
||||
),
|
||||
temperature=temperature,
|
||||
)
|
||||
|
||||
@@ -6,7 +6,6 @@ from pydantic import (
|
||||
BaseModel,
|
||||
BeforeValidator,
|
||||
Field,
|
||||
ValidationError,
|
||||
field_validator,
|
||||
model_validator,
|
||||
)
|
||||
@@ -141,7 +140,7 @@ class LlmResponseQuality(BaseModel):
|
||||
'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).'
|
||||
'"style": photo has severe style issues (beyond personal taste).'
|
||||
)
|
||||
),
|
||||
]
|
||||
@@ -150,7 +149,7 @@ class LlmResponseQuality(BaseModel):
|
||||
@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")
|
||||
raise ValueError("lower-than-ok rating needs a reason")
|
||||
elif self.rating == "ok" and self.reason is not None:
|
||||
raise ValidationError("ok rating needs no reason")
|
||||
raise ValueError("ok rating needs no reason")
|
||||
return self
|
||||
|
||||
@@ -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):
|
||||
|
||||
+5
-2
@@ -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():
|
||||
|
||||
Reference in New Issue
Block a user