Spaces:
Runtime error
Runtime error
File size: 3,724 Bytes
68964c2 |
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 |
from app.core.fact_management import save_user_fact, get_user_fact
from app.core.db_setup import feedback_collection
from app.core.logging_setup import logger
def save_feedback(conversation_id, user_input, arina_reply, reason):
"""Save user feedback including reasoning."""
try:
if not conversation_id or not user_input or not arina_reply or not reason:
raise ValueError("All feedback fields must be provided.")
new_feedback = {
"conversation_id": conversation_id,
"user_input": user_input,
"arina_reply": arina_reply,
"reason": reason
}
feedback_collection.insert_one(new_feedback)
logger.info("Feedback saved to database.")
except ValueError as ve:
logger.error(f"Validation error: {ve}")
except Exception as e:
logger.error(f"Unexpected error: {e}")
def analyze_feedback(conversation_id):
"""Analyze past feedback and adjust response behavior."""
try:
feedbacks = feedback_collection.find({"conversation_id": conversation_id}, {"reason": 1, "_id": 0})
except Exception as e:
logger.error(f"Unexpected error: {e}")
return
feedbacks = list(feedbacks)
if not feedbacks:
logger.info("No feedback found for this conversation.")
return
issue_counts = {}
# Process feedback to detect common complaints
for feedback in feedbacks:
reason = feedback["reason"]
words = reason.lower().split()
for word in words:
if word in issue_counts:
issue_counts[word] += 1
else:
issue_counts[word] = 1
# Identify most common complaints
common_issues = sorted(issue_counts, key=issue_counts.get, reverse=True)[:3]
# Apply personalized feedback adjustments
try:
user_facts = {fact: get_user_fact(fact) for fact in ["response_tone", "response_length", "clarity"]}
for issue in common_issues:
if "robotic" in issue and user_facts["response_tone"] != "casual":
save_user_fact("response_tone", "casual")
elif ("too short" in issue or "brief" in issue) and user_facts["response_length"] != "long":
save_user_fact("response_length", "long")
elif "confusing" in issue and user_facts["clarity"] != "improve":
save_user_fact("clarity", "improve")
except Exception as e:
logger.error(f"Error updating user fact: {e}")
def apply_feedback_adjustments(messages):
"""Modify responses based on stored user feedback preferences."""
try:
response_tone = get_user_fact("response_tone") or "neutral"
response_length = get_user_fact("response_length") or "default"
clarity = get_user_fact("clarity") or "normal"
if not response_tone or not response_length or not clarity:
logger.info("⚠️ Some user preferences are missing, using defaults.")
for message in messages:
if message["role"] == "system":
if response_tone == "casual":
message["content"] += " Make your tone friendly and engaging."
if response_length == "long":
message["content"] += " Provide detailed and expanded answers."
if clarity == "improve":
message["content"] += " Make sure your response is clear and easy to understand."
except TypeError as e:
logger.error(f"🚨 TypeError in apply_feedback_adjustments: {e}")
raise
except Exception as e:
logger.error(f"🚨 Unexpected error in apply_feedback_adjustments: {e}")
raise
return messages |