Norovox-Tools / agentic_runtime.py
emrevrg's picture
Upload agentic_runtime.py with huggingface_hub
47d1c78 verified
Raw History Blame Contribute Delete
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]))