File size: 3,085 Bytes
938949f | 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 | """
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], # truncate for storage
"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"),
}
|