Python+AI 学习路线：从基础语法到 LLM 应用实战

分阶段学习指南

🟢 阶段 0：Python 基础（2-4 周）

学习目标：掌握 Python 核心语法和编程思维

核心知识点

• Python 基础：数据类型、控制流程、函数、面向对象、文件操作、异常处理
• int, float, str：基本数值与字符串处理
• list, dict, tuple, set：常用数据结构
• if/else, for/while 循环：逻辑控制
• 函数定义, lambda 表达式：代码复用
• 装饰器基础：高级特性
• 类与对象，继承与多态：OOP 核心

必学代码示例

示例 1：列表推导式 这里我们对比传统写法与 Pythonic 写法，后者更简洁高效。

# 传统方式
squares = []
for i in range(10):
    squares.append(i ** 2)

# Pythonic 方式
squares = [i ** 2 for i in range(10)]

# 带条件的列表推导式
even_squares = [i ** 2 for i in range(10) if i % 2 == 0]
print(even_squares)  # [0, 4, 16, 36, 64]

示例 2：字典操作 注意 defaultdict 在处理计数时的便利性。

# 字典推导式
word_count = "hello world hello python"
counts = {word: word_count.split().count(word) for word in set(word_count.split())}
print(counts)  # {'hello': 2, 'world': 1, 'python': 1}

# defaultdict 使用
from collections import defaultdict
counts = defaultdict(int)
for word in word_count.split():
    counts[word] += 1

示例 3：上下文管理器 使用 with 语句能自动管理资源，即使发生异常也能安全关闭文件。

# 正确的文件操作方式
with open('data.txt', 'r', encoding='utf-8') as f:
    content = f.read()
# 自动关闭文件，即使发生异常

# 自定义上下文管理器
from contextlib import contextmanager

@contextmanager
def timer():
    import time
    start = time.time()
    yield
    print(f"耗时：{time.time() - start:.2f}秒")

with timer():
    sum(range(1000000))

🔵 阶段 1：数据科学基础（4-6 周）

学习目标：掌握数据处理、分析和可视化技能

核心技能树

• NumPy：数组创建、运算、广播机制、线性代数
• Pandas：Series 操作、DataFrame 操作、数据清洗、分组、合并
• 可视化：Matplotlib、Seaborn、Plotly

NumPy 实战代码

import numpy as np

# 数组创建
arr = np.array([[1, 2, 3], [4, 5, 6]])
zeros = np.zeros((3, 4))
random = np.random.randn(3, 3)

# 数组运算
print(arr * 2)          # 元素级乘法
print(arr @ arr.T)      # 矩阵乘法
print(np.dot(arr, arr.T))  # 矩阵乘法

# 广播机制
a = np.array([[1, 2, 3], [4, 5, 6]])  # (2, 3)
b = np.array([10, 20, 30])            # (3,)
print(a + b)  # b 广播到 (2, 3)

# 实用函数
print(np.mean(arr, axis=1))  # 按行求均值
print(np.argmax(arr, axis=1))  # 按行找最大值索引

Pandas 数据处理实战

import pandas as pd

# 创建 DataFrame
data = {
    'name': ['张三', '李四', '王五', '赵六'],
    'age': [25, 30, 35, 28],
    'city': ['北京', '上海', '深圳', '杭州'],
    'salary': [15000, 20000, 25000, 18000]
}
df = pd.DataFrame(data)

# 数据筛选
high_salary = df[df['salary'] > 18000]
beijing = df[df['city'] == '北京']

# 数据分组
city_stats = df.groupby('city').agg({'salary': ['mean', 'max', 'count']})

# 数据排序
df_sorted = df.sort_values('salary', ascending=False)

# 数据合并
df2 = pd.DataFrame({'name': ['张三', '李四'], 'department': ['技术', '产品']})
merged = pd.merge(df, df2, on='name', how='left')
print(city_stats)

数据可视化示例

import matplotlib.pyplot as plt
import seaborn as sns

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 创建子图
fig, axes = plt.subplots(2, 2, figsize=(12, 10))

# 1. 柱状图
axes[0, 0].bar(df['name'], df['salary'])
axes[0, 0].set_title('薪资对比')
axes[0, 0].set_xlabel('姓名')
axes[0, 0].set_ylabel('薪资')

