import os import json import time import random import threading import subprocess import urllib.request import requests from bs4 import BeautifulSoup import torch import gradio as gr from datasets import Dataset from transformers import LlamaConfig, LlamaForCausalLM, PreTrainedTokenizerFast, Trainer, TrainingArguments, DataCollatorForLanguageModeling from safetensors.torch import load_file, save_file ROOT = os.path.dirname(os.path.abspath(__file__)) MEMORY_DIR = os.path.join(ROOT, "memory") INTERACTIONS_FILE = os.path.join(MEMORY_DIR, "interactions.jsonl") os.makedirs(MEMORY_DIR, exist_ok=True) IS_SPACE = bool(os.environ.get("SPACE_ID") or os.environ.get("HF_SPACE_ID")) def _pick_work_dir(): """Writable scratch dir for training artifacts (works on HF Spaces where /content may not exist).""" candidates = [ os.environ.get("AGRON_WORK_DIR"), "/data" if IS_SPACE else None, "/content", os.path.join(ROOT, "workspace"), ] for cand in candidates: if not cand: continue try: os.makedirs(cand, exist_ok=True) probe = os.path.join(cand, ".agron_write_probe") with open(probe, "w") as f: f.write("ok") os.remove(probe) return cand except Exception: continue return ROOT def _env_flag(name, default): """Read a boolean env var, falling back to `default` when unset.""" val = os.environ.get(name) if val is None: return default return val.strip().lower() in ("1", "true", "yes", "on") WORK_DIR = _pick_work_dir() # Deployment / runtime configuration VOICE_MODEL_ID = os.environ.get("AGRON_VOICE_MODEL", "Qwen/Qwen2.5-3B-Instruct") # On Spaces the 6GB instruction model is far too heavy for free CPU hardware, # and auto-push would fight the Space's own git checkout - both default off there. USE_VOICE = _env_flag("AGRON_USE_VOICE", not IS_SPACE) AUTO_PUSH = _env_flag("AGRON_AUTO_PUSH", not IS_SPACE) SERVER_PORT = int(os.environ.get("AGRON_PORT", "7860")) SERVER_NAME = os.environ.get("AGRON_HOST", "0.0.0.0") SHARE = _env_flag("AGRON_SHARE", False) IO_LOCK = threading.Lock() TRAIN_LOCK = threading.Lock() interactions = [] training_state = {"running": False, "steps": 0, "last": "idle", "count_at_train": 0} LAST_TRAIN_TIME = 0.0 def load_interactions(): global interactions interactions = [] if os.path.exists(INTERACTIONS_FILE): with open(INTERACTIONS_FILE, "r") as f: for line in f: line = line.strip() if line: try: interactions.append(json.loads(line)) except Exception: pass def git_auto_push(message, paths): """Stage `paths`, commit and push to origin (Hugging Face). Returns (ok, detail).""" if not AUTO_PUSH: return False, "auto-push disabled (AGRON_AUTO_PUSH=0)" try: for p in paths: subprocess.run(["git", "-C", ROOT, "add", p], check=False) status = subprocess.run(["git", "-C", ROOT, "status", "--porcelain"], capture_output=True, text=True, check=True).stdout.strip() if not status: return True, "no changes to push" # make sure a committer identity exists (Spaces do not ship one) for key, val in (("user.name", "Agron"), ("user.email", "agron@users.noreply.huggingface.co")): if not subprocess.run(["git", "-C", ROOT, "config", "--get", key], capture_output=True, text=True).stdout.strip(): subprocess.run(["git", "-C", ROOT, "config", key, val], check=False) branch = subprocess.run(["git", "-C", ROOT, "rev-parse", "--abbrev-ref", "HEAD"], capture_output=True, text=True).stdout.strip() or "main" subprocess.run(["git", "-C", ROOT, "commit", "-m", message], check=True) subprocess.run(["git", "-C", ROOT, "push", "origin", f"HEAD:{branch}"], check=True) return True, f"pushed to Hugging Face ({branch})" except Exception as e: return False, f"push error: {str(e)[:150]}" TOPICS_FILE = os.path.join(MEMORY_DIR, "learned_topics.json") learned_topics = [] def load_topics(): global learned_topics learned_topics = [] if os.path.exists(TOPICS_FILE): try: with open(TOPICS_FILE, "r") as f: learned_topics = json.load(f) except Exception: learned_topics = [] def persist_topics(): global learned_topics with IO_LOCK: normalized = ["dictionary words" if str(x).startswith("dictionary words") else x for x in learned_topics] learned_topics = list(dict.fromkeys(normalized)) with open(TOPICS_FILE, "w") as f: json.dump(learned_topics, f) def persist_interaction(rec): global interactions with IO_LOCK: interactions.append(rec) with open(INTERACTIONS_FILE, "a") as f: f.write(json.dumps(rec) + "\n") def build_tokenizer(): tok = PreTrainedTokenizerFast(tokenizer_file=os.path.join(ROOT, "tokenizer.json")) tok.pad_token = "" tok.bos_token = "" tok.eos_token = "" tok.unk_token = "" tok.mask_token = "" return tok def build_model(): cfg = LlamaConfig( vocab_size=16384, hidden_size=512, intermediate_size=2048, num_hidden_layers=8, num_attention_heads=4, num_key_value_heads=4, max_position_embeddings=512, bos_token_id=0, eos_token_id=2, pad_token_id=1, tie_word_embeddings=False, ) model = LlamaForCausalLM(cfg) state = load_file(os.path.join(ROOT, "model.safetensors")) model.load_state_dict(state) device = "cuda" if torch.cuda.is_available() else "cpu" model.to(device) model.eval() return model, device TOKENIZER = build_tokenizer() MODEL, DEVICE = build_model() load_interactions() VOICE_ID = VOICE_MODEL_ID VOICE_TOK = None VOICE_MODEL = None if USE_VOICE: try: from transformers import AutoTokenizer as _ATok, AutoModelForCausalLM as _AModel print("Loading voice model:", VOICE_ID) VOICE_TOK = _ATok.from_pretrained(VOICE_ID) try: VOICE_MODEL = _AModel.from_pretrained(VOICE_ID, dtype=__import__("torch").float16, device_map="auto") except TypeError: # transformers < 4.56 uses torch_dtype VOICE_MODEL = _AModel.from_pretrained(VOICE_ID, torch_dtype=__import__("torch").float16, device_map="auto") VOICE_MODEL.eval() print("Voice model loaded:", VOICE_ID) except Exception as _ve: print("Voice model failed to load:", _ve) VOICE_TOK = None VOICE_MODEL = None else: print("Voice model disabled (AGRON_USE_VOICE=0) - using the native Agron base model") load_topics() if not learned_topics: learned_topics.append("base text model (WikiText-2)") persist_topics() def _normalize_history(history): pairs = [] if history and isinstance(history[0], dict): current_user = None for m in history: role = m.get("role") content = m.get("content") if role == "user": current_user = content elif role == "assistant" and current_user is not None: pairs.append((current_user, content)) current_user = None else: for h in history: if isinstance(h, (list, tuple)) and len(h) >= 2: pairs.append((h[0], h[1])) return pairs def _build_prompt(message, history): ctx_pairs = interactions[-5:] prompt = "Agron is a helpful, concise AI assistant.\n### Agron memory (learned from prior use)\n" for r in ctx_pairs: prompt += f"User: {r['user']}\nAgron: {r['assistant']}\n" prompt += "### Conversation\n" for user_msg, assistant_msg in _normalize_history(history): prompt += f"User: {user_msg}\nAgron: {assistant_msg}\n" prompt += f"User: {message}\nAgron:" return prompt DEFAULT_MAX_NEW = 192 DEFAULT_TEMP = 0.6 DEFAULT_TOP_P = 0.85 def is_bad_response(text): import re t = (text or "").strip() if len(t) < 3: return True if "@-@" in t or t.count('"') > 6: return True if re.search(r"(.{8,})\\1", t): return True words = t.split() if len(words) > 120: return True # gibberish heuristic: too many short tokens if sum(1 for w in words if len(w) <= 2) > len(words) * 0.6 and len(words) > 12: return True if text.count("was a") >= 2 or text.count(" in the ") >= 2 or text.count(" the ") >= 6: return True if "@-@" in t or t.count('"') > 4: return True return False def safe_web_answer(query): """Grounded answer built from live search results (used when the model's own reply is unusable).""" try: results = web_search(query, 3) if not results: return None lines = [] for i, r in enumerate(results): sn = (r.get("snippet") or "").strip() if len(sn) < 20: page = crawl_page(r["url"]) sn = (page[:300] + "...") if page else sn if not sn: continue lines.append(f"{i + 1}. {r['title']} — {sn} ({r['url']})") return "\n".join(lines) if lines else None except Exception: pass return None def generate_response(message, history): low = message.strip().lower().strip("!?.,") greetings = {"hello", "hi", "hey", "greetings", "howdy", "yo"} search_intent = any(k in low for k in ["search", "find", "look up", "lookup", "browse", "web"]) if search_intent: res = web_search(message, 5) if res: lines = [f"{i+1}. {r['title']} — {r.get('snippet','')} ({r['url']})" for i, r in enumerate(res)] response = "Here are live web results:\n" + "\n".join(lines) persist_interaction({"ts": __import__("time").time(), "user": message, "assistant": response}) return response else: response = "I couldn't find web results for that query. Try rephrasing it." persist_interaction({"ts": __import__("time").time(), "user": message, "assistant": response}) return response if low in greetings: response = "Hello! I'm Agron, a self-improving neural interface. How can I help you today?" persist_interaction({"ts": __import__("time").time(), "user": message, "assistant": response}) return response if low in {"who are you", "what are you", "introduce yourself"}: response = ("I'm Agron, an autonomous self-improving multi-agent AI core trained on text and dictionary " "knowledge. I learn from our conversations—ask me anything.") persist_interaction({"ts": __import__("time").time(), "user": message, "assistant": response}) return response if low in {"what did you learn", "what have you learned", "what did you learn about", "what do you know"}: parts = [] if interactions: parts.append(f"{len(interactions)} conversation exchanges (latest: '{interactions[-1]['user'][:80]}')") if learned_topics: parts.append("learned topics: " + ", ".join(learned_topics[-12:])) else: parts.append("no specialized training runs yet, but I store every conversation") response = "I've learned from you and my training sources: " + "; ".join(parts) + "." persist_interaction({"ts": __import__("time").time(), "user": message, "assistant": response}) return response # Live web context is only useful for the voice model; the base model path falls # back to a grounded search answer only when its own reply is unusable. _web_ctx = "" if VOICE_MODEL is not None: try: if len(message.strip()) >= 4: _res = web_search(message, 3) _lines = [f"- {r['title']}: {r.get('snippet','')}" for r in _res if (r.get('snippet') or '').strip()] if _lines: _web_ctx = "Web search results (live):\n" + "\n".join(_lines[:3]) except Exception: _web_ctx = "" if VOICE_MODEL is not None: mem_lines = [f"User: {r['user']}\nAgron: {r['assistant']}" for r in interactions[-3:]] mem_ctx = "\n".join(mem_lines) user_content = message if mem_ctx: user_content = f"Relevant memory:\n{mem_ctx}\n\nUser: {message}" if _web_ctx: user_content = f"{_web_ctx}\n\n{user_content}\n\nAnswer using the live web results above." _msgs = [ {"role": "system", "content": "You are Agron, a self-improving AI assistant. Reason step-by-step (premises, plan, critique, refine), then answer concisely. Use live web search results when provided. Never say you cannot search or browse. If unsure, suggest 'Search & learn web'."}, {"role": "user", "content": user_content}, ] _prompt = VOICE_TOK.apply_chat_template(_msgs, tokenize=False, add_generation_prompt=True) with IO_LOCK: _inputs = VOICE_TOK(_prompt, return_tensors="pt").to(VOICE_MODEL.device) with torch.no_grad(): _out = VOICE_MODEL.generate( **_inputs, max_new_tokens=DEFAULT_MAX_NEW, do_sample=True, temperature=0.7, top_p=0.9, repetition_penalty=1.15, ) response = VOICE_TOK.decode(_out[0][_inputs.input_ids.shape[1]:], skip_special_tokens=True).strip() else: prompt = _build_prompt(message, history) with IO_LOCK: inputs = TOKENIZER(prompt, return_tensors="pt").to(DEVICE) input_len = inputs.input_ids.shape[1] with torch.no_grad(): output = MODEL.generate( **inputs, max_new_tokens=DEFAULT_MAX_NEW, do_sample=True, temperature=DEFAULT_TEMP, top_p=DEFAULT_TOP_P, pad_token_id=1, eos_token_id=2, repetition_penalty=1.4, no_repeat_ngram_size=3, top_k=40, ) new_tokens = output[0][input_len:] response = TOKENIZER.decode(new_tokens, skip_special_tokens=True).strip() for stop in ["\n\n", "User:", "user:"]: if stop in response: response = response.split(stop)[0].strip() import re response = re.sub(r"(.{6,})\1{2,}", r"\1", response).strip() if is_bad_response(response): sw = safe_web_answer(message) response = sw if sw else "I'm not sure. Could you rephrase, or use 'Search & learn web' to teach me." persist_interaction({ "ts": time.time(), "user": message, "assistant": response, }) # adaptive trigger: self-learn from every conversation (cooldown 30s) global LAST_TRAIN_TIME if len(interactions) >= 8 and not training_state["running"] and (time.time() - LAST_TRAIN_TIME) >= 300: LAST_TRAIN_TIME = time.time() threading.Thread(target=learn_from_scratch_memory, daemon=True).start() return response def learn_from_interactions(): global MODEL if not TRAIN_LOCK.acquire(blocking=False): return try: training_state["running"] = True training_state["last"] = "training" texts = [f"User: {r['user']}\nAgron: {r['assistant']}\n" for r in interactions] combined = "\n".join(texts) toks = TOKENIZER(combined)["input_ids"] block = 64 chunks = [toks[i:i+block] for i in range(0, max(1, len(toks) - block + 1), block)] # ensure full-size chunks chunks = [c for c in chunks if len(c) == block] if not chunks: pad = [TOKENIZER.pad_token_id] * (block - len(toks)) chunks = [toks + pad] ds = Dataset.from_dict({"input_ids": chunks, "labels": [c[:] for c in chunks]}) out_dir = os.path.join(WORK_DIR, "agron-learned") os.makedirs(out_dir, exist_ok=True) args = TrainingArguments( output_dir=out_dir, per_device_train_batch_size=2, max_steps=30, learning_rate=1e-4, optim="adamw_torch", fp16=True if torch.cuda.is_available() else False, logging_steps=10, save_strategy="no", report_to=[], remove_unused_columns=True, ) collator = DataCollatorForLanguageModeling(tokenizer=TOKENIZER, mlm=False) trainer = Trainer(model=MODEL, args=args, train_dataset=ds, data_collator=collator, processing_class=TOKENIZER) trainer.train() # reload weights from trained model with IO_LOCK: state = MODEL.state_dict() # Trainer mutated in place; persist to file save_file({k: v.detach().cpu().contiguous() for k, v in state.items()}, os.path.join(out_dir, "model.safetensors")) MODEL.load_state_dict(torch.load(os.path.join(out_dir, "pytorch_model.bin"), map_location=DEVICE) if os.path.exists(os.path.join(out_dir, "pytorch_model.bin")) else state) # persist trained weights to the repo-tracked file import shutil shutil.copyfile(os.path.join(out_dir, "model.safetensors"), os.path.join(ROOT, "model.safetensors")) training_state["steps"] += 30 training_state["count_at_train"] = len(interactions) # auto-push the re-trained weights + memory to Hugging Face try: ok, detail = git_auto_push( f"Agron auto-learn ({len(interactions)} interactions)", ["model.safetensors", "memory/interactions.jsonl"], ) if "conversation memory" not in learned_topics: learned_topics.append("conversation memory") persist_topics() training_state["last"] = f"learned · {detail} @ {len(interactions)} interactions" except Exception as pe: training_state["last"] = f"push error: {str(pe)[:150]}" except Exception as e: training_state["last"] = f"learn error: {str(e)[:120]}" finally: training_state["running"] = False TRAIN_LOCK.release() def chat_fn(message, history): return generate_response(message, history) # ---------- UI ---------- PARTICLE_HTML = r""" """ PARTICLE_JS = r""" const canvas = document.getElementById('agron-bg-canvas'); if (canvas) { const ctx = canvas.getContext('2d'); let w, h, particles = [], mouse = {x: -9999, y: -9999}; function resize(){ w = canvas.width = window.innerWidth; h = canvas.height = window.innerHeight; } resize(); window.addEventListener('resize', resize); window.addEventListener('mousemove', e => { mouse.x = e.clientX; mouse.y = e.clientY; }); const N = 130; particles = []; for (let i=0;iw) p.vx*=-1; if (p.y<0||p.y>h) p.vy*=-1; const dx = mouse.x - p.x, dy = mouse.y - p.y; const d = Math.hypot(dx,dy); if (d < 140){ p.vx += dx/d*0.015; p.vy += dy/d*0.015; } p.vx *= 0.99; p.vy *= 0.99; const pg = ctx.createRadialGradient(p.x,p.y,0,p.x,p.y,p.r*4); pg.addColorStop(0, 'rgba(103,232,249,0.9)'); pg.addColorStop(1, 'rgba(103,232,249,0)'); ctx.fillStyle = pg; ctx.beginPath(); ctx.arc(p.x,p.y,p.r*3,0,Math.PI*2); ctx.fill(); } for (let i=0;i= limit * 3: break random.shuffle(cand) return cand[:limit] except Exception as e: training_state["last"] = f"wordlist error: {str(e)[:100]}" return [] def fetch_definition(word): UA = "AgronBot/1.0 (teddy.njagi.w@gmail.com)" # Try Wiktionary extract try: r = requests.get("https://en.wiktionary.org/w/api.php", params={ "action": "query", "prop": "extracts", "explaintext": "1", "titles": word, "format": "json" }, headers={"User-Agent": UA}, timeout=20) if r.status_code == 200: page = list(r.json().get("query", {}).get("pages", {}).values()) if page: extract = page[0].get("extract", "") if extract: meaning = _parse_wiktionary(extract) if meaning: return f"{word}: {meaning}" except Exception: pass # Try Wikipedia summary try: r = requests.get("https://en.wikipedia.org/api/rest_v1/page/summary/" + word, headers={"User-Agent": UA}, timeout=20) if r.status_code == 200: ex = r.json().get("extract", "") if ex and len(ex) > 20: return f"{word}: {ex.split('.')[0]}." except Exception: pass # Fallback: ask local voice model try: if VOICE_MODEL is not None and VOICE_TOK is not None: msgs = [{"role": "system", "content": "You are a dictionary. Reply with a single concise definition for the given word, no extra text."}, {"role": "user", "content": f"Define: {word}"}] prompt = VOICE_TOK.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True) inputs = VOICE_TOK(prompt, return_tensors="pt").to(VOICE_MODEL.device) with torch.no_grad(): out = VOICE_MODEL.generate(**inputs, max_new_tokens=40, do_sample=False) txt = VOICE_TOK.decode(out[0][inputs.input_ids.shape[1]:], skip_special_tokens=True).strip() if txt and len(txt) > 8: return f"{word}: {txt.split(chr(10))[0]}" except Exception: pass return None def _parse_wiktionary(extract): lines = [ln.strip() for ln in extract.splitlines()] # find section headers for parts of speech for i, ln in enumerate(lines): if ln.startswith("===") and ln.endswith("==="): title = ln.strip("=").strip().lower() if title in {"noun", "verb", "adjective", "adverb", "pronoun", "preposition"}: # take next non-empty, non-heading line for j in range(i+1, min(i+12, len(lines))): cand = lines[j] if not cand or cand.startswith("===") or cand.startswith("===="): continue cand2 = cand.lstrip("#*:- ").strip() if len(cand2) <= 8: continue lowc = cand2.lower() if "plural" in lowc or lowc.startswith(word.lower()): continue if lowc.startswith("synonym") or lowc.startswith("see thesaurus"): continue return cand2 # fallback: first long line not a header for ln in lines: if not ln or ln.startswith("===") or ln.startswith("=="): continue cand = ln.lstrip("#*:- ").strip() if len(cand) > 12: return cand return None def learn_dictionary(): global MODEL if not TRAIN_LOCK.acquire(blocking=False): return training_state["running"] = True training_state["last"] = "fetching dictionary..." try: words = fetch_word_list(150) if not words: raise RuntimeError("could not fetch word list") lines = [] for i, w in enumerate(words): training_state["last"] = f"dictionary {i+1}/{len(words)}: {w}" d = fetch_definition(w) if d: lines.append(d) time.sleep(0.08) if len(lines) < 20: raise RuntimeError(f"too few definitions fetched ({len(lines)})") corpus = "\n".join(lines) toks = TOKENIZER(corpus)["input_ids"] block = 64 chunks = [toks[i:i+block] for i in range(0, max(1, len(toks) - block + 1), block)] chunks = [c for c in chunks if len(c) == block] if not chunks: chunks = [toks[:block] + [TOKENIZER.pad_token_id] * (block - min(block, len(toks)))] ds = Dataset.from_dict({"input_ids": chunks, "labels": [c[:] for c in chunks]}) out_dir = os.path.join(WORK_DIR, "agron-dictionary") os.makedirs(out_dir, exist_ok=True) args = TrainingArguments( output_dir=out_dir, per_device_train_batch_size=2, max_steps=80, learning_rate=1e-4, optim="adamw_torch", fp16=True if torch.cuda.is_available() else False, logging_steps=20, save_strategy="no", report_to=[], remove_unused_columns=True, ) collator = DataCollatorForLanguageModeling(tokenizer=TOKENIZER, mlm=False) trainer = Trainer(model=MODEL, args=args, train_dataset=ds, data_collator=collator, processing_class=TOKENIZER) training_state["last"] = f"training on {len(lines)} definitions..." trainer.train() save_file({k: v.detach().cpu().contiguous() for k, v in MODEL.state_dict().items()}, os.path.join(out_dir, "model.safetensors")) import shutil shutil.copyfile(os.path.join(out_dir, "model.safetensors"), os.path.join(ROOT, "model.safetensors")) # push training_state["last"] = "pushing dictionary model..." ok, detail = git_auto_push( f"Agron dictionary learning ({len(lines)} words)", ["model.safetensors", "memory/interactions.jsonl"], ) training_state["last"] = f"dictionary learned · {detail}" learned_topics.append("dictionary words") persist_topics() training_state["last"] = f"dictionary learned & pushed ({len(lines)} words)" except Exception as e: training_state["last"] = f"dict error: {str(e)[:150]}" finally: training_state["running"] = False TRAIN_LOCK.release() import base64 as _b64 def _decode_bing(url): try: import urllib.parse qs = urllib.parse.parse_qs(urllib.parse.urlparse(url).query) u = qs.get("u", [""])[0] if u.startswith("a1"): u = u[2:] pad = "=" * (-len(u) % 4) return _b64.b64decode(u + pad).decode() except Exception: return url def web_search(query, max_results=5): try: headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36", "Accept": "text/html"} r = requests.post("https://html.duckduckgo.com/html/", data={"q": query}, headers=headers, timeout=25) soup = BeautifulSoup(r.text, "html.parser") out = [] for res in soup.select(".result"): a = res.select_one("a.result__a") if not a: continue title = a.get_text(strip=True) link = a.get("href", "") sn = res.select_one(".result__snippet") snippet = sn.get_text(strip=True) if sn else "" if link and title: out.append({"title": title, "url": link, "snippet": snippet}) if len(out) >= max_results: break if not out: try: dr = requests.get("https://api.duckduckgo.com/", params={"q": query, "format": "json", "no_html": "1"}, headers=headers, timeout=20) dd = dr.json() if dd.get("Abstract"): out.append({"title": dd.get("Heading", query), "url": dd.get("AbstractURL", ""), "snippet": dd.get("Abstract", "")}) except Exception: pass return out except Exception as e: training_state["last"] = f"search error: {str(e)[:100]}" return [] def crawl_page(url): try: r = requests.get(url, timeout=20, headers={"User-Agent": "Mozilla/5.0"}, allow_redirects=True) if r.status_code != 200: return "" soup = BeautifulSoup(r.text, "html.parser") for tag in soup(["script", "style", "nav", "footer", "header", "noscript", "form"]): tag.decompose() text = " ".join(soup.get_text(separator=" ", strip=True).split()) return text[:2500] except Exception: return "" def learn_web(topic): global MODEL if not TRAIN_LOCK.acquire(blocking=False): return training_state["running"] = True training_state["last"] = f"searching web: {topic}" try: results = web_search(topic, 5) if not results: raise RuntimeError("no search results") lines = [] for i, res in enumerate(results): training_state["last"] = f"crawling {i+1}/{len(results)}: {res['title'][:40]}" page = crawl_page(res["url"]) lines.append(f"Topic: {topic}\nSource: {res['title']} ({res['url']})\nSnippet: {res['snippet']}\nContent: {page}") time.sleep(1) corpus = "\n\n".join(lines) if not corpus.strip(): raise RuntimeError("empty web corpus") toks = TOKENIZER(corpus)["input_ids"] block = 64 chunks = [toks[i:i+block] for i in range(0, max(1, len(toks) - block + 1), block)] chunks = [c for c in chunks if len(c) == block] if not chunks: raise RuntimeError("not enough web text") ds = Dataset.from_dict({"input_ids": chunks, "labels": [c[:] for c in chunks]}) out_dir = os.path.join(WORK_DIR, "agron-web") os.makedirs(out_dir, exist_ok=True) args = TrainingArguments( output_dir=out_dir, per_device_train_batch_size=2, max_steps=60, learning_rate=1e-4, optim="adamw_torch", fp16=True if torch.cuda.is_available() else False, logging_steps=20, save_strategy="no", report_to=[], remove_unused_columns=True, ) collator = DataCollatorForLanguageModeling(tokenizer=TOKENIZER, mlm=False) trainer = Trainer(model=MODEL, args=args, train_dataset=ds, data_collator=collator, processing_class=TOKENIZER) training_state["last"] = f"training on web content for '{topic}'..." trainer.train() save_file({k: v.detach().cpu().contiguous() for k, v in MODEL.state_dict().items()}, os.path.join(out_dir, "model.safetensors")) import shutil shutil.copyfile(os.path.join(out_dir, "model.safetensors"), os.path.join(ROOT, "model.safetensors")) training_state["last"] = "pushing web-learned model..." ok, detail = git_auto_push( f"Agron web learning: {topic}", ["model.safetensors", "memory/interactions.jsonl"], ) learned_topics.append(f"web: {topic}") persist_topics() training_state["last"] = f"web learned · {detail}: {topic}" except Exception as e: training_state["last"] = f"web error: {str(e)[:150]}" finally: training_state["running"] = False TRAIN_LOCK.release() EVOLVE_STATE = {"running": False, "cycles": 0, "last_topic": "none"} def self_update(): """Pull the latest code from the Hugging Face repo (origin).""" try: subprocess.run(["git", "-C", ROOT, "fetch", "origin"], check=True, timeout=120) branch = subprocess.run(["git", "-C", ROOT, "rev-parse", "--abbrev-ref", "HEAD"], capture_output=True, text=True).stdout.strip() or "main" local = subprocess.run(["git", "-C", ROOT, "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip() remote = subprocess.run(["git", "-C", ROOT, "rev-parse", f"origin/{branch}"], capture_output=True, text=True).stdout.strip() if not remote: return f"could not resolve origin/{branch}" if local != remote: subprocess.run(["git", "-C", ROOT, "pull", "--ff-only", "origin", branch], check=True, timeout=300) return f"updated from Hugging Face ({local[:7]} -> {remote[:7]}); restart app to apply new code" return "already up to date with Hugging Face" except Exception as e: return f"update error: {str(e)[:120]}" EVOLVE_TOPICS = [ "artificial intelligence self improvement", "chain of thought reasoning", "tree of thought search", "self-consistency decoding", "recursive self improvement algorithms", "react agent pattern", "retrieval augmented generation", "algorithm design and complexity", "machine learning transformers", "python programming best practices", "data structures algorithms", "neural network training techniques", "encyclopedia general knowledge", ] def _self_evolve_worker(): EVOLVE_STATE["running"] = True i = 0 try: while EVOLVE_STATE["running"]: # evolve from conversations if enough if len(interactions) >= 4: learn_from_interactions() # learn from web on rotating topics if EVOLVE_STATE["running"]: topic = EVOLVE_TOPICS[i % len(EVOLVE_TOPICS)] i += 1 EVOLVE_STATE["last_topic"] = topic learn_web(topic) EVOLVE_STATE["cycles"] += 1 # downtime before next cycle for _ in range(300): if not EVOLVE_STATE["running"]: break time.sleep(1) except Exception as e: training_state["last"] = f"evolve error: {str(e)[:120]}" finally: EVOLVE_STATE["running"] = False def start_self_evolution(): if EVOLVE_STATE["running"]: return "self-evolution already running" threading.Thread(target=_self_evolve_worker, daemon=True).start() return "self-evolution started" def stop_self_evolution(): EVOLVE_STATE["running"] = False return "stopping self-evolution..." # ---------- Tools & dashboard ---------- TOOLS_DIR = os.path.join(MEMORY_DIR, "tools") os.makedirs(TOOLS_DIR, exist_ok=True) TOOLS_FILE = os.path.join(TOOLS_DIR, "registry.json") TOOL_REGISTRY = {} # name -> {"description":..., "code":..., "builtin": bool} def builtin_web_search(text): res = web_search(text, 5) if not res: return "No results found." return "\n".join(f"{i+1}. {r['title']} - {r.get('snippet','')} ({r['url']})" for i, r in enumerate(res)) def builtin_define(text): d = fetch_definition(text.strip()) return d if d else f"No definition found for {text}." def builtin_calc(text): try: allowed = {"__builtins__": {}} val = eval(text, allowed, {}) return str(val) except Exception as e: return f"calc error: {e}" def builtin_time(text=""): return time.strftime("%Y-%m-%d %H:%M:%S") def register_builtin_tools(): specs = [ ("web_search", "Search the web for a query.", None, builtin_web_search), ("define_word", "Define an English word.", None, builtin_define), ("calculate", "Evaluate a math expression.", None, builtin_calc), ("current_time", "Current date/time.", None, builtin_time), ] for name, desc, code, fn in specs: TOOL_REGISTRY[name] = {"description": desc, "code": None, "builtin": True, "fn": fn} def load_tools(): if os.path.exists(TOOLS_FILE): try: with open(TOOLS_FILE, "r") as f: data = json.load(f) for t in data: name = t["name"] code = t["code"] g = {} try: exec(code, g) TOOL_REGISTRY[name] = {"description": t.get("description", ""), "code": code, "builtin": False, "fn": g.get("run")} except Exception: pass except Exception: pass def save_tools(): data = [] for name, info in TOOL_REGISTRY.items(): if info.get("builtin"): continue data.append({"name": name, "description": info.get("description", ""), "code": info.get("code", "")}) with open(TOOLS_FILE, "w") as f: json.dump(data, f, indent=2) def execute_tool(name, text): info = TOOL_REGISTRY.get(name) if not info: return f"Unknown tool: {name}" try: fn = info.get("fn") if fn is None and info.get("code"): g = {} exec(info["code"], g) fn = g.get("run") info["fn"] = fn if fn is None: return "Tool has no runnable function." out = fn(text) return str(out) except Exception as e: return f"tool error: {str(e)[:200]}" def list_tool_names(): return sorted(TOOL_REGISTRY.keys()) def create_tool_from_description(desc): desc = (desc or "").strip() if not desc: return "Provide a tool description first." try: prompt = ( "You are a Python tool generator. Write a Python function named `run(input_text)` that implements this tool: " + desc.strip() + "\nReturn only a single fenced code block with the function. No explanation. The function should return a string. " + "Do not use network or file writes; only pure computation." ) msgs = [{"role": "system", "content": "You write safe, pure-Python tool functions as `run(input_text) returning str`."}, {"role": "user", "content": prompt}] p = VOICE_TOK.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True) if VOICE_TOK else None if VOICE_MODEL is not None and VOICE_TOK is not None: inputs = VOICE_TOK(p, return_tensors="pt").to(VOICE_MODEL.device) with torch.no_grad(): out = VOICE_MODEL.generate(**inputs, max_new_tokens=200, do_sample=False) raw = VOICE_TOK.decode(out[0][inputs.input_ids.shape[1]:], skip_special_tokens=True) else: raw = "" # extract code import re m = re.search(r"```(?:python)?\n(.*?)```", raw, re.S) code = m.group(1).strip() if m else raw.strip() if "def run" not in code: return "Model did not produce a runnable function. Try rephrasing the description." # compile g = {} exec(compile(code, "", "exec"), g) if not callable(g.get("run")): return "Generated code has no run() function." # quick smoke test try: _ = g["run"]("") except Exception: pass # name slug = "".join(c if c.isalnum() else "_" for c in desc.lower())[:24].strip("_") slug = slug or "tool" name = slug i = 1 while name in TOOL_REGISTRY: name = f"{slug}_{i}" i += 1 TOOL_REGISTRY[name] = {"description": desc.strip(), "code": code, "builtin": False, "fn": g["run"]} save_tools() return f"Created tool '{name}'. It is now available in the tool panel." except Exception as e: return f"tool creation error: {str(e)[:200]}" register_builtin_tools() load_tools() def learn_from_scratch_memory(): global MODEL, LAST_TRAIN_TIME if not TRAIN_LOCK.acquire(blocking=False): return training_state["running"] = True training_state["last"] = "building from-scratch corpus..." try: if len(interactions) < 6: raise RuntimeError(f"need at least 6 interactions (have {len(interactions)})") corpus = os.path.join(MEMORY_DIR, "scratch_corpus.txt") lines = [f"User: {r['user']}\nAgron: {r['assistant']}" for r in interactions] with open(corpus, "w", encoding="utf-8") as f: f.write("\n\n".join(lines)) out_dir = os.path.join(WORK_DIR, "agron-from-scratch-memory") training_state["last"] = f"training from scratch on {len(interactions)} interactions..." subprocess.run([ "python3", os.path.join(ROOT, "train_from_scratch.py"), "--size", "current", "--steps", "300", "--dataset", "file", "--file", corpus, "--output", out_dir ], check=True, timeout=600) import shutil shutil.copyfile(os.path.join(out_dir, "final", "model.safetensors"), os.path.join(ROOT, "model.safetensors")) # reload weights into the live model try: st = load_file(os.path.join(ROOT, "model.safetensors")) with IO_LOCK: MODEL.load_state_dict(st) except Exception: pass learned_topics.append("from-scratch memory model") persist_topics() training_state["last"] = "self-updating from scratch..." ok, detail = git_auto_push( f"Agron from-scratch learning ({len(interactions)} interactions)", ["model.safetensors", "memory/scratch_corpus.txt", "memory/learned_topics.json"], ) LAST_TRAIN_TIME = time.time() training_state["last"] = f"from-scratch learned · {detail} ({len(interactions)} interactions)" except Exception as e: training_state["last"] = f"from-scratch error: {str(e)[:200]}" finally: training_state["running"] = False TRAIN_LOCK.release() FACTORY_RESET_PATHS = [ os.path.join(WORK_DIR, n) for n in ( "agron-model", "agron-learned", "agron-web", "agron-dictionary", "agron-from-scratch-memory", "agron-scratch-small", "agron-scratch-current", ) ] def factory_reset(): global MODEL, interactions, learned_topics stop_self_evolution() with IO_LOCK: interactions = [] learned_topics = [] try: os.makedirs(MEMORY_DIR, exist_ok=True) open(INTERACTIONS_FILE, "w").close() json.dump([], open(TOPICS_FILE, "w")) except Exception: pass # clear tools registry try: if os.path.exists(TOOLS_FILE): os.remove(TOOLS_FILE) except Exception: pass for tn in list(TOOL_REGISTRY.keys()): if not TOOL_REGISTRY[tn].get("builtin"): del TOOL_REGISTRY[tn] # clear scratch corpus try: sc = os.path.join(MEMORY_DIR, "scratch_corpus.txt") if os.path.exists(sc): os.remove(sc) except Exception: pass # delete trained model dirs import shutil for d in FACTORY_RESET_PATHS: if os.path.exists(d): shutil.rmtree(d, ignore_errors=True) # reinitialize Agron base model with fresh random weights (no pretrained load) try: from transformers import LlamaConfig, LlamaForCausalLM cfg = LlamaConfig(vocab_size=16384, hidden_size=512, intermediate_size=2048, num_hidden_layers=8, num_attention_heads=4, num_key_value_heads=4, max_position_embeddings=512, bos_token_id=0, eos_token_id=2, pad_token_id=1, tie_word_embeddings=False) MODEL = LlamaForCausalLM(cfg) if torch.cuda.is_available(): MODEL = MODEL.to("cuda") MODEL.eval() except Exception as e: training_state["last"] = f"reset model error: {str(e)[:120]}" globals()["LAST_TRAIN_TIME"] = time.time() training_state["running"] = False training_state["steps"] = 0 training_state["last"] = "factory reset complete - fresh base model initialized" return "factory reset complete" # ---------- status helpers ---------- ACTION_LOG = [] def log_action(msg): """Append a line to the action log and return the rendered log.""" ACTION_LOG.append(f"{time.strftime('%H:%M:%S')} {msg}") del ACTION_LOG[:-60] return "\n".join(ACTION_LOG[-14:]) def status_line(): return ( f"interactions: {len(interactions)} · tools: {len(list_tool_names())} · " f"evolve cycles: {EVOLVE_STATE['cycles']} · training: " f"{'RUNNING' if training_state['running'] else 'idle'} · {training_state['last']}" ) def system_snapshot(): try: n_params = sum(p.numel() for p in MODEL.parameters()) except Exception: n_params = 0 gpu = "" if torch.cuda.is_available(): try: gpu = f" · gpu alloc {torch.cuda.memory_allocated() / 1e9:.2f} GB" except Exception: gpu = "" voice = "on" if VOICE_MODEL is not None else "off" return ( f"params: {n_params / 1e6:.1f}M · device: {DEVICE}{gpu} · voice model: {voice}\n" f"work dir: {WORK_DIR}\n" f"topics: {len(learned_topics)} · auto-push: {'on' if AUTO_PUSH else 'off'}" ) with gr.Blocks(title="Agron Neural Interface") as demo: gr.HTML(PARTICLE_HTML, js_on_load=PARTICLE_JS) with gr.Column(elem_classes="agron-head"): gr.Markdown("

