Spaces:
Running
Running
| """ | |
| NeuraPrompt Agent — Feedback Module (v1.0) | |
| ============================================ | |
| Collects user feedback on AI responses. Stored in MongoDB (neuraprompt DB, | |
| separate collection). Reuses main.py's MongoClient connection. | |
| Collections: | |
| neuraprompt.feedback — all feedback types in one collection with a "type" field | |
| Feedback types: | |
| - "binary" — thumbs up/down on a specific message | |
| - "rating" — 1-5 stars for helpfulness, accuracy, relevance | |
| - "tags" — categorical tags (inaccurate, creative, etc.) | |
| - "text" — free text "what went wrong?" | |
| - "correction" — preferred answer (RLHF training data) | |
| - "product" — bug report, feature request, safety concern | |
| Endpoints: | |
| POST /api/feedback — Submit feedback | |
| GET /api/feedback/{uid} — Get user's feedback history (for admin) | |
| GET /api/feedback/stats — Aggregate feedback stats (for admin) | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import sys | |
| import logging | |
| from datetime import datetime, timezone | |
| from typing import Optional, Dict, Any, List | |
| from fastapi import APIRouter, Request, HTTPException, Query | |
| from fastapi.responses import JSONResponse | |
| from pydantic import BaseModel, Field | |
| log = logging.getLogger("agent.feedback") | |
| router = APIRouter(prefix="/api/feedback", tags=["feedback"]) | |
| # ── MONGODB (reuse main.py's connection) ────────────────────────────────────── | |
| _db = None | |
| def _get_db(): | |
| """Get the neuraprompt database. Reuses main.py's MongoClient.""" | |
| global _db | |
| if _db is not None: | |
| return _db | |
| try: | |
| main_mod = sys.modules.get("main") | |
| if main_mod and hasattr(main_mod, "neuraprompt_db"): | |
| _db = main_mod.neuraprompt_db | |
| return _db | |
| except Exception: | |
| pass | |
| # Fallback: own connection | |
| try: | |
| from pymongo import MongoClient | |
| from pymongo.server_api import ServerApi | |
| MONGO_URL = os.environ.get("MONGO_URI", "") | |
| client = MongoClient( | |
| MONGO_URL, ssl=True, | |
| tlsAllowInvalidCertificates=False, | |
| tlsCAFile="/etc/ssl/certs/ca-certificates.crt", | |
| server_api=ServerApi("1"), | |
| ) | |
| _db = client["neuraprompt"] | |
| return _db | |
| except Exception as e: | |
| log.error(f"[Feedback] DB connection failed: {e}") | |
| return None | |
| # ── PYDANTIC MODELS ─────────────────────────────────────────────────────────── | |
| class FeedbackRequest(BaseModel): | |
| user_id: str = Field(..., description="Firebase UID") | |
| feedback_type: str = Field(..., description="binary | rating | tags | text | correction | product") | |
| message_content: str = Field(default="", description="The AI message being feedbacked (truncated)") | |
| user_message: str = Field(default="", description="What the user asked") | |
| # Binary | |
| thumbs: Optional[str] = Field(default=None, description="up | down") | |
| # Rating (1-5) | |
| helpfulness: Optional[int] = Field(default=None, description="1-5") | |
| accuracy: Optional[int] = Field(default=None, description="1-5") | |
| relevance: Optional[int] = Field(default=None, description="1-5") | |
| # Tags | |
| tags: Optional[List[str]] = Field(default=None, description="List of tag names") | |
| # Free text | |
| comment: Optional[str] = Field(default=None, description="Free text feedback") | |
| # Correction | |
| preferred_answer: Optional[str] = Field(default=None, description="What the AI should have said") | |
| # Product feedback | |
| product_type: Optional[str] = Field(default=None, description="bug | feature | safety") | |
| # Metadata | |
| mode: Optional[str] = Field(default=None, description="fast | web | agent | kype") | |
| model_used: Optional[str] = Field(default=None, description="Model that generated the response") | |
| # ── ENDPOINTS ───────────────────────────────────────────────────────────────── | |
| async def submit_feedback(req: FeedbackRequest): | |
| """Submit user feedback. Stored in neuraprompt.feedback collection.""" | |
| db = _get_db() | |
| if db is None: | |
| return JSONResponse({"error": "Database unavailable"}, status_code=503) | |
| try: | |
| doc = { | |
| "user_id": req.user_id, | |
| "feedback_type": req.feedback_type, | |
| "message_content": req.message_content[:500], # truncate for storage | |
| "user_message": req.user_message[:500], | |
| "thumbs": req.thumbs, | |
| "helpfulness": req.helpfulness, | |
| "accuracy": req.accuracy, | |
| "relevance": req.relevance, | |
| "tags": req.tags, | |
| "comment": req.comment[:2000] if req.comment else None, | |
| "preferred_answer": req.preferred_answer[:2000] if req.preferred_answer else None, | |
| "product_type": req.product_type, | |
| "mode": req.mode, | |
| "model_used": req.model_used, | |
| "created_at": datetime.now(timezone.utc), | |
| "status": "new", # new | reviewed | resolved | |
| } | |
| result = db.feedback.insert_one(doc) | |
| log.info(f"[Feedback] Saved {req.feedback_type} feedback from {req.user_id}: {result.inserted_id}") | |
| return { | |
| "status": "success", | |
| "feedback_id": str(result.inserted_id), | |
| "message": "Thank you for your feedback!" | |
| } | |
| except Exception as e: | |
| log.error(f"[Feedback] Submit failed: {e}") | |
| return JSONResponse({"error": str(e)}, status_code=500) | |
| async def get_user_feedback(user_id: str): | |
| """Get a user's feedback history.""" | |
| db = _get_db() | |
| if db is None: | |
| return JSONResponse({"error": "Database unavailable"}, status_code=503) | |
| try: | |
| feedbacks = [] | |
| cursor = db.feedback.find({"user_id": user_id}).sort("created_at", -1).limit(50) | |
| for f in cursor: | |
| f["_id"] = str(f["_id"]) | |
| if "created_at" in f and hasattr(f["created_at"], "isoformat"): | |
| f["created_at"] = f["created_at"].isoformat() | |
| feedbacks.append(f) | |
| return {"feedbacks": feedbacks, "count": len(feedbacks)} | |
| except Exception as e: | |
| return JSONResponse({"error": str(e)}, status_code=500) | |
| async def get_feedback_stats(): | |
| """Get aggregate feedback statistics. For admin dashboard.""" | |
| db = _get_db() | |
| if db is None: | |
| return JSONResponse({"error": "Database unavailable"}, status_code=503) | |
| try: | |
| total = db.feedback.count_documents({}) | |
| thumbs_up = db.feedback.count_documents({"thumbs": "up"}) | |
| thumbs_down = db.feedback.count_documents({"thumbs": "down"}) | |
| # Tag distribution | |
| tag_pipeline = [ | |
| {"$unwind": "$tags"}, | |
| {"$group": {"_id": "$tags", "count": {"$sum": 1}}}, | |
| {"$sort": {"count": -1}}, | |
| ] | |
| tag_data = list(db.feedback.aggregate(tag_pipeline)) | |
| tag_dist = {t["_id"]: t["count"] for t in tag_data} | |
| # Type distribution | |
| type_pipeline = [ | |
| {"$group": {"_id": "$feedback_type", "count": {"$sum": 1}}}, | |
| ] | |
| type_data = list(db.feedback.aggregate(type_pipeline)) | |
| type_dist = {t["_id"]: t["count"] for t in type_data} | |
| # Average ratings | |
| avg_pipeline = [ | |
| {"$match": {"helpfulness": {"$gte": 1}}}, | |
| {"$group": { | |
| "_id": None, | |
| "avg_helpfulness": {"$avg": "$helpfulness"}, | |
| "avg_accuracy": {"$avg": "$accuracy"}, | |
| "avg_relevance": {"$avg": "$relevance"}, | |
| "count": {"$sum": 1}, | |
| }}, | |
| ] | |
| avg_data = list(db.feedback.aggregate(avg_pipeline)) | |
| avg_ratings = avg_data[0] if avg_data else {} | |
| return { | |
| "total_feedback": total, | |
| "thumbs_up": thumbs_up, | |
| "thumbs_down": thumbs_down, | |
| "thumb_ratio": round(thumbs_up / max(thumbs_up + thumbs_down, 1), 2), | |
| "tag_distribution": tag_dist, | |
| "type_distribution": type_dist, | |
| "average_ratings": { | |
| "helpfulness": round(avg_ratings.get("avg_helpfulness", 0), 1), | |
| "accuracy": round(avg_ratings.get("avg_accuracy", 0), 1), | |
| "relevance": round(avg_ratings.get("avg_relevance", 0), 1), | |
| "count": avg_ratings.get("count", 0), | |
| }, | |
| } | |
| except Exception as e: | |
| return JSONResponse({"error": str(e)}, status_code=500) | |