File size: 3,084 Bytes
eff511c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | from __future__ import annotations
import json
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from config import FEEDBACK_LOG_PATH
FEEDBACK_CATEGORIES = [
"Doğru ve yararlı",
"Yanlış kaynak",
"Eksik cevap",
"Kendini tekrar ediyor",
"Bağlamı kaçırdı",
"Başka konuya geçti",
"Çok uzun",
"Çok kısa",
"Anlaşılır değil",
"Diğer",
]
REGRESSION_REVIEW_CATEGORIES = {
"Yanlış kaynak",
"Eksik cevap",
"Kendini tekrar ediyor",
"Bağlamı kaçırdı",
"Başka konuya geçti",
"Anlaşılır değil",
}
def save_feedback(chat_history: Any, rating: int | float, category: str, comment: str) -> str:
user_message, assistant_message = _last_turn(chat_history)
if not user_message and not assistant_message:
return "Kaydedilecek bir diyalog bulunamadı."
try:
numeric_rating = int(float(rating))
except Exception:
numeric_rating = 0
numeric_rating = max(1, min(numeric_rating, 10))
normalized_category = category or "Diğer"
record = {
"feedback_id": str(uuid.uuid4()),
"created_at": datetime.now(timezone.utc).isoformat(),
"rating": numeric_rating,
"category": normalized_category,
"comment": (comment or "").strip(),
"user_message": user_message,
"assistant_message": assistant_message,
"needs_regression_review": _needs_regression_review(numeric_rating, normalized_category),
}
_append_jsonl(FEEDBACK_LOG_PATH, record)
return "Geri bildirim kaydedildi. Teşekkürler."
def _last_turn(chat_history: Any) -> tuple[str, str]:
if not chat_history:
return "", ""
for item in reversed(chat_history):
if isinstance(item, dict):
role = item.get("role")
if role == "assistant":
assistant = str(item.get("content", "") or "").strip()
user = _previous_user_message(chat_history, item)
return user, assistant
if isinstance(item, (list, tuple)) and len(item) >= 2:
return str(item[0] or "").strip(), str(item[1] or "").strip()
return "", ""
def _previous_user_message(chat_history: Any, assistant_item: dict) -> str:
try:
assistant_index = chat_history.index(assistant_item)
except ValueError:
assistant_index = len(chat_history)
for item in reversed(chat_history[:assistant_index]):
if isinstance(item, dict) and item.get("role") == "user":
return str(item.get("content", "") or "").strip()
return ""
def _append_jsonl(path: Path, record: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as file:
file.write(json.dumps(record, ensure_ascii=False) + "\n")
def _needs_regression_review(rating: int, category: str) -> bool:
return rating <= 5 or category in REGRESSION_REVIEW_CATEGORIES
|