AI Agent在智能感知系统中的数据预处理与优化策略

import numpy as np
import pandas as pd
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.ensemble import IsolationForest
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score

class DataPreprocessingAgent:
    def __init__(self):
        self.imputer = SimpleImputer(strategy="mean")
        self.scaler = StandardScaler()
        self.pca = PCA(n_components=5)
        self.outlier_detector = IsolationForest(contamination=0.05)
    
    def clean_data(self, X):
        # 填补缺失值
        X_imputed = self.imputer.fit_transform(X)
        # 去除异常值
        mask = self.outlier_detector.fit_predict(X_imputed) == 1
        return X_imputed[mask]
    
    def optimize_features(self, X):
        # 标准化
        X_scaled = self.scaler.fit_transform(X)
        # PCA降维
        return self.pca.fit_transform(X_scaled)
    
    def preprocess(self, X):
        X_clean = self.clean_data(X)
        X_opt = self.optimize_features(X_clean)
        return X_opt

# ===== 模拟数据集 =====
np.random.seed(42)
X = np.random.randn(500, 10) * 5
y = np.random.randint(0, 2, 500)

# 注入缺失值
X.ravel()[np.random.choice(X.size, 50, replace=False)] = np.nan

# 使用AI Agent预处理
agent = DataPreprocessingAgent()
X_processed = agent.preprocess(X)

# 划分数据集并训练模型
X_train, X_test, y_train, y_test = train_test_split(X_processed, y, test_size=0.3, random_state=42)
clf = RandomForestClassifier()
clf.fit(X_train, y_train)

# 评估
y_pred = clf.predict(X_test)
print("Accuracy:", accuracy_score(y_test, y_pred))
print("F1 Score:", f1_score(y_test, y_pred))

class BaseModule:
    def process(self, X):
        raise NotImplementedError

# ===== 各类模块实现 =====
class MissingValueHandler(BaseModule):
    def __init__(self, strategy="mean"):
        from sklearn.impute import SimpleImputer
        self.imputer = SimpleImputer(strategy=strategy)
    
    def process(self, X):
        return self.imputer.fit_transform(X)

class OutlierRemover(BaseModule):
    def __init__(self, contamination=0.05):
        from sklearn.ensemble import IsolationForest
        self.detector = IsolationForest(contamination=contamination)
    
    def process(self, X):
        mask = self.detector.fit_predict(X) == 1
        return X[mask]

class FeatureScaler(BaseModule):
    def __init__(self):
        from sklearn.preprocessing import StandardScaler
        self.scaler = StandardScaler()
    
    def process(self, X):
        return self.scaler.fit_transform(X)

class DimensionalityReducer(BaseModule):
    def __init__(self, method="pca", n_components=5):
        from sklearn.decomposition import PCA
        self.reducer = PCA(n_components=n_components) if method == "pca" else None
    
    def process(self, X):
        return self.reducer.fit_transform(X)

# ===== Agent组装器 =====
class ModularDataAgent:
    def __init__(self, modules):
        self.modules = modules
    
    def preprocess(self, X):
        for module in self.modules:
            X = module.process(X)
        return X

# 使用示例
modules = [
    MissingValueHandler(strategy="mean"),
    OutlierRemover(contamination=0.05),
    FeatureScaler(),
    DimensionalityReducer(method="pca", n_components=5)
]

agent = ModularDataAgent(modules)
X_processed = agent.preprocess(X)