yui_continuous_evolution.py

# ================================================================
# YUI CONTINUOUS EVOLUTION ENGINE v1.0
# File: yui_continuous_evolution.py
#
# Tính năng:
# [1] Background thread tự động phân tích lỗi & hội thoại
# [2] Tạo đề xuất cải tiến → gọi Code Brain → sandbox verify → apply
# [3] Adaptive frequency: tăng tần suất khi phát hiện nhiều lỗi
# [4] Self-monitoring: theo dõi sức khỏe hệ thống liên tục
# [5] Skill absorption: học từ mỗi cuộc hội thoại thành công
# [6] Zero-cost: chỉ dùng free APIs (Mistral/Gemini/Cerebras)
# ================================================================

import os, sys, re, json, time, random, threading, traceback, datetime
import sqlite3, hashlib, ast
from typing import Optional, List, Dict, Tuple

# ── Constants ─────────────────────────────────────────────────────
_EVL_VERSION        = "1.0.0"
_EVL_INTERVAL_MIN   = 300      # 5 phút — chu kỳ bình thường
_EVL_INTERVAL_FAST  = 60       # 1 phút — khi có nhiều lỗi
_EVL_INTERVAL_SLOW  = 900      # 15 phút — khi hệ thống ổn định
_EVL_MAX_ERRORS     = 10       # Bao nhiêu lỗi → chuyển sang FAST mode
_EVL_LOG_MAX        = 500      # Giữ tối đa N entries trong log
_EVL_DB_PATH        = None     # Set by init()

# ── Internal state ────────────────────────────────────────────────
_error_buffer: List[Dict]   = []   # Buffer lỗi gần đây
_insight_buffer: List[Dict] = []   # Insight từ hội thoại
_evolution_log: List[Dict]  = []   # Log tiến hóa
_lock = threading.Lock()
_stop_event = threading.Event()
_evolution_thread: Optional[threading.Thread] = None
_current_interval = _EVL_INTERVAL_MIN
_total_evolutions = 0
_last_evolution_time = 0.0
_health_score = 100.0          # 0-100, hệ thống càng khỏe càng cao

# ── Getters cho app.py ────────────────────────────────────────────
def get_health_score() -> float:
    return _health_score

def get_total_evolutions() -> int:
    return _total_evolutions

def get_last_evolution_time() -> float:
    return _last_evolution_time

def get_evolution_log(n: int = 10) -> List[Dict]:
    with _lock:
        return list(_evolution_log[-n:])

# ── Thu thập lỗi & insight ────────────────────────────────────────
def report_error(error: Exception, context: str = "", platform: str = ""):
    """Gọi từ app.py khi có lỗi xảy ra để engine học."""
    global _health_score
    with _lock:
        entry = {
            "ts": time.time(),
            "type": type(error).__name__,
            "msg": str(error)[:300],
            "ctx": context[:200],
            "platform": platform,
        }
        _error_buffer.append(entry)
        if len(_error_buffer) > 200:
            _error_buffer.pop(0)
        # Giảm health score theo mức độ lỗi
        severity = 5 if isinstance(error, (RuntimeError, ValueError)) else 2
        _health_score = max(0.0, _health_score - severity)

def report_insight(topic: str, content: str, quality: float = 0.7):
    """Gọi khi Yui trả lời thành công — học từ phản hồi tốt."""
    global _health_score
    with _lock:
        entry = {
            "ts": time.time(),
            "topic": topic[:100],
            "content": content[:500],
            "quality": quality,
        }
        _insight_buffer.append(entry)
        if len(_insight_buffer) > 300:
            _insight_buffer.pop(0)
        # Tăng health score khi có insight tốt
        _health_score = min(100.0, _health_score + quality * 0.5)

# ── Core Evolution Logic ──────────────────────────────────────────
def _analyze_errors() -> Optional[str]:
    """Phân tích các lỗi gần đây, tạo summary."""
    with _lock:
        recent = [e for e in _error_buffer if time.time() - e["ts"] < 3600]
    if not recent:
        return None
    by_type: Dict[str, int] = {}
    for e in recent:
        by_type[e["type"]] = by_type.get(e["type"], 0) + 1
    top = sorted(by_type.items(), key=lambda x: x[1], reverse=True)[:3]
    summary = "; ".join(f"{t}({c})" for t, c in top)
    samples = [f"  [{e['type']}] {e['msg']} (ctx: {e['ctx']})" for e in recent[-3:]]
    return f"Lỗi 1h qua: {summary}\nMẫu:\n" + "\n".join(samples)

