书架 · Python 学习系列 · 06 · 数据科学与机器学习基础下一章：07 Web 服务 · FastAPI →

06 · 数据科学与机器学习基础

本章目标：会用 numpy 做数组运算、pandas 做数据处理、matplotlib 画图、scikit-learn 跑完整训练评估流程、pytorch 写第一个训练循环。并搞清楚定位问题：什么场景该训模型，什么场景调 API 就够。

前置：01、02（本章大量用到推导式、切片、广播思维）。

目录

1. 定位：你到底需不需要"训模型"
2. numpy：向量化运算的基石
3. pandas：DataFrame 数据处理
4. matplotlib：快速可视化
5. scikit-learn：传统 ML 全流程
6. pytorch：第一个训练循环
7. 自测清单与练习
8. 参考与出处

1. 定位：你到底需不需要训模型

☕ 给 Java 开发者的定位：本章技能是"数据在手时的分析武器"，不是 AI 应用的必修前置。你的主路线（LLM 应用/Agent）在前四章，这章按需取用。但 numpy 的"向量化思维"和 pandas 的数据处理，做 embedding/RAG 数据清洗时一定用得上。

安装（本章全家桶）：

uv add numpy pandas matplotlib scikit-learn
uv add torch          # Windows 默认装 CPU 版；GPU 版按 pytorch.org 官网选择器安装（见文末出处）

2. numpy：向量化运算的基石

2.1 为什么快

numpy 数组（ndarray）是连续内存的同构类型矩阵，运算走 C 级循环——"写循环不如写向量表达式"：

import numpy as np

data = [1, 2, 3, 4]

# ❌ Python 循环：慢 100 倍
result = [x * 2 for x in data]

# ✅ 向量化
arr = np.array(data)
arr * 2                  # array([2, 4, 6, 8])

2.2 核心操作

import numpy as np

a = np.array([[1, 2, 3],
              [4, 5, 6]])       # 2 行 3 列

a.shape          # (2, 3)
a.dtype          # int64
a.ndim           # 2

# 创建
np.zeros((3, 4)); np.ones(5); np.arange(0, 10, 2); np.linspace(0, 1, 5)
rng = np.random.default_rng(42)         # 现代随机 API，带种子可复现
rng.normal(size=(2, 3))                 # 正态分布

# 索引与切片（list 语法的矩阵版）
a[0, 1]          # 2        第 0 行第 1 列
a[:, 0]          # array([1, 4])   第 0 列（所有行）
a[0, :]          # array([1, 2, 3]) 第 0 行
a[a > 3]         # array([4, 5, 6]) 布尔掩码筛选 ⭐

# 广播（broadcasting）：形状不同的数组自动对齐运算
a = np.array([1, 2, 3])
a + 10                    # array([11, 12, 13])  标量广播到每个元素
m = np.ones((2, 3))
m + a                     # (2,3) + (3,) → 每行都加 a ⭐RAG 余弦相似度就是这么算的（第 04 章）

# 聚合
a.sum(); a.mean(); a.max(axis=0)      # axis=0 沿列压、axis=1 沿行压
np.argmax(a)                          # 最大值下标

# 线性代数（embedding 检索的核心）
u @ v                    # 点积
np.linalg.norm(u)        # 模长

☕ ≈ Java 无直接对应（ND4J/DL4J 生态小众）。心智模型：能用一行矩阵表达就绝不写 for——这也是 pytorch 张量的语法基础。

3. pandas：DataFrame 数据处理

DataFrame = 带标签的二维表（Excel/SQL 表的编程形态），数据清洗的事实标准。

import pandas as pd

# 构造
df = pd.DataFrame({
    "name": ["张三", "李四", "王五", "赵六"],
    "dept": ["dev", "ops", "dev", "hr"],
    "salary": [25, 18, 30, 15],
    "age": [30, 25, 35, 28],
})

# 看
df.head(2)          # 前两行
df.info()           # 行列、类型、非空计数
df.describe()       # 数值列统计摘要

# 选
df["salary"]                    # 单列（是 Series）
df[["name", "salary"]]          # 多列
df.loc[0, "name"]               # 按标签
df.iloc[0, 0]                   # 按下标
df[df["salary"] > 20]           # 条件筛选 ⭐
df[(df["dept"] == "dev") & (df["age"] < 32)]    # 多条件 & | ~（每个条件必须加括号！）

