Einführung
Die durchschnittliche Zeit, die ein Datenwissenschaftler mit der Vorbereitung und Bereinigung von Daten verbringt, beträgt etwa 45 % der Arbeitszeit. Diese Schätzung wird in zahlreichen Branchenumfragen immer wieder bestätigt. Die Aufgaben, die diese Zeit in Anspruch nehmen, umfassen das Profilieren von Spalten, das Kennzeichnen von Nullwerten, das Ausführen identischer explorativer Datenanalysen (EDA), das Durchführen von Grid-Search für Hyperparameter und das Schreiben von Monitoring-Checks. Diese Aufgaben sind so formelhaft, dass sie expliziten Regeln folgen und somit automatisierbar sind.
Genau das macht sie für Agenten geeignet. Agentische Workflows ersetzen nicht den Datenwissenschaftler, sondern übernehmen die prozeduralen Aufgaben, sodass sich die Fachkräfte auf die entscheidenden Aspekte konzentrieren können: die Beurteilung, ob ein Modell sinnvoll ist, ob ein Merkmal tatsächlich informativ ist und ob eine Erkenntnis eine geschäftliche Entscheidung rechtfertigt. Plattformen wie Databricks haben bereits damit begonnen, agentische Datenwissenschaftsfunktionen in ihre Kerninfrastruktur zu integrieren, wobei ihr Agenten-Framework speziell darauf ausgelegt ist, „die Zeit von der Frage zur Erkenntnis zu verkürzen“. Dies ist die Richtung, in die sich Produktionsteams im Bereich Daten bewegen.
Dieser Artikel behandelt fünf konkrete agentische Workflows, einen für jede Hauptphase einer Datenwissenschafts-Pipeline. Jeder Workflow umfasst ein reales Szenario, getestete Code-Muster und die Designentscheidungen, die in der Produktion von Bedeutung sind.
Voraussetzungen
Alle fünf Workflows setzen Python 3.10+ sowie Kenntnisse in pandas, scikit-learn und grundlegender Nutzung von APIs für große Sprachmodelle (LLM) voraus. Die spezifischen Paketanforderungen sind unter jedem Workflow aufgeführt. Für die Toolaufrufmuster benötigen Sie entweder einen OpenAI-API-Schlüssel oder einen lokalen Endpunkt (Ollama, vLLM), der eine OpenAI-kompatible API bereitstellt.
Wichtige Pakete, die in allen Workflows verwendet werden
- openai
- pandas
- numpy
- scipy
- scikit-learn
- lightgbm
- shap
- pydantic
Workflow 1: Automatisierter Explorativer Datenanalyse-Agent
Was ersetzt wird: Manuelles Laden von Daten, Berechnung von Zusammenfassungsstatistiken, Visualisierung von Verteilungen, Überprüfung von Nullwerten, Erkennung von Ausreißern und das Verfassen von Ergebnissen. Bei jedem Datensatz wird immer dasselbe Skript mit unterschiedlichen Spaltennamen verwendet.
Was der Agent stattdessen tut: Er lädt den Datensatz, führt ein vollständiges Profil durch, kennzeichnet Probleme nach Schweregrad und erstellt einen strukturierten Markdown-Bericht. Ein Mensch überprüft die Ergebnisse und entscheidet, wie weiter verfahren wird. Der Agent übernimmt alles, was vor dieser Überprüfung liegt.
Architektur
Der Agent nutzt eine Reasoning and Acting (ReAct)-Schleife mit zwei Werkzeugen: profile_dataset erzeugt Zusammenfassungsstatistiken pro Spalte, und flag_issues klassifiziert Probleme nach Schweregrad. Der Agent synthetisiert beide Ausgaben in einem strukturierten Bericht durch einen einzigen Aufruf des Sprachmodells. Die entscheidende Designentscheidung besteht darin, wie der Agent mit der Ausgabe von flag_issues umgeht; er überlegt, welche Probleme umsetzbar sind, bevor er berichtet, sodass die Ausgabe eine priorisierte Liste und kein Rohdump ist.
Code-Muster
# eda_agent.py
# Voraussetzungen: pip install openai pandas scipy
# Ausführen: python eda_agent.py
import json
import pandas as pd
from scipy import stats
from openai import OpenAI
from dataclasses import dataclass
client = OpenAI() # Verwendet die Umgebungsvariable OPENAI_API_KEY
@dataclass
class ColumnIssue:
column: str
issue_type: str # null_rate | skewness | dtype | high_correlation
severity: str # low | medium | high
detail: str
def profile_dataset(df: pd.DataFrame) -> dict:
\"\"\"
Generiere Statistiken pro Spalte.
In der Produktion sollte dies durch ydata-profiling für reichhaltigere Ausgaben ersetzt werden.
\"\"\"
profile = {}
for col in df.columns:
col_stats = {
\"dtype\": str(df[col].dtype),
\"null_rate\": df[col].isnull().mean(),
\"n_unique\": df[col].nunique(),
}
if pd.api.types.is_numeric_dtype(df[col]):
col_stats[\"skewness\"] = float(df[col].skew())
col_stats[\"mean\"] = float(df[col].mean())
col_stats[\"std\"] = float(df[col].std())
elif df[col].dtype == \"object\":
non_null = df[col].dropna()
numeric_coerced = pd.to_numeric(non_null, errors=\"coerce\")
col_stats[\"looks_numeric\"] = bool(len(non_null) > 0 and numeric_coerced.notna().mean() > 0.9)
profile[col] = col_stats
return profile
def flag_issues(profile: dict) -> list[ColumnIssue]:
\"\"\"
Kennzeichne Datenqualitätsprobleme aus einem Spaltenprofil.
Schweregrade: hoch = benötigt sofortige Aufmerksamkeit, mittel = wert zu überprüfen.
\"\"\"
issues = []
for col, stats_dict in profile.items():
null_rate = stats_dict.get(\"null_rate\", 0.0)
if null_rate > 0.15:
issues.append(ColumnIssue(col, \"null_rate\", \"high\",
f\"{null_rate:.0%} der Werte fehlen\"))
elif null_rate > 0.05:
issues.append(ColumnIssue(col, \"null_rate\", \"medium\",
f\"{null_rate:.0%} der Werte fehlen\"))
skewness = abs(stats_dict.get(\"skewness\", 0.0))
if skewness > 5.0:
issues.append(ColumnIssue(col, \"skewness\", \"high\",
f\"Extreme Schiefe={skewness:.1f} -- erwägen Sie eine Log-Transformation\"))
elif skewness > 2.0:
issues.append(ColumnIssue(col, \"skewness\", \"medium\",
f\"Moderate Schiefe={skewness:.1f}\"))
# Objektspalten mit ausschließlich numerischen Werten sind wahrscheinlich falsch codiert
if stats_dict[\"dtype\"] == \"object\" and stats_dict.get(\"looks_numeric\", False):
issues.append(ColumnIssue(col, \"dtype\", \"medium\",
\"Numerische Werte als Strings gespeichert\"))
return issues
def run_eda_agent(df: pd.DataFrame, dataset_description: str) -> str:
\"\"\"
Führen Sie die EDA-Agentenschleife aus.
Der Agent entscheidet, welche Werkzeuge in welcher Reihenfolge aufgerufen werden,
und erstellt dann einen strukturierten Bericht, der seine Ergebnisse zusammenfasst.
\"\"\"
profile = profile_dataset(df)
issues = flag_issues(profile)
# Formatieren Sie die Probleme für den Agenten
issues_text = \"\\n\".join(
f\"- [{i.severity.upper()}] {i.column}: {i.issue_type} -- {i.detail}\"
for i in issues
) or \"Keine Probleme festgestellt.\"
prompt = f\"\"\"Sie sind ein leitender Datenwissenschaftler, der einen Datensatz für ein Datenwissenschaftsprojekt überprüft.
Datensatz: {dataset_description}
Spaltenprofil (Zusammenfassungsstatistiken):
{json.dumps(profile, indent=2)}
Erkannte Probleme:
{issues_text}
Schreiben Sie einen strukturierten EDA-Bericht mit diesen Abschnitten:
1. DATENSATZ-ÜBERSICHT -- Form, dtypes, allgemeine Qualitätsbewertung (1-2 Sätze)
2. HOCHPRIORITÄRE PROBLEME -- Punkte, die vor dem Modellieren Maßnahmen erfordern
3. MITTELPRIORITÄRE PROBLEME -- Punkte, die überwacht werden sollten
4. EMPFOHLENE NÄCHSTE SCHRITTE -- geordnete Liste von 3-5 spezifischen Maßnahmen
Seien Sie direkt. Priorisieren Sie Umsetzbarkeit über Vollständigkeit.\"\"\"\n response = client.chat.completions.create(
model=\"gpt-4o-mini\",
messages=[{\"role\": \"user\", \"content\": prompt}],
temperature=0.2, # Niedrige Temperatur für konsistente strukturierte Ausgaben
)
return response.choices[0].message.content
# ── Führen Sie es aus ────────────────────────────────────────────────────────────────────
if __name__ == \"__main__\":
# Beispiel: Einzelhandels-Transaktionsdaten
import numpy as np
np.random.seed(42)
n = 5000
df = pd.DataFrame({
\"revenue\": np.random.exponential(scale=200, size=n), # rechtsschief
\"customer_age\": np.random.normal(40, 12, n),
\"created_at\": pd.date_range(\"2024-01-01\", periods=n, freq=\"h\").astype(str),
\"region_code\": np.random.choice([\"US\", \"EU\", \"APAC\", None], size=n, p=[0.5, 0.3, 0.1, 0.1]),
\"session_count\": np.where(np.random.rand(n) < 0.5, np.random.poisson(3, n), None),
})
result = run_eda_agent(df, \"Retail transaction data\")
print(result)
Reales Szenario
Einzelhandels-Transaktionsdaten, 5.000 Zeilen, 8 Spalten. Der Agent kennzeichnet den Umsatz als hochprioritär (extreme rechtsschiefe bei 7,3), die Sitzungsanzahl als hochprioritär (22 % Nullrate) und das Erstellungsdatum als mittelprioritär (Datum als String gespeichert). Er empfiehlt eine Log-Transformation für den Umsatz, ein Nullindikator-Feature für die Sitzungsanzahl und das Parsen des Erstellungsdatums, um Merkmale wie Stunde des Tages und Wochentag zu extrahieren. All dies wird in weniger als 30 Sekunden sichtbar. Ein Mensch überprüft den Bericht und handelt gemäß den Empfehlungen, ohne Zeit mit manuellen Diagnosen zu verbringen.
Workflow 2: Agentische Merkmalsentwicklung und -auswahl
Was ersetzt wird: Manuelles Brainstorming von Interaktionsmerkmalen, Schreiben des Transformationscodes, Evaluierung jedes Kandidaten mit einem Basismodell, Entfernen derjenigen, die keinen Beitrag leisten, und Dokumentation, was überlebt hat und warum.
Was der Agent stattdessen tut: Er schlägt Kandidatenmerkmale basierend auf dem Datenprofil und dem Kontext des Fachgebiets vor, generiert den Transformationscode, bewertet jeden Kandidaten gegen ein schnelles Basismodell und entfernt Merkmale, die unter einem konfigurierbaren Wichtigkeitsschwellenwert liegen, mit einer schriftlichen Begründung für jede Entscheidung. Weitere Informationen zu leistungsstarken Python-Dekoratoren finden Sie in unserem Artikel über leistungsstarke Python-Dekoratoren.
Architektur
Zwei Phasen, ein Agent. Die Generierungsphase nutzt das LLM, um Kandidatenmerkmale aus einer strukturierten Beschreibung des Datensatzes und der Vorhersageaufgabe vorzuschlagen. Die Auswahlphase bewertet jeden Kandidaten, indem ein LightGBM-Klassifikator mit 5-facher Kreuzvalidierung (CV) trainiert wird und die Merkmalswichtigkeit mithilfe von SHapley Additive exPlanations (SHAP) berechnet wird. Merkmale unter dem Schwellenwert werden entfernt. Der Agent überlegt sich die Wichtigkeitsscores, bevor er entfernt; er erkennt Fälle, in denen ein Merkmal global schwach aussieht, aber für ein bestimmtes Segment ein Signal trägt.
Code-Muster
# feature_agent.py
# Voraussetzungen: pip install openai lightgbm shap scikit-learn pandas numpy
# Ausführen: python feature_agent.py
import json
import numpy as np
import pandas as pd
from openai import OpenAI
from sklearn.model_selection import cross_val_score
from sklearn.preprocessing import LabelEncoder
import lightgbm as lgb
client = OpenAI()
def generate_feature_candidates(
column_descriptions: dict[str, str],
target: str,
task_type: str = "classification",
n_candidates: int = 10,
) -> list[dict]:
\"\"\"
Fragen Sie das LLM, um Kandidatenmerkmale basierend auf Spaltenbeschreibungen und der Vorhersageaufgabe vorzuschlagen.
Gibt eine Liste von Diktaten mit 'name', 'formula' und 'rationale' zurück.
\"\"\"
prompt = f\"\"\"Sie sind ein leitender ML-Ingenieur, der Merkmalsentwicklung für eine {task_type}-Aufgabe durchführt.
Zielvariable: {target}
Verfügbare Spalten:
{json.dumps(column_descriptions, indent=2)}
Schlagen Sie {n_candidates} Kandidatenmerkmale vor, die wahrscheinlich die Modellleistung verbessern.
Für jedes Merkmal geben Sie an:
- name: ein snake_case-Merkmalsname
- formula: wie man es aus den verfügbaren Spalten berechnet (pandas-Ausdruck)
- rationale: ein Satz, warum dieses Merkmal hilfreich sein könnte
Geben Sie ein JSON-Objekt mit einem einzigen Schlüssel "features" zurück, das ein Array von Objekten enthält,
jedes mit den Schlüsseln: name, formula, rationale.
Geben Sie NUR gültiges JSON zurück -- keine Erklärung außerhalb des JSON.\"\"\"\n response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0.4,
)
result = json.loads(response.choices[0].message.content)
return result.get("features", result.get("candidates", []))
def evaluate_and_prune(
df: pd.DataFrame,
candidate_features: list[dict],
target_col: str,
importance_threshold: float = 0.01,
) -> tuple[list[str], list[str], dict[str, float]]:
\"\"\"
Fügen Sie Kandidatenmerkmale zum DataFrame hinzu, trainieren Sie ein schnelles LightGBM-Basismodell,
extrahieren Sie Merkmalswichtigkeiten und entfernen Sie unter dem Schwellenwert.
Gibt (behaltene Merkmale, entfernte Merkmale, Wichtigkeitsscores) zurück.
\"\"\"
feature_df = df.copy()
added = []
for candidate in candidate_features:
try:
# Bewerten Sie die Formelzeichenfolge -- verwenden Sie in der Produktion eine sichere eval-Sandbox
feature_df[candidate[\"name\"]] = feature_df.eval(candidate[\"formula\"])
added.append(candidate[\"name\"])
except Exception as e:
# Formel fehlgeschlagen -- überspringen Sie diesen Kandidaten
print(f\" Übersprungen '{candidate['name']}': {e}\")
if not added:
return [], [], {}
X = feature_df[added].fillna(0)
y = df[target_col]
model = lgb.LGBMClassifier(n_estimators=100, random_state=42, verbose=-1)
model.fit(X, y)
importance_scores = dict(zip(added, model.feature_importances_ / model.feature_importances_.sum()))
kept = [f for f in added if importance_scores.get(f, 0) >= importance_threshold]
pruned = [f for f in added if importance_scores.get(f, 0) < importance_threshold]
return kept, pruned, importance_scores
def explain_selection(kept: list[str], pruned: list[str], scores: dict[str, float]) -> str:
\"\"\"
Fragen Sie den Agenten, seine Auswahlentscheidungen in einfacher Sprache zu erklären.
\"\"\"
prompt = f\"\"\"Sie überprüfen die Ergebnisse der Merkmalsauswahl für eine ML-Pipeline.
Behaltene Merkmale (über dem Wichtigkeitsschwellenwert):
{json.dumps({f: round(scores.get(f, 0), 4) for f in kept}, indent=2)}
Entfernte Merkmale (unter dem Schwellenwert):
{json.dumps({f: round(scores.get(f, 0), 4) for f in pruned}, indent=2)}
Schreiben Sie eine 3-5 Satz Zusammenfassung des Auswahl-Ergebnisses.
Hinweise zu überraschenden Entfernungen oder unerwartet wichtigen Merkmalen.
Schlagen Sie ein weiteres Merkmal vor, das basierend auf dem, was überlebt hat, getestet werden sollte.\"\"\"\n response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
)
return response.choices[0].message.content
if __name__ == \"__main__\":
column_descriptions = {
\"days_since_login\": \"Anzahl der Tage seit dem letzten Login des Kunden\",
\"plan_tier\": \"Abonnementstufe: basic, pro oder enterprise\",
\"support_tickets_90d\": \"Anzahl der in den letzten 90 Tagen eröffneten Support-Tickets\",
\"monthly_spend\": \"Durchschnittliche monatliche Ausgaben des Kunden in USD\",
}
candidates = generate_feature_candidates(
column_descriptions, target=\"churned\", task_type=\"classification\", n_candidates=10
)
# In der Produktion hier echte Kundendaten laden
np.random.seed(42)
n = 3000
df = pd.DataFrame({
\"days_since_login\": np.random.randint(0, 90, n),
```
```html