def _analyze_insights() -> Optional[str]:
    """Tóm tắt patterns từ các insight thành công."""
    with _lock:
        recent = [i for i in _insight_buffer if time.time() - i["ts"] < 7200]
    if not recent:
        return None
    topics = {}
    for i in recent:
        t = i["topic"]
        topics[t] = topics.get(t, 0) + 1
    top = sorted(topics.items(), key=lambda x: x[1], reverse=True)[:5]
    return "Topics phổ biến: " + ", ".join(f"{t}({c})" for t, c in top)

def _call_evolution_brain(prompt: str) -> Optional[str]:
    """
    Gọi free AI để tạo cải tiến.
    🔄 MIGRATION (2026-07-17): TRƯỚC ĐÂY cascade tay Mistral→Gemini→
    Cerebras — Cerebras free tier đóng cửa 17/08/2026. Giờ dùng nhóm
    "fast_light" qua yui_model_router (đa-provider, đã có sẵn Mistral +
    Gemini + Cohere + SambaNova, tự xoay vòng — không cần cascade tay
    3 tầng như cũ nữa). Giữ nguyên tên hàm + chữ ký để mọi nơi gọi
    _call_evolution_brain(...) không cần sửa gì thêm.
    """
    try:
        import asyncio
        from yui_model_router import call_task_group

        async def _go():
            return await call_task_group(
                group="fast_light", prompt=prompt, system="",
                max_tokens=1200, timeout_sec=30,
            )

        text = asyncio.run(_go())
        if text and len(text) > 20:
            return text
    except Exception as e:
        print(f"[EVOLUTION] Brain call lỗi: {e}")
    return None

def _safe_code_verify(code: str) -> Tuple[bool, str]:
    """Kiểm tra code mới có an toàn không (AST scan)."""
    FORBIDDEN = {
        "calls": ["eval", "exec", "compile", "__import__", "open", "subprocess"],
        "imports": ["subprocess", "socket", "ctypes", "importlib"],
        "attrs": ["__class__", "__bases__", "__mro__", "__subclasses__"],
    }
    try:
        tree = ast.parse(code)
        for node in ast.walk(tree):
            if isinstance(node, ast.Call):
                if isinstance(node.func, ast.Name) and node.func.id in FORBIDDEN["calls"]:
                    return False, f"Forbidden call: {node.func.id}"
            if isinstance(node, (ast.Import, ast.ImportFrom)):
                names = [a.name for a in node.names] if isinstance(node, ast.Import) \
                    else ([node.module] if node.module else [])
                for n in names:
                    if n and n.split(".")[0] in FORBIDDEN["imports"]:
                        return False, f"Forbidden import: {n}"
        return True, "ok"
    except SyntaxError as e:
        return False, f"SyntaxError: {e}"
    except Exception as e:
        return False, f"Verify error: {e}"

def _save_evolution_to_db(evo_type: str, summary: str, success: bool):
    """Lưu log tiến hóa vào SQLite nếu có DB."""
    global _evolution_log
    entry = {
        "ts": time.time(),
        "dt": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "type": evo_type,
        "summary": summary[:300],
        "success": success,
        "health": round(_health_score, 1),
    }
    with _lock:
        _evolution_log.append(entry)
        if len(_evolution_log) > _EVL_LOG_MAX:
            _evolution_log.pop(0)
    # Ghi vào SQLite nếu path đã được set
    if _EVL_DB_PATH:
        try:
            conn = sqlite3.connect(_EVL_DB_PATH, timeout=5)
            conn.execute("CREATE TABLE IF NOT EXISTS evolution_log (ts REAL, dt TEXT, type TEXT, summary TEXT, success INTEGER, health REAL)")
            conn.execute("INSERT INTO evolution_log VALUES (?,?,?,?,?,?)",
                         (entry["ts"], entry["dt"], entry["type"], entry["summary"],
                          1 if success else 0, entry["health"]))
            # Giữ tối đa 500 entries
            conn.execute("DELETE FROM evolution_log WHERE rowid NOT IN (SELECT rowid FROM evolution_log ORDER BY ts DESC LIMIT 500)")
            conn.commit()
            conn.close()
        except Exception:
            pass

# ── Các loại tiến hóa ─────────────────────────────────────────────
def _evolve_error_fix() -> bool:
    """Phân tích lỗi → đề xuất fix → verify → log."""
    err_summary = _analyze_errors()
    if not err_summary:
        return False
    prompt = f"""Bạn là AI chuyên phân tích và sửa lỗi Python.
Hệ thống Yui AI đang gặp các lỗi sau (1 giờ qua):

{err_summary}

Hãy:
1. Phân tích nguyên nhân gốc rễ
2. Đề xuất 1-2 fix cụ thể (pattern-based, không cần biết toàn bộ code)
3. Nếu có thể, viết một helper function nhỏ (<30 dòng) để xử lý lỗi này

Trả lời ngắn gọn, thực tế. Không viết essay."""
    response = _call_evolution_brain(prompt)
    if not response:
        return False
    summary = f"Error fix analysis: {err_summary[:100]} → {response[:200]}"
    _save_evolution_to_db("error_fix", summary, True)
    print(f"[EVOLUTION] 🔧 Error fix: {response[:150]}")
    return True