# 2. 散点图
axes[0, 1].scatter(df['age'], df['salary'], s=100, alpha=0.6)
axes[0, 1].set_title('年龄与薪资关系')
axes[0, 1].set_xlabel('年龄')
axes[0, 1].set_ylabel('薪资')

# 3. 饼图（城市分布）
city_counts = df['city'].value_counts()
axes[1, 0].pie(city_counts, labels=city_counts.index, autopct='%1.1f%%')
axes[1, 0].set_title('城市分布')

# 4. 箱线图
axes[1, 1].boxplot(df['salary'])
axes[1, 1].set_title('薪资分布')

plt.tight_layout()
plt.savefig('visualization.png', dpi=300)
plt.show()

🟡 阶段 2：机器学习（6-8 周）

学习目标：理解 ML 原理，掌握 Scikit-learn 实战

ML 算法分类图

• 监督学习：回归（线性、决策树、随机森林）、分类（逻辑回归、SVM、XGBoost）
• 无监督学习：聚类（K-Means、DBSCAN）、降维（PCA、t-SNE）
• 强化学习：策略优化

经典算法实现

1. 线性回归完整流程

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error, r2_score
from sklearn.preprocessing import StandardScaler

# 生成模拟数据
np.random.seed(42)
n_samples = 1000
# 特征：面积、房间数、房龄
X = np.random.randn(n_samples, 3)
X[:, 0] = X[:, 0] * 50 + 100  # 面积：50-150㎡
X[:, 1] = np.abs(X[:, 1]) * 2 + 1  # 房间数：1-5 间
X[:, 2] = np.abs(X[:, 2]) * 10 + 1  # 房龄：1-30 年

# 真实价格 = 面积*1000 + 房间数*50000 - 房龄*2000 + 噪声
y = (X[:, 0] * 1000 + X[:, 1] * 50000 - X[:, 2] * 2000 + np.random.randn(n_samples) * 50000)

# 划分数据集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 特征标准化
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

# 训练模型
model = LinearRegression()
model.fit(X_train_scaled, y_train)

# 预测
y_pred = model.predict(X_test_scaled)

# 评估
mse = mean_squared_error(y_test, y_pred)
r2 = r2_score(y_test, y_pred)
print(f"均方误差 MSE: {mse:.2f}")
print(f"决定系数 R²: {r2:.4f}")
print(f"系数：{model.coef_}")
print(f"截距：{model.intercept_:.2f}")

# 可视化预测结果
plt.figure(figsize=(10, 6))
plt.scatter(y_test, y_pred, alpha=0.5)
plt.plot([y.min(), y.max()], [y.min(), y.max()], 'r--', lw=2)
plt.xlabel('真实价格')
plt.ylabel('预测价格')
plt.title('房价预测：真实值 vs 预测值')
plt.grid(True, alpha=0.3)
plt.show()

2. 分类算法对比

from sklearn.datasets import make_classification
from sklearn.model_selection import cross_val_score
from sklearn.linear_model import LogisticRegression
from sklearn.svm import SVC
from sklearn.tree import DecisionTreeClassifier
from sklearn.ensemble import RandomForestClassifier
from sklearn.naive_bayes import GaussianNB
from sklearn.neighbors import KNeighborsClassifier

# 生成分类数据
X, y = make_classification(
    n_samples=1000, n_features=20, n_informative=15,
    n_redundant=5, n_classes=3, random_state=42
)

# 定义模型
models = {
    '逻辑回归': LogisticRegression(max_iter=1000),
    'SVM': SVC(),
    '决策树': DecisionTreeClassifier(),
    '随机森林': RandomForestClassifier(n_estimators=100),
    '朴素贝叶斯': GaussianNB(),
    'KNN': KNeighborsClassifier()
}

# 交叉验证评估
results = {}
for name, model in models.items():
    scores = cross_val_score(model, X, y, cv=5, scoring='accuracy')
    results[name] = {'mean': scores.mean(), 'std': scores.std()}
    print(f"{name}: {scores.mean():.4f} (+/- {scores.std():.4f})")