# 改
df["tax"] = df["salary"] * 0.1                  # 派生列（向量化，不写循环）
df.loc[df["dept"] == "hr", "dept"] = "people"   # 条件更新

# 分组聚合 —— ≈ SQL 的 GROUP BY / Java 的 Collectors.groupingBy + downstream
df.groupby("dept")["salary"].mean()             # 各部门平均薪资
df.groupby("dept").agg(avg_salary=("salary", "mean"),
                       max_age=("age", "max"))  # 多指标

# 排序 / 缺失值 / 去重
df.sort_values("salary", ascending=False)
df.isna().sum()                     # 每列缺失计数
df.dropna(subset=["age"])
df["salary"].fillna(df["salary"].median())
df.drop_duplicates(subset=["name"])

# 合并 —— ≈ SQL JOIN
pd.merge(df, depts_df, on="dept", how="left")

# 读写
df = pd.read_csv("data.csv", encoding="utf-8")
df.to_csv("out.csv", index=False)
df = pd.read_excel("报表.xlsx", sheet_name="Sheet1")   # 依赖 openpyxl

☕ pandas ≈ SQL + Stream 的合体：筛选=WHERE、groupby=GROUP BY、merge=JOIN、assign=SELECT 派生列。链式写法也流行：df.query("salary > 20").groupby("dept")[["salary"]].mean().round(1)。

4. matplotlib：快速可视化

import matplotlib.pyplot as plt

# 中文显示（Windows）
plt.rcParams["font.sans-serif"] = ["Microsoft YaHei"]
plt.rcParams["axes.unicode_minus"] = False

x = np.linspace(0, 10, 100)

fig, axes = plt.subplots(1, 2, figsize=(10, 4))   # 1 行 2 图
axes[0].plot(x, np.sin(x), label="sin")
axes[0].set(title="曲线", xlabel="x", ylabel="y")
axes[0].legend()
axes[1].bar(["dev", "ops", "hr"], [3, 2, 1])
axes[1].set_title("人数")
plt.tight_layout()
plt.savefig("report.png", dpi=150)    # 存文件（服务器/脚本环境）
plt.show()                            # 弹窗显示（本地）

日常记住四种图就够：plot 趋势、bar 对比、hist 分布、scatter 相关性。交互式探索用 df.plot()（pandas 内置快捷方式）。

5. scikit-learn：传统 ML 全流程

sklearn 统一了所有模型的 API 形状：fit / predict / score——换模型 = 换一个类名：

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix

# 数据集（真实场景用 pandas 读自己的数据；这里用 sklearn 自带的鸢尾花 150×4）
from sklearn.datasets import load_iris
X, y = load_iris(return_X_y=True)             # X: (n_samples, n_features)，y: 标签

# ① 切分：训练/测试集（严禁用测试集调参——数据泄漏）
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y)

# ② 预处理：标准化（仅用训练集的统计量 fit，防泄漏）
scaler = StandardScaler()
X_train_s = scaler.fit_transform(X_train)
X_test_s = scaler.transform(X_test)           # ⚠️ 测试集只 transform 不 fit

# ③ 训练
model = RandomForestClassifier(n_estimators=200, random_state=42)
model.fit(X_train_s, y_train)

# ④ 评估
y_pred = model.predict(X_test_s)
print(classification_report(y_test, y_pred))  # precision/recall/f1
print(confusion_matrix(y_test, y_pred))

# ⑤ 使用
model.predict_proba(new_sample)               # 概率输出

进阶一件套（交叉验证 + 网格搜索）：

from sklearn.model_selection import GridSearchCV, cross_val_score
cross_val_score(model, X_train_s, y_train, cv=5)      # 5 折交叉验证
grid = GridSearchCV(RandomForestClassifier(random_state=42),
                    {"n_estimators": [100, 200, 400], "max_depth": [5, 10, None]},
                    cv=5, scoring="f1_macro")
grid.fit(X_train_s, y_train)
grid.best_params_, grid.best_score_

☕ 面向对象直觉：所有 estimator 都实现 fit/predict，≈ 实现统一接口的策略模式；Pipeline 把预处理+模型串成一条链防泄漏：Pipeline([("scaler", StandardScaler()), ("clf", RandomForestClassifier())]).fit(X_train, y_train)。

