Delete server/universal_agent.py

server/universal_agent.py

@@ -1,145 +0,0 @@
-"""Универсальный агент с поддержкой 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}"