# 可视化对比
plt.figure(figsize=(12, 6))
names = list(results.keys())
means = [results[name]['mean'] for name in names]
stds = [results[name]['std'] for name in names]
plt.bar(names, means, yerr=stds, alpha=0.8, capsize=5)
plt.ylabel('准确率')
plt.title('不同分类算法性能对比（5 折交叉验证）')
plt.ylim(0.7, 1.0)
plt.grid(axis='y', alpha=0.3)
plt.xticks(rotation=15)
plt.show()

3. 聚类算法实现

from sklearn.cluster import KMeans, DBSCAN
from sklearn.datasets import make_blobs
from sklearn.metrics import silhouette_score

# 生成聚类数据
X, _ = make_blobs(n_samples=500, centers=4, cluster_std=1.5, random_state=42)

# K-Means 聚类（需要指定簇数）
kmeans = KMeans(n_clusters=4, random_state=42, n_init=10)
kmeans_labels = kmeans.fit_predict(X)
kmeans_silhouette = silhouette_score(X, kmeans_labels)

# DBSCAN 聚类（自动发现簇数）
dbscan = DBSCAN(eps=1.5, min_samples=10)
dbscan_labels = dbscan.fit_predict(X)
n_clusters = len(set(dbscan_labels)) - (1 if -1 in dbscan_labels else 0)

print(f"K-Means: 发现 4 个簇，轮廓系数={kmeans_silhouette:.3f}")
print(f"DBSCAN: 发现{n_clusters}个簇")

# 肘部法则确定最佳 K 值
inertias = []
K_range = range(2, 11)
for K in K_range:
    kmeans = KMeans(n_clusters=K, random_state=42, n_init=10)
    kmeans.fit(X)
    inertias.append(kmeans.inertia_)

plt.figure(figsize=(12, 5))
# 肘部图
plt.subplot(1, 2, 1)
plt.plot(K_range, inertias, 'bo-')
plt.xlabel('K 值')
plt.ylabel('惯性（Inertia）')
plt.title('肘部法则确定最佳 K 值')
plt.grid(True, alpha=0.3)

# 聚类结果可视化
plt.subplot(1, 2, 2)
plt.scatter(X[:, 0], X[:, 1], c=kmeans_labels, cmap='viridis', alpha=0.6)
plt.scatter(kmeans.cluster_centers_[:, 0], kmeans.cluster_centers_[:, 1], c='red', s=200, marker='X', label='质心')
plt.title('K-Means 聚类结果')
plt.legend()
plt.tight_layout()
plt.show()

🟠 阶段 3：深度学习（8-12 周）

学习目标：掌握 PyTorch，理解深度学习原理

深度学习框架选择

• PyTorch：研究首选，动态图，Python 原生风格
• TensorFlow：工业部署，TF-Serving, TFLite 移动端
• JAX：函数式编程，自动微分，高性能计算

PyTorch 实战代码

1. 构建神经网络

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
import matplotlib.pyplot as plt

# 检查 CUDA 可用性
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f"使用设备：{device}")

