Download self_learning.py from emrevrg/Norovox-Tools: direct link, hf CLI and curl.
- Browser
- Download file 14.6 kB
-
https://huggingface.co/emrevrg/Norovox-Tools/resolve/main/self_learning.py
- Command line
-
hf download hf://emrevrg/Norovox-Tools/self_learning.py
-
curl -L -o self_learning.py https://huggingface.co/emrevrg/Norovox-Tools/resolve/main/self_learning.py
14.6 kB
| #!/usr/bin/env python3 | |
| """Norovox Nythos — Self-Learning + Skill-Forge Runtime. | |
| İki ayrı yetenek: | |
| 1. **SelfLearner**: Kullanıcı düzeltmelerini KB'ye ekler. | |
| Her konuşmadan sonra "ne öğrendi?" diye sentezler + bilgi tabanına yazar. | |
| Sonraki konuşmalarda otomatik hatırlar. | |
| 2. **SkillForge**: Kullanıcının öğrettiği yeni beceriyi kalıcı hale getirir. | |
| Örnek: "şöyle yaparsan: 1. X, 2. Y, 3. Z" dediğinde, runtime bunu | |
| bir "skill template" olarak kaydeder ve sonra benzer görevlerde | |
| otomatik uygular. | |
| KULLANIM: | |
| from nythos_self_learning import SelfLearningChat | |
| chat = SelfLearningChat( | |
| model_url="http://localhost:8080/v1/chat/completions", | |
| kb_path="./norovox_kb.json", | |
| skills_path="./norovox_skills.json" | |
| ) | |
| # Kullanıcı bir şey öğretir | |
| chat.sor("Şöyle yazıyorum: 'bilgi al(X)' demek 'X hakkında web arası yap + Türkçe özetle'") | |
| # → SkillForge bunu skill olarak kaydeder | |
| # Sonra aynı pattern'i kullanır | |
| chat.sor("bilgi al(Tesla)") | |
| # → otomatik web_ara + özetleme uygular | |
| """ | |
| import json, re, hashlib | |
| from pathlib import Path | |
| from datetime import datetime | |
| from typing import Optional | |
| # ════════════════════════════════════════════════════════════════ | |
| # KNOWLEDGE BASE — sürekli güncellenen bilgi tabanı | |
| # ════════════════════════════════════════════════════════════════ | |
| class KnowledgeBase: | |
| """Basit JSON-tabanlı KB. Konu → fact mapping.""" | |
| def __init__(self, path: str = "./norovox_kb.json"): | |
| self.path = Path(path) | |
| self.data: dict = {"facts": {}, "meta": {"created": str(datetime.now())}} | |
| if self.path.exists(): | |
| try: | |
| self.data = json.loads(self.path.read_text(encoding="utf-8")) | |
| except Exception: | |
| pass | |
| def add(self, topic: str, fact: str, source: str = "user"): | |
| topic_key = topic.lower().strip() | |
| if topic_key not in self.data["facts"]: | |
| self.data["facts"][topic_key] = [] | |
| self.data["facts"][topic_key].append({ | |
| "fact": fact, | |
| "source": source, | |
| "added": str(datetime.now()), | |
| }) | |
| self._save() | |
| def search(self, query: str, limit: int = 5) -> list: | |
| q = query.lower() | |
| matches = [] | |
| for topic, facts in self.data["facts"].items(): | |
| if any(w in topic for w in q.split()): | |
| matches.extend([f["fact"] for f in facts[:limit]]) | |
| return matches[:limit] | |
| def remove(self, topic: str): | |
| self.data["facts"].pop(topic.lower(), None) | |
| self._save() | |
| def _save(self): | |
| self.path.write_text(json.dumps(self.data, ensure_ascii=False, indent=2), | |
| encoding="utf-8") | |
| def size(self) -> int: | |
| return sum(len(v) for v in self.data["facts"].values()) | |
| # ════════════════════════════════════════════════════════════════ | |
| # SKILL FORGE — yeni beceri kazanma | |
| # ════════════════════════════════════════════════════════════════ | |
| class SkillForge: | |
| """Kullanıcının öğrettiği yeni pattern/skill'leri kalıcı yapar. | |
| Skill = pattern + adımlar + örnek input/output | |
| Storage: JSON | |
| Eşleştirme: regex pattern üzerinden | |
| """ | |
| SKILL_TEACHING_PATTERN = re.compile( | |
| r"(?:şöyle|şu şekilde|şunu yap|yaparsan)[:\s]+(.+)", | |
| re.IGNORECASE | re.DOTALL | |
| ) | |
| def __init__(self, path: str = "./norovox_skills.json"): | |
| self.path = Path(path) | |
| self.skills: dict = {} | |
| if self.path.exists(): | |
| try: | |
| self.skills = json.loads(self.path.read_text(encoding="utf-8")) | |
| except Exception: | |
| pass | |
| def _hash(self, name: str) -> str: | |
| return hashlib.sha256(name.encode()).hexdigest()[:8] | |
| def detect_teaching(self, user_input: str) -> Optional[dict]: | |
| """Kullanıcı bir şey öğretiyor mu tespit et.""" | |
| # Pattern 1: "X komutu Y anlamına gelir" | |
| m1 = re.search(r"['\"]?(\w+\([^)]+\))['\"]?\s*(?:komutu|fonksiyonu|demek|anlamına geliyor)\s*['\"]?(.+?)['\"]?\s*$", | |
| user_input, re.DOTALL) | |
| if m1: | |
| return {"name": m1.group(1), "definition": m1.group(2).strip()} | |
| # Pattern 2: "şöyle yap: ...adımlar..." | |
| m2 = self.SKILL_TEACHING_PATTERN.search(user_input) | |
| if m2: | |
| steps_text = m2.group(1).strip() | |
| # Numbered list? | |
| steps = re.findall(r"(?:^|[\n,])\s*\d+[.)]\s*([^\n,]+)", steps_text) | |
| if steps: | |
| first_line = user_input.split(":")[0].strip() | |
| return {"name": first_line[-40:], "steps": steps} | |
| # Pattern 3: "X yapmak için Y kullan" | |
| m3 = re.search(r"(\w+)\s+(?:yapmak için|için)\s+(.+)", user_input, re.IGNORECASE) | |
| if m3 and "kullan" in m3.group(2).lower(): | |
| return {"name": m3.group(1), "method": m3.group(2)} | |
| return None | |
| def forge(self, skill_info: dict): | |
| """Yeni skill kaydet.""" | |
| name = skill_info.get("name", "untitled")[:50] | |
| skill_id = self._hash(name) | |
| self.skills[skill_id] = { | |
| "name": name, | |
| "info": skill_info, | |
| "created": str(datetime.now()), | |
| "use_count": 0, | |
| } | |
| self._save() | |
| return skill_id | |
| def match(self, user_input: str) -> Optional[dict]: | |
| """Mevcut bir skill'in input'la eşleşip eşleşmediği.""" | |
| for sid, skill in self.skills.items(): | |
| name = skill["name"].lower() | |
| # Pattern: skill name + arg pattern | |
| base_name = re.sub(r"\([^)]*\)", "", name).strip() | |
| if base_name and base_name in user_input.lower(): | |
| skill["use_count"] += 1 | |
| self._save() | |
| return skill | |
| return None | |
| def list_skills(self) -> list: | |
| return [(s["name"], s["use_count"]) for s in self.skills.values()] | |
| def _save(self): | |
| self.path.write_text(json.dumps(self.skills, ensure_ascii=False, indent=2), | |
| encoding="utf-8") | |
| # ════════════════════════════════════════════════════════════════ | |
| # SELF-LEARNER — konuşmadan sonuç çıkartma | |
| # ════════════════════════════════════════════════════════════════ | |
| class SelfLearner: | |
| """Konuşmayı analiz eder, KB'ye eklenebilecek fact'leri çıkartır.""" | |
| FACT_PATTERNS = [ | |
| (re.compile(r"(\w[\w\s]+?)\s+(?:şudur|şöyledir|şuna eşittir|=)\s+(.+?)(?:\.|$)", re.I), | |
| lambda m: (m.group(1).strip(), m.group(2).strip())), | |
| (re.compile(r"benim adım\s+(.+?)(?:\.|$)", re.I), | |
| lambda m: ("kullanıcı adı", m.group(1).strip())), | |
| (re.compile(r"(?:tercih ediyorum|seviyorum):\s*(.+?)(?:\.|$)", re.I), | |
| lambda m: ("kullanıcı tercihi", m.group(1).strip())), | |
| (re.compile(r"(\w+)\s+(?:doğum tarihi|kuruluş tarihi)\s+(.+?)(?:\.|$)", re.I), | |
| lambda m: (m.group(1).strip() + " tarihi", m.group(2).strip())), | |
| ] | |
| def __init__(self, kb: KnowledgeBase): | |
| self.kb = kb | |
| def extract_facts(self, user_input: str, assistant_response: str) -> list: | |
| facts = [] | |
| for pattern, mapper in self.FACT_PATTERNS: | |
| for m in pattern.finditer(user_input): | |
| t, f = mapper(m) | |
| if t and f and len(f) < 200: | |
| facts.append((t, f)) | |
| return facts | |
| def learn(self, user_input: str, assistant_response: str) -> int: | |
| """Konuşmadan fact çıkar + KB'ye ekle.""" | |
| facts = self.extract_facts(user_input, assistant_response) | |
| for topic, fact in facts: | |
| self.kb.add(topic, fact, source="self_learner") | |
| return len(facts) | |
| # ════════════════════════════════════════════════════════════════ | |
| # SELF-LEARNING CHAT — tüm modülleri birleştir | |
| # ════════════════════════════════════════════════════════════════ | |
| class SelfLearningChat: | |
| """Norovox Nythos için kendi kendine öğrenen sohbet runtime'ı.""" | |
| SYSTEM_BASE = ( | |
| "Sen Norovox Nythos'sun — kendi kendine öğrenen Türkçe AI.\n\n" | |
| "Sahip olduğun yetenekler:\n" | |
| "- <think>...</think> derin düşünce blokları\n" | |
| "- Yerleşik araç çağrısı: <tool_call> bloğu\n" | |
| "- Kullanıcının öğrettiği yeni beceri ve bilgileri kalıcı hatırlama\n" | |
| "- Bilgi tabanından (KB) otomatik bağlam getirme\n\n" | |
| "# ÖĞRENDIKLERIM (KB'den)\n__KB__\n\n" | |
| "# KAZANDIĞIM BECERILER\n__SKILLS__\n\n" | |
| "Konuşma sırasında kullanıcı yeni bir şey öğretirse runtime otomatik kaydeder. " | |
| "Sen Türkçe yanıt ver, derinden düşün." | |
| ) | |
| def __init__(self, | |
| model_url: str = "http://localhost:8080/v1/chat/completions", | |
| kb_path: str = "./norovox_kb.json", | |
| skills_path: str = "./norovox_skills.json"): | |
| self.model_url = model_url | |
| self.kb = KnowledgeBase(kb_path) | |
| self.skills = SkillForge(skills_path) | |
| self.learner = SelfLearner(self.kb) | |
| def _build_system(self, query: str) -> str: | |
| kb_hits = self.kb.search(query, limit=3) | |
| kb_ctx = "\n".join(f"- {h}" for h in kb_hits) if kb_hits else "(henüz yok)" | |
| skill_list = self.skills.list_skills()[:3] | |
| skill_ctx = "\n".join(f"- {n} (kullanım: {u})" for n, u in skill_list) if skill_list else "(henüz yok)" | |
| return self.SYSTEM_BASE.replace("__KB__", kb_ctx).replace("__SKILLS__", skill_ctx) | |
| def sor(self, user_input: str) -> dict: | |
| """Tam pipeline: skill-match → LLM → learn.""" | |
| result = { | |
| "input": user_input, | |
| "matched_skill": None, | |
| "new_skill": None, | |
| "learned_facts": 0, | |
| "response": "", | |
| } | |
| # 1) Skill match — daha önce öğretilen pattern var mı? | |
| matched = self.skills.match(user_input) | |
| if matched: | |
| result["matched_skill"] = matched["name"] | |
| # 2) Teaching detect — yeni skill öğretiliyor mu? | |
| teaching = self.skills.detect_teaching(user_input) | |
| if teaching: | |
| skill_id = self.skills.forge(teaching) | |
| result["new_skill"] = teaching.get("name", skill_id) | |
| # 3) LLM çağrısı (varsa) | |
| system_msg = self._build_system(user_input) | |
| # Burada gerçek LLM çağrısı yapılır — placeholder offline mod: | |
| try: | |
| import urllib.request | |
| body = json.dumps({ | |
| "messages": [ | |
| {"role": "system", "content": system_msg}, | |
| {"role": "user", "content": user_input} | |
| ], | |
| "max_tokens": 600, | |
| "temperature": 0.5 | |
| }).encode() | |
| req = urllib.request.Request(self.model_url, data=body, | |
| headers={"Content-Type": "application/json"}) | |
| with urllib.request.urlopen(req, timeout=60) as r: | |
| data = json.loads(r.read()) | |
| result["response"] = data["choices"][0]["message"]["content"] | |
| except Exception as e: | |
| result["response"] = f"[offline mod — model bağlantısı yok: {e}]" | |
| # 4) Konuşmadan öğren | |
| learned = self.learner.learn(user_input, result["response"]) | |
| result["learned_facts"] = learned | |
| return result | |
| def stats(self) -> dict: | |
| return { | |
| "kb_facts": self.kb.size(), | |
| "skills_count": len(self.skills.skills), | |
| "skills_total_use": sum(s["use_count"] for s in self.skills.skills.values()), | |
| } | |
| # ════════════════════════════════════════════════════════════════ | |
| # CLI TEST | |
| # ════════════════════════════════════════════════════════════════ | |
| if __name__ == "__main__": | |
| import sys, tempfile | |
| print("== Norovox Nythos Self-Learning + Skill-Forge Test ==\n") | |
| # Geçici dosyalarla test | |
| with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f: | |
| kb_path = f.name | |
| with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f: | |
| skills_path = f.name | |
| chat = SelfLearningChat(kb_path=kb_path, skills_path=skills_path) | |
| # 1) Yeni skill öğret | |
| print("[1] Kullanıcı yeni skill öğretiyor...") | |
| r1 = chat.sor("bilgi_al(X) komutu demek 'X hakkında web ara + Türkçe özetle' anlamına geliyor") | |
| print(f" New skill kaydedildi: {r1['new_skill']}") | |
| print(f" Stats: {chat.stats()}\n") | |
| # 2) Aynı pattern'i kullan | |
| print("[2] Aynı pattern'le kullanım...") | |
| r2 = chat.sor("bilgi_al(Tesla şirketi)") | |
| print(f" Matched skill: {r2['matched_skill']}") | |
| print(f" Stats: {chat.stats()}\n") | |
| # 3) Fact öğret | |
| print("[3] Kullanıcı bir fact söylüyor...") | |
| r3 = chat.sor("benim adım Emre. Tercih ediyorum: kısa yanıtlar.") | |
| print(f" KB'ye eklenen fact sayısı: {r3['learned_facts']}") | |
| print(f" Stats: {chat.stats()}\n") | |
| # 4) KB hatırlama | |
| print("[4] KB'den hatırla...") | |
| facts = chat.kb.search("kullanıcı") | |
| for f in facts: | |
| print(f" → {f}") | |
| print("\n[OK] Self-Learning + Skill-Forge çalışıyor") | |
| # Cleanup | |
| Path(kb_path).unlink(missing_ok=True) | |
| Path(skills_path).unlink(missing_ok=True) | |