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