import requests import json import subprocess import os import re import platform import time import logging import sys import importlib import shutil from datetime import datetime # Configurare logging pentru afișare în timp real class RealTimeLogger: def __init__(self, name, level=logging.INFO): self.logger = logging.getLogger(name) self.logger.setLevel(level) # Handler pentru consolă console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(level) # Format cu timestamp formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s', datefmt='%H:%M:%S') console_handler.setFormatter(formatter) self.logger.addHandler(console_handler) # Handler pentru fișier log_dir = "logs" if not os.path.exists(log_dir): os.makedirs(log_dir) log_file = os.path.join(log_dir, f"activity_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log") file_handler = logging.FileHandler(log_file, encoding='utf-8') file_handler.setLevel(level) file_handler.setFormatter(formatter) self.logger.addHandler(file_handler) def info(self, message): self.logger.info(message) def warning(self, message): self.logger.warning(message) def error(self, message): self.logger.error(message) def success(self, message): self.logger.info(f"✅ {message}") # Definirea directoarelor și fișierelor BASE_DIR = os.path.dirname(os.path.abspath(__file__)) SCRIPTS_DIR = os.path.join(BASE_DIR, "agenți") TOOLS_DIR = os.path.join(BASE_DIR, "unelte") SOLUTION_DIR = os.path.join(BASE_DIR, "soluții") TEMP_DIR = os.path.join(BASE_DIR, "temp") LOG_DIR = os.path.join(BASE_DIR, "logs") # Inițializare logger logger = RealTimeLogger("ai_ollama") # Lista modelelor disponibile în instalarea Ollama AVAILABLE_MODELS = [ "mxbai-embed-large:latest", "qwq:latest", "qwen2-math:7b", "qwen2.5-coder:32b", "gemma3:27b", "gemma3:latest", "llama3:latest", "llama3.2:latest" ] # Categorizarea modelelor instalate MODEL_CATEGORIES = { "coding": ["qwen2.5-coder:32b", "llama3:latest"], "math": ["qwen2-math:7b"], "embedding": ["mxbai-embed-large:latest"], "general": ["gemma3:27b", "gemma3:latest", "llama3:latest", "llama3.2:latest", "qwq:latest"] } # Funcții utilitare pentru lucrul cu modelele Ollama def check_ollama_connection(): """Verifică dacă serverul Ollama este disponibil""" try: response = requests.get("http://localhost:11434/api/tags", timeout=5) if response.status_code == 200: return True return False except requests.exceptions.ConnectionError: return False except Exception as e: print(f"Eroare la verificarea conexiunii Ollama: {str(e)}") return False def get_available_models(): """Obține lista de modele disponibile de la serverul Ollama""" try: response = requests.get("http://localhost:11434/api/tags", timeout=5) if response.status_code == 200: data = response.json() models = data.get("models", []) return [model["name"] for model in models] return [] except Exception as e: print(f"Eroare la obținerea modelelor: {str(e)}") return [] def select_best_model(task_type, available_models=None): """Selectează cel mai bun model pentru un anumit tip de sarcină""" if available_models is None: available_models = get_available_models() # Dacă nu sunt modele disponibile, folosim lista statică if not available_models: available_models = AVAILABLE_MODELS # Selectăm modelul în funcție de tipul de sarcină if task_type == "coding": for model in MODEL_CATEGORIES["coding"]: if model in available_models: return model elif task_type == "math": for model in MODEL_CATEGORIES["math"]: if model in available_models: return model # Pentru sarcini generale, preferăm modelele mai mari preferred_general = ["gemma3:27b", "qwen2.5-coder:32b", "llama3.2:latest", "llama3:latest", "gemma3:latest"] for model in preferred_general: if model in available_models: return model # Dacă nu găsim un model potrivit, returnăm primul disponibil return available_models[0] if available_models else "gemma3:latest" def query_ollama(model, prompt, max_retries=3, timeout=60, system_prompt=None): """Interogare model Ollama cu suport pentru retry și afișare în timp real""" url = "http://localhost:11434/api/generate" payload = { "model": model, "prompt": prompt, "stream": True } # Adăugăm system prompt dacă este specificat if system_prompt: payload["system"] = system_prompt logger.info(f"Interogare model: {model}") logger.info(f"Prompt: {prompt[:100]}...") full_response = "" retries = 0 while retries < max_retries: try: # Stream response pentru afișare în timp real with requests.post(url, json=payload, stream=True, timeout=timeout) as response: if response.status_code != 200: logger.error(f"Eroare HTTP: {response.status_code}") retries += 1 time.sleep(2) continue sys.stdout.write(f"\n[Răspuns de la {model}]: ") for line in response.iter_lines(): if line: chunk = json.loads(line.decode('utf-8')) text_chunk = chunk.get("response", "") full_response += text_chunk sys.stdout.write(text_chunk) sys.stdout.flush() sys.stdout.write("\n\n") return full_response except requests.exceptions.Timeout: logger.warning(f"Timeout la interogarea modelului {model}. Încercare {retries+1}/{max_retries}") retries += 1 except Exception as e: logger.error(f"Excepție la interogare: {e}") retries += 1 time.sleep(2) logger.error(f"Nu s-a putut obține un răspuns după {max_retries} încercări") return None def extract_python_code(text): """Extrage codul Python dintr-un răspuns""" if not text: return None # Încercăm mai multe pattern-uri pentru a găsi codul Python patterns = [ r"```python\s*(.*?)\s*```", # Standard markdown r"```\s*(.*?)\s*```", # Orice bloc de cod r"(?:(?:def|import|class|if)\s+.*?(?:\n.*?)+)(?:\n\n|$)", # Pattern de cod Python ] for pattern in patterns: matches = re.findall(pattern, text, re.DOTALL) if matches: code = matches[0].strip() # Verificare de bază dacă este cod Python valid if 'import' in code or 'def ' in code or 'print(' in code: logger.info("Cod Python extras cu succes") return code # Încercăm să obținem tot textul dacă pare a fi cod Python if 'import' in text and ('def ' in text or 'print(' in text): logger.info("Extragere cod din text complet") return text.strip() logger.error("Nu s-a găsit cod Python valid în răspuns") return None def detect_task_type(task): """Detectează tipul de sarcină pentru a alege modelul potrivit""" task_lower = task.lower() # Detectare sarcini de programare/cod coding_keywords = [ "scrie un script", "scrie un program", "dezvoltă un program", "creează un program", "script python", "cod python", "program python", "implementează", "algoritm", "automatizează", "funcție", "clasă", "aplicație", "API", "backend" ] # Detectare sarcini matematice math_keywords = [ "calculează", "rezolvă", "ecuație", "matematică", "formula", "algebră", "geometrie", "statistică", "probabilitate", "derivată", "integrală", "matrice" ] # Detectare sarcini generale general_keywords = [ "explică", "descrie", "povestește", "cum să", "ce este", "cine este", "când", "de ce", "unde", "analizează", "compară", "rezumă", "enumeră" ] # Calculăm scorurile pentru fiecare tip coding_score = sum(2 for keyword in coding_keywords if keyword in task_lower) math_score = sum(2 for keyword in math_keywords if keyword in task_lower) general_score = sum(1 for keyword in general_keywords if keyword in task_lower) # Returnăm tipul cu scorul cel mai mare if coding_score > math_score and coding_score > general_score: return "coding" elif math_score > coding_score and math_score > general_score: return "math" else: return "general" def run_script(script_path, env_vars=None, timeout=30): """Rulează un script Python cu captare output""" logger.info(f"Rulare script: {script_path}") # Adăugăm variabile de mediu dacă sunt specificate env = os.environ.copy() if env_vars: env.update(env_vars) try: # Utilizăm subprocess.Popen pentru a afișa output-ul în timp real process = subprocess.Popen( [sys.executable, script_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, env=env ) # Citim output-ul în timp real stdout_chunks = [] stderr_chunks = [] start_time = time.time() while process.poll() is None: # Verificăm timeout if time.time() - start_time > timeout: process.kill() logger.error(f"Script terminat forțat după {timeout} secunde (timeout)") return False, "Timeout", "" # Citim stdout stdout_line = process.stdout.readline() if stdout_line: stdout_chunks.append(stdout_line) sys.stdout.write(f"[Script Output]: {stdout_line}") sys.stdout.flush() # Citim stderr stderr_line = process.stderr.readline() if stderr_line: stderr_chunks.append(stderr_line) sys.stderr.write(f"[Script Error]: {stderr_line}") sys.stderr.flush() # Evităm consumul excesiv de CPU if not stdout_line and not stderr_line: time.sleep(0.1) # Citim orice output rămas stdout_remainder = process.stdout.read() if stdout_remainder: stdout_chunks.append(stdout_remainder) sys.stdout.write(f"[Script Output]: {stdout_remainder}") stderr_remainder = process.stderr.read() if stderr_remainder: stderr_chunks.append(stderr_remainder) sys.stderr.write(f"[Script Error]: {stderr_remainder}") return_code = process.returncode stdout_content = "".join(stdout_chunks) stderr_content = "".join(stderr_chunks) if return_code == 0: logger.success(f"Script rulat cu succes (cod retur: {return_code})") return True, stdout_content, stderr_content else: logger.error(f"Script eșuat cu codul {return_code}") logger.error(f"Stderr: {stderr_content}") return False, stdout_content, stderr_content except Exception as e: logger.error(f"Excepție la rularea scriptului: {e}") return False, "", str(e) def save_code_to_file(code, filename=None): """Salvează codul într-un fișier""" if not filename: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"script_{timestamp}.py" if not os.path.exists(SCRIPTS_DIR): os.makedirs(SCRIPTS_DIR) file_path = os.path.join(SCRIPTS_DIR, filename) try: with open(file_path, 'w', encoding='utf-8') as f: f.write(code) logger.success(f"Cod salvat în: {file_path}") return file_path except Exception as e: logger.error(f"Eroare la salvarea codului: {e}") return None def save_solution(task, solution, source, output_file=None): """Salvează soluția finală""" if not output_file: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_file = f"soluție_{timestamp}.txt" if not os.path.exists(SOLUTION_DIR): os.makedirs(SOLUTION_DIR) solution_path = os.path.join(SOLUTION_DIR, output_file) try: with open(solution_path, 'w', encoding='utf-8') as f: f.write(f"Sarcina: {task}\n\n") f.write(f"Soluție (generată de {source}):\n") f.write("="*50 + "\n") f.write(solution) f.write("\n" + "="*50 + "\n") f.write(f"Data și ora: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") logger.info(f"Soluție salvată în: {solution_path}") return solution_path except Exception as e: logger.error(f"Eroare la salvarea soluției: {e}") return None def open_file(file_path): """Deschide un fișier""" if not os.path.exists(file_path): logger.error(f"Fișierul nu există: {file_path}") return system = platform.system() try: if system == 'Windows': os.startfile(file_path) elif system == 'Darwin': # macOS subprocess.run(['open', file_path]) elif system == 'Linux': subprocess.run(['xdg-open', file_path]) else: logger.error("Nu pot deschide fișierul automat pe acest sistem.") except Exception as e: logger.error(f"Eroare la deschiderea fișierului: {e}") class DirectQueryTool: """Unealtă pentru interogarea directă a modelelor Ollama""" def __init__(self): self.models = get_available_models() if not self.models: self.models = AVAILABLE_MODELS # Folosim lista statică dacă nu putem obține modelele self.logger = RealTimeLogger("direct_query_tool") def show_available_models(self): """Afișează modelele disponibile""" self.logger.info("Modele disponibile:") for i, model in enumerate(self.models, 1): print(f"{i}. {model}") def select_model_interactive(self): """Permite utilizatorului să selecteze un model""" self.show_available_models() while True: try: choice = input(f"\nSelectează un model (1-{len(self.models)}): ") choice = int(choice) if 1 <= choice <= len(self.models): model = self.models[choice-1] self.logger.info(f"Model selectat: {model}") return model else: print(f"Te rog selectează un număr între 1 și {len(self.models)}") except ValueError: print("Te rog introdu un număr valid") def query(self, prompt, model=None, system_prompt=None): """Interogare directă a unui model""" if not model: task_type = detect_task_type(prompt) model = select_best_model(task_type, self.models) self.logger.info(f"Interogare {model}: {prompt[:50]}...") response = query_ollama(model, prompt, system_prompt=system_prompt) if response: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") solution_file = save_solution( prompt, response, f"Model: {model}", f"răspuns_{timestamp}.txt" ) print("\nRăspuns salvat în:", solution_file) return response, solution_file return None, None class CodeGeneratorTool: """Unealtă pentru generarea de cod Python folosind modelele Ollama""" def __init__(self): self.logger = RealTimeLogger("code_generator_tool") self.models = [] for model in get_available_models(): if model in MODEL_CATEGORIES["coding"]: self.models.append(model) # Dacă nu găsim modele de coding, folosim lista statică if not self.models: self.models = MODEL_CATEGORIES["coding"] def generate_code(self, task, model=None): """Generează cod Python pentru o sarcină""" if not model: # Alegem cel mai bun model pentru cod model = next((m for m in self.models), "qwen2.5-coder:32b") self.logger.info(f"Generare cod cu {model}: {task[:50]}...") # Construim prompt-ul pentru generarea de cod prompt = f"""Generează un program Python complet care să rezolve următoarea sarcină: {task} Returnează doar codul Python fără explicații. Utilizează blocuri markdown ```python pentru cod. """ system_prompt = "Ești un expert în programare Python. Generează cod eficient, bine structurat și comentat." # Interogăm modelul response = query_ollama(model, prompt, system_prompt=system_prompt) if not response: self.logger.error("Nu s-a putut genera codul") return None, None # Extragem codul Python din răspuns code = extract_python_code(response) if not code: self.logger.error("Nu s-a putut extrage codul Python din răspuns") return response, None # Salvăm codul într-un fișier file_path = save_code_to_file(code) if file_path: return code, file_path return code, None def run_generated_code(self, file_path): """Rulează codul generat""" if not file_path or not os.path.exists(file_path): self.logger.error("Fișierul nu există") return False, "", "" self.logger.info(f"Rulare cod din: {file_path}") return run_script(file_path, timeout=60) def generate_and_run(self, task, model=None): """Generează și rulează cod pentru o sarcină""" code, file_path = self.generate_code(task, model) if not file_path: self.logger.error("Nu s-a putut genera sau salva codul") return False print("\nCod generat:") print("-" * 50) print(code) print("-" * 50) # Întreabă utilizatorul dacă dorește să ruleze codul run_code = input("\nDoriți să rulați codul? (y/n): ").lower() if run_code == 'y': success, stdout, stderr = self.run_generated_code(file_path) if success: self.logger.success("Cod rulat cu succes") return True else: self.logger.error("Eroare la rularea codului") self.logger.error(stderr) # Întreabă dacă utilizatorul dorește să corecteze codul fix_code = input("\nDoriți să încercați să corectați codul? (y/n): ").lower() if fix_code == 'y': return self.fix_and_run(task, stderr, code, file_path) return False return True def fix_and_run(self, task, error, original_code, file_path): """Încearcă să corecteze și să ruleze din nou codul""" self.logger.info("Încercare de corectare a codului...") # Alegem cel mai bun model pentru corectarea codului model = next((m for m in self.models), "qwen2.5-coder:32b") # Construim prompt-ul pentru corectarea codului prompt = f"""Corectează următorul cod Python care a generat eroarea: Eroare: {error} Cod original: ```python {original_code} ``` Returnează doar codul corectat, fără explicații. Utilizează blocuri markdown ```python pentru cod. """ system_prompt = "Ești un expert în debugging Python. Corectează erori și optimizează codul." # Interogăm modelul response = query_ollama(model, prompt, system_prompt=system_prompt) if not response: self.logger.error("Nu s-a putut corecta codul") return False # Extragem codul Python din răspuns fixed_code = extract_python_code(response) if not fixed_code: self.logger.error("Nu s-a putut extrage codul Python corectat din răspuns") return False print("\nCod corectat:") print("-" * 50) print(fixed_code) print("-" * 50) # Salvăm codul corectat with open(file_path, 'w', encoding='utf-8') as f: f.write(fixed_code) self.logger.info(f"Cod corectat salvat în: {file_path}") # Întreabă utilizatorul dacă dorește să ruleze codul corectat run_fixed = input("\nDoriți să rulați codul corectat? (y/n): ").lower() if run_fixed == 'y': success, stdout, stderr = self.run_generated_code(file_path) if success: self.logger.success("Cod corectat rulat cu succes") return True else: self.logger.error("Eroare la rularea codului corectat") return False return True def main(): """Funcția principală a programului""" print("-" * 60) print("AI Ollama - Asistent bazat pe modelele instalate") print("-" * 60) print(f"Modele disponibile: {', '.join(AVAILABLE_MODELS)}") print("-" * 60) # Verifică dacă Ollama este disponibil if not check_ollama_connection(): print("Eroare: Nu s-a putut conecta la serverul Ollama.") print("Asigură-te că Ollama este pornit și rulează la adresa http://localhost:11434") return # Inițializează uneltele direct_query = DirectQueryTool() code_generator = CodeGeneratorTool() while True: print("\nOpțiuni:") print("1. Interogare directă") print("2. Generare și rulare cod") print("3. Afișare modele disponibile") print("4. Ieșire") choice = input("\nAlege o opțiune (1-4): ") if choice == "1": # Interogare directă prompt = input("\nIntroduceți întrebarea sau sarcina: ") if prompt: direct_query.query(prompt) elif choice == "2": # Generare și rulare cod task = input("\nDescrierea sarcinii de programare: ") if task: code_generator.generate_and_run(task) elif choice == "3": # Afișare modele disponibile print("\nModele disponibile:") models = get_available_models() if not models: models = AVAILABLE_MODELS for i, model in enumerate(models, 1): print(f"{i}. {model}") print("\nCategorii de modele:") for category, models in MODEL_CATEGORIES.items(): print(f"- {category.capitalize()}: {', '.join(models)}") elif choice == "4": print("\nLa revedere!") break else: print("\nOpțiune invalidă. Te rog alege din nou.") if __name__ == "__main__": # Creăm directoarele necesare for directory in [SCRIPTS_DIR, TOOLS_DIR, SOLUTION_DIR, TEMP_DIR, LOG_DIR]: if not os.path.exists(directory): os.makedirs(directory) main()