Spaces:
Running
Running
File size: 4,350 Bytes
8e30b6a c0426da 8e30b6a c0426da da164cc c0426da dcb5a1a 8e30b6a da164cc dcb5a1a c0426da da164cc c0426da da164cc c0426da da164cc c0426da da164cc c0426da da164cc c0426da da164cc 8e30b6a c0426da 8e30b6a da164cc 8e30b6a da164cc dcb5a1a da164cc 8e30b6a da164cc | 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 | import logging
import time
import gc
from typing import Dict, Any
from app.config_manager import get_active_text_model, is_multi_model_enabled
from app.core.config import get_settings
from app.utils.exceptions import SetupRequiredError
from transformers import pipeline
logger = logging.getLogger(__name__)
# Słownik do keszowania klasyfikatorów w RAM (zapobiega ciągłemu przeładowywaniu przy multi-modelu)
_loaded_classifiers = {}
def _load_model(target_model_name: str):
global _loaded_classifiers
if target_model_name in _loaded_classifiers:
return _loaded_classifiers[target_model_name]
logger.info(f"Model {target_model_name} nie jest załadowany. Ładowanie do RAM...")
gc.collect()
_loaded_classifiers[target_model_name] = pipeline(
"text-classification",
model=target_model_name,
device=-1
)
logger.info(f"Model {target_model_name} został pomyślnie załadowany.")
return _loaded_classifiers[target_model_name]
async def analyze_text(text: str, guild_id: str) -> Dict[str, Any]:
start_time = time.time()
settings = get_settings()
# Sprawdzamy, czy włączony jest tryb wielomodelowy
multi_model_active = is_multi_model_enabled(guild_id)
if multi_model_active:
models_to_run = settings.AVAILABLE_MODELS.get("text", [])
if not models_to_run:
raise ValueError("Brak zdefiniowanych modeli tekstowych w ustawieniach systemu.")
logger.info(f"Rozpoczęcie wielomodelowej analizy tekstu dla serwera {guild_id} ({len(models_to_run)} modeli)")
individual_results = []
for m in models_to_run:
try:
classifier = _load_model(m)
result = classifier(text)
label = result[0]["label"]
score = result[0]["score"]
is_fake = label.lower() in ["fake", "ai", "chatgpt", "label_1", "machine-generated"]
individual_results.append({
"model": m,
"is_deepfake": is_fake,
"confidence": round(score, 3)
})
except Exception as e:
logger.error(f"Błąd modelu {m} podczas wielomodelowej analizy: {e}")
if not individual_results:
raise ValueError("Żaden z modeli tekstowych nie dokonał pomyślnej analizy.")
# Agregacja: Głosowanie większościowe
fake_votes = sum(1 for r in individual_results if r["is_deepfake"])
is_deepfake = fake_votes > (len(individual_results) / 2)
# Pewność: Średnia pewność wszystkich modeli
confidence = sum(r["confidence"] for r in individual_results) / len(individual_results)
analysis_time = time.time() - start_time
return {
"is_deepfake": is_deepfake,
"confidence": round(confidence, 3),
"analysis_time": round(analysis_time, 3),
"used_model": "Multi-Model Workflow (Ensemble)",
"details": individual_results # Przekazujemy szczegóły do bota
}
else:
# Tradycyjna analiza pojedynczego modelu
active_model = get_active_text_model(guild_id)
if not active_model:
logger.warning(f"Zablokowano zapytanie! Serwer {guild_id} nie ma skonfigurowanego modelu.")
raise SetupRequiredError(
f"Serwer o ID '{guild_id}' nie został jeszcze skonfigurowany. "
"Użyj komendy setup na Discordzie przed wykonaniem analizy."
)
logger.info(f"Rozpoczęcie analizy tekstu dla serwera {guild_id} przy użyciu modelu: {active_model}")
classifier = _load_model(active_model)
result = classifier(text)
label = result[0]["label"]
score = result[0]["score"]
is_deepfake = label.lower() in ["fake", "ai", "chatgpt", "label_1", "machine-generated"]
confidence = score
analysis_time = time.time() - start_time
return {
"is_deepfake": is_deepfake,
"confidence": round(confidence, 3),
"analysis_time": round(analysis_time, 3),
"used_model": active_model,
"details": None
} |