utils.py

import pandas as pd
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

# Globale Analyseinstanz
_analyzer = None

def get_analyzer():
    """Singleton-Muster zur Vermeidung der Modellneuinitialisierung bei jedem Aufruf"""
    global _analyzer
    if _analyzer is None:
        _analyzer = CSVAnalysierer()
    return _analyzer

class CSVAnalysierer:
    def __init__(self):
        self.model_name = "mistralai/Mistral-7B-Instruct-v0.2"
        try:
            print("Modell wird initialisiert...")
            # Tokenizer-Initialisierung mit spezifischer Konfiguration
            self.tokenizer = AutoTokenizer.from_pretrained(
                self.model_name,
                trust_remote_code=True,
                use_fast=False
            )
            
            # Padding-Token-Konfiguration
            if self.tokenizer.pad_token is None:
                self.tokenizer.pad_token = self.tokenizer.eos_token
                self.tokenizer.padding_side = "right"

            # Modell-Initialisierung
            self.model = AutoModelForCausalLM.from_pretrained(
                self.model_name,
                torch_dtype=torch.float16,
                device_map="auto",
                trust_remote_code=True
            )
            
            # Sicherstellen, dass das Modell das Padding-Token kennt
            self.model.config.pad_token_id = self.tokenizer.pad_token_id
            print("Modell erfolgreich initialisiert!")
            
        except Exception as e:
            print(f"Initialisierungsfehler: {str(e)}")
            raise

    def kontext_vorbereiten(self, df: pd.DataFrame) -> str:
        """Bereitet den Kontext mit den DataFrame-Daten vor."""
        try:
            kontext = "E-Mail-Informationen:\n\n"
            
            # DataFrame in String umwandeln und fehlende Werte behandeln
            df_str = df.fillna("Keine Angabe").astype(str)
            
            # Jede Zeile verarbeiten
            for index in range(len(df_str)):
                zeile = df_str.iloc[index]
                kontext += f"E-Mail {index + 1}:\n"
                for spalte in df_str.columns:
                    kontext += f"{spalte}: {zeile[spalte]}\n"
                kontext += "---\n"
            
            return kontext.strip()
            
        except Exception as e:
            raise Exception(f"Fehler bei der Kontextvorbereitung: {str(e)}")

    def antwort_generieren(self, kontext: str, frage: str) -> str:
        """Generiert eine Antwort auf die Frage unter Verwendung des Kontexts."""
        prompt = f"""<s>[INST] Sie sind ein deutscher Assistent für Facility Management Datenanalyse. 
Analysieren Sie die folgenden E-Mail-Daten:

{kontext}

Frage: {frage}

Wichtige Anweisungen:
1. Antworten Sie AUSSCHLIEßLICH auf Deutsch
2. Beziehen Sie sich auf die konkreten E-Mail-Daten (Datum, Absender, Betreff)
3. Fassen Sie den relevanten Inhalt präzise zusammen
4. Bei Bedarf erstellen Sie eine strukturierte Übersicht der wichtigsten Punkte

Ihre deutsche Antwort: [/INST]"""

        try:
            eingabe = self.tokenizer(
                prompt,
                return_tensors="pt",
                padding=True,
                truncation=True,
                max_length=2048,
                pad_to_multiple_of=8,
                return_attention_mask=True
            ).to(self.model.device)
            
            with torch.no_grad():
                ausgabe = self.model.generate(
                    input_ids=eingabe["input_ids"],
                    attention_mask=eingabe["attention_mask"],
                    max_new_tokens=512,
                    temperature=0.7,
                    top_p=0.95,
                    repetition_penalty=1.15,
                    do_sample=True,
                    num_beams=1,
                    pad_token_id=self.tokenizer.pad_token_id,
                    eos_token_id=self.tokenizer.eos_token_id
                )
            
            antwort = self.tokenizer.decode(ausgabe[0], skip_special_tokens=True)
            antwort = antwort.split("[/INST]")[-1].strip()
            
            return antwort
            
        except Exception as e:
            return f"Generierungsfehler: {str(e)}"

def csv_analysieren(df: pd.DataFrame, frage: str) -> str:
    """Hauptfunktion zur CSV-Analyse und Fragenbeantwortung."""
    try:
        analysierer = get_analyzer()  # Verwendet die einzige Instanz
        kontext = analysierer.kontext_vorbereiten(df)
        antwort = analysierer.antwort_generieren(kontext, frage)
        return antwort
        
    except Exception as e:
        return f"Fehler bei der Analyse: {str(e)}"