Spaces:
Running
Running
File size: 5,306 Bytes
143d94c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 |
#!/usr/bin/env python3
"""
Обгортка для MAI-DX що виправляє проблеми з Rich форматуванням
"""
import os
import sys
import warnings
# Налаштування середовища
os.environ.update({
"SWARMS_VERBOSITY": "ERROR",
"RICH_TRACEBACK": "0",
"SWARMS_SHOW_PANEL": "true",
"SWARMS_AUTO_PRINT": "true",
"PYTHONWARNINGS": "ignore"
})
def patch_rich_and_swarms():
"""Патчимо Rich та Swarms для уникнення помилок виводу"""
try:
# Патчимо Rich Console
import rich.console
original_print = rich.console.Console.print
def safe_print(self, *args, **kwargs):
"""Безпечний print що ігнорує None значення"""
try:
# Фільтруємо None значення
filtered_args = [arg for arg in args if arg is not None]
if filtered_args:
return original_print(self, *filtered_args, **kwargs)
except Exception:
# Ігноруємо всі помилки виводу
pass
rich.console.Console.print = safe_print
# Патчимо Swarms formatter
try:
import swarms.utils.formatter as formatter
# # Заміняємо _print_panel на заглушку
# def dummy_print_panel(*args, **kwargs):
# pass
# if hasattr(formatter, 'Formatter'):
# formatter.Formatter._print_panel = dummy_print_panel
# Патчимо print_panel_token_usage
def safe_print_panel_token_usage(self, *args, **kwargs):
try:
# Просто виводимо базову інформацію без форматування
if args and hasattr(args[0], '__dict__'):
data = args[0].__dict__ if hasattr(args[0], '__dict__') else str(args[0])
print(f"[Token Usage] {data}")
except:
pass
if hasattr(formatter, 'print_panel_token_usage'):
formatter.print_panel_token_usage = safe_print_panel_token_usage
except ImportError:
pass
# Патчимо threading для уникнення помилок в потоках
import threading
original_thread_init = threading.Thread.__init__
def safe_thread_init(self, *args, **kwargs):
"""Обгортаємо target функцію в try/except"""
original_target = kwargs.get('target')
if original_target:
def safe_target(*args, **kwargs):
try:
return original_target(*args, **kwargs)
except Exception:
# Ігноруємо помилки в потоках
pass
kwargs['target'] = safe_target
return original_thread_init(self, *args, **kwargs)
threading.Thread.__init__ = safe_thread_init
except Exception as e:
print(f"Warning: Could not patch Rich/Swarms: {e}")
# Застосовуємо патчі перед імпортом MAI-DX
patch_rich_and_swarms()
# Тепер можна безпечно імпортувати MAI-DX
try:
from mai_dx import MaiDxOrchestrator
MAI_DX_AVAILABLE = True
except ImportError as e:
MAI_DX_AVAILABLE = False
print(f"MAI-DX not available: {e}")
class SafeMaiDxOrchestrator:
"""Безпечна обгортка для MaiDxOrchestrator"""
def __init__(self, *args, **kwargs):
if not MAI_DX_AVAILABLE:
raise ImportError("MAI-DX is not available")
# Додаткові патчі безпосередньо перед створенням
patch_rich_and_swarms()
# Створюємо оригінальний orchestrator
self.orchestrator = MaiDxOrchestrator(*args, **kwargs)
def run(self, *args, **kwargs):
"""Виконати діагностику з додатковим захистом"""
# Переконуємося що патчі активні
patch_rich_and_swarms()
# Тимчасово змінюємо stderr щоб приховати помилки Rich
import io
original_stderr = sys.stderr
sys.stderr = io.StringIO()
try:
# Виконуємо оригінальний run
result = self.orchestrator.run(*args, **kwargs)
return result
finally:
# Відновлюємо stderr
sys.stderr = original_stderr
def __getattr__(self, name):
"""Проксі для всіх інших методів"""
return getattr(self.orchestrator, name)
# Експортуємо класи
__all__ = ['SafeMaiDxOrchestrator', 'MAI_DX_AVAILABLE', 'patch_rich_and_swarms'] |