# 定义神经网络
class NeuralNetwork(nn.Module):
    def __init__(self, input_size, hidden_size, num_classes):
        super(NeuralNetwork, self).__init__()
        self.layer1 = nn.Linear(input_size, hidden_size)
        self.relu = nn.ReLU()
        self.layer2 = nn.Linear(hidden_size, hidden_size // 2)
        self.layer3 = nn.Linear(hidden_size // 2, num_classes)
        self.dropout = nn.Dropout(0.2)

    def forward(self, x):
        out = self.layer1(x)
        out = self.relu(out)
        out = self.dropout(out)
        out = self.layer2(out)
        out = self.relu(out)
        out = self.dropout(out)
        out = self.layer3(out)
        return out

# 超参数
input_size = 784  # MNIST 图像 28x28
hidden_size = 256
num_classes = 10
num_epochs = 10
batch_size = 100
learning_rate = 0.001

# 初始化模型
model = NeuralNetwork(input_size, hidden_size, num_classes).to(device)

# 损失函数和优化器
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# 模拟数据
X_train = torch.randn(1000, input_size).to(device)
y_train = torch.randint(0, num_classes, (1000,)).to(device)
train_dataset = TensorDataset(X_train, y_train)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

# 训练循环
train_losses = []
for epoch in range(num_epochs):
    model.train()
    epoch_loss = 0
    for i, (images, labels) in enumerate(train_loader):
        # 前向传播
        outputs = model(images)
        loss = criterion(outputs, labels)
        
        # 反向传播和优化
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        epoch_loss += loss.item()
    
    avg_loss = epoch_loss / len(train_loader)
    train_losses.append(avg_loss)
    print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {avg_loss:.4f}')

# 绘制训练曲线
plt.figure(figsize=(10, 5))
plt.plot(train_losses, marker='o')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.title('训练损失曲线')
plt.grid(True, alpha=0.3)
plt.show()

2. CNN 图像分类

import torch
import torch.nn as nn
import torch.nn.functional as F

class CNN(nn.Module):
    def __init__(self, num_classes=10):
        super(CNN, self).__init__()
        # 第一个卷积块
        self.conv1 = nn.Conv2d(1, 32, kernel_size=3, padding=1)
        self.bn1 = nn.BatchNorm2d(32)
        self.conv2 = nn.Conv2d(32, 32, kernel_size=3, padding=1)
        self.bn2 = nn.BatchNorm2d(32)
        self.pool1 = nn.MaxPool2d(2, 2)
        
        # 第二个卷积块
        self.conv3 = nn.Conv2d(32, 64, kernel_size=3, padding=1)
        self.bn3 = nn.BatchNorm2d(64)
        self.conv4 = nn.Conv2d(64, 64, kernel_size=3, padding=1)
        self.bn4 = nn.BatchNorm2d(64)
        self.pool2 = nn.MaxPool2d(2, 2)
        
        # 全连接层
        self.fc1 = nn.Linear(64 * 7 * 7, 256)
        self.dropout = nn.Dropout(0.5)
        self.fc2 = nn.Linear(256, num_classes)

    def forward(self, x):
        # 第一个卷积块
        x = F.relu(self.bn1(self.conv1(x)))
        x = F.relu(self.bn2(self.conv2(x)))
        x = self.pool1(x)
        
        # 第二个卷积块
        x = F.relu(self.bn3(self.conv3(x)))
        x = F.relu(self.bn4(self.conv4(x)))
        x = self.pool2(x)
        
        # 展平
        x = x.view(-1, 64 * 7 * 7)
        
        # 全连接层
        x = F.relu(self.fc1(x))
        x = self.dropout(x)
        x = self.fc2(x)
        return x

# 模型结构可视化
model = CNN()
print(model)

# 统计参数量
total_params = sum(p.numel() for p in model.parameters())
trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
print(f"总参数量：{total_params:,}")
print(f"可训练参数量：{trainable_params:,}")

3. Transformer 注意力机制

import torch
import torch.nn as nn
import math

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super(MultiHeadAttention, self).__init__()
        assert d_model % num_heads == 0
        self.d_model = d_model
        self.num_heads = num_heads
        self.d_k = d_model // num_heads
        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)

    def scaled_dot_product_attention(self, Q, K, V, mask=None):
        scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)
        if mask is not None:
            scores = scores.masked_fill(mask == 0, -1e9)
        attention_weights = F.softmax(scores, dim=-1)
        output = torch.matmul(attention_weights, V)
        return output, attention_weights

    def forward(self, query, key, value, mask=None):
        batch_size = query.size(0)
        
        # 线性变换并分割成多头
        Q = self.W_q(query).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
        K = self.W_k(key).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
        V = self.W_v(value).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
        
        # 计算注意力
        x, attention_weights = self.scaled_dot_product_attention(Q, K, V, mask)
        
        # 拼接多头
        x = x.transpose(1, 2).contiguous().view(batch_size, -1, self.d_model)
        
        # 最终线性变换
        output = self.W_o(x)
        return output, attention_weights

# 使用示例
d_model = 512
num_heads = 8
seq_length = 10
batch_size = 4

# 创建模拟输入
x = torch.randn(batch_size, seq_length, d_model)

# 创建注意力层
mha = MultiHeadAttention(d_model, num_heads)
output, attention = mha(x, x, x)
print(f"输入形状：{x.shape}")
print(f"输出形状：{output.shape}")
print(f"注意力权重形状：{attention.shape}")

🔴 阶段 4：NLP 与 LLM 应用（6-8 周）

