Spaces:
Sleeping
Sleeping
File size: 2,974 Bytes
9788b7f |
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 99 100 101 102 103 104 105 106 |
"""
utils/feedback.py
Unified feedback handler for Clinical Research Chatbot.
Includes:
1️⃣ Feedback Queue (unanswered/low-confidence queries)
2️⃣ User Voting (👍 Helpful / 👎 Not Helpful)
"""
import os
import json
from datetime import datetime
# ----------------------------
# File Paths
# ----------------------------
FEEDBACK_QUEUE_LOG = "logs/feedback_queue.jsonl"
FEEDBACK_VOTES_LOG = "logs/feedback_votes.jsonl"
# ----------------------------
# Feedback Queue (for Admin Review)
# ----------------------------
def log_feedback(query: str, notes: str = "", sources=None):
"""
Store unanswered or low-confidence queries for admin review.
Saves to JSONL (one entry per line).
"""
entry = {
"timestamp": datetime.utcnow().isoformat(),
"query": query,
"notes": notes,
"sources": sources or [],
}
os.makedirs(os.path.dirname(FEEDBACK_QUEUE_LOG), exist_ok=True)
with open(FEEDBACK_QUEUE_LOG, "a", encoding="utf-8") as f:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
print(f"📝 Feedback queued for admin review: {query}")
def load_feedback(limit: int = 20):
"""
Load last N feedback entries for admin dashboard.
"""
if not os.path.exists(FEEDBACK_QUEUE_LOG):
return []
with open(FEEDBACK_QUEUE_LOG, "r", encoding="utf-8") as f:
lines = f.readlines()
entries = [json.loads(line) for line in lines]
return entries[-limit:]
def clear_feedback():
"""
Clear feedback log (admin only).
"""
if os.path.exists(FEEDBACK_QUEUE_LOG):
os.remove(FEEDBACK_QUEUE_LOG)
print("🗑️ Feedback log cleared.")
# ----------------------------
# User Voting (for “Helpful / Not Helpful”)
# ----------------------------
def save_vote_feedback(query: str, vote: str, context=None):
"""
Log user votes (👍 / 👎) on chatbot responses.
"""
entry = {
"timestamp": datetime.utcnow().isoformat(),
"query": query,
"vote": vote,
"context": context or {},
}
os.makedirs(os.path.dirname(FEEDBACK_VOTES_LOG), exist_ok=True)
try:
with open(FEEDBACK_VOTES_LOG, "a", encoding="utf-8") as f:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
print(f"🗳️ User voted '{vote}' for query: {query}")
except Exception as e:
print(f"⚠️ Failed to save vote feedback: {e}")
def load_votes(limit: int = 50):
"""
Load last N user votes for analysis.
"""
if not os.path.exists(FEEDBACK_VOTES_LOG):
return []
with open(FEEDBACK_VOTES_LOG, "r", encoding="utf-8") as f:
lines = f.readlines()
entries = [json.loads(line) for line in lines]
return entries[-limit:]
def clear_votes():
"""
Clear user voting log (admin only).
"""
if os.path.exists(FEEDBACK_VOTES_LOG):
os.remove(FEEDBACK_VOTES_LOG)
print("🗑️ User vote feedback cleared.")
|