server/universal_agent.py

"""Универсальный агент с поддержкой Open Interpreter"""

import json
import urllib.request
import re
from typing import Dict, List, Optional, Any
from .models import Role, ModelConfig
from .config import API_KEY, API_BASE, HF_TOKEN
from .file_manager import FILE_MANAGER
from .internet_agent import INTERNET_AGENT
from .notification_system import NOTIFICATIONS
from .process_manager import PROCESS_MANAGER
from .state import STATE

class UniversalAgent:
    def __init__(self, role: Role, model: ModelConfig, use_interpreter: bool = False):
        self.role = role
        self.model = model
        self.use_interpreter = use_interpreter
        self.conversation_history: List[Dict[str, str]] = []
        self.file_manager = FILE_MANAGER
        self.internet = INTERNET_AGENT
        self.notifications = NOTIFICATIONS
        self.tools = {
            "search_web": self.internet.search_web,
            "fetch_page": self.internet.fetch_page,
            "analyze_website": self.internet.analyze_website,
            "read_file": self.file_manager.read_file,
            "save_file": self.file_manager.save_file,
            "list_files": self.file_manager.list_files,
            "analyze_file": self.file_manager.analyze_file,
        }
    
    def _add_tools_to_task(self, task: str) -> str:
        tools_desc = "\n\nAVAILABLE TOOLS:\n"
        for name, func in self.tools.items():
            tools_desc += f"- {name}: {func.__doc__ or 'No description'}\n"
        tools_desc += "\nUse tools when needed. Return results in natural language."
        return task + tools_desc
    
    def execute(self, task: str, sys_prompt_override: str = None, chat_id: str = None, 
                history: List[Dict[str, str]] = None, mode: str = "chat") -> str:
        if self.use_interpreter and mode in ("skill", "build"):
            return self._execute_with_interpreter(task, chat_id, mode)
        return self._execute_with_api(task, sys_prompt_override, chat_id, history, mode)
    
    def _execute_with_interpreter(self, task: str, chat_id: str = None, mode: str = "chat") -> str:
        try:
            from interpreter import interpreter
            configure_interpreter_for_model(self.model.name)
            base_instructions = interpreter.custom_instructions or ""
            interpreter.custom_instructions = base_instructions + f"\n\nCURRENT ROLE: {self.role.name}\n{self.role.prompt}"
            messages = interpreter.chat(task, display=False)
            interpreter.custom_instructions = base_instructions
            if messages and len(messages) > 0:
                return messages[-1].get("content", "Done")
            return "No response from interpreter"
        except Exception as e:
            return f"Interpreter error: {e}. Falling back to API..."
    
    def _execute_with_api(self, task: str, sys_prompt_override: str = None, chat_id: str = None,
                          history: List[Dict[str, str]] = None, mode: str = "chat") -> str:
        system_prompt = sys_prompt_override or self.role.prompt
        messages = [{"role": "system", "content": system_prompt}]
        if history:
            for h in history[-10:]:
                messages.append({"role": h.get("role", "user"), "content": h.get("content", "")})
        messages.append({"role": "user", "content": task})
        
        if self.model.provider == "hf":
            return self._call_hf(messages, chat_id)
        return self._call_openai_compatible(messages, chat_id)
    
    def _call_openai_compatible(self, messages: List[Dict], chat_id: str = None) -> str:
        url = f"{API_BASE}/chat/completions"
        data = json.dumps({
            "model": self.model.endpoint,
            "messages": messages,
            "temperature": 0.7,
            "max_tokens": self.model.max_tokens
        }).encode('utf-8')
        req = urllib.request.Request(url, data=data, headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        })
        try:
            with urllib.request.urlopen(req, timeout=45) as response:
                res = json.loads(response.read().decode('utf-8'))
                return res['choices'][0]['message']['content']
        except Exception as primary_error:
            return self._tiered_fallback(messages, primary_error, chat_id)
    
    def _call_hf(self, messages: List[Dict], chat_id: str = None) -> str:
        url = "https://api-inference.huggingface.co/v1/chat/completions"
        data = json.dumps({
            "model": self.model.endpoint,
            "messages": messages,
            "temperature": 0.7,
            "max_tokens": self.model.max_tokens
        }).encode('utf-8')
        req = urllib.request.Request(url, data=data, headers={
            "Authorization": f"Bearer {HF_TOKEN}",
            "Content-Type": "application/json"
        })
        try:
            with urllib.request.urlopen(req, timeout=60) as response:
                res = json.loads(response.read().decode('utf-8'))
                return res['choices'][0]['message']['content']
        except Exception as e:
            return f"HF API Error: {e}"
    
    def _tiered_fallback(self, messages: List[Dict], primary_error, chat_id: str = None) -> str:
        if chat_id:
            from .telegram_utils import send_tg
            send_tg(chat_id, f"⚠️ {self.model.name} failed: {primary_error}. Trying fallback...")
        tried = [self.model.name]
        current = self.model.name
        while True:
            next_model = STATE.get_next_tier_model(current)
            if not next_model or next_model in tried:
                break
            tried.append(next_model)
            model_cfg = STATE.models.get(next_model)
            if not model_cfg:
                break
            try:
                agent = UniversalAgent(self.role, model_cfg)
                result = agent._call_hf(messages, chat_id) if model_cfg.provider == "hf" else agent._call_openai_compatible(messages, chat_id)
                return result + f"\n\n_(Fallback via {next_model})_"
            except Exception as e:
                current = next_model
                continue
        return self._fallback_hf(messages, primary_error, chat_id, tried)
    
    def _fallback_hf(self, messages: List[Dict], primary_error, chat_id: str = None, tried_models: List[str] = None) -> str:
        if not HF_TOKEN:
            return f"API Error: {primary_error}. No HF_TOKEN for backup."
        fallback = STATE.models.get("hf_fallback")
        if not fallback:
            return f"API Error: {primary_error}. HF fallback not configured."
        try:
            result = self._call_hf(messages, chat_id)
            return result + "\n\n_(Context saved via HF Serverless)_"
        except Exception as hf_error:
            return f"Both systems failed.\n1. API: {primary_error}\n2. HF Fallback: {hf_error}"