学习目标：掌握现代 NLP 技术，熟练使用大语言模型

NLP 技术发展时间线

• 2017: Transformer
• 2018: BERT/GPT
• 2019: GPT-2
• 2020: GPT-3
• 2022: ChatGPT
• 2023: GPT-4/Llama2
• 2024: 多模态大模型

Transformers 实战

1. 使用预训练模型

from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch

# 情感分析
sentiment_pipeline = pipeline("sentiment-analysis", model="distilbert-base-uncased-finetuned-sst-2-english")
texts = [
    "I love this product! It's amazing!",
    "This is the worst experience ever.",
    "It's okay, nothing special."
]
results = sentiment_pipeline(texts)
for text, result in zip(texts, results):
    print(f"文本：{text}")
    print(f"情感：{result['label']}, 置信度：{result['score']:.4f}\n")

# 文本分类
tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese")
model = AutoModelForSequenceClassification.from_pretrained("bert-base-chinese", num_labels=3)  # 正面、中性、负面

# 编码文本
text = "这家餐厅的菜品味道很好，服务也很周到！"
inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)

# 预测
with torch.no_grad():
    outputs = model(**inputs)
    predictions = torch.nn.functional.softmax(outputs.logits, dim=-1)
    predicted_class = torch.argmax(predictions).item()
    labels = ["负面", "中性", "正面"]
    print(f"预测类别：{labels[predicted_class]}")
    print(f"置信度：{predictions[0][predicted_class]:.4f}")

2. RAG 检索增强生成

from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

# 1. 加载文档
loader = TextLoader('knowledge_base.txt')
documents = loader.load()

# 2. 文本切分
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500, chunk_overlap=50, length_function=len
)
splits = text_splitter.split_documents(documents)

# 3. 创建向量存储
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)
vectorstore = FAISS.from_documents(splits, embeddings)

# 4. 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity", search_kwargs={"k": 3}
)

# 5. 自定义提示模板
prompt_template = """使用以下上下文信息来回答问题。如果不知道答案，就说不知道，不要编造答案。
上下文信息：{context}
问题：{question}
答案:"""
PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# 6. 创建 QA 链
qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(temperature=0),
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True,
    chain_type_kwargs={"prompt": PROMPT}
)

# 7. 查询
query = "如何办理社保卡？"
result = qa_chain({"query": query})
print(f"问题：{query}")
print(f"答案：{result['result']}")
print("\n参考文档:")
for doc in result['source_documents']:
    print(f"- {doc.page_content[:100]}...")

3. 简单的 ChatGLM 对话示例

from transformers import AutoTokenizer, AutoModel
import torch

# 加载模型
tokenizer = AutoTokenizer.from_pretrained("THUDM/chatglm3-6b", trust_remote_code=True)
model = AutoModel.from_pretrained("THUDM/chatglm3-6b", trust_remote_code=True).half().cuda()
model = model.eval()

# 对话历史
history = []
response = "你好！我是智能助手小政，有什么可以帮助您的吗？"
print(f"助手：{response}")

while True:
    user_input = input("\n用户：")
    if user_input.lower() in ['退出', 'exit', 'quit']:
        break
    
    # 添加政务系统人设
    system_prompt = "你是一个政务服务大厅的智能引导员，名叫'小政'。"
    response, history = model.chat(
        tokenizer,
        f"{system_prompt}\n用户：{user_input}",
        history=history,
        max_length=2048,
        temperature=0.7
    )
    print(f"小政：{response}")
    history.append((user_input, response))

🟣 阶段 5：实战项目（持续进行）

项目推荐

端到端项目示例：智能文档问答系统

project_structure.txt
"""
智能文档问答系统
│
├── data/           # 数据目录
│   ├── documents/  # 原始文档
│   └── vectorstore/ # 向量存储
│
├── src/            # 源代码
│   ├── config.py   # 配置文件
│   ├── loader.py   # 文档加载
│   ├── embeddings.py # 向量化
│   ├── retriever.py # 检索器
│   ├── generator.py # 生成器
│   └── api.py      # API 接口
│
├── app.py          # 主应用
├── requirements.txt # 依赖
└── README.md       # 说明文档
"""

config.py

import os
from dataclasses import dataclass
from typing import Optional

