Node-Brain-BU3 / utils.py
broadfield-dev's picture
Create utils.py
081e43b verified
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