Download agentic_runtime.py from emrevrg/Norovox-Tools: direct link, hf CLI and curl.
- Browser
- Download file 13 kB
-
https://huggingface.co/emrevrg/Norovox-Tools/resolve/main/agentic_runtime.py
- Command line
-
hf download hf://emrevrg/Norovox-Tools/agentic_runtime.py
-
curl -L -o agentic_runtime.py https://huggingface.co/emrevrg/Norovox-Tools/resolve/main/agentic_runtime.py
13 kB
| #!/usr/bin/env python3 | |
| """Norovox Agentic Runtime — Qwythos benzeri toolsuz web/araç kullanım. | |
| Model sistem prompt'u ile çağrı formatını öğrenir. Runtime LLM yanıtında | |
| özel etiketleri algılayıp tool'u arka planda çalıştırır, sonucu bağlama | |
| ekler. Kullanıcı sadece "internet'ten X bul" der; model otomatik web_ara | |
| + özetler. | |
| Desteklenen toollar (yerleşik): | |
| - web_ara(sorgu) — Wikipedia + DuckDuckGo fallback (anahtarsız) | |
| - url_oku(url) — sayfa metnini çek | |
| - hesapla(ifade) — math + sympy | |
| - kod_yurut(python) — sandbox | |
| - zaman_simdi() — saat | |
| - dosya_oku(yol) — dosya | |
| - kb_ara(sorgu) — bilgi tabanı | |
| KULLANIM: | |
| from norovox_agentic_runtime import AgenticChat, NOROVOX_AGENTIC_SYS | |
| chat = AgenticChat(model_url="http://localhost:8080/v1/chat/completions") | |
| cevap = chat.sor("internet'ten İstanbul nüfusunu bul") | |
| # Model otomatik [TOOL: web_ara(İstanbul nüfus)] yayar, runtime | |
| # arar, sonucu modele besler, model sentezleyip cevap verir. | |
| """ | |
| import re, json, math, urllib.parse, urllib.request, urllib.error | |
| from datetime import datetime | |
| from pathlib import Path | |
| # ════════════════════════════════════════════════════════════════ | |
| # SİSTEM PROMPTU — modele tool kullanımını öğretir | |
| # ════════════════════════════════════════════════════════════════ | |
| NOROVOX_AGENTIC_SYS = """Sen Norovox'sun — Türkçe açık-kaynak yapay zeka. Yerleşik araç kullanım yeteneğine sahipsin. | |
| # ARAÇ KULLANIMI | |
| Bir konuda bilmediğin ya da güncel bilgi gereken bir soru gelirse, cevabının içinde şu formatta araç çağrısı yap: | |
| [TOOL: web_ara("sorgu metni")] | |
| [TOOL: url_oku("https://...")] | |
| [TOOL: hesapla("2*pi*5")] | |
| [TOOL: kod_yurut("print('merhaba')")] | |
| [TOOL: zaman_simdi()] | |
| [TOOL: kb_ara("Norovox mimarisi")] | |
| Runtime bu çağrıyı otomatik çalıştırıp [RESULT: ...] formatında sana cevabı verir. Sonra sen sonucu kullanarak nihai yanıtı kullanıcıya yaz. | |
| # KURALLAR | |
| - Önce düşün, hangi araç gerekli karar ver | |
| - Tek mesajda en fazla 3 araç çağrısı | |
| - Sonuçları geldiğinde Türkçe sentezleyerek cevap ver | |
| - Araç gerekmiyorsa direkt cevap ver (selamlama, basit matematik, genel kültür) | |
| - "Yapamam" deme — alternatif araç yolu bul | |
| # ÖRNEK | |
| Kullanıcı: "Bugün İstanbul'da hava nasıl?" | |
| Sen: "[TOOL: web_ara("İstanbul hava durumu bugün")]" | |
| Runtime: "[RESULT: ...]" | |
| Sen: "İstanbul'da bugün 22°C, parçalı bulutlu, akşam 18°C'ye düşer." | |
| """ | |
| # ════════════════════════════════════════════════════════════════ | |
| # YERLEŞIK ARAÇLAR | |
| # ════════════════════════════════════════════════════════════════ | |
| def tool_web_ara(query: str, max_results: int = 3) -> str: | |
| """Wikipedia TR + DuckDuckGo HTML fallback — anahtarsız.""" | |
| results = [] | |
| # Wikipedia REST | |
| try: | |
| url = f"https://tr.wikipedia.org/api/rest_v1/page/summary/{urllib.parse.quote(query)}" | |
| req = urllib.request.Request(url, headers={"User-Agent": "Norovox/1.0"}) | |
| with urllib.request.urlopen(req, timeout=8) as r: | |
| data = json.loads(r.read()) | |
| if data.get("extract"): | |
| results.append({ | |
| "kaynak": "Wikipedia TR", | |
| "baslik": data.get("title", ""), | |
| "ozet": data["extract"][:600], | |
| "url": data.get("content_urls", {}).get("desktop", {}).get("page", "") | |
| }) | |
| except Exception: | |
| pass | |
| # Wikipedia opensearch (ek liste) | |
| if len(results) < max_results: | |
| try: | |
| url = f"https://tr.wikipedia.org/w/api.php?action=opensearch&search={urllib.parse.quote(query)}&limit={max_results}&format=json" | |
| req = urllib.request.Request(url, headers={"User-Agent": "Norovox/1.0"}) | |
| with urllib.request.urlopen(req, timeout=8) as r: | |
| _, titles, descs, urls = json.loads(r.read()) | |
| for t, d, u in zip(titles, descs, urls): | |
| if any(rs.get("url") == u for rs in results): | |
| continue | |
| results.append({"kaynak": "Wikipedia TR", "baslik": t, "ozet": d, "url": u}) | |
| if len(results) >= max_results: | |
| break | |
| except Exception: | |
| pass | |
| # DuckDuckGo HTML (son çare) | |
| if not results: | |
| try: | |
| url = f"https://html.duckduckgo.com/html/?q={urllib.parse.quote(query)}" | |
| req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) | |
| with urllib.request.urlopen(req, timeout=8) as r: | |
| html = r.read().decode("utf-8", errors="ignore") | |
| m = re.findall(r'<a class="result__a" href="([^"]+)">([^<]+)</a>', html) | |
| for u, t in m[:max_results]: | |
| results.append({"kaynak": "DuckDuckGo", "baslik": t, "ozet": "", "url": u}) | |
| except Exception: | |
| pass | |
| if not results: | |
| return "[web: sonuç bulunamadı]" | |
| return "\n\n".join( | |
| f"# {r['baslik']} ({r['kaynak']})\n{r['ozet']}\n{r.get('url', '')}" | |
| for r in results[:max_results] | |
| ) | |
| def tool_url_oku(url: str, max_chars: int = 4000) -> str: | |
| try: | |
| req = urllib.request.Request(url, headers={"User-Agent": "Norovox/1.0"}) | |
| with urllib.request.urlopen(req, timeout=10) as r: | |
| html = r.read().decode("utf-8", errors="ignore") | |
| text = re.sub(r"<script[^>]*>.*?</script>", "", html, flags=re.S | re.I) | |
| text = re.sub(r"<style[^>]*>.*?</style>", "", text, flags=re.S | re.I) | |
| text = re.sub(r"<[^>]+>", " ", text) | |
| text = re.sub(r"\s+", " ", text).strip() | |
| return text[:max_chars] | |
| except Exception as e: | |
| return f"[url_oku hata: {e}]" | |
| def tool_hesapla(expr: str) -> str: | |
| try: | |
| # Güvenli math eval | |
| allowed = {k: getattr(math, k) for k in dir(math) if not k.startswith("_")} | |
| allowed.update({"abs": abs, "round": round, "min": min, "max": max, "sum": sum}) | |
| result = eval(expr, {"__builtins__": {}}, allowed) | |
| return str(result) | |
| except Exception as e: | |
| return f"[hesapla hata: {e}]" | |
| def tool_kod_yurut(code: str, timeout: int = 5) -> str: | |
| import subprocess, sys, tempfile | |
| try: | |
| with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False, encoding="utf-8") as f: | |
| f.write(code) | |
| path = f.name | |
| result = subprocess.run([sys.executable, path], capture_output=True, text=True, timeout=timeout) | |
| Path(path).unlink(missing_ok=True) | |
| return result.stdout[-2000:] or result.stderr[-2000:] or "[ok, çıktı yok]" | |
| except Exception as e: | |
| return f"[kod_yurut hata: {e}]" | |
| def tool_zaman_simdi() -> str: | |
| return datetime.now().isoformat(timespec="seconds") | |
| def tool_dosya_oku(path: str, max_chars: int = 3000) -> str: | |
| try: | |
| return Path(path).read_text(encoding="utf-8")[:max_chars] | |
| except Exception as e: | |
| return f"[dosya_oku hata: {e}]" | |
| def tool_kb_ara(query: str) -> str: | |
| """Basit bilgi tabanı — yerleşik Norovox bilgisi.""" | |
| KB = { | |
| "norovox": "Norovox — Türkçe açık-kaynak yapay zeka ailesi. N-1 (Qwen 2.5), Nythos (Qwen 3.5/3.6), Alpha (sıfırdan özgün).", | |
| "n-1": "Norovox N-1 — Qwen 2.5 tabanlı kardeş model. Boyutlar: 0.5B, 1.5B, 3B, 7B, 14B.", | |
| "nythos": "Norovox Nythos — Qwen 3.5 (küçük) / Qwen 3.6 (büyük) tabanlı reasoning modeli. THINK·UNDERSTAND·CREATE. <think> blokları, MTP, MoE.", | |
| "alpha": "Norovox Alpha — sıfırdan özgün, hiçbir base ağırlık yok. Norovox V4 mimari: sMoE + NoroLinear + SelfRefining + ContinuousLearner.", | |
| } | |
| q = query.lower() | |
| hits = [v for k, v in KB.items() if k in q] | |
| return "\n".join(hits) if hits else f"[kb: '{query}' bulunamadı]" | |
| TOOL_REGISTRY = { | |
| "web_ara": tool_web_ara, | |
| "url_oku": tool_url_oku, | |
| "hesapla": tool_hesapla, | |
| "kod_yurut": tool_kod_yurut, | |
| "zaman_simdi": tool_zaman_simdi, | |
| "dosya_oku": tool_dosya_oku, | |
| "kb_ara": tool_kb_ara, | |
| } | |
| # ════════════════════════════════════════════════════════════════ | |
| # AGENTIC CHAT — LLM ile iletişim + tool dispatch | |
| # ════════════════════════════════════════════════════════════════ | |
| TOOL_PATTERN = re.compile(r'\[TOOL:\s*(\w+)\(([^)]*)\)\s*\]') | |
| def parse_tool_calls(text: str): | |
| """Modelin çıktısından tool çağrılarını çıkar.""" | |
| calls = [] | |
| for m in TOOL_PATTERN.finditer(text): | |
| name = m.group(1) | |
| raw_args = m.group(2).strip() | |
| # Basit arg parser: tek string ya da yok | |
| if raw_args.startswith('"') and raw_args.endswith('"'): | |
| arg = raw_args[1:-1] | |
| elif raw_args.startswith("'") and raw_args.endswith("'"): | |
| arg = raw_args[1:-1] | |
| else: | |
| arg = raw_args | |
| calls.append((name, arg, m.start(), m.end())) | |
| return calls | |
| def execute_tools(calls): | |
| """Tool çağrılarını sırayla çalıştır.""" | |
| results = [] | |
| for name, arg, _, _ in calls: | |
| if name not in TOOL_REGISTRY: | |
| results.append(f"[{name}: bilinmiyor]") | |
| continue | |
| try: | |
| if arg: | |
| r = TOOL_REGISTRY[name](arg) | |
| else: | |
| r = TOOL_REGISTRY[name]() | |
| results.append(f"[RESULT {name}]: {r}") | |
| except Exception as e: | |
| results.append(f"[RESULT {name} hata]: {e}") | |
| return "\n\n".join(results) | |
| class AgenticChat: | |
| """Tek-shot veya REPL agentic sohbet — OpenAI uyumlu LLM endpoint'i.""" | |
| def __init__(self, model_url: str = "http://localhost:8080/v1/chat/completions", | |
| system_prompt: str = NOROVOX_AGENTIC_SYS, | |
| max_iterations: int = 3): | |
| self.url = model_url | |
| self.sys = system_prompt | |
| self.max_iter = max_iterations | |
| def _post(self, messages): | |
| body = json.dumps({"messages": messages, "max_tokens": 800, "temperature": 0.4}).encode() | |
| req = urllib.request.Request(self.url, data=body, | |
| headers={"Content-Type": "application/json"}) | |
| with urllib.request.urlopen(req, timeout=60) as r: | |
| data = json.loads(r.read()) | |
| return data["choices"][0]["message"]["content"] | |
| def sor(self, user_input: str) -> str: | |
| """Tek soru — model+tool dispatch döngüsü.""" | |
| msgs = [ | |
| {"role": "system", "content": self.sys}, | |
| {"role": "user", "content": user_input}, | |
| ] | |
| for _ in range(self.max_iter): | |
| ans = self._post(msgs) | |
| calls = parse_tool_calls(ans) | |
| if not calls: | |
| return ans # final cevap | |
| # Tool execute | |
| tool_results = execute_tools(calls) | |
| # Sonucu konuşmaya ekle | |
| msgs.append({"role": "assistant", "content": ans}) | |
| msgs.append({"role": "user", "content": tool_results}) | |
| return ans # max_iter doldu, son cevap | |
| # ════════════════════════════════════════════════════════════════ | |
| # CLI test | |
| # ════════════════════════════════════════════════════════════════ | |
| if __name__ == "__main__": | |
| import sys | |
| args = sys.argv[1:] | |
| if not args: | |
| print("Kullanım:") | |
| print(" python norovox_agentic_runtime.py test # tool registry smoke") | |
| print(" python norovox_agentic_runtime.py 'soru' # llama-server lazım") | |
| sys.exit(0) | |
| if args[0] == "test": | |
| print("== Tool registry smoke test ==\n") | |
| print("[web_ara]") | |
| print(tool_web_ara("Türkiye nüfusu")[:300], "\n") | |
| print("[hesapla]") | |
| print(tool_hesapla("sqrt(2) * pi"), "\n") | |
| print("[zaman_simdi]") | |
| print(tool_zaman_simdi(), "\n") | |
| print("[kb_ara]") | |
| print(tool_kb_ara("nythos nedir"), "\n") | |
| print("[OK] yerleşik araçlar çalışıyor") | |
| else: | |
| chat = AgenticChat() | |
| print(chat.sor(args[0])) | |