@dataclass
class Config:
    # API 密钥
    OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
    
    # 模型配置
    EMBEDDING_MODEL: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
    LLM_MODEL: str = "gpt-3.5-turbo"
    LLM_TEMPERATURE: float = 0.7
    LLM_MAX_TOKENS: int = 1000
    
    # 向量存储配置
    CHUNK_SIZE: int = 500
    CHUNK_OVERLAP: int = 50
    VECTOR_DB_PATH: str = "data/vectorstore"
    
    # 检索配置
    TOP_K: int = 3
    SIMILARITY_THRESHOLD: float = 0.7
    
    # API 配置
    API_HOST: str = "0.0.0.0"
    API_PORT: int = 8000

loader.py

from typing import List
from langchain.document_loaders import (
    TextLoader, PyPDFLoader, DirectoryLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document

class DocumentLoader:
    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
            separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""]
        )

    def load_text(self, file_path: str) -> List[Document]:
        """加载文本文件"""
        loader = TextLoader(file_path, encoding='utf-8')
        documents = loader.load()
        return self.text_splitter.split_documents(documents)

    def load_pdf(self, file_path: str) -> List[Document]:
        """加载 PDF 文件"""
        loader = PyPDFLoader(file_path)
        documents = loader.load()
        return self.text_splitter.split_documents(documents)

    def load_directory(self, directory: str, glob: str = "**/*.txt") -> List[Document]:
        """加载目录下的所有文档"""
        loader = DirectoryLoader(directory, glob=glob)
        documents = loader.load()
        return self.text_splitter.split_documents(documents)

api.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn

app = FastAPI(title="智能文档问答系统", version="1.0.0")

class QueryRequest(BaseModel):
    question: str
    top_k: Optional[int] = 3

class QueryResponse(BaseModel):
    answer: str
    sources: List[str]
    confidence: float

