7c9d851a9a
All checks were successful
Build and Publish Docker Image / build-and-push (push) Successful in 5m3s
92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from app.config import get_request_settings
|
|
from app.llm_adapters import build_adapter, coerce_json_text
|
|
from app.models import ClassificationResult, ClassifyRequest, EmailData
|
|
|
|
VALID_CATEGORIES = {
|
|
"action_required",
|
|
"question",
|
|
"fyi",
|
|
"newsletter",
|
|
"promotional",
|
|
"automated",
|
|
"alert",
|
|
"uncategorized",
|
|
}
|
|
VALID_PRIORITIES = {"high", "medium", "low"}
|
|
|
|
|
|
async def classify_email(request: ClassifyRequest) -> ClassificationResult:
|
|
clean_email = _clean_email(request.email_data)
|
|
settings = get_request_settings(
|
|
provider=request.provider,
|
|
model=request.model,
|
|
base_url=request.base_url,
|
|
api_key=request.api_key,
|
|
temperature=request.temperature,
|
|
)
|
|
adapter = build_adapter(settings)
|
|
|
|
attempts = 0
|
|
while attempts < settings.max_retries:
|
|
raw_response = await adapter.classify(clean_email)
|
|
try:
|
|
payload = json.loads(coerce_json_text(raw_response))
|
|
result = _normalize_result(payload)
|
|
if result.needs_action and not result.task_description:
|
|
attempts += 1
|
|
continue
|
|
return result
|
|
except (json.JSONDecodeError, ValueError, TypeError):
|
|
attempts += 1
|
|
|
|
return ClassificationResult(
|
|
needs_action=False,
|
|
category="uncategorized",
|
|
priority="low",
|
|
task_description=None,
|
|
reasoning="System failed to classify after multiple attempts.",
|
|
confidence=0.0,
|
|
)
|
|
|
|
|
|
def _clean_email(email: EmailData) -> EmailData:
|
|
from app.helpers.clean_email_html import clean_email_html
|
|
from app.helpers.extract_latest_message import extract_latest_message
|
|
from app.helpers.remove_disclaimer import remove_disclaimer
|
|
|
|
return EmailData(
|
|
subject=email.subject,
|
|
body=remove_disclaimer(clean_email_html(extract_latest_message(email.body))),
|
|
)
|
|
|
|
|
|
def _normalize_result(data: dict[str, Any]) -> ClassificationResult:
|
|
needs_action = bool(data.get("needs_action", False))
|
|
category = str(data.get("category", "uncategorized") or "uncategorized").lower()
|
|
if category not in VALID_CATEGORIES:
|
|
category = "uncategorized"
|
|
priority = str(data.get("priority", "low") or "low").lower()
|
|
if priority not in VALID_PRIORITIES:
|
|
priority = "low"
|
|
task_description = data.get("task_description")
|
|
if task_description is not None:
|
|
task_description = str(task_description).strip() or None
|
|
if needs_action and not task_description:
|
|
raise ValueError("task_description required when needs_action is true")
|
|
reasoning = str(data.get("reasoning", "") or "").strip() or "No reasoning provided."
|
|
confidence_raw = data.get("confidence", 0.0)
|
|
confidence = max(0.0, min(1.0, float(confidence_raw)))
|
|
return ClassificationResult(
|
|
needs_action=needs_action,
|
|
category=category,
|
|
priority=priority,
|
|
task_description=task_description,
|
|
reasoning=reasoning,
|
|
confidence=confidence,
|
|
)
|