本文目录导读：
我来为您提供几个常见的Python模型调用案例，包括scikit-learn、PyTorch和TensorFlow等框架。
scikit-learn 模型调用示例
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, accuracy_score
import joblib
# 创建或加载数据
def create_sample_data():
X = np.random.randn(100, 4)
y = np.random.randint(0, 2, 100)
return X, y
# 训练模型并保存
def train_and_save_model():
X, y = create_sample_data()
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 创建和训练模型
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
# 保存模型
joblib.dump(model, 'random_forest_model.joblib')
# 评估模型
y_pred = model.predict(X_test)
print("准确率:", accuracy_score(y_test, y_pred))
print("分类报告:\n", classification_report(y_test, y_pred))
return model
# 加载并使用模型
def load_and_predict(model_path='random_forest_model.joblib'):
# 加载模型
model = joblib.load(model_path)
# 新数据预测
new_data = np.array([[0.5, -0.2, 0.1, 0.3]])
prediction = model.predict(new_data)
probability = model.predict_proba(new_data)
print(f"预测结果: {prediction[0]}")
print(f"预测概率: {probability}")
return prediction
# 使用示例
if __name__ == "__main__":
# 训练模型
model = train_and_save_model()
# 预测新数据
result = load_and_predict()
PyTorch 模型调用示例
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
# 定义简单的神经网络
class SimpleNN(nn.Module):
def __init__(self, input_size, hidden_size, num_classes):
super(SimpleNN, self).__init__()
self.layer1 = nn.Linear(input_size, hidden_size)
self.relu = nn.ReLU()
self.layer2 = nn.Linear(hidden_size, num_classes)
def forward(self, x):
out = self.layer1(x)
out = self.relu(out)
out = self.layer2(out)
return out
# 训练模型
def train_pytorch_model():
# 创建数据
X = torch.randn(100, 10)
y = torch.randint(0, 2, (100,))
dataset = TensorDataset(X, y)
dataloader = DataLoader(dataset, batch_size=16, shuffle=True)
# 初始化模型
model = SimpleNN(input_size=10, hidden_size=50, num_classes=2)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
# 训练
num_epochs = 10
for epoch in range(num_epochs):
for batch_X, batch_y in dataloader:
optimizer.zero_grad()
outputs = model(batch_X)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
if (epoch + 1) % 2 == 0:
print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.4f}')
# 保存模型
torch.save(model.state_dict(), 'pytorch_model.pth')
print("模型已保存为 pytorch_model.pth")
return model
# 加载和使用模型
def load_pytorch_model(model_path='pytorch_model.pth'):
# 创建相同结构的模型
model = SimpleNN(input_size=10, hidden_size=50, num_classes=2)
model.load_state_dict(torch.load(model_path))
model.eval() # 切换到评估模式
# 新数据预测
new_data = torch.randn(1, 10)
with torch.no_grad(): # 不需要计算梯度
output = model(new_data)
_, predicted = torch.max(output, 1)
print(f"预测结果: {predicted.item()}")
print(f"各分类概率: {torch.softmax(output, dim=1)}")
return predicted.item()
# 使用示例
if __name__ == "__main__":
model = train_pytorch_model()
result = load_pytorch_model()
TensorFlow/Keras 模型调用示例
import tensorflow as tf
from tensorflow import keras
import numpy as np
# 创建和训练模型
def train_tensorflow_model():
# 创建数据
X_train = np.random.randn(100, 10)
y_train = np.random.randint(0, 2, 100)
# 构建模型
model = keras.Sequential([
keras.layers.Dense(64, activation='relu', input_shape=(10,)),
keras.layers.Dropout(0.3),
keras.layers.Dense(32, activation='relu'),
keras.layers.Dense(2, activation='softmax')
])
# 编译模型
model.compile(optimizer='adam',
loss='sparse_categorical_crossentropy',
metrics=['accuracy'])
# 训练模型
history = model.fit(X_train, y_train,
epochs=10,
batch_size=16,
validation_split=0.2,
verbose=1)
# 保存模型
model.save('tensorflow_model.h5')
print("模型已保存为 tensorflow_model.h5")
return model
# 加载和使用模型
def load_tensorflow_model(model_path='tensorflow_model.h5'):
# 加载模型
model = keras.models.load_model(model_path)
# 新数据预测
new_data = np.random.randn(3, 10)
predictions = model.predict(new_data)
print("预测结果:")
for i, pred in enumerate(predictions):
predicted_class = np.argmax(pred)
confidence = pred[predicted_class]
print(f"样本 {i+1}: 类别 {predicted_class}, 置信度 {confidence:.4f}")
return predictions
# 使用示例
if __name__ == "__main__":
model = train_tensorflow_model()
predictions = load_tensorflow_model()
通用模型调用模式（推荐）
# model_handler.py
import json
import numpy as np
from typing import Any, Dict, List, Union
import joblib
class ModelHandler:
"""通用模型处理器"""
def __init__(self, model_path: str, model_type: str = 'sklearn'):
self.model_path = model_path
self.model_type = model_type
self.model = None
self.load_model()
def load_model(self):
"""加载模型"""
if self.model_type == 'sklearn':
self.model = joblib.load(self.model_path)
elif self.model_type == 'pytorch':
# PyTorch加载逻辑
pass
elif self.model_type == 'tensorflow':
# TensorFlow加载逻辑
pass
else:
raise ValueError(f"不支持的模型类型: {self.model_type}")
def predict(self, data: Union[np.ndarray, List]) -> Dict[str, Any]:
"""预测"""
if isinstance(data, list):
data = np.array(data)
predictions = self.model.predict(data)
# 尝试获取预测概率
try:
probabilities = self.model.predict_proba(data)
has_probability = True
except:
probabilities = None
has_probability = False
result = {
'predictions': predictions.tolist(),
'probabilities': probabilities.tolist() if has_probability else None,
'model_type': self.model_type
}
return result
def batch_predict(self, data_list: List[Union[np.ndarray, List]]) -> List[Dict]:
"""批量预测"""
return [self.predict(data) for data in data_list]
# 调用示例
if __name__ == "__main__":
# 创建模型实例
handler = ModelHandler('random_forest_model.joblib', 'sklearn')
# 单个预测
single_result = handler.predict([[0.5, -0.2, 0.1, 0.3]])
print("单个预测:", json.dumps(single_result, indent=2))
# 批量预测
batch_data = [
[0.5, -0.2, 0.1, 0.3],
[0.1, 0.2, -0.5, 0.8],
[-0.3, 0.4, 0.6, -0.1]
]
batch_results = handler.batch_predict(batch_data)
print("批量预测结果:")
for i, result in enumerate(batch_results):
print(f"样本 {i+1}: {result['predictions']}")
配置驱动的模型调用
# model_config.yaml model: name: "my_model" type: "sklearn" # sklearn, pytorch, tensorflow path: "./models/random_forest_model.joblib" inference: batch_size: 32 device: "cpu" # cpu, cuda precision: "float32" preprocessing: normalize: True feature_names: ["feature1", "feature2", "feature3", "feature4"]
# config_based_model.py
import yaml
import numpy as np
import joblib
class ConfigBasedModel:
def __init__(self, config_path: str):
with open(config_path, 'r') as f:
self.config = yaml.safe_load(f)
self.model = self._load_model()
self.preprocessing_config = self.config.get('preprocessing', {})
self.inference_config = self.config.get('inference', {})
def _load_model(self):
model_path = self.config['model']['path']
model_type = self.config['model']['type']
if model_type == 'sklearn':
return joblib.load(model_path)
else:
raise ValueError(f"Unknown model type: {model_type}")
def _preprocess(self, data: np.ndarray) -> np.ndarray:
if self.preprocessing_config.get('normalize', False):
# 添加你的预处理逻辑
data = (data - np.mean(data, axis=0)) / np.std(data, axis=0)
return data
def predict(self, data: np.ndarray):
data = self._preprocess(data)
return self.model.predict(data)
# 使用示例
if __name__ == "__main__":
model = ConfigBasedModel('model_config.yaml')
data = np.random.randn(5, 4)
predictions = model.predict(data)
print("预测结果:", predictions)
关键要点
- 模型保存：使用适当的格式（joblib、pickle、h5等）
- 错误处理：添加try-catch处理加载失败情况
- 输入验证：检查输入数据的格式和类型
- 批处理支持：支持单样本和批量预测
- 日志记录：添加日志便于调试
- 配置管理：使用配置文件管理模型参数
选择哪种模式取决于您的具体需求和应用场景。