6. pytorch：第一个训练循环

pytorch 2.x 是深度学习框架事实标准（训练侧）。核心抽象：张量（Tensor）+ 自动求导（autograd）+ 神经网络模块（nn.Module）。

6.1 张量与 autograd

import torch

x = torch.tensor([1.0, 2.0, 3.0])
m = torch.ones(3, 3)
x.device                    # cpu（GPU 版：x.to("cuda")）

w = torch.tensor(2.0, requires_grad=True)    # 声明需要求梯度
y = w ** 2                                  # 前向
y.backward()                                # 反向传播：自动算 dy/dw
w.grad                                      # tensor(4.) = 2*w

6.2 手写线性回归训练循环（理解一切深度学习框架的骨架）

"""y = 3x + 0.5 的最小完整训练：定义模型 → 损失 → 优化器 → 循环（前向/反向/更新）"""
import torch
import torch.nn as nn

torch.manual_seed(42)
X = torch.linspace(0, 1, 100).reshape(-1, 1)
y = 3 * X + 0.5 + 0.05 * torch.randn(X.shape)    # 带噪声的数据

model = nn.Linear(in_features=1, out_features=1)  # 待学的 w、b 就在里面
loss_fn = nn.MSELoss()                            # 损失：均方误差
optimizer = torch.optim.SGD(model.parameters(), lr=0.1)   # 优化器

for epoch in range(200):                          # ← 所有 DL 训练都是这个循环
    pred = model(X)                               # ① 前向
    loss = loss_fn(pred, y)                       # ② 算损失
    optimizer.zero_grad()                         # ③ 清旧梯度（pytorch 默认累加）
    loss.backward()                               # ④ 反向求梯度
    optimizer.step()                              # ⑤ 更新参数
    if epoch % 50 == 0:
        print(f"epoch={epoch} loss={loss.item():.4f}")

print(model.weight, model.bias)     # ≈ weight 3.0、bias 0.5 —— 学出来了

把这 5 步循环刻进脑子：前向 → 损失 → 清梯度 → 反向 → 更新。换成 CNN/Transformer，变的只是 model 的结构，循环一字不改。

6.3 nn.Module 自定义模型

import torch
import torch.nn as nn

class MLP(nn.Module):                       # 所有模型都继承 nn.Module
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(10, 64), nn.ReLU(),
            nn.Linear(64, 1),
        )
    def forward(self, x):                   # 只定义前向，梯度自动
        return self.net(x)

GPU（本地有 N 卡时）：model.to("cuda") + 数据 X.to("cuda")，其余代码不变。Windows 下 uv add torch 默认 CPU 版；CUDA 版按 pytorch 官网选择器给的命令装。

7. 自测清单与练习

• [ ] 向量化比循环快的原因？"能用矩阵就不写 for"用 numpy 怎么表达布尔筛选？
• [ ] (2,3) + (3,) 广播结果是什么形状？RAG 余弦相似度用了哪个矩阵运算？
• [ ] pandas 多条件筛选为什么每个条件要加括号？groupby 对应 SQL 什么？
• [ ] sklearn 五步流程？为什么测试集只能 transform 不能 fit？
• [ ] pytorch 训练循环 5 步？zero_grad() 为什么必须？
• [ ] 你的业务里哪类问题该调 API、哪类该 sklearn、哪类才需要 pytorch？

练习 1（numpy）：生成 100 个 3 维随机向量，实现余弦相似度矩阵，找出与查询向量最相近的 top-3（就是第 04 章 RAG 的内核）。 练习 2（pandas）：读一个真实 CSV（如公司导出的日志），完成：缺失值统计 → 按某维度 groupby 聚合 → 输出前 10 行到新 CSV。 练习 3（sklearn）：用 sklearn 自带的 load_iris 走完五步流程，对比 LogisticRegression 和 RandomForest 的报告。 练习 4（pytorch）：把 6.2 改成用 MLP（6.3 结构）拟合 y = sin(2πx)，观察 200/1000/5000 轮的拟合效果差别。

8. 参考与出处

以下均为官方一手来源（访问日期：2026-09-05）：

⬅️ 返回目录 ｜ ➡️ 下一章：07-Web服务-FastAPI