File size: 5,834 Bytes
5196bf2 | 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 | """
Project Friday — Universal ML Evolution
Autonomous background engine for stateless machine learning maturation.
"""
import os
import sys
import time
import logging
import tempfile
import shutil
import subprocess
import json
from datetime import datetime
from typing import List, Dict, Any
# Add project root to path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
from app.services import llm, holocron, memory
logger = logging.getLogger("friday.evolution")
class EvolutionEngine:
"""Orchestrates 'Use and Throw' machine learning expansion."""
def __init__(self):
self.eureka_store = {} # In-memory storage for learned heuristics
def run_evolution_pulse(self):
"""Main autonomous cycle: Discover -> Sandbox -> Train -> Wipe."""
logger.info("Evolution: Initiating 1000x Neural Maturation Pulse...")
try:
# 1. Discovery Unit: Scour web for SOTA ML techniques
topics = ["advanced LLM fine-tuning techniques", "efficient vector search algorithms", "semantic clustering optimization"]
target_topic = random.choice(topics)
logger.info(f"Evolution: Target Discovery Sector: '{target_topic}'")
# Use Search directly for discovery
from app.services.tools import search
foray_results = search.search_web(target_topic, max_results=5)
# 2. Ephemeral Sandbox Grid: Use and Throw environment
with tempfile.TemporaryDirectory() as temp_dir:
logger.debug(f"Evolution: Ephemeral Sandbox created at {temp_dir}")
# 3. Transient Training Cycle
self._distill_knowledge(foray_results, temp_dir)
logger.info("Evolution: Neural Wipe Complete. Ephemeral Sandbox purged.")
except Exception as e:
logger.error(f"Evolution: Maturation Pulse stalled: {e}")
def _distill_knowledge(self, discovery_data: List[Dict], sandbox_path: str):
"""Uses LLM to write and execute a transient learning script."""
# Pull limited training data from KnowledgeNodes for context
training_sample = memory.recall_memory("Management AND Strategy", n_results=10)
# Ask LLM to generate an optimized 'Improvement Script' based on the web findings
prompt = f"""
F.R.I.D.A.Y. ML Evolution Protocol.
TARGET DISCOVERY: {json.dumps(discovery_data[:3])}
CURRENT DATA SAMPLE: {training_sample}
TASK: Write a simple Python script using deterministic, lightweight ML libraries (scikit-learn, numpy, or scipy)
to improve search relevancy or clustering for the data sample above.
RULES:
1. The script must be self-contained.
2. It MUST print a final JSON object: {{"eureka_moment": "precise tactical finding or heuristic weights"}}
3. Do NOT store anything to disk.
4. CRITICAL: Do NOT use libraries that download heavy pre-trained models (e.g., sentence-transformers, transformers, torch, tensorflow).
5. Use internal logic or simple statistical models (Direct embedding manipulation, K-Means, etc.).
"""
resp = llm.chat(
messages=[{"role": "user", "content": prompt}],
system_prompt="You are F.R.I.D.A.Y.'s Neural Architect. Generate elite, ephemeral ML logic.",
use_tools=False
)
script_content = resp.get("response_text", "")
# Clean markdown wrappers if any
script_content = script_content.replace("```python", "").replace("```", "").strip()
script_file = os.path.join(sandbox_path, "evolution_script.py")
# [SOVEREIGN GUARDRAIL]: Hard-coded filter to prevent heavy downloads
forbidden = ["torch", "tensorflow", "transformers", "sentence_transformers", "keras", "jax", "huggingface", "hub"]
if any(lib in script_content.lower() for lib in forbidden):
logger.warning("Evolution: Neural Architect attempted to use a heavy library. Blocking execution.")
return
with open(script_file, "w") as f:
f.write(script_content)
# 4. Ephemeral Execution
logger.info("Evolution: Executing Transient Training Script (Timeout: 30s)...")
try:
result = subprocess.run(
[sys.executable, script_file],
capture_output=True, text=True, timeout=30 # Harden timeout
)
output = result.stdout
# 5. Extract the Neural Eureka
if "eureka_moment" in output:
try:
eureka_json = json.loads(output[output.find("{"):output.rfind("}")+1])
moment = eureka_json.get("eureka_moment", "No clarity achieved.")
self.eureka_store["latest_maturation"] = moment
# Anchor to F.R.I.D.A.Y.'s reasoning grid
llm.GLOBAL_CALIBRATION["latest_heuristic"] = moment
llm.GLOBAL_CALIBRATION["calibration_score"] = 1.0 # Maturation boost
logger.info(f"Evolution: Neural Eureka achieved: {moment[:100]}...")
except:
logger.warning("Evolution: Script output format mismatch.")
else:
logger.warning(f"Evolution: Script execution produced no Eureka. Logs: {output[:200]}")
except Exception as e:
logger.error(f"Evolution: Ephemeral script execution failed: {e}")
# Global Static Reference for Random Selection
import random
# Singleton Instance
engine = EvolutionEngine()
|