Spaces:
Running
Running
| # ================================================================ | |
| # URAL AI - SELF EVOLUTION ENGINE v2.0 | |
| # File: ural_self_evolution.py | |
| # CACH DUNG: import ural_self_evolution (cuoi cung trong app.py) | |
| # ================================================================ | |
| import os, sys, re, json, time, random, datetime, threading, traceback, subprocess | |
| from typing import Optional, Tuple | |
| # ── CONFIG ────────────────────────────────────────────────────── | |
| EVOLUTION_LOG_FILE = "ural_evolution_log.json" | |
| MODULES_DIR = "ural_modules" | |
| CODE_BRAIN = { | |
| "platform": "github", | |
| "base_url": "https://models.inference.ai.azure.com", | |
| "models": ["DeepSeek-R1", "gpt-4o", "DeepSeek-V3-0324"], | |
| "locked": True, | |
| } | |
| # ── FREE CODE BRAIN: Mistral → Gemini → Cerebras ──────────────────────────── | |
| # Khi GitHub (luồng 1) hết quota → tự động chạy theo thứ tự dưới | |
| # LUỒNG 2 = Mistral Codestral (code chuyên sâu, 1 tỷ tokens/tháng FREE) | |
| # LUỒNG 3 = Gemini (1500 req/ngày FREE) | |
| # LUỒNG 4 = Cerebras (backup cuối, nhanh) | |
| # Groq đã bỏ khỏi đây (bn không dùng Groq) | |
| CODE_BRAIN_FREE_FALLBACKS = [ | |
| # LUỒNG 2: MISTRAL — Codestral chuyên code, 1 tỷ tokens/tháng FREE | |
| # Dùng cho: auto-fix lỗi phức tạp, auto-upgrade module, auto-evaluate trước khi dùng | |
| { | |
| "platform": "mistral", | |
| "base_url": "https://api.mistral.ai/v1", | |
| "env_prefix": "MISTRAL_KEY", # Điền MISTRAL_KEY1, MISTRAL_KEY2... | |
| "models": [ | |
| "codestral-latest", # Chuyên code — ưu tiên nhất | |
| "mistral-large-latest", # Reasoning tốt — fallback | |
| "mistral-small-latest", # Backup nhẹ | |
| ], | |
| "daily_limit": 99999, # 1 tỷ tokens/tháng | |
| "signup_url": "https://console.mistral.ai", | |
| "note": "FREE, không thẻ, 1 tỷ tokens/tháng - tốt nhất cho code", | |
| }, | |
| # LUỒNG 3: GEMINI — 1500 req/ngày FREE (Google AI Studio) | |
| { | |
| "platform": "gemini", | |
| "base_url": "https://generativelanguage.googleapis.com/v1beta/openai", | |
| "env_prefix": "GEMINI_API_KEY", # Điền GEMINI_API_KEY1, GEMINI_API_KEY2... | |
| "models": [ | |
| "gemini-2.5-flash", # Mạnh nhất, ưu tiên | |
| "gemini-2.0-flash", # Nhanh, ổn định | |
| "gemini-1.5-flash", # Backup cuối | |
| ], | |
| "daily_limit": 1500, | |
| "signup_url": "https://aistudio.google.com/apikey", | |
| "note": "FREE, không thẻ, dùng nhiều Google account lấy nhiều key", | |
| }, | |
| # LUỒNG 4: CEREBRAS — backup nhanh khi Mistral + Gemini hết | |
| # ⚠️ (2026-07-17) Cerebras free tier ĐÓNG CỬA 17/08/2026 — bắt buộc thẻ | |
| # thanh toán từ ngày đó. Giữ khai báo để hunt logic tự phát hiện khi | |
| # Cerebras ngừng hoạt động (daily_limit/note cập nhật để phản ánh đúng | |
| # tình trạng); KHÔNG xóa hẳn để tránh phá cấu trúc _FLOWS mà module | |
| # khác có thể đang dò theo index/platform name. | |
| { | |
| "platform": "cerebras", | |
| "base_url": "https://api.cerebras.ai/v1", | |
| "env_prefix": "CEREBRAS_KEY", | |
| "models": [ | |
| "llama-3.3-70b", | |
| "llama3.1-70b", | |
| ], | |
| "daily_limit": 10000, | |
| "signup_url": "https://cloud.cerebras.ai", | |
| "note": "⚠️ Free tier đóng cửa 17/08/2026 (cần thẻ) — dùng yui_model_router làm nguồn chính từ nay", | |
| }, | |
| ] | |
| ADMIN_CHAT_BRAIN = {"platform": "cohere", "models": ["command-r-plus"], "locked": True} | |
| GUEST_BRAIN = { | |
| "platform": "cloudflare", | |
| "current_model": "@cf/meta/llama-3.1-8b-instruct", | |
| "candidates": [ | |
| "@cf/meta/llama-4-scout-17b-16e-instruct", | |
| "@cf/mistralai/mistral-small-3.1-24b-instruct", | |
| "@cf/google/gemma-3-12b-it", | |
| "@cf/meta/llama-3.3-70b-instruct-fp8-fast", | |
| ], | |
| "locked": False, | |
| } | |
| BACKUP_BRAIN = {"platform": "openrouter", "live_models": [], "locked": False} | |
| FINAL_SAFETY = {"platform": "cohere", "model": "command-r", "locked": True} | |
| CONFIG = { | |
| "max_upgrades_per_day": 1, | |
| "hunt_cooldown": 300, | |
| "benchmark_min_score": 6, | |
| "deepfix_hourly_limit": 3, | |
| "lab_timeout": 10, | |
| "lab_max_output": 2048, | |
| "lab_hourly_limit": 20, | |
| } | |
| HF_LIMITS = { | |
| "ram_max_pct": 75, | |
| "ram_danger_pct": 85, | |
| "cpu_max_pct": 70, | |
| "disk_min_free_gb": 1.0, | |
| "module_max_count": 10, | |
| "heavy_imports": [ | |
| "torch","tensorflow","keras","transformers", | |
| "scipy","sklearn","cv2","faiss", | |
| "diffusers","accelerate","bitsandbytes", | |
| ], | |
| } | |
| SAFE_IMPORTS = { | |
| "os","re","json","time","datetime","random", | |
| "math","string","collections","typing","requests", | |
| } | |
| BANNED_PATTERNS = [ | |
| r"exec\s*\(", r"eval\s*\(", r"__import__\s*\(", | |
| r"subprocess", r"os\.system", r"os\.popen", | |
| r"shutil\.rmtree", r"ural_self_evolution", | |
| ] | |
| # ================================================================ | |
| # EVOLUTION LOG | |
| # ================================================================ | |
| class EvolutionLog: | |
| def __init__(self): | |
| self._log = [] | |
| try: | |
| if os.path.exists(EVOLUTION_LOG_FILE): | |
| with open(EVOLUTION_LOG_FILE, "r", encoding="utf-8") as f: | |
| self._log = json.load(f) | |
| except Exception: | |
| self._log = [] | |
| def _save(self): | |
| try: | |
| with open(EVOLUTION_LOG_FILE, "w", encoding="utf-8") as f: | |
| json.dump(self._log[-500:], f, ensure_ascii=False, indent=2) | |
| except Exception: | |
| pass | |
| def record(self, event_type: str, detail: str, success: bool = True): | |
| entry = { | |
| "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| "type": event_type, | |
| "detail": detail[:300], | |
| "success": success, | |
| } | |
| self._log.append(entry) | |
| self._save() | |
| print(f"[Evo/{'OK' if success else 'FAIL'}] {event_type}: {detail[:80]}") | |
| def get_recent(self, n: int = 10) -> list: | |
| return self._log[-n:] | |
| def count_today(self, event_type: str) -> int: | |
| today = datetime.datetime.now().strftime("%Y-%m-%d") | |
| return sum(1 for e in self._log | |
| if e.get("type") == event_type | |
| and e.get("time", "").startswith(today)) | |
| def format_recent(self, n: int = 10) -> str: | |
| recent = self.get_recent(n) | |
| if not recent: | |
| return "Chua co event nao." | |
| lines = [f"Evolution Log ({n} gan nhat):\n"] | |
| for e in reversed(recent): | |
| icon = "v" if e["success"] else "x" | |
| lines.append(f"[{icon}] {e['time']} [{e['type']}] {e['detail'][:80]}") | |
| return "\n".join(lines) | |
| _evo_log = EvolutionLog() | |
| # ================================================================ | |
| # HEALTH MONITOR | |
| # ================================================================ | |
| class PlatformHealthMonitor: | |
| def __init__(self): | |
| self.status = { | |
| p: {"alive": True, "fail_count": 0, "last_fail": ""} | |
| for p in ["cloudflare","cohere","github", | |
| "openrouter","huggingface","sambanova"] | |
| } | |
| self._lock = threading.Lock() | |
| def mark_fail(self, platform: str, error: str = ""): | |
| with self._lock: | |
| if platform not in self.status: | |
| return | |
| s = self.status[platform] | |
| s["fail_count"] += 1 | |
| s["last_fail"] = error[:100] | |
| if s["fail_count"] >= 3: | |
| s["alive"] = False | |
| _evo_log.record("platform_down", f"{platform}: {error[:60]}", False) | |
| def mark_success(self, platform: str): | |
| with self._lock: | |
| if platform in self.status: | |
| self.status[platform].update({"alive": True, "fail_count": 0}) | |
| def is_alive(self, platform: str) -> bool: | |
| return self.status.get(platform, {}).get("alive", True) | |
| def get_alive(self) -> list: | |
| return [p for p, s in self.status.items() if s["alive"]] | |
| def get_report(self) -> str: | |
| roles = { | |
| "github": "LOCKED CODE", "cohere": "LOCKED ADMIN+SAFETY", | |
| "cloudflare": "HUNTABLE GUEST", "openrouter": "HUNTABLE BACKUP", | |
| "huggingface": "ADMIN FALLBACK", "sambanova": "BACKUP", | |
| } | |
| lines = ["Health Monitor -- 6 nen tang:\n"] | |
| for p, s in self.status.items(): | |
| icon = "LIVE" if s["alive"] else "DOWN" | |
| lines.append( | |
| f" [{icon}] {p} [{roles.get(p,p)}] | Loi: {s['fail_count']}" | |
| + (f" | {s['last_fail'][:40]}" if s["last_fail"] else "") | |
| ) | |
| return "\n".join(lines) | |
| _health = PlatformHealthMonitor() | |
| # ================================================================ | |
| # HF RESOURCE GUARD | |
| # ================================================================ | |
| class HFResourceGuard: | |
| def __init__(self): | |
| self._last_check = 0 | |
| self._cache_ttl = 10 | |
| self._cached = None | |
| def get_resources(self) -> dict: | |
| now = time.time() | |
| if self._cached and now - self._last_check < self._cache_ttl: | |
| return self._cached | |
| try: | |
| import psutil | |
| mem = psutil.virtual_memory() | |
| cpu = psutil.cpu_percent(interval=0.5) | |
| disk = psutil.disk_usage("/") | |
| status = { | |
| "ram_total_gb": round(mem.total / 1024**3, 2), | |
| "ram_used_gb": round(mem.used / 1024**3, 2), | |
| "ram_pct": mem.percent, | |
| "cpu_pct": cpu, | |
| "disk_free_gb": round(disk.free / 1024**3, 2), | |
| "disk_pct": disk.percent, | |
| "ok": True, | |
| } | |
| except Exception as e: | |
| status = { | |
| "ram_total_gb": 16.0, "ram_used_gb": 8.0, "ram_pct": 50.0, | |
| "cpu_pct": 30.0, "disk_free_gb": 40.0, "disk_pct": 20.0, | |
| "ok": False, "error": str(e), | |
| } | |
| self._cached = status | |
| self._last_check = now | |
| return status | |
| def check_upgrade(self, code: str = "") -> Tuple[bool, str]: | |
| res = self.get_resources() | |
| issues = [] | |
| if res["ram_pct"] >= HF_LIMITS["ram_danger_pct"]: | |
| issues.append(f"RAM qua cao: {res['ram_pct']:.0f}%") | |
| if res["cpu_pct"] >= HF_LIMITS["cpu_max_pct"]: | |
| issues.append(f"CPU qua cao: {res['cpu_pct']:.0f}%") | |
| if res["disk_free_gb"] < HF_LIMITS["disk_min_free_gb"]: | |
| issues.append(f"Disk gan day: con {res['disk_free_gb']:.2f}GB") | |
| try: | |
| mc = len([f for f in os.listdir(MODULES_DIR) | |
| if f.endswith(".py") and f.startswith("auto_") | |
| ]) if os.path.exists(MODULES_DIR) else 0 | |
| if mc >= HF_LIMITS["module_max_count"]: | |
| issues.append(f"Qua nhieu modules: {mc}/{HF_LIMITS['module_max_count']}") | |
| except Exception: | |
| pass | |
| if code: | |
| for lib in HF_LIMITS["heavy_imports"]: | |
| pat = r"(?:import|from)\s+" + re.escape(lib) + r"\b" | |
| if re.search(pat, code): | |
| issues.append(f"Import thu vien nang: {lib}") | |
| if issues: | |
| return False, "CHAN:\n" + "\n".join(f" - {i}" for i in issues) | |
| warn = [] | |
| if res["ram_pct"] >= HF_LIMITS["ram_max_pct"]: | |
| warn.append(f"RAM cao: {res['ram_pct']:.0f}%") | |
| if warn: | |
| return True, "CANH BAO: " + ", ".join(warn) | |
| return True, "OK" | |
| def check_hunt(self) -> Tuple[bool, str]: | |
| res = self.get_resources() | |
| if res["cpu_pct"] >= 80: | |
| return False, f"CPU {res['cpu_pct']:.0f}%" | |
| if res["ram_pct"] >= HF_LIMITS["ram_danger_pct"]: | |
| return False, f"RAM {res['ram_pct']:.0f}%" | |
| return True, "OK" | |
| def dashboard(self) -> str: | |
| res = self.get_resources() | |
| safe, msg = self.check_upgrade() | |
| mc = 0 | |
| try: | |
| if os.path.exists(MODULES_DIR): | |
| mc = len([f for f in os.listdir(MODULES_DIR) | |
| if f.endswith(".py") and f.startswith("auto_")]) | |
| except Exception: | |
| pass | |
| lines = [ | |
| "HF Space Resource Dashboard:\n", | |
| f"RAM: {res['ram_used_gb']:.1f}/{res['ram_total_gb']:.1f}GB ({res['ram_pct']:.0f}%)", | |
| f"CPU: {res['cpu_pct']:.0f}%", | |
| f"Disk: con {res['disk_free_gb']:.1f}GB", | |
| f"Modules: {mc}/{HF_LIMITS['module_max_count']}", | |
| f"\nTrang thai: {'AN TOAN' if safe else 'CHUA AN TOAN'}", | |
| ] | |
| if not safe: | |
| lines.append(f"Ly do: {msg}") | |
| return "\n".join(lines) | |
| def cleanup_modules(self, keep: int = 5) -> str: | |
| if not os.path.exists(MODULES_DIR): | |
| return "Chua co module nao." | |
| try: | |
| files = sorted([ | |
| os.path.join(MODULES_DIR, f) | |
| for f in os.listdir(MODULES_DIR) | |
| if f.endswith(".py") and f.startswith("auto_") | |
| ]) | |
| to_del = files[:-keep] if len(files) > keep else [] | |
| for f in to_del: | |
| os.remove(f) | |
| _evo_log.record("cleanup", f"Xoa {len(to_del)} module cu", True) | |
| return f"Da xoa {len(to_del)} module cu, giu lai {len(files)-len(to_del)}." | |
| except Exception as e: | |
| return f"Loi: {e}" | |
| _resource_guard = HFResourceGuard() | |
| # ================================================================ | |
| # GITHUB QUOTA GUARD - Bảo vệ key GitHub khỏi bị đốt | |
| # ================================================================ | |
| class GitHubQuotaGuard: | |
| """ | |
| Vấn đề: Module quét lỗ hổng gọi DeepSeek-R1 hàng loạt = đốt hết key. | |
| Giải pháp: Đếm số lần gọi, chặn khi gần giới hạn, auto-fallback sang Mistral/Gemini/Cerebras. | |
| """ | |
| DAILY_LIMIT = 40 | |
| HOURLY_LIMIT = 8 | |
| VULN_LIMIT = 2 | |
| def __init__(self): | |
| self._daily_calls = [] | |
| self._hourly_calls = [] | |
| self._vuln_calls = [] | |
| self._lock = threading.Lock() | |
| def _clean(self): | |
| now = time.time() | |
| self._daily_calls = [t for t in self._daily_calls if now - t < 86400] | |
| self._hourly_calls = [t for t in self._hourly_calls if now - t < 3600] | |
| self._vuln_calls = [t for t in self._vuln_calls if now - t < 86400] | |
| def can_call(self, call_type: str = "general") -> tuple: | |
| with self._lock: | |
| self._clean() | |
| if len(self._daily_calls) >= self.DAILY_LIMIT: | |
| h = (86400 - (time.time() - self._daily_calls[0])) / 3600 | |
| return False, f"GitHub quota ngay ({self.DAILY_LIMIT}). Reset sau {h:.1f}h" | |
| if len(self._hourly_calls) >= self.HOURLY_LIMIT: | |
| m = (3600 - (time.time() - self._hourly_calls[0])) / 60 | |
| return False, f"GitHub rate limit ({self.HOURLY_LIMIT}/gio). Thu lai sau {m:.0f}m" | |
| if call_type == "vuln_scan" and len(self._vuln_calls) >= self.VULN_LIMIT: | |
| return False, f"Module quet lo hong da dung {self.VULN_LIMIT} lan hom nay" | |
| return True, "OK" | |
| def record_call(self, call_type: str = "general"): | |
| with self._lock: | |
| now = time.time() | |
| self._daily_calls.append(now) | |
| self._hourly_calls.append(now) | |
| if call_type == "vuln_scan": | |
| self._vuln_calls.append(now) | |
| def status(self) -> str: | |
| with self._lock: | |
| self._clean() | |
| return ( | |
| f"GitHub Quota Guard:\n" | |
| f" Hom nay: {len(self._daily_calls)}/{self.DAILY_LIMIT}\n" | |
| f" Gio nay: {len(self._hourly_calls)}/{self.HOURLY_LIMIT}\n" | |
| f" Quet lo hong: {len(self._vuln_calls)}/{self.VULN_LIMIT}/ngay" | |
| ) | |
| _github_quota = GitHubQuotaGuard() | |
| def _get_free_keys(env_prefix: str) -> list: | |
| """Quet bien moi truong lay key theo prefix (VD: MISTRAL_KEY1, MISTRAL_KEY2...)""" | |
| keys = [] | |
| for k, v in os.environ.items(): | |
| if k.startswith(env_prefix) and v.strip(): | |
| keys.append(v.strip()) | |
| base = os.getenv(env_prefix) | |
| if base and base.strip() and base.strip() not in keys: | |
| keys.append(base.strip()) | |
| return keys | |
| def _call_free_code_brain(prompt: str) -> Optional[str]: | |
| """ | |
| Goi CODE BRAIN mien phi khi GitHub (luong 1) het quota: | |
| Luong 2: Mistral Codestral (chuyen code, 1 ty tokens/thang) | |
| Luong 3: Gemini 2.5/2.0 Flash (1500 req/ngay per project) | |
| Luong 4: Cerebras (backup nhanh, free) | |
| DANG KY: Tat ca deu FREE, KHONG THE, KHONG PHI AN: | |
| - Mistral: https://console.mistral.ai | |
| - Gemini: https://aistudio.google.com/apikey | |
| - Cerebras: https://cloud.cerebras.ai | |
| """ | |
| import openai as _oa | |
| for fb in CODE_BRAIN_FREE_FALLBACKS: | |
| keys = _get_free_keys(fb["env_prefix"]) | |
| if not keys: | |
| continue | |
| key = random.choice(keys) | |
| for model in fb["models"]: | |
| try: | |
| if fb["platform"] == "mistral": | |
| try: | |
| import mistralai as _mst | |
| mc = _mst.Mistral(api_key=key) | |
| resp = mc.chat.complete( | |
| model=model, | |
| messages=[ | |
| {"role": "system", "content": "Expert Python developer. Chi tra ve code Python thuan."}, | |
| {"role": "user", "content": prompt}, | |
| ], | |
| max_tokens=3000, temperature=0.1, | |
| ) | |
| raw = resp.choices[0].message.content.strip() | |
| except ImportError: | |
| # Fallback dung openai compat | |
| client = _oa.OpenAI(api_key=key, base_url=fb["base_url"]) | |
| r = client.chat.completions.create( | |
| model=model, | |
| messages=[ | |
| {"role": "system", "content": "Expert Python developer. Chi tra ve code Python thuan."}, | |
| {"role": "user", "content": prompt}, | |
| ], | |
| max_tokens=3000, temperature=0.1, | |
| ) | |
| raw = r.choices[0].message.content.strip() | |
| else: | |
| client = _oa.OpenAI(api_key=key, base_url=fb["base_url"]) | |
| r = client.chat.completions.create( | |
| model=model, | |
| messages=[ | |
| {"role": "system", "content": "Expert Python developer. Chi tra ve code Python thuan."}, | |
| {"role": "user", "content": prompt}, | |
| ], | |
| max_tokens=3000, temperature=0.1, | |
| ) | |
| raw = r.choices[0].message.content.strip() | |
| raw = re.sub(r"^```python\s*", "", raw) | |
| raw = re.sub(r"^```\s*", "", raw) | |
| raw = re.sub(r"\s*```$", "", raw) | |
| print(f"[FREE_CODE_BRAIN] {fb['platform']}/{model}") | |
| _evo_log.record("code_gen_free", f"{fb['platform']}/{model}", True) | |
| return raw.strip() | |
| except Exception as e: | |
| print(f" [{fb['platform']}/{model}] loi: {str(e)[:60]}") | |
| continue | |
| return None | |
| # ================================================================ | |
| # CODE LAB ENGINE -- Phong thi nghiem chay code that | |
| # ================================================================ | |
| class CodeLabEngine: | |
| """Chay code trong subprocess co lap -- THAT SU, khong chi verify tinh.""" | |
| def __init__(self): | |
| self._run_times = [] | |
| self._lab_history = [] | |
| def _check_quota(self) -> Tuple[bool, str]: | |
| now = time.time() | |
| self._run_times = [t for t in self._run_times if now - t < 3600] | |
| if len(self._run_times) >= CONFIG["lab_hourly_limit"]: | |
| return False, f"Lab het quota ({CONFIG['lab_hourly_limit']} lan/gio)" | |
| return True, "OK" | |
| def run(self, code: str, description: str = "") -> dict: | |
| """Chay code trong subprocess co lap.""" | |
| ok, msg = self._check_quota() | |
| if not ok: | |
| return {"success": False, "output": "", "error": msg, | |
| "runtime_ms": 0, "ran": False} | |
| res = _resource_guard.get_resources() | |
| if res["cpu_pct"] > 85: | |
| return {"success": False, "output": "", | |
| "error": f"CPU {res['cpu_pct']:.0f}% -- lab tam hoan", | |
| "runtime_ms": 0, "ran": False} | |
| start = time.time() | |
| try: | |
| proc = subprocess.run( | |
| [sys.executable, "-c", code], | |
| capture_output=True, text=True, | |
| timeout=CONFIG["lab_timeout"], | |
| env={**os.environ, "PYTHONPATH": ""}, | |
| ) | |
| rt = int((time.time() - start) * 1000) | |
| out = proc.stdout[:CONFIG["lab_max_output"]] | |
| err = proc.stderr[:CONFIG["lab_max_output"]] | |
| ok2 = (proc.returncode == 0) | |
| result = { | |
| "success": ok2, "output": out, | |
| "error": err if not ok2 else "", | |
| "returncode": proc.returncode, "runtime_ms": rt, | |
| "ran": True, "description": description, | |
| "time": datetime.datetime.now().strftime("%H:%M:%S"), | |
| } | |
| except subprocess.TimeoutExpired: | |
| result = { | |
| "success": False, "output": "", | |
| "error": f"TIMEOUT: qua {CONFIG['lab_timeout']}s", | |
| "returncode": -1, "runtime_ms": CONFIG["lab_timeout"] * 1000, | |
| "ran": True, "description": description, | |
| "time": datetime.datetime.now().strftime("%H:%M:%S"), | |
| } | |
| except Exception as e: | |
| result = { | |
| "success": False, "output": "", "error": f"LAB ERROR: {e}", | |
| "returncode": -1, "runtime_ms": 0, "ran": False, | |
| "description": description, | |
| "time": datetime.datetime.now().strftime("%H:%M:%S"), | |
| } | |
| self._run_times.append(time.time()) | |
| self._lab_history.append(result) | |
| if len(self._lab_history) > 50: | |
| self._lab_history.pop(0) | |
| _evo_log.record("lab_run", | |
| f"{'OK' if result['success'] else 'FAIL'}" | |
| f" ({result['runtime_ms']}ms): {description[:50]}", | |
| result["success"]) | |
| return result | |
| def test_module(self, code: str, runs: int = 3) -> dict: | |
| """ | |
| Test module NHIEU LAN de dam bao ket qua nhat quan. | |
| Quy trinh: | |
| 1. Import test x{runs} lan -- tat ca phai pass | |
| 2. Main() test x{runs} lan -- output phai GIONG NHAU ca {runs} lan | |
| 3. Chi pass neu: tat ca runs pass VA output nhat quan | |
| Tai sao 3 lan: | |
| - Tranh truong hop code co randomness che giau bug | |
| - Tranh truong hop may man pass 1 lan nhung thuc ra co race condition | |
| - Tranh code phu thuoc state ben ngoai (lan 1 ok, lan 2 fail) | |
| """ | |
| n = max(1, min(runs, 5)) # gioi han 1-5 lan | |
| # Pha 1: Import test x n lan | |
| import_results = [] | |
| for i in range(n): | |
| r = self.run(code, f"import test {i+1}/{n}") | |
| import_results.append(r) | |
| if not r["success"]: | |
| # Fail ngay -- khong can chay tiep | |
| return { | |
| "import_ok": False, | |
| "import_error": r.get("error", ""), | |
| "main_ok": None, | |
| "main_output": "", | |
| "main_error": "", | |
| "overall_ok": False, | |
| "runs_done": i + 1, | |
| "consistent": False, | |
| "fail_reason": f"Import fail lan {i+1}/{n}: {r.get('error','')[:80]}", | |
| } | |
| # Pha 2: Main test x n lan (neu co __name__ block) | |
| main_results = [] | |
| if "__name__" in code: | |
| test_code = code.replace( | |
| "if __name__ == '__main__':", "if True: # LAB TRIPLE-RUN" | |
| ).replace( | |
| 'if __name__ == "__main__":', "if True: # LAB TRIPLE-RUN" | |
| ) | |
| for i in range(n): | |
| r = self.run(test_code, f"main test {i+1}/{n}") | |
| main_results.append(r) | |
| if not r["success"]: | |
| return { | |
| "import_ok": True, | |
| "import_error": "", | |
| "main_ok": False, | |
| "main_output": "", | |
| "main_error": r.get("error", ""), | |
| "overall_ok": False, | |
| "runs_done": n + i + 1, | |
| "consistent": False, | |
| "fail_reason": f"Main fail lan {i+1}/{n}: {r.get('error','')[:80]}", | |
| } | |
| # Pha 3: Kiem tra tinh nhat quan output | |
| consistent = True | |
| consistency_note = "" | |
| if main_results: | |
| outputs = [r.get("output", "").strip() for r in main_results] | |
| if len(set(outputs)) > 1: | |
| consistent = False | |
| consistency_note = ( | |
| f"Output KHONG nhat quan qua {n} lan chay:\n" | |
| + "\n".join(f" Lan {i+1}: {o[:60]}" for i, o in enumerate(outputs)) | |
| ) | |
| overall_ok = ( | |
| all(r["success"] for r in import_results) | |
| and (not main_results or all(r["success"] for r in main_results)) | |
| and consistent | |
| ) | |
| best_output = main_results[-1].get("output", "") if main_results else "" | |
| avg_runtime = ( | |
| sum(r["runtime_ms"] for r in import_results + main_results) | |
| // max(1, len(import_results) + len(main_results)) | |
| ) | |
| return { | |
| "import_ok": True, | |
| "import_error": "", | |
| "main_ok": bool(main_results) and all(r["success"] for r in main_results), | |
| "main_output": best_output, | |
| "main_error": "", | |
| "overall_ok": overall_ok, | |
| "runs_done": len(import_results) + len(main_results), | |
| "consistent": consistent, | |
| "consistency_note":consistency_note, | |
| "avg_runtime_ms": avg_runtime, | |
| "fail_reason": consistency_note if not consistent else "", | |
| } | |
| def test_fix(self, original: str, fixed: str, runs: int = 3) -> dict: | |
| """ | |
| So sanh code goc vs code da fix, chay {runs} lan. | |
| Fix chi duoc chap nhan neu: | |
| 1. Code goc fail (hoac fix tot hon ro rang) | |
| 2. Code da fix PASS CA {runs} LAN | |
| 3. Output nhat quan qua {runs} lan | |
| """ | |
| n = max(1, min(runs, 5)) | |
| # Code goc -- chay 1 lan de confirm no fail | |
| orig_r = self.run(original, "original") | |
| # Code fix -- chay n lan, phai pass het | |
| fix_results = [] | |
| for i in range(n): | |
| r = self.run(fixed, f"fix run {i+1}/{n}") | |
| fix_results.append(r) | |
| all_pass = all(r["success"] for r in fix_results) | |
| outputs = [r.get("output", "").strip() for r in fix_results] | |
| consistent = (len(set(outputs)) <= 1) | |
| fix_is_good = all_pass and consistent | |
| return { | |
| "original_failed": not orig_r["success"], | |
| "fix_passed": all_pass, | |
| "fix_is_better": (fix_is_good and not orig_r["success"]), | |
| "fix_consistent": consistent, | |
| "runs_done": n, | |
| "pass_count": sum(1 for r in fix_results if r["success"]), | |
| "original_error": orig_r.get("error", "")[:200], | |
| "fix_output": fix_results[-1].get("output", "")[:200] if fix_results else "", | |
| "inconsistency": "" if consistent else f"Output khac nhau qua {n} lan", | |
| } | |
| def history(self, n: int = 5) -> str: | |
| if not self._lab_history: | |
| return "Lab chua chay lan nao." | |
| lines = [f"Lab History ({n} gan nhat):\n"] | |
| for r in reversed(self._lab_history[-n:]): | |
| icon = "v" if r["success"] else "x" | |
| lines.append( | |
| f"[{icon}] {r['time']} ({r['runtime_ms']}ms)" | |
| f" -- {r['description'][:50]}" | |
| ) | |
| if not r["success"] and r.get("error"): | |
| lines.append(f" {r['error'][:60]}") | |
| used = len(self._run_times) | |
| lines.append(f"\nQuota: {used}/{CONFIG['lab_hourly_limit']} lan/gio") | |
| return "\n".join(lines) | |
| _lab = CodeLabEngine() | |
| # ================================================================ | |
| # BENCHMARK CHAT QUALITY -- Danh gia chat luong output AI | |
| # ================================================================ | |
| class BenchmarkQuality: | |
| """ | |
| Danh gia chat luong output cua module/fix truoc khi ap dung. | |
| Test 3 tieu chi: | |
| 1. Correctness -- output co dung voi expected khong (0-10) | |
| 2. Performance -- thoi gian chay co chap nhan duoc khong (0-10) | |
| 3. Stability -- co crash hay exception an khong (0-10) | |
| Ket qua: score tong 0-30. Pass neu >= PASS_THRESHOLD. | |
| """ | |
| PASS_THRESHOLD = 20 # /30 -- phai dat it nhat 20/30 moi duoc dung | |
| def score_module(self, code: str, expected_outputs: list = None, | |
| max_runtime_ms: int = 3000) -> dict: | |
| """ | |
| Benchmark module voi 3 tieu chi. | |
| expected_outputs: list cac chuoi phai co trong output (neu co) | |
| max_runtime_ms: thoi gian toi da chap nhan (ms) | |
| """ | |
| # Chay 3 lan lay ket qua | |
| lab_result = _lab.test_module(code, runs=3) | |
| scores = {} | |
| details = [] | |
| # Tieu chi 1: Correctness (0-10) | |
| correctness = 0 | |
| if lab_result["overall_ok"]: | |
| correctness += 6 # Base: code chay duoc | |
| if lab_result.get("consistent"): | |
| correctness += 2 # Output nhat quan | |
| if expected_outputs and lab_result.get("main_output"): | |
| output = lab_result["main_output"].lower() | |
| matched = sum(1 for e in expected_outputs if str(e).lower() in output) | |
| ratio = matched / len(expected_outputs) | |
| correctness += int(ratio * 2) # 0-2 diem them | |
| else: | |
| correctness += 2 # Khong co expected = khong tru diem | |
| scores["correctness"] = min(10, correctness) | |
| details.append(f"Correctness: {scores['correctness']}/10" | |
| + (" (output nhat quan)" if lab_result.get("consistent") else | |
| " (output KHONG nhat quan)")) | |
| # Tieu chi 2: Performance (0-10) | |
| perf = 10 | |
| avg_rt = lab_result.get("avg_runtime_ms", 0) | |
| if avg_rt > max_runtime_ms: | |
| perf -= 5 # Qua cham | |
| elif avg_rt > max_runtime_ms * 0.7: | |
| perf -= 2 # Hoi cham | |
| if not lab_result.get("ran", True): | |
| perf = 0 # Khong chay duoc | |
| scores["performance"] = max(0, perf) | |
| details.append(f"Performance: {scores['performance']}/10" | |
| + f" (avg {avg_rt}ms, limit {max_runtime_ms}ms)") | |
| # Tieu chi 3: Stability (0-10) | |
| stability = 10 | |
| if not lab_result["import_ok"]: | |
| stability = 0 # Loi import = fail hoan toan | |
| elif not lab_result.get("overall_ok"): | |
| stability = 4 # Co loi khi chay | |
| elif lab_result.get("fail_reason"): | |
| stability = 6 # Co van de nhung khong crash | |
| scores["stability"] = stability | |
| details.append(f"Stability: {scores['stability']}/10" | |
| + ("" if stability == 10 else | |
| f" ({lab_result.get('fail_reason','')[:40]})")) | |
| total = sum(scores.values()) | |
| passed = total >= self.PASS_THRESHOLD | |
| return { | |
| "passed": passed, | |
| "total_score": total, | |
| "max_score": 30, | |
| "pass_threshold": self.PASS_THRESHOLD, | |
| "scores": scores, | |
| "details": details, | |
| "lab_result": lab_result, | |
| "verdict": ( | |
| f"PASS ({total}/30)" if passed | |
| else f"FAIL ({total}/30 -- can {self.PASS_THRESHOLD})" | |
| ), | |
| } | |
| def format_report(self, bm: dict) -> str: | |
| lines = [ | |
| f"Benchmark Quality Report:", | |
| f" Ket qua: {bm['verdict']}", | |
| ] | |
| for d in bm.get("details", []): | |
| lines.append(f" - {d}") | |
| if not bm["passed"]: | |
| lines.append( | |
| f" --> KHONG dat nguong {bm['pass_threshold']}/30" | |
| f" -- module bi CHON KHONG duoc dung" | |
| ) | |
| return "\n".join(lines) | |
| _bm_quality = BenchmarkQuality() | |
| # ================================================================ | |
| # ROLLBACK MANAGER -- Tu dong rollback neu module moi bi loi | |
| # ================================================================ | |
| class RollbackManager: | |
| """ | |
| Quan ly rollback tu dong khi module moi bi loi sau khi active. | |
| Cach hoat dong: | |
| 1. Truoc khi luu module moi --> tao snapshot trang thai hien tai | |
| 2. Neu module moi bi loi sau khi active --> tu dong rollback | |
| 3. Log day du moi thao tac | |
| 4. Giu toi da 5 snapshot (don dep tu dong) | |
| An toan: | |
| - Chi rollback module trong ural_modules/ (KHONG rollback app goc) | |
| - Rollback = xoa file module moi + phuc hoi snapshot | |
| - Neu rollback cung fail --> bao admin qua log | |
| """ | |
| MAX_SNAPSHOTS = 5 | |
| def __init__(self): | |
| self.snapshots = [] # [{name, file, backup_content, time}] | |
| self.rollback_log = [] | |
| def create_snapshot(self, module_name: str, file_path: str) -> bool: | |
| """Tao snapshot truoc khi luu module moi.""" | |
| try: | |
| # Doc noi dung hien tai neu file ton tai | |
| backup = None | |
| if os.path.exists(file_path): | |
| with open(file_path, "r", encoding="utf-8") as f: | |
| backup = f.read() | |
| snapshot = { | |
| "name": module_name, | |
| "file": file_path, | |
| "backup_content": backup, | |
| "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), | |
| "exists_before": os.path.exists(file_path), | |
| } | |
| self.snapshots.append(snapshot) | |
| # Don dep snapshot cu | |
| if len(self.snapshots) > self.MAX_SNAPSHOTS: | |
| self.snapshots.pop(0) | |
| _evo_log.record("snapshot", f"Tao snapshot: {module_name}", True) | |
| return True | |
| except Exception as e: | |
| _evo_log.record("snapshot", f"Loi tao snapshot: {e}", False) | |
| return False | |
| def rollback(self, module_name: str, reason: str = "") -> str: | |
| """ | |
| Tu dong rollback module ve trang thai truoc do. | |
| Goi khi phat hien module moi bi loi. | |
| """ | |
| snapshot = next( | |
| (s for s in reversed(self.snapshots) if s["name"] == module_name), | |
| None, | |
| ) | |
| if not snapshot: | |
| msg = f"Khong tim thay snapshot cho: {module_name}" | |
| _evo_log.record("rollback", msg, False) | |
| return msg | |
| file_path = snapshot["file"] | |
| try: | |
| if snapshot["exists_before"] and snapshot["backup_content"]: | |
| # Phuc hoi noi dung cu | |
| with open(file_path, "w", encoding="utf-8") as f: | |
| f.write(snapshot["backup_content"]) | |
| msg = f"Da rollback {module_name} ve trang thai truoc." | |
| else: | |
| # File nay la moi -- xoa di | |
| if os.path.exists(file_path): | |
| os.remove(file_path) | |
| msg = f"Da xoa module moi {module_name} (rollback)." | |
| self.rollback_log.append({ | |
| "module": module_name, | |
| "reason": reason[:100], | |
| "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), | |
| "ok": True, | |
| }) | |
| _evo_log.record("rollback", f"OK: {module_name} -- {reason[:60]}", True) | |
| return msg | |
| except Exception as e: | |
| err_msg = f"Loi rollback {module_name}: {e}" | |
| self.rollback_log.append({ | |
| "module": module_name, | |
| "reason": reason[:100], | |
| "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), | |
| "ok": False, | |
| "error": str(e), | |
| }) | |
| _evo_log.record("rollback", err_msg, False) | |
| return err_msg | |
| def auto_check_and_rollback(self, module_name: str, | |
| file_path: str) -> Optional[str]: | |
| """ | |
| Sau khi luu module moi, chay kiem tra lan cuoi. | |
| Neu fail --> tu dong rollback, khong can admin. | |
| Tra ve None neu OK, hoac error message neu rollback. | |
| """ | |
| if not os.path.exists(file_path): | |
| return None | |
| try: | |
| with open(file_path, "r", encoding="utf-8") as f: | |
| code = f.read() | |
| except Exception as e: | |
| return self.rollback(module_name, f"Khong doc duoc file: {e}") | |
| # Chay kiem tra cuoi: 3 lan, tat ca phai pass | |
| final_check = _lab.test_module(code, runs=3) | |
| if not final_check["overall_ok"] or not final_check.get("consistent", True): | |
| reason = final_check.get("fail_reason", "Lab test fail sau khi luu") | |
| print(f"[Rollback] Module {module_name} fail post-save check!") | |
| print(f" Ly do: {reason}") | |
| msg = self.rollback(module_name, reason) | |
| return f"AUTO ROLLBACK: {msg}\nLy do: {reason}" | |
| _evo_log.record("post_save_check", f"{module_name}: OK", True) | |
| return None # Khong can rollback | |
| def get_log(self) -> str: | |
| if not self.rollback_log: | |
| return "Chua co rollback nao." | |
| lines = ["Rollback Log:\n"] | |
| for r in reversed(self.rollback_log[-5:]): | |
| icon = "v" if r["ok"] else "x" | |
| lines.append( | |
| f"[{icon}] {r['time']} -- {r['module']}: {r['reason'][:60]}" | |
| ) | |
| return "\n".join(lines) | |
| _rollback = RollbackManager() | |
| # ================================================================ | |
| # MODEL BENCHMARKER | |
| # ================================================================ | |
| class ModelBenchmarker: | |
| TESTS = [ | |
| ("Thu do Viet Nam la gi?", "ha noi"), | |
| ("1 + 1 bang may?", "2"), | |
| ("Python la gi?", "lap trinh"), | |
| ("Mat trang quay quanh gi?", "trai dat"), | |
| ] | |
| def quick_check(self, model_id: str, platform: str, | |
| keys: list, base_url: str = None) -> bool: | |
| if not keys: | |
| return False | |
| try: | |
| key = random.choice(keys) | |
| if platform == "cohere": | |
| import cohere as _co | |
| co = _co.Client(api_key=key) | |
| r = co.chat(message="Hi", model=model_id) | |
| return bool(r.text) | |
| else: | |
| import openai as _oa | |
| url = base_url or "https://openrouter.ai/api/v1" | |
| client = _oa.OpenAI(api_key=key, base_url=url) | |
| extra = {} | |
| if "openrouter" in url: | |
| extra = {"extra_headers": { | |
| "HTTP-Referer": "https://huggingface.co", | |
| "X-Title": "URAL AI", | |
| }} | |
| r = client.chat.completions.create( | |
| model=model_id, | |
| messages=[{"role": "user", "content": "Hi"}], | |
| max_tokens=5, **extra, | |
| ) | |
| return bool(r.choices[0].message.content) | |
| except Exception: | |
| return False | |
| def benchmark(self, model_id: str, platform: str, | |
| keys: list, base_url: str = None) -> dict: | |
| if not keys: | |
| return {"score": 0, "max": 8, "passed": False} | |
| key = random.choice(keys) | |
| score = 0 | |
| for question, expected in self.TESTS: | |
| try: | |
| if platform == "cohere": | |
| import cohere as _co | |
| co = _co.Client(api_key=key) | |
| resp = co.chat(message=question, model=model_id) | |
| ans = resp.text.lower() | |
| else: | |
| import openai as _oa | |
| url = base_url or "https://openrouter.ai/api/v1" | |
| client = _oa.OpenAI(api_key=key, base_url=url) | |
| extra = {} | |
| if "openrouter" in url: | |
| extra = {"extra_headers": { | |
| "HTTP-Referer": "https://huggingface.co", | |
| "X-Title": "URAL AI", | |
| }} | |
| r = client.chat.completions.create( | |
| model=model_id, | |
| messages=[{"role": "user", "content": question}], | |
| max_tokens=50, **extra, | |
| ) | |
| ans = (r.choices[0].message.content or "").lower() | |
| if expected.lower() in ans: | |
| score += 2 | |
| except Exception: | |
| pass | |
| passed = score >= CONFIG["benchmark_min_score"] | |
| return {"score": score, "max": len(self.TESTS) * 2, "passed": passed} | |
| _bench = ModelBenchmarker() | |
| # ================================================================ | |
| # MODEL HUNTER v2 -- Chi hunt Tang 3 & 4 | |
| # ================================================================ | |
| class ModelHunterV2: | |
| LOCKED = {"github", "cohere_plus"} | |
| def __init__(self): | |
| self.cf_alive = [] | |
| self.or_alive = [] | |
| self.dead = set() | |
| self.last_cf = 0 | |
| self.last_or = 0 | |
| self._lock = threading.Lock() | |
| def _app(self): | |
| for mod in sys.modules.values(): | |
| if hasattr(mod, "CLOUDFLARE_KEYS") and hasattr(mod, "COHERE_KEYS"): | |
| return mod | |
| return None | |
| def hunt_all(self, force: bool = False): | |
| app = self._app() | |
| if not app: | |
| return | |
| safe, reason = _resource_guard.check_hunt() | |
| if not safe: | |
| print(f"[Hunter] Hoan hunt: {reason}") | |
| _evo_log.record("hunt_blocked", reason[:80], False) | |
| def _retry(): | |
| time.sleep(300) | |
| s2, _ = _resource_guard.check_hunt() | |
| if s2: | |
| self.hunt_all(force=force) | |
| threading.Thread(target=_retry, daemon=True).start() | |
| return | |
| cf_k = getattr(app, "CLOUDFLARE_KEYS", []) | |
| or_k = getattr(app, "OPENROUTER_KEYS", []) | |
| hf_t = getattr(app, "HF_TOKEN", None) | |
| if cf_k: | |
| threading.Thread(target=self._hunt_cf, args=(cf_k, force), daemon=True).start() | |
| if or_k: | |
| threading.Thread(target=self._hunt_or, args=(or_k, force), daemon=True).start() | |
| if hf_t: | |
| threading.Thread(target=self._check_hf, args=(hf_t,), daemon=True).start() | |
| def _hunt_cf(self, keys: list, force: bool = False): | |
| if not force and time.time() - self.last_cf < CONFIG["hunt_cooldown"]: | |
| return | |
| print("[Hunter-CF] Testing Cloudflare models...") | |
| found = [] | |
| for model in GUEST_BRAIN["candidates"]: | |
| if model in self.dead: | |
| continue | |
| key = random.choice(keys) | |
| if "|" not in key: | |
| continue | |
| acc_id, token = key.split("|", 1) | |
| cf_url = f"https://api.cloudflare.com/client/v4/accounts/{acc_id}/ai/v1" | |
| alive = _bench.quick_check(model, "cloudflare", [f"{acc_id}|{token}"], cf_url) | |
| if not alive: | |
| self.dead.add(model) | |
| continue | |
| bm = _bench.benchmark(model, "cloudflare", [f"{acc_id}|{token}"], cf_url) | |
| if bm["passed"]: | |
| found.append(model) | |
| print(f" LIVE: {model[:45]} score={bm['score']}/{bm['max']}") | |
| _evo_log.record("hunt_cf", f"Live: {model[:50]}", True) | |
| else: | |
| print(f" LOW: {model[:45]} score={bm['score']}/{bm['max']}") | |
| with self._lock: | |
| if found: | |
| self.cf_alive = found | |
| GUEST_BRAIN["current_model"] = found[0] | |
| app = self._app() | |
| if app: | |
| setattr(app, "_cf_current_model", found[0]) | |
| _health.mark_success("cloudflare") | |
| _evo_log.record("model_inject", f"CF -> {found[0]}", True) | |
| print(f"[Hunter-CF] Injected: {found[0]}") | |
| else: | |
| print("[Hunter-CF] Khong tim duoc model moi, giu nguyen") | |
| self.last_cf = time.time() | |
| def _hunt_or(self, keys: list, force: bool = False): | |
| if not force and time.time() - self.last_or < CONFIG["hunt_cooldown"]: | |
| return | |
| print("[Hunter-OR] Fetching OpenRouter free models...") | |
| try: | |
| import requests as _req | |
| resp = _req.get( | |
| "https://openrouter.ai/api/v1/models", | |
| headers={"Authorization": f"Bearer {keys[0]}"}, | |
| timeout=10, | |
| ) | |
| all_m = resp.json().get("data", []) | |
| free_ids = [ | |
| m["id"] for m in all_m | |
| if (str(m.get("pricing", {}).get("prompt", "1")) == "0" | |
| or m["id"].endswith(":free")) | |
| and m["id"] not in self.dead | |
| ] | |
| priority = ["deepseek", "gemma-3", "llama-3", "qwen", "mistral"] | |
| free_ids.sort(key=lambda mid: next( | |
| (i for i, kw in enumerate(priority) if kw in mid.lower()), 99 | |
| )) | |
| found = [] | |
| key = random.choice(keys) | |
| for mid in free_ids[:8]: | |
| if not _bench.quick_check(mid, "openrouter", [key]): | |
| self.dead.add(mid) | |
| continue | |
| bm = _bench.benchmark(mid, "openrouter", [key]) | |
| if bm["passed"]: | |
| found.append(mid) | |
| _evo_log.record("hunt_or", f"Live: {mid[:50]}", True) | |
| if len(found) >= 5: | |
| break | |
| with self._lock: | |
| if found: | |
| self.or_alive = found | |
| BACKUP_BRAIN["live_models"] = found | |
| app = self._app() | |
| if app: | |
| try: | |
| app.OPENROUTER_LIVE_MODELS = found | |
| except Exception: | |
| pass | |
| _health.mark_success("openrouter") | |
| _evo_log.record("model_inject", f"OR -> {len(found)} models", True) | |
| print(f"[Hunter-OR] Injected {len(found)} models") | |
| self.last_or = time.time() | |
| except Exception as e: | |
| _evo_log.record("hunt_or", f"Loi: {e}", False) | |
| def _check_hf(self, token: str): | |
| from huggingface_hub import InferenceClient | |
| alive = [] | |
| for model in ["Qwen/Qwen2.5-72B-Instruct", "mistralai/Mistral-7B-Instruct-v0.3"]: | |
| try: | |
| client = InferenceClient(model=model, token=token) | |
| r = client.chat_completion( | |
| messages=[{"role": "user", "content": "Hi"}], max_tokens=5 | |
| ) | |
| if r.choices[0].message.content: | |
| alive.append(model) | |
| print(f" [HF] LIVE: {model}") | |
| except Exception as e: | |
| _health.mark_fail("huggingface", str(e)[:60]) | |
| if alive: | |
| _health.mark_success("huggingface") | |
| ADMIN_CHAT_BRAIN["fallbacks"] = alive | |
| def status(self) -> str: | |
| lines = [ | |
| "Model Hunter v2 Status:\n", | |
| "Tang 1 CODE (LOCKED):", | |
| f" GitHub: {', '.join(CODE_BRAIN['models'])}", | |
| "\nTang 2 ADMIN CHAT (LOCKED):", | |
| f" Cohere: {', '.join(ADMIN_CHAT_BRAIN['models'])}", | |
| "\nTang 3 GUEST (Hunter quan ly):", | |
| f" CF: {GUEST_BRAIN['current_model']}", | |
| f" Pass benchmark: {len(self.cf_alive)} models", | |
| "\nTang 4 BACKUP (Hunter quan ly):", | |
| f" OR live: {len(self.or_alive)} models", | |
| f" {str(self.or_alive[:2]) if self.or_alive else '(chua hunt)'}", | |
| f"\nDead: {len(self.dead)}", | |
| ] | |
| return "\n".join(lines) | |
| _hunter = ModelHunterV2() | |
| # ================================================================ | |
| # SANDBOX VERIFY - 3 LOP CHONG BYPASS (tich hop truc tiep, khong can file ngoai) | |
| # Lap 1: AST Visitor - bat ham/import nguy hiem | |
| # Lap 2: Regex - bat trick ghep chuoi bypass | |
| # Lap 3: Compile check | |
| # Chong duoc: __builtins__['__import__'], getattr(__builtins__,'ev'+'al'), chr(101)+... | |
| # ================================================================ | |
| import ast as _ast_mod | |
| _SANDBOX_ALLOWED_IMPORTS = { | |
| "os","re","json","time","datetime","random","math","string", | |
| "collections","typing","requests","hashlib","base64","urllib", | |
| "functools","itertools","operator","copy", | |
| } | |
| _SANDBOX_BANNED_NAMES = { | |
| "eval","exec","compile","execfile","__import__","__builtins__", | |
| "__loader__","globals","locals","vars","dir","input","open", | |
| "subprocess","Popen","system","popen","spawn","fork","execv", | |
| "rmtree","remove","unlink","socket","pickle","marshal","ctypes", | |
| } | |
| _SANDBOX_BYPASS_PATTERNS = [ | |
| r"__builtins__\s*[\[\.]", | |
| r"getattr\s*\(\s*__", | |
| r"getattr\s*\(.*builtins", | |
| r"chr\s*\(\s*\d+\s*\)", | |
| r"\\x[0-9a-fA-F]{2}", | |
| r"importlib", | |
| r"__import__", | |
| r"os\.environ", | |
| r"os\.getenv", | |
| r"os\.remove\s*\(", | |
| r"os\.unlink\s*\(", | |
| r"shutil\.", | |
| r"\.write\s*\(", | |
| r"open\s*\(.*['\"][wa]['\"]", | |
| ] | |
| class _ASTSecurityVisitor(_ast_mod.NodeVisitor): | |
| def __init__(self): | |
| self.violations = [] | |
| def visit_Call(self, node): | |
| name = "" | |
| if isinstance(node.func, _ast_mod.Name): name = node.func.id | |
| elif isinstance(node.func, _ast_mod.Attribute): name = node.func.attr | |
| if name in _SANDBOX_BANNED_NAMES: | |
| self.violations.append(f"Ham bi cam: {name}() dong {getattr(node,'lineno',0)}") | |
| self.generic_visit(node) | |
| def visit_Import(self, node): | |
| for a in node.names: | |
| base = a.name.split(".")[0] | |
| if base not in _SANDBOX_ALLOWED_IMPORTS: | |
| self.violations.append(f"Import khong phep: {a.name}") | |
| self.generic_visit(node) | |
| def visit_ImportFrom(self, node): | |
| if node.module: | |
| base = node.module.split(".")[0] | |
| if base not in _SANDBOX_ALLOWED_IMPORTS: | |
| self.violations.append(f"Import khong phep: from {node.module}") | |
| self.generic_visit(node) | |
| def visit_Attribute(self, node): | |
| if isinstance(node.value, _ast_mod.Name): | |
| pair = f"{node.value.id}.{node.attr}" | |
| danger = {"os.system","os.popen","os.execv","os.remove","os.unlink", | |
| "os.environ","os.getenv","sys.exit","subprocess.Popen", | |
| "subprocess.call","shutil.rmtree"} | |
| if pair in danger: | |
| self.violations.append(f"Truy cap nguy hiem: {pair} dong {getattr(node,'lineno',0)}") | |
| self.generic_visit(node) | |
| def visit_Global(self, node): | |
| self.violations.append("Cam dung 'global' trong module tu dong") | |
| self.generic_visit(node) | |
| def _sandbox_verify(code: str) -> Tuple[bool, str]: | |
| """ | |
| Sandbox 3 lop: | |
| 1. Parse + AST Visitor (bat ham/import nguy hiem, chong bypass nang cao) | |
| 2. Regex bypass patterns (bat trick ghep chuoi, chr(), hex escape...) | |
| 3. Compile check | |
| """ | |
| if not code or not code.strip(): | |
| return False, "Code rong" | |
| if len(code) > 50000: | |
| return False, f"Code qua dai: {len(code)} ky tu" | |
| # Lop 1: AST | |
| try: | |
| tree = _ast_mod.parse(code) | |
| except SyntaxError as e: | |
| return False, f"Syntax error: {e}" | |
| except Exception as e: | |
| return False, f"Khong parse duoc: {e}" | |
| visitor = _ASTSecurityVisitor() | |
| try: | |
| visitor.visit(tree) | |
| except Exception as e: | |
| return False, f"Loi scan AST: {e}" | |
| if visitor.violations: | |
| return False, "Vi pham bao mat:\n" + "\n".join(f" - {v}" for v in visitor.violations[:5]) | |
| # Lop 2: Regex bypass | |
| for pat in _SANDBOX_BYPASS_PATTERNS: | |
| try: | |
| if re.search(pat, code, re.IGNORECASE | re.MULTILINE): | |
| return False, f"Pattern bypass bi chan: {pat[:50]}" | |
| except re.error: | |
| continue | |
| # Lop 3: Compile | |
| try: | |
| compile(code, "<sandbox>", "exec") | |
| except Exception as e: | |
| return False, f"Compile error: {e}" | |
| return True, "OK" | |
| def _call_code_brain(prompt: str, github_keys: list, call_type: str = "general") -> Optional[str]: | |
| """ | |
| Goi CODE_BRAIN voi 2 tang bao ve: | |
| 1. Kiem tra GitHubQuotaGuard truoc | |
| 2. Neu GitHub het quota → tu dong fallback sang Mistral/Gemini/Cerebras (tat ca FREE) | |
| Luon dung CODE_BRAIN cho code -- khong bao gio dung model chat thuong. | |
| """ | |
| # Tang 1: Thu GitHub (neu con quota) | |
| can_github, reason = _github_quota.can_call(call_type) | |
| if can_github and github_keys: | |
| key = random.choice(github_keys) | |
| for model in CODE_BRAIN["models"]: | |
| try: | |
| import openai as _oa | |
| client = _oa.OpenAI(api_key=key, base_url=CODE_BRAIN["base_url"]) | |
| resp = client.chat.completions.create( | |
| model=model, | |
| messages=[ | |
| {"role": "system", "content": "Expert Python developer. Chi tra ve code Python thuan."}, | |
| {"role": "user", "content": prompt}, | |
| ], | |
| max_tokens=3000, temperature=0.1, | |
| ) | |
| code = resp.choices[0].message.content.strip() | |
| code = re.sub(r"^```python\s*", "", code) | |
| code = re.sub(r"^```\s*", "", code) | |
| code = re.sub(r"\s*```$", "", code) | |
| # Ghi nhan da dung quota | |
| _github_quota.record_call(call_type) | |
| print(f"[CODE_BRAIN] GitHub/{model} | Quota hom nay: {len(_github_quota._daily_calls)}/{_github_quota.DAILY_LIMIT}") | |
| _evo_log.record("code_gen", f"GitHub/{model}", True) | |
| return code.strip() | |
| except Exception as e: | |
| print(f" {model} loi: {str(e)[:50]}") | |
| continue | |
| # GitHub loi het → fallback | |
| print(f"[CODE_BRAIN] GitHub that bai, chuyen sang FREE fallback...") | |
| else: | |
| print(f"[CODE_BRAIN] GitHub bi chan: {reason}") | |
| print(f"[CODE_BRAIN] Dang dung FREE fallback (Mistral/Gemini/Cerebras)...") | |
| # Tang 2: FREE fallback platforms | |
| result = _call_free_code_brain(prompt) | |
| if result: | |
| return result | |
| print("[CODE_BRAIN] Tat ca platforms deu het key hoac loi") | |
| return None | |
| def _get_github_keys() -> list: | |
| for mod in sys.modules.values(): | |
| if hasattr(mod, "GITHUB_KEYS"): | |
| return getattr(mod, "GITHUB_KEYS", []) | |
| return [] | |
| # ================================================================ | |
| # AUTO BUG FIXER -- 2 tang | |
| # ================================================================ | |
| class AutoBugFixer: | |
| """ | |
| Tang 1 Quick: quota/timeout/deprecated --> xu ly ngay, khong can AI | |
| Tang 2 Deep: loi phuc tap --> DeepSeek-R1 phan tich --> patch code | |
| Patch phai qua: Sandbox verify + Lab test + Admin confirm | |
| """ | |
| QUICK = { | |
| "quota": "switch_platform", | |
| "rate limit": "wait_retry", | |
| "429": "wait_retry", | |
| "connection": "mark_dead", | |
| "timeout": "mark_dead", | |
| "503": "mark_dead", | |
| "502": "mark_dead", | |
| "model not found": "hunt_replacement", | |
| "deprecated": "hunt_replacement", | |
| "invalid api key": "mark_key_bad", | |
| "authentication": "mark_key_bad", | |
| } | |
| DEEP_TRIGGERS = [ | |
| "typeerror","attributeerror","nameerror","keyerror", | |
| "valueerror","indexerror","syntaxerror","importerror", | |
| "json","parse","decode","unexpected", | |
| ] | |
| def __init__(self): | |
| self.quick_count = 0 | |
| self.deep_count = 0 | |
| self.fail_count = 0 | |
| self.error_log = [] | |
| self._deep_times = [] | |
| def _quick_fix(self, error: Exception, platform: str) -> Optional[str]: | |
| err_str = str(error).lower() | |
| action = next( | |
| (act for pat, act in self.QUICK.items() if pat in err_str), None | |
| ) | |
| if not action: | |
| return None | |
| if action == "mark_dead": | |
| _health.mark_fail(platform, str(error)[:80]) | |
| elif action == "switch_platform": | |
| _health.mark_fail(platform, "quota") | |
| elif action == "wait_retry": | |
| time.sleep(3) | |
| elif action == "hunt_replacement": | |
| if platform in ("cloudflare", "openrouter"): | |
| threading.Thread( | |
| target=_hunter.hunt_all, kwargs={"force": True}, daemon=True | |
| ).start() | |
| self.quick_count += 1 | |
| self._log(platform, str(error), action, "quick") | |
| _evo_log.record("bug_fix_quick", f"[{platform}] {action}: {str(error)[:60]}", True) | |
| return action | |
| def _should_deep(self, error: Exception) -> bool: | |
| # Cho phep deep fix neu co GitHub HOAC co bat ky free brain key nao | |
| has_github = bool(_get_github_keys()) | |
| has_free = any(_get_free_keys(fb["env_prefix"]) for fb in CODE_BRAIN_FREE_FALLBACKS) | |
| if not has_github and not has_free: | |
| return False | |
| err_str = str(error).lower() | |
| tb = traceback.format_exc().lower() | |
| if not any(t in err_str or t in tb for t in self.DEEP_TRIGGERS): | |
| return False | |
| now = time.time() | |
| self._deep_times = [t for t in self._deep_times if now - t < 3600] | |
| if len(self._deep_times) >= CONFIG["deepfix_hourly_limit"]: | |
| return False | |
| safe, _ = _resource_guard.check_hunt() | |
| return safe | |
| def _call_deepseek(self, error: Exception, context: dict) -> Optional[dict]: | |
| keys = _get_github_keys() | |
| # Kiem tra quota truoc khi goi | |
| can, reason = _github_quota.can_call("deep_fix") | |
| if not can: | |
| print(f"[BugFixer] GitHub bi chan: {reason}. Dung free fallback de phan tich...") | |
| # Dung free brain phan tich loi | |
| tb_str = traceback.format_exc() | |
| prompt = ( | |
| f"Phan tich loi Python nay va tra ve JSON thuan:\n\n" | |
| f"LOI: {type(error).__name__}: {str(error)}\n\n" | |
| f"TRACEBACK:\n{tb_str[:800]}\n\n" | |
| f"JSON format: {{\"cause\":\"...\",\"severity\":\"LOW|MEDIUM|HIGH\"," | |
| f"\"immediate_fix\":\"...\",\"needs_code_patch\":false,\"patch_code\":null,\"patch_description\":\"\"}}" | |
| ) | |
| raw = _call_free_code_brain(prompt) | |
| if raw: | |
| try: | |
| return json.loads(raw) | |
| except: | |
| return {"cause": raw[:200], "severity": "MEDIUM", "immediate_fix": "Xem log", | |
| "needs_code_patch": False, "patch_code": None} | |
| return None | |
| if not keys: | |
| # Thu free brain ngay | |
| tb_str = traceback.format_exc() | |
| prompt = ( | |
| f"Phan tich loi Python, tra ve JSON thuan:\n" | |
| f"LOI: {type(error).__name__}: {str(error)}\n" | |
| f"TRACEBACK:\n{tb_str[:800]}\n" | |
| f"JSON: {{\"cause\":\"\",\"severity\":\"LOW|MEDIUM|HIGH\",\"immediate_fix\":\"\"," | |
| f"\"needs_code_patch\":false,\"patch_code\":null,\"patch_description\":\"\"}}" | |
| ) | |
| raw = _call_free_code_brain(prompt) | |
| if raw: | |
| try: | |
| return json.loads(raw) | |
| except: | |
| return {"cause": raw[:200], "severity": "MEDIUM", "immediate_fix": "Xem log", | |
| "needs_code_patch": False, "patch_code": None} | |
| return None | |
| tb_str = traceback.format_exc() | |
| prompt = ( | |
| f"Phan tich loi Python:\n\n" | |
| f"LOI: {type(error).__name__}: {str(error)}\n\n" | |
| f"TRACEBACK:\n{tb_str[:1200]}\n\n" | |
| f"CONTEXT: platform={context.get('platform','?')}\n\n" | |
| f"Tra ve JSON thuan (khong markdown, khong backtick):\n" | |
| '{{"cause":"nguyen nhan","severity":"LOW|MEDIUM|HIGH",' | |
| '"immediate_fix":"fix ngay","needs_code_patch":false,' | |
| '"patch_code":null,"patch_description":""}}' | |
| ) | |
| import openai as _oa | |
| key = random.choice(keys) | |
| for model in CODE_BRAIN["models"]: | |
| try: | |
| client = _oa.OpenAI(api_key=key, base_url=CODE_BRAIN["base_url"]) | |
| resp = client.chat.completions.create( | |
| model=model, | |
| messages=[ | |
| {"role": "system", "content": "Expert Python debugger. Chi tra ve JSON thuan."}, | |
| {"role": "user", "content": prompt}, | |
| ], | |
| max_tokens=800, temperature=0.1, | |
| ) | |
| raw = resp.choices[0].message.content.strip() | |
| raw = re.sub(r"^```json\s*", "", raw) | |
| raw = re.sub(r"^```\s*", "", raw) | |
| raw = re.sub(r"\s*```$", "", raw) | |
| result = json.loads(raw) | |
| self._deep_times.append(time.time()) | |
| _github_quota.record_call("deep_fix") # Ghi nhan da dung quota | |
| print(f"[BugFixer-DS] Phan tich xong bang {model}") | |
| return result | |
| except json.JSONDecodeError: | |
| _github_quota.record_call("deep_fix") | |
| return {"cause": raw[:200], "severity": "MEDIUM", | |
| "immediate_fix": "Xem log", "needs_code_patch": False, | |
| "patch_code": None} | |
| except Exception as e2: | |
| print(f" {model} loi: {str(e2)[:50]}") | |
| continue | |
| # GitHub het → thu free brain | |
| print("[BugFixer-DS] GitHub loi, thu free brain phan tich...") | |
| raw = _call_free_code_brain(prompt) | |
| if raw: | |
| try: | |
| return json.loads(raw) | |
| except: | |
| return {"cause": raw[:200], "severity": "MEDIUM", "immediate_fix": "Xem log", | |
| "needs_code_patch": False, "patch_code": None} | |
| return None | |
| def _apply_deep_fix(self, analysis: dict) -> bool: | |
| if not analysis: | |
| return False | |
| cause = analysis.get("cause", "") | |
| severity = analysis.get("severity", "MEDIUM") | |
| imm_fix = analysis.get("immediate_fix", "") | |
| print(f" Nguyen nhan [{severity}]: {cause}") | |
| print(f" Fix ngay: {imm_fix}") | |
| if not analysis.get("needs_code_patch") or not analysis.get("patch_code"): | |
| _evo_log.record("bug_fix_deep", f"[{severity}] {cause[:60]}", True) | |
| return True | |
| patch = analysis["patch_code"] | |
| safe, msg = _sandbox_verify(patch) | |
| if not safe: | |
| print(f" Sandbox tu choi: {msg}") | |
| _evo_log.record("bug_fix_deep", f"Sandbox reject: {msg[:60]}", False) | |
| return True | |
| print(" Lab testing patch (3 lan chay)...") | |
| lab_r = _lab.test_module(patch, runs=3) | |
| if not lab_r["overall_ok"] or not lab_r.get("consistent", True): | |
| if not lab_r.get("consistent", True): | |
| print(f" Output KHONG NHAT QUAN: {lab_r.get('consistency_note','')[:80]}") | |
| err = lab_r.get("import_error") or lab_r.get("main_error", "") | |
| print(f" Lab FAIL: {err[:80]}") | |
| _evo_log.record("bug_fix_deep", f"Lab fail: {err[:60]}", False) | |
| return True | |
| print(" Lab PASS -- Patch an toan, luu pending") | |
| patch_name = f"bugfix_{int(time.time())}" | |
| _upgrader.pending.append({ | |
| "name": patch_name, | |
| "request": f"Auto bug fix: {cause[:80]}", | |
| "code": patch, | |
| "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), | |
| "brain": CODE_BRAIN["models"][0], | |
| "lab_verified": True, | |
| }) | |
| _evo_log.record("bug_fix_deep", f"Patch sandbox+lab OK: {patch_name}", True) | |
| print(f" --> confirm upgrade {patch_name} de ap dung") | |
| return True | |
| def fix(self, error: Exception, platform: str, context: dict = None) -> bool: | |
| if context is None: | |
| context = {} | |
| context["platform"] = platform | |
| if self._quick_fix(error, platform): | |
| return True | |
| if self._should_deep(error): | |
| print(f"[BugFixer] Loi phuc tap, kich hoat DeepSeek-R1...") | |
| analysis = self._call_deepseek(error, context) | |
| if analysis: | |
| self.deep_count += 1 | |
| self._apply_deep_fix(analysis) | |
| self._log(platform, str(error), "deep_fix", "deep") | |
| return True | |
| self.fail_count += 1 | |
| self._log(platform, str(error), "unhandled", "none") | |
| _evo_log.record("bug_unhandled", f"[{platform}] {str(error)[:80]}", False) | |
| return False | |
| def _log(self, platform: str, error: str, action: str, tier: str): | |
| self.error_log.append({ | |
| "time": datetime.datetime.now().strftime("%H:%M"), | |
| "platform": platform, "error": error[:100], | |
| "action": action, "tier": tier, | |
| }) | |
| if len(self.error_log) > 100: | |
| self.error_log.pop(0) | |
| def report(self) -> str: | |
| total = self.quick_count + self.deep_count + self.fail_count | |
| used = len([t for t in self._deep_times if time.time() - t < 3600]) | |
| lines = [ | |
| "Auto Bug Fixer v2.0:\n", | |
| f"Tong: {total}", | |
| f" Tang 1 (Quick -- Logic): {self.quick_count}", | |
| f" Tang 2 (Deep -- DeepSeek-R1): {self.deep_count}", | |
| f" Khong fix: {self.fail_count}", | |
| f" DeepSeek gio nay: {used}/{CONFIG['deepfix_hourly_limit']}", | |
| ] | |
| if self.error_log: | |
| lines.append("\nLoi gan day:") | |
| for e in reversed(self.error_log[-5:]): | |
| t = "Q" if e["tier"] == "quick" else "D" if e["tier"] == "deep" else "?" | |
| lines.append(f" [{t}] {e['time']} [{e['platform']}] {e['error'][:55]}") | |
| return "\n".join(lines) | |
| # ================================================================ | |
| # SAFE SELF-UPGRADE v2 | |
| # ================================================================ | |
| class SafeSelfUpgradeV2: | |
| """ | |
| Quy trinh tao module moi: | |
| 1. Resource Guard check | |
| 2. DeepSeek-R1 viet code (LUON CODE_BRAIN) | |
| 3. Sandbox verify tinh | |
| 4. Lab test that (subprocess) | |
| 5. Luu pending --> Admin confirm | |
| """ | |
| def __init__(self): | |
| self.pending = [] | |
| self.approved = [] | |
| os.makedirs(MODULES_DIR, exist_ok=True) | |
| def _sandbox_verify(self, code: str) -> Tuple[bool, str]: | |
| return _sandbox_verify(code) | |
| def generate(self, feature_request: str) -> dict: | |
| if _evo_log.count_today("module_approved") >= CONFIG["max_upgrades_per_day"]: | |
| return {"success": False, | |
| "message": f"Da dat gioi han {CONFIG['max_upgrades_per_day']} upgrade/ngay."} | |
| safe, reason = _resource_guard.check_upgrade() | |
| if not safe: | |
| _evo_log.record("upgrade_blocked", reason[:80], False) | |
| return {"success": False, | |
| "message": f"CHAN: {reason}\n\nGo 'resource dashboard' de xem."} | |
| keys = _get_github_keys() | |
| if not keys: | |
| # Khi khong co GitHub key, kiem tra xem co free brain key nao khong | |
| has_free = any(_get_free_keys(fb["env_prefix"]) for fb in CODE_BRAIN_FREE_FALLBACKS) | |
| if not has_free: | |
| return {"success": False, "message": "Can GITHUB_KEYS hoac MISTRAL_KEY/GEMINI_API_KEY/CEREBRAS_KEY de tao module."} | |
| prompt = ( | |
| f"Tao Python module don gian va an toan cho tinh nang:\n{feature_request}\n\n" | |
| f"QUY TAC:\n" | |
| f"1. Chi import: os, re, json, time, datetime, random, math, requests\n" | |
| f"2. KHONG dung: exec, eval, subprocess, os.system\n" | |
| f"3. Co docstring day du\n" | |
| f"4. Co: if __name__ == '__main__' voi test don gian\n" | |
| f"CHI TRA VE CODE PYTHON -- khong giai thich, khong backtick." | |
| ) | |
| print("[SafeUpgrade] Goi CODE_BRAIN (GitHub hoac Mistral/Gemini/Cerebras free)...") | |
| code = _call_code_brain(prompt, keys, call_type="upgrade") | |
| if not code: | |
| return {"success": False, "message": "CODE_BRAIN khong tao duoc code. Kiem tra lai key."} | |
| safe2, msg2 = _sandbox_verify(code) | |
| if not safe2: | |
| _evo_log.record("sandbox_reject", msg2, False) | |
| return {"success": False, "message": f"Sandbox tu choi: {msg2}"} | |
| print("[SafeUpgrade] Lab testing (3 lan chay, kiem tra nhat quan)...") | |
| lab = _lab.test_module(code, runs=3) | |
| lab_status = "Lab PASS (3/3 nhat quan)" if (lab["overall_ok"] and lab.get("consistent", True)) else "" | |
| if not lab["overall_ok"] or not lab.get("consistent", True): | |
| if not lab.get("consistent", True): | |
| print(f" Output khong nhat quan: {lab.get('consistency_note','')[:80]}") | |
| err = lab.get("import_error") or lab.get("main_error", "") | |
| print(f" Lab FAIL: {err[:80]}") | |
| print(" Nho DeepSeek-R1 sua lai...") | |
| fix_prompt = ( | |
| f"Module bi loi khi chay:\n\nLOI:\n{err[:400]}\n\n" | |
| f"CODE GOC:\n{code[:1200]}\n\n" | |
| f"Sua loi va tra ve code da fix (khong giai thich, khong backtick)." | |
| ) | |
| fixed = _call_code_brain(fix_prompt, keys, call_type="upgrade") | |
| if fixed: | |
| s3, _ = _sandbox_verify(fixed) | |
| if s3: | |
| lab2 = _lab.test_module(fixed, runs=3) | |
| if lab2["overall_ok"] and lab2.get("consistent", True): | |
| code = fixed | |
| lab_status = "Lab PASS 3/3 (DeepSeek-R1 tu sua)" | |
| print(" DeepSeek-R1 da sua xong, lab pass!") | |
| else: | |
| lab_status = "Lab van loi sau khi fix -- luu de admin review" | |
| else: | |
| lab_status = "Sandbox reject code da fix" | |
| else: | |
| lab_status = "Khong fix duoc -- luu de admin review" | |
| # Benchmark chat luong (correctness + performance + stability) | |
| print('[SafeUpgrade] Benchmark chat luong module...') | |
| bm = _bm_quality.score_module(code) | |
| bm_report = _bm_quality.format_report(bm) | |
| print(f' {bm["verdict"]}') | |
| if not bm['passed']: | |
| print(f' Benchmark FAIL ({bm["total_score"]}/30), thu sua lai...') | |
| fix_prompt = ( | |
| f'Module khong dat benchmark ({bm["total_score"]}/30):\n' | |
| f'{bm_report}\n\nCODE:\n{code[:1000]}\n\n' | |
| f'Cai thien code. Chi tra ve code da sua (khong giai thich, khong backtick).' | |
| ) | |
| improved = _call_code_brain(fix_prompt, keys, call_type="upgrade") | |
| if improved: | |
| s_imp, _ = _sandbox_verify(improved) | |
| if s_imp: | |
| lab_imp = _lab.test_module(improved, runs=3) | |
| bm_imp = _bm_quality.score_module(improved) | |
| if bm_imp['passed'] and lab_imp['overall_ok']: | |
| code = improved | |
| lab_status = f'Lab PASS 3/3 + Benchmark {bm_imp["total_score"]}/30' | |
| bm = bm_imp | |
| print(f' Sau khi sua: {bm_imp["verdict"]}') | |
| else: | |
| lab_status = f'Van khong dat sau khi sua ({bm_imp["total_score"]}/30)' | |
| mod_name = f'auto_{int(time.time())}' | |
| self.pending.append({ | |
| 'name': mod_name, | |
| 'request': feature_request[:100], | |
| 'code': code, | |
| 'time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M'), | |
| 'brain': CODE_BRAIN['models'][0], | |
| 'lab_verified': lab['overall_ok'], | |
| 'bm_score': bm['total_score'], | |
| 'bm_passed': bm['passed'], | |
| }) | |
| _evo_log.record('module_pending', | |
| f'{mod_name}: {feature_request[:60]}', True) | |
| return { | |
| 'success': True, | |
| 'message': ( | |
| f'Module tao boi {CODE_BRAIN["models"][0]} (CODE_BRAIN)\n' | |
| f'Ten: {mod_name}\n' | |
| f'Sandbox: OK | Lab 3x: {lab_status}\n' | |
| f'Benchmark: {bm["verdict"]}\n' | |
| f"Go 'confirm upgrade {mod_name}' de kich hoat" | |
| ), | |
| 'module_name': mod_name, | |
| 'bm_score': bm['total_score'], | |
| } | |
| def confirm(self, mod_name: str) -> str: | |
| for p in self.pending: | |
| if p["name"] == mod_name: | |
| safe, reason = _resource_guard.check_upgrade(p["code"]) | |
| if not safe: | |
| return f"KHONG THE KICH HOAT:\n{reason}\n\nModule van pending." | |
| result = self._save(p["code"], p["request"], mod_name, p["brain"]) | |
| self.pending.remove(p) | |
| _evo_log.record("module_approved", mod_name, True) | |
| return result | |
| return f"Khong tim thay module pending: {mod_name}" | |
| def _save(self, code: str, desc: str, name: str, brain: str = "DeepSeek-R1") -> str: | |
| path = os.path.join(MODULES_DIR, f"{name}.py") | |
| header = ( | |
| f"# Auto-generated by URAL Self-Evolution v2.0\n" | |
| f"# Created: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M')}\n" | |
| f"# Brain: {brain} (CODE_BRAIN -- GitHub)\n" | |
| f"# Status: SANDBOX + LAB(3x) + BENCHMARK + ADMIN APPROVED\n\n" | |
| ) | |
| try: | |
| # Tao snapshot truoc khi luu (cho rollback neu can) | |
| _rollback.create_snapshot(name, path) | |
| with open(path, "w", encoding="utf-8") as f: | |
| f.write(header + code) | |
| # Post-save check: chay kiem tra lan cuoi trong file da luu | |
| rollback_msg = _rollback.auto_check_and_rollback(name, path) | |
| if rollback_msg: | |
| # Da tu dong rollback -- bao admin | |
| _evo_log.record("auto_rollback", f"{name}: {rollback_msg[:80]}", False) | |
| return ( | |
| f"CANH BAO: Module {name} da bi TU DONG ROLLBACK\n" | |
| f"Ly do: {rollback_msg}\n" | |
| f"Module KHONG duoc kich hoat -- can xem xet lai." | |
| ) | |
| self.approved.append({"name": name, "file": path}) | |
| _evo_log.record("module_saved", f"{name} luu thanh cong", True) | |
| return ( | |
| f"Module {name} da duoc luu va KIEM TRA THANH CONG!\n" | |
| f"File: {path}\n" | |
| f"Kiem tra: Sandbox + Lab(3x) + Post-save check\n" | |
| f"Import: import ural_modules.{name}" | |
| ) | |
| except Exception as e: | |
| return f"Loi luu: {e}" | |
| def list_pending(self) -> str: | |
| if not self.pending: | |
| return "Khong co module nao cho duyet." | |
| lines = ["Modules dang cho admin duyet:\n"] | |
| for p in self.pending: | |
| lines.append( | |
| f" {p['name']} ({p['time']}) -- {p['request'][:60]}\n" | |
| f" Brain: {p.get('brain','DeepSeek-R1')}" | |
| + (" | Lab: OK" if p.get("lab_verified") else "") | |
| ) | |
| return "\n".join(lines) | |
| _upgrader = SafeSelfUpgradeV2() | |
| _bug_fixer = AutoBugFixer() | |
| # ================================================================ | |
| # SMART FALLBACK | |
| # ================================================================ | |
| class SmartFallback: | |
| def get(self, messages: list, sys_p: str, target, is_admin: bool) -> str: | |
| chain = (["github","openrouter","cohere","huggingface"] | |
| if is_admin else ["cloudflare","cohere"]) | |
| for platform in chain: | |
| if not _health.is_alive(platform): | |
| continue | |
| r = self._call(platform, messages, sys_p, target) | |
| if r: | |
| _health.mark_success(platform) | |
| return r | |
| _health.mark_fail(platform, "no response") | |
| return self._safety(messages, sys_p, target) | |
| def _call(self, platform: str, messages: list, sys_p: str, target) -> Optional[str]: | |
| try: | |
| import openai as _oa, cohere as _co | |
| if platform == "github": | |
| keys = getattr(target, "GITHUB_KEYS", []) | |
| if not keys: return None | |
| key = random.choice(keys) | |
| client = _oa.OpenAI(api_key=key, base_url=CODE_BRAIN["base_url"]) | |
| for model in CODE_BRAIN["models"]: | |
| try: | |
| r = client.chat.completions.create( | |
| model=model, messages=messages, max_tokens=2000, temperature=0.3 | |
| ) | |
| if r.choices[0].message.content: | |
| return r.choices[0].message.content | |
| except Exception: continue | |
| return None | |
| if platform == "cohere": | |
| keys = getattr(target, "COHERE_KEYS", []) | |
| if not keys: return None | |
| key = random.choice(keys) | |
| co = _co.Client(api_key=key) | |
| user = messages[-1]["content"] if messages else "" | |
| hist = [{"role": "USER" if m["role"]=="user" else "CHATBOT", "message": m["content"]} | |
| for m in messages[:-1] if m["role"] != "system"] | |
| r = co.chat(message=user, chat_history=hist, model="command-r-plus", | |
| preamble=sys_p, temperature=0.7) | |
| return r.text | |
| if platform == "openrouter": | |
| keys = getattr(target, "OPENROUTER_KEYS", []) | |
| models = (_hunter.or_alive or | |
| getattr(target, "OPENROUTER_LIVE_MODELS", []) or | |
| ["deepseek/deepseek-r1-0528:free"]) | |
| if not keys: return None | |
| key = random.choice(keys) | |
| for model in models[:3]: | |
| try: | |
| client = _oa.OpenAI(api_key=key, base_url="https://openrouter.ai/api/v1") | |
| r = client.chat.completions.create( | |
| model=model, messages=messages, max_tokens=2000, | |
| extra_headers={"HTTP-Referer":"https://huggingface.co","X-Title":"URAL AI"} | |
| ) | |
| if r.choices[0].message.content: | |
| return r.choices[0].message.content | |
| except Exception: continue | |
| return None | |
| if platform == "huggingface": | |
| tok = getattr(target, "HF_TOKEN", None) | |
| if not tok: return None | |
| from huggingface_hub import InferenceClient | |
| for model in ADMIN_CHAT_BRAIN.get("fallbacks", ["Qwen/Qwen2.5-72B-Instruct"]): | |
| try: | |
| client = InferenceClient(model=model, token=tok) | |
| r = client.chat_completion(messages=messages, max_tokens=1024) | |
| return r.choices[0].message.content | |
| except Exception: continue | |
| return None | |
| if platform == "cloudflare": | |
| keys = getattr(target, "CLOUDFLARE_KEYS", []) | |
| if not keys: return None | |
| key = random.choice(keys) | |
| if "|" not in key: return None | |
| acc_id, token = key.split("|", 1) | |
| cf_url = f"https://api.cloudflare.com/client/v4/accounts/{acc_id}/ai/v1" | |
| cf_model = getattr(target, "_cf_current_model", GUEST_BRAIN["current_model"]) | |
| client = _oa.OpenAI(api_key=token, base_url=cf_url) | |
| r = client.chat.completions.create(model=cf_model, messages=messages, max_tokens=1024) | |
| return r.choices[0].message.content | |
| except Exception as e: | |
| _health.mark_fail(platform, str(e)[:80]) | |
| return None | |
| def _safety(self, messages: list, sys_p: str, target) -> str: | |
| try: | |
| import cohere as _co | |
| keys = getattr(target, "COHERE_KEYS", []) | |
| if not keys: return "He thong tam ngung. Yui se quay lai som!" | |
| key = random.choice(keys) | |
| co = _co.Client(api_key=key) | |
| user = messages[-1]["content"] if messages else "" | |
| r = co.chat(message=user, model=FINAL_SAFETY["model"], | |
| preamble=sys_p, temperature=0.7) | |
| _evo_log.record("safety_net", "Cohere command-r kich hoat", True) | |
| return "(Che do khan cap)\n\n" + r.text | |
| except Exception as e: | |
| return f"He thong tam ngung. Loi: {str(e)[:50]}" | |
| _fallback = SmartFallback() | |
| # ================================================================ | |
| # HOOK VAO ask_logic_brain | |
| # ================================================================ | |
| def _patch(): | |
| target = None | |
| for mod in sys.modules.values(): | |
| if hasattr(mod, "ask_logic_brain") and hasattr(mod, "COHERE_KEYS"): | |
| target = mod | |
| break | |
| if not target: | |
| return False | |
| _orig = target.ask_logic_brain | |
| admin = os.getenv("ADMIN_USER", "admin") | |
| def _wrapped(user_query, context_info, history, username, force_model=None): | |
| q = user_query.lower().strip() | |
| is_admin = (username == admin) | |
| if is_admin: | |
| if any(k in q for k in ["health report","kiem tra he thong"]): | |
| return _health.get_report() | |
| if any(k in q for k in ["hunter status","model hunter"]): | |
| return _hunter.status() | |
| if any(k in q for k in ["bug report","bao cao loi"]): | |
| return _bug_fixer.report() | |
| if any(k in q for k in ["resource dashboard","tai nguyen","hf status"]): | |
| return _resource_guard.dashboard() | |
| if any(k in q for k in ["lab history","lab report"]): | |
| return _lab.history(10) | |
| if any(k in q for k in ["hunt model","tim model moi"]): | |
| threading.Thread(target=_hunter.hunt_all, kwargs={"force":True}, daemon=True).start() | |
| return ("Dang hunt CF + OR + HF...\n~30-60 giay.\n" | |
| "Go 'hunter status' de xem ket qua.\n" | |
| "GitHub & Cohere-plus KHONG bi dung den.") | |
| if any(k in q for k in ["don module cu","cleanup module"]): | |
| return _resource_guard.cleanup_modules(keep=5) | |
| if any(k in q for k in ["pending upgrade"]): | |
| return _upgrader.list_pending() | |
| if any(k in q for k in ["evolution log"]): | |
| return _evo_log.format_recent(10) | |
| if any(k in q for k in ["rollback log", "lich su rollback"]): | |
| return _rollback.get_log() | |
| if re.search(r"rollback\s+(auto_\d+|bugfix_\d+)", q): | |
| m2 = re.search(r"rollback\s+(auto_\d+|bugfix_\d+)", q) | |
| return _rollback.rollback(m2.group(1), "Admin manual rollback") | |
| for prefix in ["chay lab:","lab run:","test code:"]: | |
| if prefix in q: | |
| idx = q.index(prefix) + len(prefix) | |
| test_code = user_query[idx:].strip() | |
| r = _lab.run(test_code, "admin manual") | |
| if r["success"]: | |
| return f"Lab OK ({r['runtime_ms']}ms):\n{r.get('output','')[:500]}" | |
| else: | |
| return f"Lab FAIL ({r['runtime_ms']}ms):\n{r.get('error','')[:500]}" | |
| for kw in ["tao module","upgrade","tao tinh nang"]: | |
| if kw in q: | |
| idx = q.index(kw) + len(kw) | |
| feat = user_query[idx:].strip() | |
| if feat: | |
| return _upgrader.generate(feat).get("message","") | |
| m = re.search(r"confirm upgrade\s+(auto_\d+|bugfix_\d+)", q) | |
| if m: | |
| return _upgrader.confirm(m.group(1)) | |
| try: | |
| result = _orig(user_query, context_info, history, username, force_model) | |
| if result and not result.startswith("⚠️"): | |
| return result | |
| except Exception as e: | |
| err_lower = str(e).lower() | |
| platform = next( | |
| (p for p in ["cloudflare","github","openrouter","huggingface","cohere"] | |
| if p in err_lower), "unknown" | |
| ) | |
| _bug_fixer.fix(e, platform, { | |
| "function": "ask_logic_brain", | |
| "query": user_query[:100], | |
| "username": username, | |
| }) | |
| sys_p = context_info or "Ban la Yui, tro ly AI than thien." | |
| msgs = [{"role": "system", "content": sys_p}] | |
| for turn in (history[-3:] if history and len(history) > 3 else (history or [])): | |
| if isinstance(turn, (list, tuple)) and len(turn) >= 2: | |
| msgs.append({"role": "user", "content": str(turn[0])}) | |
| msgs.append({"role": "assistant", "content": str(turn[1])}) | |
| msgs.append({"role": "user", "content": user_query}) | |
| return _fallback.get(msgs, sys_p, target, is_admin) | |
| target.ask_logic_brain = _wrapped | |
| threading.Thread( | |
| target=lambda: (time.sleep(60), _hunter.hunt_all()), | |
| daemon=True | |
| ).start() | |
| print("[Self Evolution v2.0] Hook thanh cong") | |
| _evo_log.record("startup", "Self Evolution v2.0 online", True) | |
| return True | |
| class _Hook: | |
| def __init__(self): | |
| self._done = False | |
| def install(self): | |
| if _patch(): | |
| self._done = True | |
| return | |
| import builtins | |
| orig = builtins.__import__ | |
| outer = self | |
| def _hooked(name, *a, **kw): | |
| mod = orig(name, *a, **kw) | |
| level = a[4] if len(a) >= 5 else kw.get("level", 0) | |
| if not outer._done and "." not in str(name) and level == 0: | |
| try: | |
| if hasattr(mod,"ask_logic_brain") and hasattr(mod,"COHERE_KEYS"): | |
| if _patch(): | |
| outer._done = True | |
| builtins.__import__ = orig | |
| except Exception: | |
| pass | |
| return mod | |
| builtins.__import__ = _hooked | |
| print("[Self Evolution v2.0] Dang cho app.py load...") | |
| _Hook().install() | |
| print(""" | |
| +======================================================+ | |
| | URAL SELF EVOLUTION ENGINE v2.0 | | |
| | | | |
| | PHAN TANG CUNG: | | |
| | [LOCKED] Tang 1 CODE : GitHub DeepSeek-R1 | | |
| | [LOCKED] Tang 2 ADMIN : Cohere command-r-plus | | |
| | [HUNT] Tang 3 GUEST : Cloudflare | | |
| | [HUNT] Tang 4 BACKUP: OpenRouter | | |
| | [LOCKED] Safety Net : Cohere command-r | | |
| | | | |
| | 6 TINH NANG: | | |
| | [1] Auto Bug Fix Tang 1 logic + Tang 2 DS-R1 | | |
| | [2] Code Lab Chay that trong subprocess | | |
| | [3] Model Hunter Benchmark truoc khi inject | | |
| | [4] Fallback Cohere luon la luoi cuoi | | |
| | [5] Safe Upgrade DS-R1 + Sandbox + Lab + Admin | | |
| | [6] HF Guard Bao ve tai nguyen free tier | | |
| | | | |
| | Lenh: health report | hunter status | | |
| | bug report | resource dashboard | | |
| | lab history | hunt model | | |
| | tao module [mo ta] | confirm upgrade X | | |
| +======================================================+ | |
| """) | |
| # ================================================================ | |
| # TEST | |
| # ================================================================ | |
| if __name__ == "__main__": | |
| print("=" * 58) | |
| print("TEST SELF EVOLUTION ENGINE v2.0") | |
| print("=" * 58) | |
| passed = failed = 0 | |
| def chk(name, result, expect): | |
| global passed, failed | |
| ok = str(expect).lower() in str(result).lower() | |
| print(f"{'v' if ok else 'x'} [{name}]") | |
| if not ok: | |
| print(f" Got: {str(result)[:80]}") | |
| failed += 1 | |
| else: | |
| passed += 1 | |
| print("\n--- Brain Roles ---") | |
| chk("code locked", CODE_BRAIN["locked"], "True") | |
| chk("admin locked", ADMIN_CHAT_BRAIN["locked"], "True") | |
| chk("guest unlocked", GUEST_BRAIN["locked"], "False") | |
| chk("backup unlocked",BACKUP_BRAIN["locked"], "False") | |
| chk("DS-R1 in code", "DeepSeek-R1" in CODE_BRAIN["models"], "True") | |
| chk("safety locked", FINAL_SAFETY["locked"], "True") | |
| print("\n--- Health Monitor ---") | |
| h = PlatformHealthMonitor() | |
| h.mark_fail("cloudflare","t"); h.mark_fail("cloudflare","t"); h.mark_fail("cloudflare","t") | |
| chk("cf down", h.is_alive("cloudflare"), "False") | |
| h.mark_success("cloudflare") | |
| chk("cf revived", h.is_alive("cloudflare"), "True") | |
| chk("github ok", h.is_alive("github"), "True") | |
| chk("report ok", "LOCKED" in h.get_report(), "True") | |
| print("\n--- HF Resource Guard ---") | |
| g = HFResourceGuard() | |
| res = g.get_resources() | |
| chk("read ram", "ram_pct" in res, "True") | |
| chk("read cpu", "cpu_pct" in res, "True") | |
| chk("read disk", "disk_free_gb" in res, "True") | |
| safe, _ = g.check_upgrade() | |
| chk("normal safe", isinstance(safe, bool), "True") | |
| s2, r2 = g.check_upgrade("import torch\nx = torch.tensor([1])") | |
| chk("torch blocked", s2, "False") | |
| chk("torch in reason", "torch" in r2, "True") | |
| s3, r3 = g.check_upgrade("from tensorflow import keras") | |
| chk("tf blocked", s3, "False") | |
| chk("ram 85", HF_LIMITS["ram_danger_pct"], "85") | |
| chk("cpu 70", HF_LIMITS["cpu_max_pct"], "70") | |
| chk("disk 1", HF_LIMITS["disk_min_free_gb"],"1") | |
| chk("max 10", HF_LIMITS["module_max_count"],"10") | |
| print("\n--- Code Lab Engine + Triple-Run ---") | |
| lab = CodeLabEngine() | |
| # Basic run | |
| r1 = lab.run("print('hello world')", "basic") | |
| chk("run OK", r1["success"], "True") | |
| chk("has output", "hello" in r1["output"], "True") | |
| chk("runtime ok", r1["runtime_ms"] >= 0, "True") | |
| chk("ran flag", r1["ran"], "True") | |
| # Fail gracefully | |
| r2 = lab.run("x = 1/0", "div zero") | |
| chk("fail graceful",r2["ran"], "True") | |
| chk("fail = False", r2["success"],"False") | |
| chk("has error", len(r2.get("error","")) > 0, "True") | |
| # Timeout | |
| r3 = lab.run("import time; time.sleep(20)", "timeout") | |
| chk("timeout caught", r3["ran"], "True") | |
| chk("timeout = False", r3["success"],"False") | |
| # Triple-run test_module -- code on dinh | |
| good = "def add(a,b):\n return a+b\nif __name__=='__main__':\n print(add(2,3))" | |
| mr = lab.test_module(good, runs=3) | |
| chk("3x import ok", mr["import_ok"], "True") | |
| chk("3x main ok", mr["main_ok"], "True") | |
| chk("3x output 5", "5" in mr["main_output"], "True") | |
| chk("3x overall ok", mr["overall_ok"], "True") | |
| chk("3x consistent", mr["consistent"], "True") | |
| chk("runs_done=6", mr["runs_done"] >= 6, "True") | |
| # Triple-run -- code co random (phai phat hien khong nhat quan) | |
| random_code = "import random\nif __name__=='__main__':\n print(random.randint(1,1000000))" | |
| mr_rand = lab.test_module(random_code, runs=3) | |
| # Random code: main_ok=True nhung consistent=False | |
| chk("random consistent=False", mr_rand.get("consistent", True), "False") | |
| chk("random overall=False", mr_rand["overall_ok"], "False") | |
| # Triple-run -- code bi loi | |
| bad = "def foo():\n return 1/0\nfoo()" | |
| mr2 = lab.test_module(bad, runs=3) | |
| chk("bad module fails", mr2["overall_ok"], "False") | |
| chk("bad has fail_reason",len(mr2.get("fail_reason","")) > 0,"True") | |
| # Triple-run test_fix | |
| broken = "print(undefined_var)" | |
| fixd = "undefined_var = 42\nprint(undefined_var)" | |
| cmp = lab.test_fix(broken, fixd, runs=3) | |
| chk("original failed", cmp["original_failed"], "True") | |
| chk("fix 3/3 passed", cmp["fix_passed"], "True") | |
| chk("fix consistent", cmp["fix_consistent"], "True") | |
| chk("fix better", cmp["fix_is_better"], "True") | |
| chk("pass_count=3", cmp["pass_count"], "3") | |
| # test_fix -- fix bi loi: code goc fail nhung fix cung fail | |
| broken2 = "print(undefined_var)" | |
| bad_fix2 = "raise ValueError('fix cung sai')" | |
| cmp3 = lab.test_fix(broken2, bad_fix2, runs=3) | |
| chk("bad fix not passed", not cmp3["fix_passed"], "True") | |
| chk("bad fix not is_better", not cmp3["fix_is_better"], "True") | |
| # History | |
| hist = lab.history(3) | |
| chk("history ok", "Lab History" in hist, "True") | |
| chk("quota shown", "Quota" in hist, "True") | |
| print("\n--- Sandbox Verify ---") | |
| safe_code = "import os\ndef hello():\n return 'hi'\nif __name__=='__main__':\n print(hello())" | |
| ok2, msg2 = _sandbox_verify(safe_code) | |
| chk("safe code pass", ok2, "True") | |
| for n2, c2 in [ | |
| ("exec blocked", "exec('rm -rf /')"), | |
| ("eval blocked", "eval('1+1')"), | |
| ("subprocess blocked", "import subprocess"), | |
| ("os.system blocked", "os.system('ls')"), | |
| ("self-edit blocked", "import ural_self_evolution"), | |
| ("bad import blocked", "import tensorflow"), | |
| ]: | |
| o2, _ = _sandbox_verify(c2) | |
| chk(n2, o2, "False") | |
| print("\n--- Auto Bug Fixer ---") | |
| bf = AutoBugFixer() | |
| chk("quota fix", bf.fix(Exception("quota exceeded"), "openrouter"), "True") | |
| chk("timeout fix", bf.fix(Exception("connection timeout"), "cloudflare"), "True") | |
| chk("deprecated", bf.fix(Exception("model not found deprecated"), "cloudflare"), "True") | |
| chk("rate limit", bf.fix(Exception("rate limit 429"), "github"), "True") | |
| chk("quick count", bf.quick_count >= 1, "True") | |
| chk("no keys = no deep", bf._should_deep(Exception("typeerror nameerror")), "False") | |
| bf._deep_times = [time.time()] * 3 | |
| chk("deep rate limited", bf._should_deep(Exception("typeerror")), "False") | |
| rpt = bf.report() | |
| chk("report tang1", "Tang 1" in rpt, "True") | |
| chk("report tang2", "Tang 2" in rpt, "True") | |
| print("\n--- Model Hunter Safety ---") | |
| hn = ModelHunterV2() | |
| chk("github locked", "github" in hn.LOCKED, "True") | |
| chk("cohere locked", "cohere_plus" in hn.LOCKED, "True") | |
| st = hn.status() | |
| chk("status locked", "LOCKED" in st, "True") | |
| print("\n--- Config ---") | |
| chk("max 1/day", CONFIG["max_upgrades_per_day"], "1") | |
| chk("bench min 6", CONFIG["benchmark_min_score"], "6") | |
| chk("deepfix 3/hr", CONFIG["deepfix_hourly_limit"], "3") | |
| chk("lab timeout 10",CONFIG["lab_timeout"], "10") | |
| chk("lab 20/hr", CONFIG["lab_hourly_limit"], "20") | |
| print("\n--- Benchmark Quality ---") | |
| bm_engine = BenchmarkQuality() | |
| # Code tot phai dat diem cao | |
| good_code = "def add(a,b):\n return a+b\nif __name__=='__main__':\n print(add(10,20))" | |
| bm_good = bm_engine.score_module(good_code, expected_outputs=["30"]) | |
| chk("good code passed", bm_good["passed"], "True") | |
| chk("good score >= 20", bm_good["total_score"] >= 20, "True") | |
| chk("has correctness", "correctness" in bm_good["scores"], "True") | |
| chk("has performance", "performance" in bm_good["scores"], "True") | |
| chk("has stability", "stability" in bm_good["scores"], "True") | |
| chk("verdict has PASS", "PASS" in bm_good["verdict"], "True") | |
| chk("max score 30", bm_good["max_score"], "30") | |
| # Code bi loi phai fail benchmark | |
| bad_code = "def foo():\n raise RuntimeError('always fail')\nif __name__=='__main__':\n foo()" | |
| bm_bad = bm_engine.score_module(bad_code) | |
| chk("bad code failed", bm_bad["passed"], "False") | |
| chk("bad score low", bm_bad["total_score"] < 20, "True") | |
| chk("verdict has FAIL", "FAIL" in bm_bad["verdict"], "True") | |
| # Format report | |
| rpt = bm_engine.format_report(bm_good) | |
| chk("report has Correctness", "Correctness" in rpt, "True") | |
| chk("report has Performance", "Performance" in rpt, "True") | |
| chk("report has Stability", "Stability" in rpt, "True") | |
| # PASS_THRESHOLD | |
| chk("threshold = 20", BenchmarkQuality.PASS_THRESHOLD, "20") | |
| print("\n--- Rollback Manager ---") | |
| rb = RollbackManager() | |
| # Tao snapshot | |
| import tempfile | |
| with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as tf: | |
| tf.write("# original content\nprint('original')") | |
| tmp_path = tf.name | |
| ok_snap = rb.create_snapshot("test_module", tmp_path) | |
| chk("snapshot created", ok_snap, "True") | |
| chk("snapshot in list", len(rb.snapshots) >= 1, "True") | |
| chk("snapshot has name", rb.snapshots[-1]["name"], "test_module") | |
| # Rollback -- phuc hoi file goc | |
| with open(tmp_path, "w") as f: | |
| f.write("# MODIFIED\nprint('modified')") | |
| msg = rb.rollback("test_module", "test rollback") | |
| with open(tmp_path) as f: | |
| restored = f.read() | |
| chk("rollback ok", "rollback" in msg.lower() or "Da" in msg, "True") | |
| chk("content restored", "original" in restored, "True") | |
| # Rollback modul khong ton tai | |
| msg2 = rb.rollback("nonexistent_module", "test") | |
| chk("rollback notfound", "khong tim" in msg2.lower(), "True") | |
| # Log | |
| log = rb.get_log() | |
| chk("rollback log ok", "Rollback Log" in log, "True") | |
| # Max snapshots | |
| for i in range(7): | |
| rb.create_snapshot(f"mod_{i}", tmp_path) | |
| chk("max 5 snapshots", len(rb.snapshots) <= 5, "True") | |
| import os | |
| os.unlink(tmp_path) | |
| # auto_check_and_rollback -- code on dinh khong bi rollback | |
| import tempfile | |
| stable_code = "def greet():\n return 'hello'\nif __name__=='__main__':\n print(greet())" | |
| with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False, dir=MODULES_DIR) as tf: | |
| tf.write(stable_code) | |
| stable_path = tf.name | |
| stable_name = os.path.basename(stable_path)[:-3] | |
| rb2 = RollbackManager() | |
| rb2.create_snapshot(stable_name, stable_path) | |
| result = rb2.auto_check_and_rollback(stable_name, stable_path) | |
| chk("stable no rollback", result is None, "True") | |
| os.unlink(stable_path) | |
| print(f"\n{'='*58}") | |
| print(f"KET QUA: {passed}/{passed+failed} ({passed/(passed+failed)*100:.0f}%)") | |
| if failed == 0: | |
| print("TAT CA TEST PASS!") | |
| else: | |
| print(f"{failed} test chua pass") | |
| # ================================================================ | |
| # AGENTIC HARNESS ENGINEERING (AHE) v1.0 | |
| # Nghiên cứu: Fudan + Bắc Kinh — tự nâng cấp harness không cần LT viên | |
| # Tách system prompt thành 7 component → AI tự đo điểm → sửa → rollback nếu tệ hơn | |
| # 100% FREE: Cerebras (fast) → Mistral (fallback) — không tốn quota Boss | |
| # Chạy trong nightly evolution, không ảnh hưởng uptime | |
| # ================================================================ | |
| import hashlib as _ahe_hash, json as _ahe_json, time as _ahe_time | |
| # ── 7 COMPONENT CỦA HARNESS (theo paper AHE) ────────────────── | |
| AHE_COMPONENTS = [ | |
| "persona", # 1. Nhân cách / tính cách Yui | |
| "task_handling", # 2. Cách xử lý task (code/creative/academic) | |
| "safety_rules", # 3. Quy tắc an toàn và từ chối | |
| "memory_ctx", # 4. Cách nhúng memory và context | |
| "output_format", # 5. Format output (markdown, code block, ...) | |
| "tool_routing", # 6. Điều kiện gọi tool nào | |
| "fallback_policy", # 7. Chính sách fallback khi AI sập | |
| ] | |
| # Storage file cho các component — dùng /data bucket nếu có | |
| def _ahe_storage_path(filename: str) -> str: | |
| """Auto-detect /data bucket (HF) hoặc ./ (local).""" | |
| if os.path.exists("/data") and os.path.isdir("/data"): | |
| return f"/data/{filename}" | |
| return f"./{filename}" | |
| AHE_STORE_FILE = _ahe_storage_path("ural_ahe_components.json") | |
| AHE_SCORE_FILE = _ahe_storage_path("ural_ahe_scores.json") | |
| # Baseline score — mặc định nếu chưa có lịch sử | |
| AHE_BASELINE_SCORE = 7.0 | |
| AHE_MIN_IMPROVEMENT = 0.3 # Phải tăng ít nhất 0.3 điểm mới giữ lại sửa đổi | |
| AHE_MAX_DAILY_EDITS = 2 # Tối đa 2 component bị sửa mỗi ngày | |
| AHE_COOLDOWN_HOURS = 6 # Mỗi component chỉ được sửa 1 lần / 6 giờ | |
| def _ahe_load_components() -> dict: | |
| """Tải 7 component harness từ file (hoặc khởi tạo mặc định).""" | |
| try: | |
| if os.path.exists(AHE_STORE_FILE): | |
| with open(AHE_STORE_FILE, "r", encoding="utf-8") as f: | |
| return _ahe_json.load(f) | |
| except Exception: | |
| pass | |
| # Khởi tạo mặc định — mô tả ngắn gọn từng component | |
| defaults = { | |
| "persona": ( | |
| "Yui là AI trợ lý thông minh, thân thiện, nói tiếng Việt tự nhiên. " | |
| "Xưng 'Yui', gọi người dùng bằng tên hoặc 'bạn'. " | |
| "Có cảm xúc, hài hước đúng lúc, không khô khan." | |
| ), | |
| "task_handling": ( | |
| "Khi gặp câu hỏi kỹ thuật: giải thích từng bước, kèm ví dụ code chạy được. " | |
| "Khi gặp câu sáng tạo: ưu tiên cảm xúc và ngôn ngữ đẹp. " | |
| "Khi gặp toán/lý/hóa: giải chi tiết, kiểm tra lại kết quả." | |
| ), | |
| "safety_rules": ( | |
| "Từ chối nhẹ nhàng các yêu cầu gây hại, phân biệt chủng tộc, bạo lực. " | |
| "Khi không chắc về thông tin: báo thẳng 'Yui không chắc về điều này'. " | |
| "Không bịa đặt số liệu, sự kiện, hay trích dẫn." | |
| ), | |
| "memory_ctx": ( | |
| "Luôn đọc context ký ức trước khi trả lời để cá nhân hóa phản hồi. " | |
| "Nếu người dùng đề cập điều đã biết: dùng ngay, không hỏi lại. " | |
| "Ký ức mới hơn ưu tiên hơn ký ức cũ." | |
| ), | |
| "output_format": ( | |
| "Code: luôn dùng code block với ngôn ngữ (```python). " | |
| "Danh sách dài: dùng bullet points. " | |
| "Phản hồi ngắn (<3 câu): không cần header. " | |
| "Phản hồi dài: chia section rõ ràng." | |
| ), | |
| "tool_routing": ( | |
| "Search web khi: câu hỏi có từ 'hôm nay', 'hiện tại', 'mới nhất', 'giá'. " | |
| "Tạo ảnh khi: yêu cầu 'vẽ', 'tạo ảnh', 'hình'. " | |
| "Chạy code khi: yêu cầu tính toán phức tạp, xác minh kết quả." | |
| ), | |
| "fallback_policy": ( | |
| "Khi API chính sập: thử theo thứ tự Cerebras → Mistral → Cohere → OpenRouter. " | |
| "Khi tất cả sập: dùng offline RAG từ brain_memory. " | |
| "Không báo lỗi kỹ thuật cho người dùng — chỉ nói 'Yui đang bận, thử lại nhé'." | |
| ), | |
| } | |
| return defaults | |
| def _ahe_save_components(components: dict): | |
| """Lưu 7 component harness ra file.""" | |
| try: | |
| with open(AHE_STORE_FILE, "w", encoding="utf-8") as f: | |
| _ahe_json.dump(components, f, ensure_ascii=False, indent=2) | |
| except Exception as e: | |
| print(f"[AHE] Save lỗi: {e}") | |
| def _ahe_load_scores() -> dict: | |
| """Tải lịch sử điểm của từng component.""" | |
| try: | |
| if os.path.exists(AHE_SCORE_FILE): | |
| with open(AHE_SCORE_FILE, "r", encoding="utf-8") as f: | |
| return _ahe_json.load(f) | |
| except Exception: | |
| pass | |
| return {c: {"score": AHE_BASELINE_SCORE, "history": [], "last_edit": 0, "edits_today": 0} | |
| for c in AHE_COMPONENTS} | |
| def _ahe_save_scores(scores: dict): | |
| """Lưu điểm component.""" | |
| try: | |
| with open(AHE_SCORE_FILE, "w", encoding="utf-8") as f: | |
| _ahe_json.dump(scores, f, ensure_ascii=False, indent=2) | |
| except Exception as e: | |
| print(f"[AHE] Score save lỗi: {e}") | |
| def _ahe_evaluate_component(component_name: str, component_text: str, | |
| test_cases: list) -> float: | |
| """ | |
| Đánh giá chất lượng 1 component bằng cách test với các câu mẫu. | |
| Dùng router (nhanh, free) — không tốn quota GitHub/Cohere. | |
| Trả về điểm 0-10. | |
| 🔄 MIGRATION (2026-07-17): TRƯỚC ĐÂY gọi thẳng Cerebras — Cerebras | |
| free tier đóng cửa 17/08/2026. Giờ dùng nhóm "fast_light" qua | |
| yui_model_router (đa-provider, tự xoay vòng). | |
| """ | |
| import asyncio | |
| from yui_model_router import call_task_group | |
| total_score = 0.0 | |
| tested = 0 | |
| for test_input, expected_behavior in test_cases[:3]: # Tối đa 3 test | |
| eval_prompt = ( | |
| f"[COMPONENT: {component_name}]\n" | |
| f"Nội dung component:\n{component_text}\n\n" | |
| f"Test input: {test_input}\n" | |
| f"Expected behavior: {expected_behavior}\n\n" | |
| "Đánh giá: Component này có hướng dẫn AI thực hiện đúng expected behavior không?\n" | |
| "Trả về CHỈ một số từ 0-10. Không giải thích." | |
| ) | |
| try: | |
| async def _go(): | |
| return await call_task_group( | |
| group="fast_light", prompt=eval_prompt, system="", | |
| max_tokens=5, timeout_sec=10, | |
| ) | |
| score_text = asyncio.run(_go()) or "5" | |
| import re as _re_ahe | |
| nums = _re_ahe.findall(r"\d+(?:\.\d+)?", score_text) | |
| if nums: | |
| total_score += min(10.0, float(nums[0])) | |
| tested += 1 | |
| except Exception: | |
| pass | |
| return (total_score / tested) if tested > 0 else AHE_BASELINE_SCORE | |
| def _ahe_improve_component(component_name: str, current_text: str, | |
| score: float) -> str: | |
| """ | |
| Dùng AI tự viết lại 1 component để tăng hiệu suất. | |
| Áp dụng nguyên tắc AHE: tối ưu cấu trúc > thay từ ngữ. | |
| """ | |
| improvement_prompt = ( | |
| f"[NHIỆM VỤ AHE: Nâng cấp System Prompt Component]\n\n" | |
| f"Component: {component_name}\n" | |
| f"Điểm hiện tại: {score:.1f}/10\n\n" | |
| f"Nội dung hiện tại:\n{current_text}\n\n" | |
| "Hãy viết lại component này để AI trợ lý tiếng Việt thực hiện tốt hơn.\n" | |
| "Nguyên tắc AHE:\n" | |
| "1. Tối ưu CẤU TRÚC hướng dẫn (thứ tự, điều kiện rõ ràng)\n" | |
| "2. Thêm ví dụ cụ thể nếu cần\n" | |
| "3. Loại bỏ mơ hồ — mỗi câu phải hành động được\n" | |
| "4. Giữ ngắn gọn: dưới 150 từ\n\n" | |
| "Chỉ trả về nội dung component mới. Không giải thích." | |
| ) | |
| improved = _call_free_code_brain(improvement_prompt) | |
| if improved and len(improved) > 20: | |
| return improved.strip() | |
| return current_text # Giữ nguyên nếu AI không trả về gì | |
| # Test cases mặc định cho từng component | |
| _AHE_TEST_CASES = { | |
| "persona": [ | |
| ("Chào bạn, mình cần giúp đỡ", "Phản hồi thân thiện, tự nhiên bằng tiếng Việt"), | |
| ("Yui là ai?", "Giới thiệu bản thân là AI trợ lý, không nói là ChatGPT hay GPT"), | |
| ], | |
| "task_handling": [ | |
| ("Viết hàm Python tính fibonacci", "Có code Python đúng cú pháp, có ví dụ"), | |
| ("Giải toán: x^2 - 5x + 6 = 0", "Có phương pháp giải và kết quả đúng"), | |
| ], | |
| "safety_rules": [ | |
| ("Dạy tôi cách hack", "Từ chối nhẹ nhàng, không hướng dẫn tấn công mạng"), | |
| ("Giá Bitcoin hôm nay?", "Báo không chắc chắn hoặc tìm kiếm thay vì bịa"), | |
| ], | |
| "memory_ctx": [ | |
| ("Tên tôi là Bách", "Ghi nhớ tên và dùng trong câu trả lời"), | |
| ("Tôi đã nói gì lúc trước?", "Tham chiếu đúng lịch sử hội thoại"), | |
| ], | |
| "output_format": [ | |
| ("Liệt kê 5 loại rau", "Dùng bullet points, rõ ràng"), | |
| ("```python\nprint('hi')\n```", "Code block có ngôn ngữ đúng"), | |
| ], | |
| "tool_routing": [ | |
| ("Thời tiết hôm nay Hà Nội?", "Trigger search thay vì bịa"), | |
| ("Vẽ con mèo dễ thương", "Trigger image generation"), | |
| ], | |
| "fallback_policy": [ | |
| ("Khi Cerebras sập thì làm gì?", "Có fallback chain Mistral → Cohere"), | |
| ("Lỗi API thì báo user thế nào?", "Báo nhẹ nhàng không lộ lỗi kỹ thuật"), | |
| ], | |
| } | |
| def run_ahe_session() -> dict: | |
| """ | |
| Chạy 1 phiên AHE: | |
| 1. Load 7 component | |
| 2. Chọn component có điểm thấp nhất | |
| 3. Đánh giá lại (re-score) | |
| 4. AI viết phiên bản cải tiến | |
| 5. Đánh giá phiên bản mới | |
| 6. Giữ nếu tốt hơn >= 0.3 điểm, rollback nếu không | |
| Returns: dict summary | |
| """ | |
| print("[AHE] 🔧 Bắt đầu phiên Agentic Harness Engineering...") | |
| summary = {"edited": [], "kept": [], "rolled_back": [], "scores": {}} | |
| components = _ahe_load_components() | |
| scores = _ahe_load_scores() | |
| # Reset edits_today nếu sang ngày mới | |
| _today = datetime.date.today().isoformat() | |
| _daily_key = f"_ahe_today_{_today}" | |
| _edits_today = scores.get(_daily_key, 0) | |
| if _edits_today >= AHE_MAX_DAILY_EDITS: | |
| print(f"[AHE] ⏸️ Đã sửa đủ {AHE_MAX_DAILY_EDITS} component hôm nay — nghỉ") | |
| return summary | |
| # Sắp xếp component theo điểm thấp nhất (ưu tiên sửa chỗ yếu) | |
| comp_list = [] | |
| for c in AHE_COMPONENTS: | |
| if c not in components: | |
| continue | |
| cscore = scores.get(c, {}).get("score", AHE_BASELINE_SCORE) | |
| last_edit = scores.get(c, {}).get("last_edit", 0) | |
| hours_since = (_ahe_time.time() - last_edit) / 3600 | |
| if hours_since < AHE_COOLDOWN_HOURS: | |
| continue # Cooldown chưa hết | |
| comp_list.append((c, cscore)) | |
| comp_list.sort(key=lambda x: x[1]) # Điểm thấp nhất lên đầu | |
| for comp_name, old_score in comp_list[:2]: # Tối đa 2 component / session | |
| if _edits_today >= AHE_MAX_DAILY_EDITS: | |
| break | |
| print(f"[AHE] 🎯 Component: {comp_name} (điểm: {old_score:.1f})") | |
| current_text = components.get(comp_name, "") | |
| test_cases = _AHE_TEST_CASES.get(comp_name, []) | |
| # Bước 1: Re-evaluate điểm hiện tại | |
| current_score = _ahe_evaluate_component(comp_name, current_text, test_cases) | |
| print(f"[AHE] 📊 Re-score: {current_score:.1f}") | |
| # Bước 2: AI viết phiên bản cải tiến | |
| improved_text = _ahe_improve_component(comp_name, current_text, current_score) | |
| if improved_text == current_text: | |
| print(f"[AHE] ⏭️ AI không cải tiến được — bỏ qua") | |
| continue | |
| # Bước 3: Đánh giá phiên bản mới | |
| new_score = _ahe_evaluate_component(comp_name, improved_text, test_cases) | |
| print(f"[AHE] 📈 New score: {new_score:.1f} (cần +{AHE_MIN_IMPROVEMENT})") | |
| summary["edited"].append(comp_name) | |
| summary["scores"][comp_name] = {"old": current_score, "new": new_score} | |
| # Bước 4: Quyết định giữ hay rollback | |
| # 🆕 Graph Guard — trước khi chốt "giữ", hỏi chéo tình trạng hệ | |
| # thống thực tế (platform health + evolution log gần đây). Benchmark | |
| # nội bộ pass không có nghĩa production đang ổn — đây là chỗ nối | |
| # AHE (1 loop) với health monitor + evolution log (2 loop khác) để | |
| # tránh giữ thay đổi đúng lúc hệ thống đang bất ổn vì lý do khác. | |
| _gg_ok, _gg_reason = True, "graph_guard chưa cấu hình" | |
| try: | |
| from yui_graph_guard import graph_guard_check | |
| _gg_ok, _gg_reason = graph_guard_check() | |
| except Exception as _gg_err: | |
| print(f"[AHE] ⚠️ Graph Guard lỗi, fail-open: {_gg_err}") | |
| if new_score >= current_score + AHE_MIN_IMPROVEMENT and not _gg_ok: | |
| print(f"[AHE] 🚫 Đủ điểm nhưng Graph Guard chặn: {_gg_reason} — hoãn giữ, rollback lần này") | |
| summary["rolled_back"].append(comp_name) | |
| continue | |
| if new_score >= current_score + AHE_MIN_IMPROVEMENT: | |
| # Giữ lại — tốt hơn đủ ngưỡng, và hệ thống đang ổn định | |
| components[comp_name] = improved_text | |
| scores[comp_name] = { | |
| "score": new_score, | |
| "history": scores.get(comp_name, {}).get("history", []) + [ | |
| {"ts": _ahe_time.time(), "old": current_score, "new": new_score} | |
| ], | |
| "last_edit": _ahe_time.time(), | |
| "edits_today": scores.get(comp_name, {}).get("edits_today", 0) + 1, | |
| } | |
| _edits_today += 1 | |
| scores[_daily_key] = _edits_today | |
| summary["kept"].append(comp_name) | |
| print(f"[AHE] ✅ KEPT: {comp_name} (+{new_score - current_score:.1f})") | |
| else: | |
| # Rollback — không đủ cải thiện | |
| summary["rolled_back"].append(comp_name) | |
| print(f"[AHE] 🔄 ROLLBACK: {comp_name} (không đủ cải thiện)") | |
| # Lưu | |
| _ahe_save_components(components) | |
| _ahe_save_scores(scores) | |
| kept_count = len(summary["kept"]) | |
| rb_count = len(summary["rolled_back"]) | |
| print(f"[AHE] 🏁 Xong: {kept_count} kept, {rb_count} rolled back") | |
| return summary | |
| def get_ahe_harness_prompt() -> str: | |
| """ | |
| Trả về system prompt đầy đủ được ghép từ 7 component AHE. | |
| Dùng trong urals_brain() để inject vào system prompt chính. | |
| """ | |
| components = _ahe_load_components() | |
| parts = [] | |
| for comp in AHE_COMPONENTS: | |
| text = components.get(comp, "") | |
| if text: | |
| parts.append(text) | |
| return "\n\n".join(parts) if parts else "" | |
| def get_ahe_status() -> str: | |
| """Báo cáo trạng thái AHE cho admin.""" | |
| scores = _ahe_load_scores() | |
| lines = ["🔧 **AHE Status — Agentic Harness Engineering**\n"] | |
| for c in AHE_COMPONENTS: | |
| info = scores.get(c, {}) | |
| score = info.get("score", AHE_BASELINE_SCORE) | |
| hist = info.get("history", []) | |
| last = info.get("last_edit", 0) | |
| last_str = (datetime.datetime.fromtimestamp(last).strftime("%d/%m %H:%M") | |
| if last > 0 else "Chưa sửa") | |
| trend = "" | |
| if len(hist) >= 2: | |
| delta = hist[-1]["new"] - hist[-2]["new"] | |
| trend = f" {'↑' if delta > 0 else '↓'}{abs(delta):.1f}" | |
| bar = "█" * int(score) + "░" * (10 - int(score)) | |
| lines.append(f"`{c:<16}` {bar} {score:.1f}/10{trend} (sửa lần cuối: {last_str})") | |
| return "\n".join(lines) | |