@app.get("/")
async def root():
    return {"message": "智能文档问答系统 API", "version": "1.0.0", "endpoints": {"/query": "POST - 问答接口", "/health": "GET - 健康检查"}}

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    try:
        # 这里调用实际的问答逻辑
        # result = qa_chain({"query": request.question})
        return QueryResponse(
            answer="这是示例回答",
            sources=["来源 1", "来源 2"],
            confidence=0.95
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    uvicorn.run("api:app", host="0.0.0.0", port=8000, reload=True)

实战代码示例

示例 1：完整的机器学习项目

"""
项目：预测客户是否会购买理财产品
数据集：模拟银行客户数据
"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.model_selection import train_test_split, cross_val_score, GridSearchCV
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score, roc_curve
from sklearn.pipeline import Pipeline
import warnings
warnings.filterwarnings('ignore')

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# ========== 1. 数据生成 ==========
np.random.seed(42)
n_samples = 5000
data = {
    '年龄': np.random.randint(18, 70, n_samples),
    '收入': np.random.randint(3000, 50000, n_samples),
    '存款': np.random.randint(0, 1000000, n_samples),
    '债务': np.random.randint(0, 500000, n_samples),
    '信用评分': np.random.randint(300, 850, n_samples),
    '已购买产品数': np.random.randint(0, 10, n_samples),
    '上次购买天数': np.random.randint(30, 3650, n_samples),
    '职业': np.random.choice(['学生', '上班族', '个体户', '退休', '自由职业'], n_samples),
    '婚姻状况': np.random.choice(['单身', '已婚', '离异'], n_samples),
    '学历': np.random.choice(['高中', '本科', '硕士', '博士'], n_samples),
}
df = pd.DataFrame(data)

# 生成目标变量（基于特征组合）
def calc_purchase_prob(row):
    score = 0
    if 25 <= row['年龄'] <= 55: score += 20
    if row['收入'] > 15000: score += 20
    if row['存款'] > 100000: score += 20
    if row['信用评分'] > 650: score += 15
    if row['职业'] in ['上班族', '个体户']: score += 15
    return min(score + np.random.randint(-10, 10), 100) / 100

df['购买概率'] = df.apply(calc_purchase_prob, axis=1)
df['是否购买'] = (df['购买概率'] > 0.5).astype(int)

# ========== 2. 数据探索 ==========
print("=" * 50)
print("数据集基本信息")
print("=" * 50)
print(df.info())
print("\n目标变量分布:")
print(df['是否购买'].value_counts())
print(f"购买率：{df['是否购买'].mean():.2%}")

# 可视化
fig, axes = plt.subplots(2, 3, figsize=(15, 10))

# 年龄分布
axes[0, 0].hist(df[df['是否购买'] == 0]['年龄'], bins=30, alpha=0.5, label='未购买')
axes[0, 0].hist(df[df['是否购买'] == 1]['年龄'], bins=30, alpha=0.5, label='购买')
axes[0, 0].set_xlabel('年龄')
axes[0, 0].set_ylabel('人数')
axes[0, 0].legend()
axes[0, 0].set_title('年龄与购买关系')

# 收入分布
axes[0, 1].hist(df[df['是否购买'] == 0]['收入'], bins=30, alpha=0.5, label='未购买')
axes[0, 1].hist(df[df['是否购买'] == 1]['收入'], bins=30, alpha=0.5, label='购买')
axes[0, 1].set_xlabel('收入')
axes[0, 1].legend()
axes[0, 1].set_title('收入与购买关系')

# 信用评分分布
axes[0, 2].hist(df[df['是否购买'] == 0]['信用评分'], bins=30, alpha=0.5, label='未购买')
axes[0, 2].hist(df[df['是否购买'] == 1]['信用评分'], bins=30, alpha=0.5, label='购买')
axes[0, 2].set_xlabel('信用评分')
axes[0, 2].legend()
axes[0, 2].set_title('信用评分与购买关系')

# 职业分布
career_purchase = df.groupby('职业')['是否购买'].mean()
axes[1, 0].bar(career_purchase.index, career_purchase.values)
axes[1, 0].set_ylabel('购买率')
axes[1, 0].set_title('不同职业的购买率')

# 学历分布
edu_purchase = df.groupby('学历')['是否购买'].mean()
axes[1, 1].bar(edu_purchase.index, edu_purchase.values)
axes[1, 1].set_ylabel('购买率')
axes[1, 1].set_title('不同学历的购买率')

# 相关性热图
numeric_cols = ['年龄', '收入', '存款', '债务', '信用评分', '已购买产品数', '上次购买天数', '是否购买']
correlation = df[numeric_cols].corr()
sns.heatmap(correlation, annot=True, fmt='.2f', cmap='coolwarm', center=0, ax=axes[1, 2])
axes[1, 2].set_title('特征相关性热图')

plt.tight_layout()
plt.savefig('data_exploration.png', dpi=300)
plt.show()

# ========== 3. 特征工程 ==========
# 编码分类变量
le = LabelEncoder()
df['职业编码'] = le.fit_transform(df['职业'])
df['婚姻编码'] = le.fit_transform(df['婚姻状况'])
df['学历编码'] = le.fit_transform(df['学历'])

# 创建新特征
df['债务收入比'] = df['债务'] / (df['收入'] * 12 + 1)
df['存款收入比'] = df['存款'] / (df['收入'] * 12 + 1)
df['净资产'] = df['存款'] - df['债务']

# 选择特征
feature_cols = ['年龄', '收入', '存款', '债务', '信用评分', '已购买产品数', '上次购买天数', '职业编码', '婚姻编码', '学历编码', '债务收入比', '存款收入比', '净资产']
X = df[feature_cols]
y = df['是否购买']

# 划分数据集
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# ========== 4. 模型训练与评估 ==========
# 定义模型
models = {
    '逻辑回归': Pipeline([('scaler', StandardScaler()), ('model', LogisticRegression(max_iter=1000, random_state=42))]),
    '随机森林': RandomForestClassifier(n_estimators=100, random_state=42),
    '梯度提升': GradientBoostingClassifier(random_state=42)
}

# 训练和评估
results = {}
for name, model in models.items():
    print(f"\n训练 {name}...")
    model.fit(X_train, y_train)
    
    # 预测
    y_pred = model.predict(X_test)
    y_pred_proba = model.predict_proba(X_test)[:, 1]
    
    # 评估
    results[name] = {
        'model': model,
        'predictions': y_pred,
        'probabilities': y_pred_proba,
        'accuracy': model.score(X_test, y_test),
        'roc_auc': roc_auc_score(y_test, y_pred_proba)
    }
    print(f"准确率：{results[name]['accuracy']:.4f}")
    print(f"AUC: {results[name]['roc_auc']:.4f}")

# ========== 5. 最佳模型分析 ==========
best_model_name = max(results, key=lambda x: results[x]['roc_auc'])
best_model = results[best_model_name]['model']
print(f"\n最佳模型：{best_model_name}")
print("=" * 50)
print("\n分类报告:")
print(classification_report(y_test, results[best_model_name]['predictions']))

# 混淆矩阵
cm = confusion_matrix(y_test, results[best_model_name]['predictions'])
plt.figure(figsize=(8, 6))
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues')
plt.xlabel('预测标签')
plt.ylabel('真实标签')
plt.title(f'{best_model_name} - 混淆矩阵')
plt.show()

# ROC 曲线
plt.figure(figsize=(10, 6))
for name, result in results.items():
    fpr, tpr, _ = roc_curve(y_test, result['probabilities'])
    plt.plot(fpr, tpr, label=f"{name} (AUC = {result['roc_auc']:.3f})")
plt.plot([0, 1], [0, 1], 'k--', label='随机分类器')
plt.xlabel('假正率')
plt.ylabel('真正率')
plt.title('ROC 曲线对比')
plt.legend()
plt.grid(alpha=0.3)
plt.show()

# 特征重要性（树模型）
if hasattr(best_model, 'feature_importances_'):
    feature_importance = pd.DataFrame({
        'feature': feature_cols,
        'importance': best_model.feature_importances_
    }).sort_values('importance', ascending=False)
    plt.figure(figsize=(10, 6))
    plt.barh(feature_importance['feature'], feature_importance['importance'])
    plt.xlabel('重要性')
    plt.title('特征重要性')
    plt.tight_layout()
    plt.show()
    print("\n特征重要性排序:")
    print(feature_importance)

print("\n" + "=" * 50)
print("项目完成！")
print("=" * 50)

常见问题解答

Q1: 没有编程基础，能学 Python+AI 吗？

答: 完全可以！Python 是公认最适合初学者的语言。建议学习路径：

1. 花 2-3 周打好 Python 基础
2. 从简单的数据分析项目入手
3. 逐步过渡到机器学习
4. 边学边做，保持实践

Q2: 数学基础不好，能学 AI 吗？

答: 可以，但需要补充必要的数学知识：

建议: 边做项目边补数学，遇到不懂的再学。

Q3: 学习多久能找到工作？

答: 因人而异，大致时间线：

关键: 项目经验 > 理论知识，一定要做项目！

Q4: GPU 不够用怎么办？

答: 多种解决方案：

1. 云平台：Google Colab（免费 GPU）、Kaggle Notebooks（免费）、AutoDL（便宜）
2. 模型压缩：量化（Quantization）、剪枝（Pruning）、知识蒸馏（Distillation）

优化技巧：

# 减小 batch size
batch_size = 16  # 而不是 64

# 使用混合精度训练
from torch.cuda.amp import autocast, GradScaler
scaler = GradScaler()

# 梯度累积
accumulation_steps = 4

Q5: 如何保持技术更新？

答: AI 技术迭代快，建议：

1. 关注前沿：arXiv.org（论文预印本）、Hugging Face（新模型）、Twitter/X（大佬动态）
2. 实践导向：每月做一个小项目、参与 Kaggle 竞赛、贡献开源项目
3. 社区交流：加入技术 Discord/微信群、参加线下 meetup、写技术博客

总结

核心要点

1. 循序渐进：不要急于求成，按阶段学习
2. 项目驱动：理论结合实践，多做项目
3. 持续学习：AI 技术更新快，保持学习热情
4. 社区参与：加入社区，与他人交流学习
5. 定期复盘：总结经验，形成自己的知识体系

学习建议

• ✅ 每天至少写代码 30 分钟
• ✅ 每周学习一个新概念
• ✅ 每月完成一个小项目
• ✅ 每季度进行一次技术复盘
• ✅ 保持好奇心和探索精神

'学习 Python+AI 是一场马拉松，不是短跑。重要的是保持持续学习和实践，不要害怕犯错。每一个错误都是学习的机会。'