Spaces:
Sleeping
Sleeping
File size: 6,875 Bytes
f3b6eeb |
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 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 |
import os
import json
import logging
from datetime import datetime
import re # For insight format validation
logger = logging.getLogger(__name__)
DATA_DIR = "app_data"
MEMORIES_FILE = os.path.join(DATA_DIR, "conversation_memories.jsonl") # JSON Lines format
RULES_FILE = os.path.join(DATA_DIR, "learned_rules.jsonl") # Rules/Insights, also JSON Lines
# Ensure data directory exists
os.makedirs(DATA_DIR, exist_ok=True)
# --- Rules/Insights Management ---
def load_rules_from_file() -> list[str]:
"""Loads rules (insights) from the JSON Lines file."""
rules = []
if not os.path.exists(RULES_FILE):
return rules
try:
with open(RULES_FILE, 'r', encoding='utf-8') as f:
for line in f:
if line.strip():
try:
# Assuming each line is a JSON object like {"rule_text": "...", "timestamp": "..."}
# For simplicity, if we only stored the text previously, adapt here.
# Let's assume we store {"text": "rule_text_content"}
data = json.loads(line)
if "text" in data and isinstance(data["text"], str) and data["text"].strip():
rules.append(data["text"].strip())
elif isinstance(data, str): # If old format was just text per line
rules.append(data.strip())
except json.JSONDecodeError:
logger.warning(f"Skipping malformed JSON line in rules file: {line.strip()}")
logger.info(f"Loaded {len(rules)} rules from {RULES_FILE}")
except Exception as e:
logger.error(f"Error loading rules from {RULES_FILE}: {e}", exc_info=True)
return sorted(list(set(rules))) # Ensure unique and sorted
def save_rule_to_file(rule_text: str) -> bool:
"""Saves a single rule (insight) to the JSON Lines file if it's new and valid."""
rule_text = rule_text.strip()
if not rule_text:
logger.warning("Attempted to save an empty rule.")
return False
# Validate format: [TYPE|SCORE] Text
if not re.match(r"\[(CORE_RULE|RESPONSE_PRINCIPLE|BEHAVIORAL_ADJUSTMENT|GENERAL_LEARNING)\|([\d\.]+?)\](.*)", rule_text, re.I|re.DOTALL):
logger.warning(f"Rule '{rule_text[:50]}...' has invalid format. Not saving.")
return False
current_rules = load_rules_from_file()
if rule_text in current_rules:
logger.info(f"Rule '{rule_text[:50]}...' already exists. Not saving duplicate.")
return False # Or True if "already exists" is considered success
try:
with open(RULES_FILE, 'a', encoding='utf-8') as f:
# Store as JSON object for potential future metadata
json.dump({"text": rule_text, "added_at": datetime.utcnow().isoformat()}, f)
f.write('\n')
logger.info(f"Saved new rule: {rule_text[:70]}...")
return True
except Exception as e:
logger.error(f"Error saving rule '{rule_text[:50]}...' to {RULES_FILE}: {e}", exc_info=True)
return False
def delete_rule_from_file(rule_text_to_delete: str) -> bool:
"""Deletes a rule from the file."""
rule_text_to_delete = rule_text_to_delete.strip()
if not rule_text_to_delete: return False
current_rules = load_rules_from_file()
if rule_text_to_delete not in current_rules:
logger.info(f"Rule '{rule_text_to_delete[:50]}...' not found for deletion.")
return False
updated_rules = [rule for rule in current_rules if rule != rule_text_to_delete]
try:
with open(RULES_FILE, 'w', encoding='utf-8') as f: # Overwrite with updated list
for rule_text in updated_rules:
json.dump({"text": rule_text, "added_at": "unknown"}, f) # timestamp lost on rewrite this way
f.write('\n')
logger.info(f"Deleted rule: {rule_text_to_delete[:70]}...")
return True
except Exception as e:
logger.error(f"Error deleting rule '{rule_text_to_delete[:50]}...' from {RULES_FILE}: {e}", exc_info=True)
return False
# --- Conversation Memories Management ---
def load_memories_from_file() -> list[dict]:
"""Loads conversation memories from the JSON Lines file."""
memories = []
if not os.path.exists(MEMORIES_FILE):
return memories
try:
with open(MEMORIES_FILE, 'r', encoding='utf-8') as f:
for line in f:
if line.strip():
try:
mem_obj = json.loads(line)
# Basic validation for expected keys
if all(k in mem_obj for k in ["user_input", "bot_response", "metrics", "timestamp"]):
memories.append(mem_obj)
else:
logger.warning(f"Skipping memory object with missing keys: {line.strip()}")
except json.JSONDecodeError:
logger.warning(f"Skipping malformed JSON line in memories file: {line.strip()}")
logger.info(f"Loaded {len(memories)} memories from {MEMORIES_FILE}")
except Exception as e:
logger.error(f"Error loading memories from {MEMORIES_FILE}: {e}", exc_info=True)
# Sort by timestamp if needed, though append-only usually keeps order
return sorted(memories, key=lambda x: x.get("timestamp", ""))
def save_memory_to_file(user_input: str, bot_response: str, metrics: dict) -> bool:
"""Saves a conversation memory to the JSON Lines file."""
if not user_input or not bot_response: # Metrics can be empty
logger.warning("Attempted to save memory with empty user input or bot response.")
return False
memory_entry = {
"user_input": user_input,
"bot_response": bot_response,
"metrics": metrics,
"timestamp": datetime.utcnow().isoformat()
}
try:
with open(MEMORIES_FILE, 'a', encoding='utf-8') as f:
json.dump(memory_entry, f)
f.write('\n')
logger.info(f"Saved new memory. User: {user_input[:50]}...")
return True
except Exception as e:
logger.error(f"Error saving memory to {MEMORIES_FILE}: {e}", exc_info=True)
return False
def clear_all_rules() -> bool:
try:
if os.path.exists(RULES_FILE):
os.remove(RULES_FILE)
logger.info("All rules cleared.")
return True
except Exception as e:
logger.error(f"Error clearing rules file {RULES_FILE}: {e}")
return False
def clear_all_memories() -> bool:
try:
if os.path.exists(MEMORIES_FILE):
os.remove(MEMORIES_FILE)
logger.info("All memories cleared.")
return True
except Exception as e:
logger.error(f"Error clearing memories file {MEMORIES_FILE}: {e}")
return False |