Spaces:
Runtime error
Runtime error
import os | |
import json | |
import re | |
import logging | |
from memory_logic import add_rule_entry, add_memory_entry | |
logger = logging.getLogger(__name__) | |
def format_insights_for_prompt(retrieved_insights_list: list[str]) -> tuple[str, list[dict]]: | |
if not retrieved_insights_list: return "No specific guiding principles or learned insights retrieved.", [] | |
parsed = [] | |
for text in retrieved_insights_list: | |
match = re.match(r"\[(CORE_RULE|RESPONSE_PRINCIPLE|BEHAVIORAL_ADJUSTMENT|GENERAL_LEARNING)\|([\d\.]+?)\](.*)", text.strip(), re.DOTALL | re.IGNORECASE) | |
if match: parsed.append({"type": match.group(1).upper().replace(" ", "_"), "score": match.group(2), "text": match.group(3).strip(), "original": text.strip()}) | |
else: parsed.append({"type": "GENERAL_LEARNING", "score": "0.5", "text": text.strip(), "original": text.strip()}) | |
parsed.sort(key=lambda x: float(x["score"]) if x["score"].replace('.', '', 1).isdigit() else -1.0, reverse=True) | |
grouped = {"CORE_RULE": [], "RESPONSE_PRINCIPLE": [], "BEHAVIORAL_ADJUSTMENT": [], "GENERAL_LEARNING": []} | |
for p_item in parsed: grouped.get(p_item["type"], grouped["GENERAL_LEARNING"]).append(f"- (Score: {p_item['score']}) {p_item['text']}") | |
sections = [f"{k.replace('_', ' ').title()}:\n" + "\n".join(v) for k, v in grouped.items() if v] | |
return "\n\n".join(sections) if sections else "No guiding principles retrieved.", parsed | |
def load_rules_from_file(filepath: str | None, progress_callback=None): | |
if not filepath or not os.path.exists(filepath): return 0, 0, 0 | |
added, skipped, errors = 0, 0, 0 | |
with open(filepath, 'r', encoding='utf-8') as f: content = f.read() | |
if not content.strip(): return 0, 0, 0 | |
potential_rules = [] | |
if filepath.lower().endswith(".txt"): | |
potential_rules = content.split("\n\n---\n\n") | |
if len(potential_rules) == 1 and "\n" in content: potential_rules = content.splitlines() | |
elif filepath.lower().endswith(".jsonl"): | |
for line in content.splitlines(): | |
if line.strip(): | |
try: potential_rules.append(json.loads(line)) | |
except json.JSONDecodeError: errors += 1 | |
valid_rules = [r.strip() for r in potential_rules if isinstance(r, str) and r.strip()] | |
total = len(valid_rules) | |
if not total: return 0, 0, errors | |
for idx, rule_text in enumerate(valid_rules): | |
success, status_msg = add_rule_entry(rule_text) | |
if success: added += 1 | |
elif status_msg == "duplicate": skipped += 1 | |
else: errors += 1 | |
if progress_callback: progress_callback((idx + 1) / total, f"Processed {idx+1}/{total} rules...") | |
logger.info(f"Loaded rules from {filepath}: Added {added}, Skipped {skipped}, Errors {errors}.") | |
return added, skipped, errors | |
def load_memories_from_file(filepath: str | None, progress_callback=None): | |
if not filepath or not os.path.exists(filepath): return 0, 0, 0 | |
added, format_err, save_err = 0, 0, 0 | |
with open(filepath, 'r', encoding='utf-8') as f: content = f.read() | |
if not content.strip(): return 0, 0, 0 | |
mem_objects = [] | |
if filepath.lower().endswith(".json"): | |
try: | |
data = json.loads(content) | |
mem_objects = data if isinstance(data, list) else [data] | |
except json.JSONDecodeError: format_err = 1 | |
elif filepath.lower().endswith(".jsonl"): | |
for line in content.splitlines(): | |
if line.strip(): | |
try: mem_objects.append(json.loads(line)) | |
except json.JSONDecodeError: format_err += 1 | |
total = len(mem_objects) | |
if not total: return 0, format_err, 0 | |
for idx, mem in enumerate(mem_objects): | |
if isinstance(mem, dict) and all(k in mem for k in ["user_input", "bot_response", "metrics"]): | |
success, _ = add_memory_entry(mem["user_input"], mem["metrics"], mem["bot_response"]) | |
if success: added += 1 | |
else: save_err += 1 | |
else: format_err += 1 | |
if progress_callback: progress_callback((idx + 1) / total, f"Processed {idx+1}/{total} memories...") | |
logger.info(f"Loaded memories from {filepath}: Added {added}, Format Errors {format_err}, Save Errors {save_err}.") | |
return added, format_err, save_err |