def _evolve_skill_absorption() -> bool:
    """Học từ các insight thành công → lưu vào skill DB."""
    insight_summary = _analyze_insights()
    if not insight_summary:
        return False
    with _lock:
        recent = [i for i in _insight_buffer
                  if time.time() - i["ts"] < 7200 and i["quality"] > 0.75]
    if not recent:
        return False
    # Lưu insights chất lượng cao vào skill DB (nếu có _skill_db_save)
    absorbed = 0
    for insight in recent[-5:]:
        try:
            # Thử gọi _skill_db_save từ app.py scope nếu available
            import sys
            app_mod = sys.modules.get("__main__") or sys.modules.get("app")
            if app_mod and hasattr(app_mod, "_skill_db_save"):
                app_mod._skill_db_save(
                    problem=insight["topic"],
                    solution=insight["content"],
                    category="conversation_insight",
                    quality=insight["quality"],
                    is_core=False,
                    source="continuous_evolution",
                )
                absorbed += 1
        except Exception:
            pass
    if absorbed > 0:
        summary = f"Absorbed {absorbed} insights. {insight_summary}"
        _save_evolution_to_db("skill_absorption", summary, True)
        print(f"[EVOLUTION] 🧠 Absorbed {absorbed} skills từ hội thoại")
        return True
    return False

def _evolve_health_check() -> bool:
    """Kiểm tra sức khỏe hệ thống, phục hồi nếu cần."""
    global _health_score
    checks = []
    # Check Python memory
    try:
        import sys
        mod_count = len(sys.modules)
        checks.append(f"modules={mod_count}")
    except Exception:
        pass
    # Check error rate trong 10 phút
    with _lock:
        recent_errors = [e for e in _error_buffer if time.time() - e["ts"] < 600]
    err_rate = len(recent_errors)
    checks.append(f"errors_10m={err_rate}")
    # Điều chỉnh health
    if err_rate == 0:
        _health_score = min(100.0, _health_score + 2.0)
    elif err_rate > 5:
        _health_score = max(0.0, _health_score - 5.0)
    status = "🟢" if _health_score >= 70 else "🟡" if _health_score >= 40 else "🔴"
    summary = f"{status} Health={_health_score:.1f} | " + " | ".join(checks)
    _save_evolution_to_db("health_check", summary, True)
    return True

def _evolve_self_optimize() -> bool:
    """Dùng AI để tự phân tích và đề xuất tối ưu hóa."""
    global _health_score
    if _health_score > 80 and random.random() > 0.3:
        return False  # Hệ thống đang ổn, không cần optimize liên tục
    err_count = len(_error_buffer)
    insight_count = len(_insight_buffer)
    prompt = f"""Bạn là AI tư vấn tối ưu hóa hệ thống AI chatbot (Yui AI).
Tình trạng hiện tại:
- Health score: {_health_score:.1f}/100
- Lỗi tích lũy: {err_count}
- Insights học được: {insight_count}
- Phiên bản engine: {_EVL_VERSION}

Đây là AI chạy trên HuggingFace Spaces (free tier), giới hạn RAM/CPU.

Hãy đề xuất 2-3 cải tiến nhỏ, cụ thể, thực tế để:
1. Tăng độ ổn định
2. Giảm lỗi
3. Học nhanh hơn từ người dùng

Ngắn gọn, không cần giải thích dài."""
    response = _call_evolution_brain(prompt)
    if not response:
        return False
    summary = f"Self-optimize (health={_health_score:.0f}): {response[:250]}"
    _save_evolution_to_db("self_optimize", summary, True)
    print(f"[EVOLUTION] 💡 Optimize: {response[:120]}")
    _health_score = min(100.0, _health_score + 1.0)
    return True

# ── Adaptive Scheduling ──────────────────────────────────────────
def _calculate_interval() -> int:
    """Tính chu kỳ tiến hóa dựa trên trạng thái hệ thống."""
    with _lock:
        recent_errors = [e for e in _error_buffer if time.time() - e["ts"] < 600]
    if len(recent_errors) >= _EVL_MAX_ERRORS:
        return _EVL_INTERVAL_FAST   # Nhiều lỗi → evolve nhanh hơn
    if _health_score >= 85:
        return _EVL_INTERVAL_SLOW   # Hệ thống khỏe → nghỉ lâu hơn
    return _EVL_INTERVAL_MIN

