Spaces:
Runtime error
Runtime error
| 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 |