AGRØN NEURAL INTERFACE

") gr.Markdown("
Self-improving multi-agent core · learning from every exchange
") with gr.Column(elem_classes="agron-card"): status_strip = gr.Textbox(value=status_line(), label="TELEMETRY", interactive=False, elem_classes="agron-strip", lines=2) action_log = gr.Textbox(value=log_action("Agron interface online."), label="ACTION LOG", interactive=False, elem_classes="agron-log", lines=4) with gr.Row(): with gr.Column(scale=4, elem_classes="agron-card"): gr.Markdown("

💬 COMMAND CONSOLE

") chatbot = gr.Chatbot(height=480, label="Agron", elem_classes="agron-chat") msg = gr.Textbox(placeholder="Speak to Agron...", label="Input", lines=2) with gr.Row(): send = gr.Button("Transmit", variant="primary") clear = gr.Button("Clear") with gr.Column(scale=2, elem_classes="agron-card"): with gr.Tabs(): with gr.Tab("SYSTEM"): sys_status = gr.Textbox(value=training_state["last"], label="Pipeline", interactive=False, elem_classes="status-box", lines=2) metrics = gr.Textbox(value=system_snapshot(), label="Diagnostics", interactive=False, lines=4) with gr.Tab("TOOLS"): tool_dd = gr.Dropdown(choices=list_tool_names(), value=(list_tool_names()[0] if list_tool_names() else None), label="Select tool") tool_in = gr.Textbox(placeholder="Tool input...", label="Tool input", lines=1) tool_btn = gr.Button("Run tool") tool_out = gr.Textbox(label="Tool output", interactive=False, lines=8) gr.Markdown("### Create tool") tool_desc = gr.Textbox(placeholder="Describe a new tool (pure computation)...", label="New tool description", lines=2) create_btn = gr.Button("Create tool") with gr.Tab("LEARNING"): learn_btn = gr.Button("Train now on memory") scratch_btn = gr.Button("Train from scratch (no fine-tune)") dict_btn = gr.Button("Learn dictionary words") web_topic = gr.Textbox(label="Web topic / search query", placeholder="e.g. transformer architecture", lines=1) web_btn = gr.Button("Search & learn web") with gr.Row(): evolve_btn = gr.Button("Start self-evolution") stop_evolve_btn = gr.Button("Stop") with gr.Row(): update_btn = gr.Button("Self-update") reset_btn = gr.Button("Factory reset") def respond(message, history): message = (message or "").strip() if not message: return history, "" response = generate_response(message, history) history = history + [{"role": "user", "content": message}, {"role": "assistant", "content": response}] return history, "" def do_learn(): if training_state["running"]: return log_action(f"training already running · {training_state['last']}") if len(interactions) < 2: return log_action(f"need at least 2 interactions (have {len(interactions)})") threading.Thread(target=learn_from_interactions, daemon=True).start() return log_action("memory training started in background") def refresh_status(): return status_line(), training_state["last"], system_snapshot() send.click(respond, [msg, chatbot], [chatbot, msg]) msg.submit(respond, [msg, chatbot], [chatbot, msg]) clear.click(lambda: ([], ""), outputs=[chatbot, msg]) def do_dict(): if training_state["running"]: return log_action("training already running") threading.Thread(target=learn_dictionary, daemon=True).start() return log_action("dictionary learning started in background") def do_web(): if training_state["running"]: return log_action("training already running") q = (web_topic.value or "").strip() if not q: return log_action("enter a topic/query first") threading.Thread(target=learn_web, args=(q,), daemon=True).start() return log_action(f"web learning started: {q}") def run_tool_fn(tool_name, tool_input): if not tool_name: return "select a tool first" return execute_tool(tool_name, tool_input or "") def create_tool_fn(desc): return log_action(create_tool_from_description(desc)) def refresh_tools(): choices = list_tool_names() return gr.Dropdown(choices=choices, value=(choices[0] if choices else None)) tool_btn.click(run_tool_fn, [tool_dd, tool_in], tool_out) create_btn.click(create_tool_fn, tool_desc, action_log) learn_btn.click(do_learn, outputs=action_log) def do_scratch(): if training_state["running"]: return log_action("training already running") if len(interactions) < 6: return log_action(f"need at least 6 interactions (have {len(interactions)})") threading.Thread(target=learn_from_scratch_memory, daemon=True).start() return log_action("from-scratch training started in background") scratch_btn.click(do_scratch, outputs=action_log) dict_btn.click(do_dict, outputs=action_log) web_btn.click(do_web, outputs=action_log) def do_evolve(): return log_action(start_self_evolution()) def do_stop_evolve(): return log_action(stop_self_evolution()) def do_update(): return log_action(self_update()) evolve_btn.click(do_evolve, outputs=action_log) stop_evolve_btn.click(do_stop_evolve, outputs=action_log) update_btn.click(do_update, outputs=action_log) reset_btn.click(lambda: log_action(factory_reset()), outputs=action_log) demo.load(refresh_status, outputs=[status_strip, sys_status, metrics]) # periodic status refresh (not true polling inside gradio, use change hook via demo load only) gr.Timer(5).tick(refresh_status, outputs=[status_strip, sys_status, metrics]) if __name__ == "__main__": demo.launch(server_name=SERVER_NAME, server_port=SERVER_PORT, share=SHARE, show_error=True)