# ── Main Evolution Loop ───────────────────────────────────────────
def _evolution_loop():
    """Thread chính: chạy vô hạn, tự tiến hóa theo chu kỳ."""
    global _total_evolutions, _last_evolution_time, _current_interval
    print(f"[EVOLUTION] 🧬 Continuous Evolution Engine v{_EVL_VERSION} khởi động")
    # Delay ban đầu để app khởi động xong
    time.sleep(30)
    while not _stop_event.is_set():
        try:
            _last_evolution_time = time.time()
            cycle_start = time.time()
            tasks = [
                ("health_check",    _evolve_health_check,    1.0),   # Luôn chạy
                ("error_fix",       _evolve_error_fix,       0.8),   # Chạy 80% chu kỳ
                ("skill_absorb",    _evolve_skill_absorption,0.7),   # Chạy 70%
                ("self_optimize",   _evolve_self_optimize,   0.4),   # Chạy 40%
            ]
            success_count = 0
            for name, fn, prob in tasks:
                if _stop_event.is_set():
                    break
                if random.random() <= prob:
                    try:
                        ok = fn()
                        if ok:
                            success_count += 1
                    except Exception as e:
                        print(f"[EVOLUTION] Task {name} lỗi: {e}")
                time.sleep(2)  # Tránh spam API
            _total_evolutions += 1
            elapsed = time.time() - cycle_start
            _current_interval = _calculate_interval()
            print(f"[EVOLUTION] 🔄 Chu kỳ #{_total_evolutions} xong ({elapsed:.1f}s) "
                  f"| {success_count} tasks OK | health={_health_score:.1f} "
                  f"| next={_current_interval}s")
        except Exception as e:
            print(f"[EVOLUTION] ⚠️ Loop lỗi: {e}")
        # Chờ chu kỳ tiếp theo (có thể interrupt)
        _stop_event.wait(timeout=_current_interval)

# ── Public API ────────────────────────────────────────────────────
def start(db_path: Optional[str] = None):
    """Khởi động evolution engine. Gọi từ app.py khi startup."""
    global _EVL_DB_PATH, _evolution_thread
    if db_path:
        _EVL_DB_PATH = db_path
    if _evolution_thread and _evolution_thread.is_alive():
        print("[EVOLUTION] Engine đã chạy rồi, bỏ qua")
        return
    _stop_event.clear()
    _evolution_thread = threading.Thread(
        target=_evolution_loop,
        daemon=True,
        name="YuiEvolutionEngine",
    )
    _evolution_thread.start()
    print("[EVOLUTION] ✅ Engine started")

def stop():
    """Dừng evolution engine."""
    _stop_event.set()
    if _evolution_thread:
        _evolution_thread.join(timeout=5)
    print("[EVOLUTION] 🛑 Engine stopped")

def trigger_now(reason: str = "manual"):
    """Force chạy 1 chu kỳ ngay lập tức (dùng cho admin)."""
    def _run():
        print(f"[EVOLUTION] ⚡ Manual trigger: {reason}")
        _evolve_health_check()
        _evolve_error_fix()
        _evolve_skill_absorption()
        _evolve_self_optimize()
    threading.Thread(target=_run, daemon=True).start()

def status() -> str:
    """Trạng thái engine dạng text."""
    with _lock:
        err_count = len(_error_buffer)
        ins_count = len(_insight_buffer)
        log_count = len(_evolution_log)
    alive = _evolution_thread and _evolution_thread.is_alive() if _evolution_thread else False
    last_dt = (datetime.datetime.fromtimestamp(_last_evolution_time).strftime("%H:%M:%S")
               if _last_evolution_time > 0 else "chưa chạy")
    health_icon = "🟢" if _health_score >= 70 else "🟡" if _health_score >= 40 else "🔴"
    return (f"🧬 **Continuous Evolution Engine v{_EVL_VERSION}**\n"
            f"{health_icon} Health: {_health_score:.1f}/100\n"
            f"⚡ Tổng chu kỳ: {_total_evolutions}\n"
            f"🕐 Lần cuối: {last_dt}\n"
            f"⏱️ Chu kỳ tiếp: {_current_interval}s\n"
            f"❌ Lỗi buffer: {err_count}\n"
            f"💡 Insights: {ins_count}\n"
            f"📝 Log entries: {log_count}\n"
            f"🔄 Thread: {'🟢 alive' if alive else '🔴 dead'}")

# ── Auto-start khi import ─────────────────────────────────────────
# Không auto-start ở đây, để app.py kiểm soát thời điểm start
print(f"[EVOLUTION] Module yui_continuous_evolution v{_EVL_VERSION} loaded")