Python 数据科学与 AI：Pandas/Polars、PyTorch 推理与 ONNX 部署

import numpy as np

# 向量化替代循环（1000x 加速）
# ❌ 慢
result = []
for x in range(1_000_000):
    result.append(x ** 2 + 2 * x + 1)

# ✅ 快
x = np.arange(1_000_000)
result = x ** 2 + 2 * x + 1  # 广播，无 Python 循环

# 高级索引
arr = np.random.randn(1000, 1000)
mask = arr > 2  # 布尔掩码
outliers = arr[mask]  # 只取大于 2 的元素

# 内存视图（零拷贝）
sub = arr[:100, :100]  # 不复制数据

import pandas as pd

# 读取大文件的 4 个技巧
df = pd.read_csv(
    "large.csv",
    usecols=["user_id", "amount", "timestamp"],  # 只读需要的列
    dtype={"user_id": "int32", "amount": "float32"},  # 降精度
    parse_dates=["timestamp"],
    chunksize=100_000  # 分块处理
)

# 高效分组聚合
# ❌ 慢：apply 调用 Python 函数
df.groupby("category").apply(lambda x: x["amount"].sum())

# ✅ 快：向量化聚合
agg = df.groupby("category").agg({
    "amount": ["sum", "mean", "count"],
    "user_id": "nunique"
})

# 时间序列重采样
# 按小时统计订单量
df.set_index("timestamp").resample("H")["order_id"].count()

import polars as pl

# 惰性查询（查询优化类似 SQL 执行计划）
df = pl.scan_csv("huge.csv")  # 不立即读数据

result = (
    df
    .filter(pl.col("amount") > 100)
    .group_by("category")
    .agg([
        pl.col("amount").sum().alias("total"),
        pl.col("user_id").n_unique().alias("unique_users")
    ])
    .sort("total", descending=True)
    .limit(10)
    .collect()  # 这里才真正执行
)

# 与 Pandas 互操作
pandas_df = result.to_pandas()
polars_df = pl.from_pandas(pandas_df)

import torch
from torchvision import models, transforms
from PIL import Image

# 加载预训练模型（ResNet50）
model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V2)
model.eval()

# 推理模式（禁用 dropout、batchnorm 更新）
with torch.no_grad():
    # 预处理管道
    preprocess = transforms.Compose([
        transforms.Resize(256),
        transforms.CenterCrop(224),
        transforms.ToTensor(),
        transforms.Normalize(
            mean=[0.485, 0.456, 0.406],
            std=[0.229, 0.224, 0.225]
        )
    ])
    
    img = Image.open("cat.jpg")
    input_tensor = preprocess(img).unsqueeze(0)
    
    # 推理
    output = model(input_tensor)
    probabilities = torch.nn.functional.softmax(output[0], dim=0)
    
    # Top 5 预测
    top5_prob, top5_idx = torch.topk(probabilities, 5)

# 1. 半精度加速（FP16）
model.half()  # 权重转 FP16
input_tensor = input_tensor.half()

# 2. 批处理推理（摊平启动开销）
def batch_predict(model, images: list[Image.Image], batch_size: int = 32):
    results = []
    for i in range(0, len(images), batch_size):
        batch = images[i:i + batch_size]
        tensors = torch.stack([preprocess(img) for img in batch])
        with torch.no_grad():
            outputs = model(tensors)
        results.extend(outputs)
    return results

# 3. TorchScript 编译（图优化）
scripted = torch.jit.script(model)
# 或 traced = torch.jit.trace(model, example_input)

# 4. 推理服务封装
class InferenceService:
    def __init__(self, model_path: str, device: str = "cpu"):
        self.device = torch.device(device)
        self.model = torch.load(model_path, map_location=self.device)
        self.model.eval().to(self.device)
    
    @torch.inference_mode()
    def predict(self, image: Image.Image) -> dict:
        tensor = preprocess(image).unsqueeze(0).to(self.device)
        output = self.model(tensor)
        return {" logits": output.cpu().numpy().tolist()}

from transformers import pipeline

# 文本分类
classifier = pipeline(
    "sentiment-analysis",
    model="distilbert-base-uncased-finetuned-sst-2-english"
)
result = classifier("I love FastAPI!")
# [{'label': 'POSITIVE', 'score': 0.9998}]

