Python
import pandas as pd
from transformers import pipeline
import os
import json
# für CPU-Fallback:
# pip install accelerate
# und
import torch
daten = pd.read_csv("beispielstudie/data/00_goldstandard.csv")
datenliste = list(daten["text"])
backup_dir = "beispielstudie/data/backup"
os.makedirs(backup_dir, exist_ok=True)
generate_gpu = pipeline("text-generation", model="meta-llama/Meta-Llama-3.1-8B-Instruct", max_new_tokens=256, do_sample=False)
def get_cpu_pipeline():
print("Lade CPU-Pipeline...")
cpu_pipeline = pipeline(
"text-generation",
model="meta-llama/Meta-Llama-3.1-8B-Instruct",
model_kwargs={"torch_dtype": torch.bfloat16},
device_map="cpu",
max_new_tokens=256, do_sample=False
)
print("CPU-Pipeline geladen!")
return cpu_pipeline
generate_cpu = None
ergebnis = []
for idx, text in enumerate(datenliste, start=1):
backup_file = f"{backup_dir}/{idx}.csv"
if os.path.exists(backup_file):
print(f"Text {idx} bereits klassifiziert, lade aus Backup...")
try:
existing_result = pd.read_csv(backup_file)
ergebnis.append({
"text": existing_result.iloc[0]["text"],
"response": existing_result.iloc[0]["response"]
})
continue
except Exception as e:
print(f"Fehler beim Laden des Backups für Text {idx}: {e}")
print("Führe neue Klassifikation durch...")
print(f"Verarbeite Text {idx} von {len(datenliste)} ...")
instructions = [
{"role": "system",
"content": "Du bist ein trainierter Assistent für Inhaltsanalyse, der die allgemeine Stimmung beziehungsweise Tonalität von Texten analysiert. Antworte immer präzise im JSON-Format mit sentiment (positiv, neutral, negativ) und reasoning (eine Begründung auf Deutsch). Gib ausschließlich den JSON-Response, beginnend mit '{' und endend mit '}', mit diesen zwei Parametern zurück"},
{"role": "user", "content": text}
]
try:
outputs = generate_gpu(instructions)
print(f"Text {idx} erfolgreich auf GPU verarbeitet")
except Exception as e:
print(f"GPU-Fehler bei Text {idx} ({e}). Verwende CPU...")
if generate_cpu is None:
generate_cpu = get_cpu_pipeline()
outputs = generate_cpu(instructions)
print(f"Text {idx} erfolgreich auf CPU verarbeitet")
result = {
"text": text,
"response": outputs[0]["generated_text"][-1]['content']
}
ergebnis.append(result)
try:
pd.DataFrame([result]).to_csv(backup_file, index=False)
print(f"Backup für Text {idx} gespeichert")
except Exception as e:
print(f"Fehler beim Speichern des Backups für Text {idx}: {e}")
def parse_response(response):
try:
parsed = json.loads(response)
return parsed['sentiment'], parsed['reasoning']
except (json.JSONDecodeError, KeyError):
return None, None
parsed_data = [(entry['text'], *parse_response(entry['response'])) for entry in ergebnis]
parsed_data = pd.DataFrame(parsed_data, columns=['text', 'sentiment', 'reasoning'])
vergleich = pd.DataFrame(parsed_data).drop("text", axis=1).join(daten)
vergleich.to_csv("beispielstudie/data/03_universelles_Decoder-Modell.csv", index=False)- 1
- Import der notwendigen Bibliotheken.
- 2
- Laden der Daten.
- 3
- Erstellen eines Backup-Verzeichnisses für Zwischenergebnisse.
- 4
- Initialisierung der Textgenerierungs-Pipeline für die GPU.
- 5
- Definition einer Funktion zur Initialisierung der CPU-Pipeline als Fallback.
- 6
- Initialisierung einer Liste zur Speicherung der Ergebnisse.
- 7
- Iteration über die Texte in den Daten.
- 8
- Überprüfung, ob ein Backup für den aktuellen Text existiert, und Laden dieses Backups, falls vorhanden.
- 9
- Definition der Anweisungen für das Modell, einschließlich der System- und Benutzerrollen.
- 10
- Versuch, den Text auf der GPU zu verarbeiten, und bei einem Fehler Wechsel zur CPU.
- 11
- Speichern des generierten Ergebnisses in der Ergebnisliste.
- 12
- Speichern des Ergebnisses in einer Backup-Datei.
- 13
- Definition einer Funktion zum Parsen der JSON-Antwort des Modells. Bei Fehlern im generierten JSON werden None-Werte zurückgegeben.
- 14
- Parsen der Antworten und Erstellen eines DataFrames mit den Ergebnissen.
- 15
- Zusammenführen der Ergebnisse mit den Originaldaten und Speichern in einer CSV-Datei.