Spaces:
Sleeping
Sleeping
File size: 2,417 Bytes
3c25c17 | 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 | from __future__ import annotations
import json
import os
import uuid
from datetime import datetime
from pathlib import Path
from config import settings
_memory_file: Path | None = None
def _get_memory_file() -> Path:
global _memory_file
if _memory_file is None:
os.makedirs(settings.memory_dir, exist_ok=True)
_memory_file = Path(settings.memory_dir) / "problems.jsonl"
return _memory_file
def save_record(
input_type: str,
extracted_text: str,
parsed_problem: dict,
topic: str,
retrieved_chunks: list[str],
solution: str,
solution_steps: list[str],
verification: dict,
explanation: str,
user_feedback: str = "",
user_comment: str = "",
) -> str:
record_id = str(uuid.uuid4())
record = {
"id": record_id,
"timestamp": datetime.now().isoformat(),
"input_type": input_type,
"extracted_text": extracted_text,
"parsed_problem": parsed_problem,
"topic": topic,
"retrieved_chunk_sources": retrieved_chunks,
"solution": solution,
"solution_steps": solution_steps,
"verification": verification,
"explanation": explanation,
"user_feedback": user_feedback,
"user_comment": user_comment,
}
path = _get_memory_file()
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
return record_id
def get_all_records() -> list[dict]:
path = _get_memory_file()
if not path.exists():
return []
records = []
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line:
try:
records.append(json.loads(line))
except json.JSONDecodeError:
continue
return records
def update_feedback(record_id: str, feedback: str, comment: str = "") -> bool:
path = _get_memory_file()
if not path.exists():
return False
records = get_all_records()
updated = False
with open(path, "w", encoding="utf-8") as f:
for record in records:
if record.get("id") == record_id:
record["user_feedback"] = feedback
record["user_comment"] = comment
updated = True
f.write(json.dumps(record, ensure_ascii=False) + "\n")
return updated
|