# 命名实体识别
ner = pipeline("ner", model="dslim/bert-base-NER")
ner("Apple Inc. was founded by Steve Jobs in Cupertino.")

# 文本生成（LLM）
generator = pipeline(
    "text-generation",
    model="microsoft/Phi-3-mini-4k-instruct",
    torch_dtype="auto",
    device_map="auto"  # 自动分配到 GPU/CPU
)

prompt = "Write a Python function to calculate fibonacci:"
output = generator(prompt, max_new_tokens=100, temperature=0.7)

from transformers import AutoModelForCausalLM, BitsAndBytesConfig

quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,  # 4-bit 量化
    bnb_4bit_compute_dtype=torch.float16,
    bnb_4bit_quant_type="nf4",  # Normal Float 4
    bnb_4bit_use_double_quant=True,
)

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b",
    quantization_config=quantization_config,
    device_map="auto",
)
# 7B 模型从 14GB 降至 4GB

from vllm import LLM

# 比 Transformers 快 10-20x 的批处理推理
llm = LLM(model="meta-llama/Llama-2-7b", tensor_parallel_size=1)

prompts = [
    "What is Python used for?",
    "Explain async/await in one sentence.",
]
outputs = llm.generate(prompts)

for output in outputs:
    print(output.outputs[0].text)

import torch

dummy_input = torch.randn(1, 3, 224, 224)

torch.onnx.export(
    model,
    dummy_input,
    "resnet50.onnx",
    input_names=["input"],
    output_names=["output"],
    dynamic_axes={"input": {0: "batch_size"}, "output": {0: "batch_size"}},
    opset_version=17,
)

import onnxruntime as ort
import numpy as np

session = ort.InferenceSession("resnet50.onnx")

# 输入准备
input_name = session.get_inputs()[0].name
input_data = np.random.randn(1, 3, 224, 224).astype(np.float32)

# 推理
outputs = session.run(None, {input_name: input_data})

# 不同执行提供器
providers = [
    "CUDAExecutionProvider",  # GPU
    "TensorRTExecutionProvider",  # NVIDIA TensorRT
    "CPUExecutionProvider",  # 回退
]
session = ort.InferenceSession("model.onnx", providers=providers)

// Rust 代码（lib.rs）
use pyo3::prelude::*;

#[pyfunction]
fn fast_sum(numbers: Vec<f64>) -> f64 {
    numbers.iter().sum()
}

#[pymodule]
fn my_rust_lib(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(fast_sum, m)?)?;
    Ok(())
}

# Python 调用
import maturin  # pip install maturin
# maturin develop  # 编译并安装

from my_rust_lib import fast_sum

# 比 Python sum() 快 100x
result = fast_sum([1.0] * 10_000_000)

from pathlib import Path
import polars as pl
from transformers import pipeline
import torch

class DataPipeline:
    def __init__(self):
        self.classifier = pipeline(
            "sentiment-analysis",
            model="nlptown/bert-base-multilingual-uncased-sentiment",
            device=0 if torch.cuda.is_available() else -1
        )
    
    def process_csv(self, path: Path) -> pl.DataFrame:
        # 1. 读取（惰性）
        df = pl.scan_csv(path)
        
        # 2. 清洗
        df = df.filter(
            pl.col("text").is_not_null() &
            (pl.col("text").str.len_chars() > 10)
        )
        
        # 3. 采样（开发时）
        # df = df.limit(1000)
        
        # 4. 收集为 Pandas（兼容性）
        return df.collect().to_pandas()
    
    def analyze_sentiment(self, df: pd.DataFrame) -> pd.DataFrame:
        texts = df["text"].tolist()
        
        # 批处理推理
        results = self.classifier(texts, batch_size=32)
        
        df["sentiment"] = [r["label"] for r in results]
        df["confidence"] = [r["score"] for r in results]
        return df
    
    def run(self, input_path: Path, output_path: Path):
        df = self.process_csv(input_path)
        df = self.analyze_sentiment(df)
        df.to_parquet(output_path)  # 列式存储，比 CSV 快 10x

# 执行
pipeline = DataPipeline()
pipeline.run(Path("reviews.csv"), Path("reviews_analyzed.parquet"))