| """ |
| Feedback storage for the Vineyard Advisor chatbot. |
| |
| Logs user feedback (thumbs up/down, flags) to a JSON-lines file. |
| Each entry captures the query, response, tool results, rules applied, |
| and the user's feedback action. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import logging |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Optional |
|
|
| from config.settings import DATA_DIR |
|
|
| logger = logging.getLogger(__name__) |
|
|
| FEEDBACK_FILE = DATA_DIR / "advisor_feedback.jsonl" |
|
|
|
|
| def log_feedback( |
| query: str, |
| response: str, |
| feedback: str, |
| confidence: str = "", |
| sources: Optional[list[str]] = None, |
| tool_calls: Optional[list[dict]] = None, |
| rule_violations: Optional[list[dict]] = None, |
| response_mode: str = "", |
| comment: str = "", |
| ) -> None: |
| """Append a feedback entry to the JSONL file. |
| |
| Parameters |
| ---------- |
| query : str |
| The user's original question. |
| response : str |
| The chatbot's response text. |
| feedback : str |
| One of: "thumbs_up", "thumbs_down", "flag_incorrect". |
| confidence, sources, tool_calls, rule_violations, response_mode : |
| Metadata from the ChatResponse. |
| comment : str |
| Optional free-text comment from the user. |
| """ |
| entry = { |
| "timestamp": datetime.now(tz=timezone.utc).isoformat(), |
| "query": query, |
| "response": response[:500], |
| "feedback": feedback, |
| "confidence": confidence, |
| "sources": sources or [], |
| "tool_calls": [ |
| {"name": tc.get("name", ""), "args": tc.get("args", {})} |
| for tc in (tool_calls or []) |
| ], |
| "rule_violations": rule_violations or [], |
| "response_mode": response_mode, |
| "comment": comment, |
| } |
|
|
| try: |
| FEEDBACK_FILE.parent.mkdir(parents=True, exist_ok=True) |
| with open(FEEDBACK_FILE, "a") as f: |
| f.write(json.dumps(entry, default=str) + "\n") |
| logger.info("Feedback logged: %s for query: %s", feedback, query[:50]) |
| except Exception as exc: |
| logger.warning("Failed to log feedback: %s", exc) |
|
|
|
|
| def load_feedback(limit: int = 100) -> list[dict]: |
| """Load recent feedback entries.""" |
| if not FEEDBACK_FILE.exists(): |
| return [] |
|
|
| entries = [] |
| try: |
| with open(FEEDBACK_FILE) as f: |
| for line in f: |
| line = line.strip() |
| if line: |
| entries.append(json.loads(line)) |
| except Exception as exc: |
| logger.warning("Failed to load feedback: %s", exc) |
|
|
| return entries[-limit:] |
|
|
|
|
| def feedback_summary() -> dict: |
| """Return a summary of feedback stats.""" |
| entries = load_feedback(limit=10000) |
| if not entries: |
| return {"total": 0} |
|
|
| return { |
| "total": len(entries), |
| "thumbs_up": sum(1 for e in entries if e.get("feedback") == "thumbs_up"), |
| "thumbs_down": sum(1 for e in entries if e.get("feedback") == "thumbs_down"), |
| "flagged": sum(1 for e in entries if e.get("feedback") == "flag_incorrect"), |
| } |
|
|
