pip install jupyter_contrib_nbextensions
jupyter contrib nbextension install --user

pip install jupyterthemes
jt -t chesterish -fs 95 -altp -tfs 11 -nfs 115 -cellw 88% -hrs

pip install nbdime
nbdime config-git --enable

pip install runipy
runipy MyNotebook.ipynb OutputNotebook.ipynb

# clean_notebook.py
import nbformat
from nbformat import v4 as nbf

with open('input.ipynb', 'r') as f:
    nb = nbformat.read(f, as_version=4)

for cell in nb.cells:
    if cell.cell_type == 'code':
        cell.outputs = []
        cell.execution_count = None

with open('cleaned.ipynb', 'w') as f:
    nbformat.write(nb, f)

# 测量代码执行时间
%timeit [x**2 for x in range(1000)]

# 多行代码执行时间测量
%%time
result = 0
for i in range(100000):
    result += i

# 显示所有魔法命令
%lsmagic

# 执行外部 Python 脚本
%run my_script.py

# 加载外部文件内容
%load my_module.py

# 追踪代码执行过程
%prun my_function()

# 内存分析
%memit [x**2 for x in range(100000)]

# 方式一：使用 pdb 魔法命令
%pdb on
def buggy_function(x):
    result = x / 0  # 这里会触发异常
    return result

# 方式二：使用 raise 显式触发调试器
def debug_example():
    breakpoint()  # Python 3.7+ 推荐写法
    x = calculate_value()
    return x

# 方式三：异常后自动进入调试
%xmode Verbose
%pdb on

{
    "python.linting.enabled": true,
    "python.linting.pylintEnabled": true,
    "python.linting.flake8Enabled": false,
    "python.formatting.provider": "black",
    "python.analysis.typeCheckingMode": "basic",
    "python.analysis.autoImportCompletions": true,
    "python.analysis.inlayVariableTypes": true,
    "python.testing.pytestEnabled": true,
    "python.testing.unittestEnabled": false,
    "editor.formatOnSave": true,
    "editor.codeActionsOnSave": {
        "source.organizeImports": true
    },
    "files.exclude": {
        "**/__pycache__": true,
        "**/*.pyc": true,
        "**/.pytest_cache": true
    }
}

pip install black
black --line-length 88 --target-version py38 my_project/

{
    "[python]": {
        "editor.defaultFormatter": "ms-python.black-formatter",
        "editor.formatOnSave": true
    },
    "black-formatter.args": ["--line-length", "100"]
}

pip install autopep8
autopep8 --in-place --aggressive --aggressive my_file.py

pip install pylint

[MESSAGES CONTROL]
disable=C0111,  # 忽略缺失文档字符串
      C0103   # 忽略变量命名风格

[FORMAT]
max-line-length=120
indent-string='    '

[DESIGN]
max-args=8
max-locals=20

pip install flake8

[flake8]
max-line-length = 100
exclude = .git,__pycache__,build,dist
ignore = E203,E266,E501,W503
per-file-ignores = __init__.py:F401

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python: Current File",
            "type": "python",
            "request": "launch",
            "program": "${file}",
            "console": "integratedTerminal"
        },
        {
            "name": "Python: Module",
            "type": "python",
            "request": "launch",
            "module": "mymodule",
            "console": "integratedTerminal"
        },
        {
            "name": "Python: PyTorch Debug",
            "type": "python",
            "request": "launch",
            "module": "torch.distributed.launch",
            "args": [
                "--nproc_per_node=2",
                "train.py"
            ],
            "env": {
                "CUDA_VISIBLE_DEVICES": "0,1"
            },
            "console": "integratedTerminal"
        }
    ]
}

{
    "Python Class Template": {
        "prefix": "pyclass",
        "body": [
            "class ${1:ClassName}:",
            "    \"\"\"${2:Class description}.",
            "",
            "    Args:",
            "        ${3:arg1}: ${4:description}",
            "    \"\"\"",
            "",
            "    def __init__(self, $3):",
            "        self.$3 = $3",
            "",
            "    def __str__(self):",
            "        return f\"${1:ClassName}($3={self.$3})\""
        ],
        "description": "Python class template with docstring"
    },
    "Data Science Plot": {
        "prefix": "dsplot",
        "body": [
            "import matplotlib.pyplot as plt",
            "",
            "plt.figure(figsize=(10, 6))",
            "$1",
            "plt.xlabel('$2')",
            "plt.ylabel('$3')",
            "plt.title('$4')",
            "plt.grid(True)",
            "plt.tight_layout()",
            "plt.show()"
        ],
        "description": "Matplotlib plot template"
    }
}

# ~/.ssh/config
Host ai-server
    HostName 192.168.1.100
    User developer
    Port 22
    IdentityFile ~/.ssh/id_rsa
    ForwardAgent yes
    ServerAliveInterval 60
    ServerAliveCountMax 3

# 创建专用 AI 环境
conda create -n ai-dev python=3.10
conda activate ai-dev
pip install torch torchvision tensorflow jupyterlab

# 生成配置文件
jupyter notebook --generate-config

# 设置密码
python -c "from jupyter_server.auth import passwd; print(passwd('your_password'))"

# 修改 jupyter_notebook_config.py
cat >> ~/.jupyter/jupyter_notebook_config.py << EOF
c.NotebookApp.ip = '0.0.0.0'
c.NotebookApp.port = 8888
c.NotebookApp.password = 'sha1:xxx...'  # 上面生成的密码哈希
c.NotebookApp.open_browser = False
c.NotebookApp.allow_root = True
EOF

# 启动 Jupyter
jupyter notebook --no-browser

ssh -L 8888:localhost:8888 ai-server
# 然后在本地浏览器打开 http://localhost:8888

# Dockerfile.ai
FROM nvidia/cuda:11.8-cudnn8-runtime-ubuntu22.04

ENV DEBIAN_FRONTEND=noninteractive
ENV PYTHONUNBUFFERED=1

WORKDIR /workspace

RUN apt-get update && apt-get install -y \
    python3.10 \
    python3-pip \
    git \
    && rm -rf /var/lib/apt/lists/*

RUN pip3 install --no-cache-dir \
    torch==2.0.1 \
    torchvision==0.15.2 \
    jupyterlab \
    black \
    pylint

EXPOSE 8888

CMD ["jupyter", "lab", "--ip=0.0.0.0", "--port=8888", "--allow-root"]

docker build -f Dockerfile.ai -t ai-dev .
docker run --gpus all -p 8888:8888 -v $(pwd):/workspace ai-dev

工作区 (Working Directory)
    ↓ git add
暂存区 (Staging Area / Index)
    ↓ git commit
本地仓库 (Local Repository)
    ↓ git push
远程仓库 (Remote Repository)

# 基础操作
git init                    # 初始化仓库
git clone <url>             # 克隆仓库
git status                  # 查看状态
git add <file>              # 暂存文件
git add .                   # 暂存所有变更
git commit -m "message"     # 提交
git commit -am "message"    # 暂存并提交已跟踪文件

# 分支操作
git branch                  # 列出分支
git branch <name>           # 创建分支
git checkout <branch>       # 切换分支
git checkout -b <branch>    # 创建并切换
git switch <branch>         # 切换分支（现代写法）
git merge <branch>          # 合并分支
git branch -d <branch>      # 删除分支

# 远程操作
git remote -v               # 查看远程仓库
git fetch                   # 获取远程更新
git pull                    # 拉取并合并
git push                    # 推送到远程
git push -u origin main     # 首次推送设置上游

# 历史查看
git log --oneline --graph   # 简洁图形日志
git log -p <file>           # 查看文件历史
git blame <file>            # 查看文件每行最后修改
git show <commit>           # 查看提交详情

# 撤销操作
git checkout -- <file>      # 撤销工作区修改
git reset HEAD <file>       # 取消暂存
git reset --soft HEAD~1     # 撤销上次提交（保留更改）
git revert <commit>         # 创建新提交撤销指定提交

# 安装 GitHub CLI
winget install GitHub.cli

# 登录
gh auth login

# 克隆仓库
gh repo clone owner/repo

# 创建 PR
gh pr create --title "Feature: Add new model" --body "Description"

# 查看 PR 状态
gh pr status
gh pr list

# 审查代码
gh pr review 123 --approve

main (生产环境)
  └─ develop (开发分支)
       ├─ feature/model-v2 (功能分支)
       ├─ feature/new-api (功能分支)
       └─ hotfix/bug-fix (热修复分支)

# 1. 确保工作区干净
git status

# 2. 更新目标分支
git checkout develop
git pull origin develop

# 3. 合并你的分支
git merge feature/my-feature

# 4. 解决冲突后标记完成
git add <resolved-files>
git commit

# 5. 推送解决后的合并
git push origin develop

<<<<<<< HEAD
当前分支的代码
=======
合并分支的代码
>>>>>>> feature/my-feature

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg

# 虚拟环境
venv/
ENV/
env/
.venv/

# Jupyter Notebook
.ipynb_checkpoints
*.ipynb_checkpoints

# 数据文件
*.csv
*.xlsx
*.h5
*.hdf5
*.parquet
data/
datasets/
models/*.pth
models/*.pt
models/*.ckpt
!models/.gitkeep

# 训练输出
logs/
tensorboard/
wandb/
mlruns/
*.pth
checkpoints/

# 大型模型文件
*.bin
*.onnx
*.pkl
*.pkl.gz

# Git LFS（如果使用）
#*.pt filter=lfs diff=lfs merge=lfs -text
#*.pth filter=lfs diff=lfs merge=lfs -text
#*.h5 filter=lfs diff=lfs merge=lfs -text

# IDE
.vscode/
.idea/
*.swp
*.swo
*~

# OS
.DS_Store
Thumbs.db

# secrets
.env
.env.local
*.pem
*.key
credentials.json

# 安装 Git LFS
git lfs install

# 跟踪大型文件类型
git lfs track "*.pth"
git lfs track "*.pt"
git lfs track "*.h5"
git lfs track "*.csv"
git lfs track "*.parquet"

# 查看跟踪状态
git lfs ls-files

# 推送到远程
git add .gitattributes
git add model.pth
git commit -m "Add trained model"
git push

# 创建模型目录
mkdir -p models/versions

# 使用日期和哈希命名
MODEL_NAME="chatbot"
VERSION=$(date +%Y%m%d)
COMMIT_HASH=$(git rev-parse --short HEAD)
MODEL_PATH="models/versions/${MODEL_NAME}_${VERSION}_${COMMIT_HASH}.pt"

# 保存模型
python -c "import torch; torch.save(model.state_dict(), '${MODEL_PATH}')"

# 创建模型清单
cat > "models/versions/${MODEL_NAME}_${VERSION}_${COMMIT_HASH}.json" << EOF
{
    "model_path": "${MODEL_PATH}",
    "commit": "${COMMIT_HASH}",
    "created_at": "$(date -Iseconds)",
    "metrics": {
        "accuracy": 0.95,
        "loss": 0.05
    },
    "config": {
        "hidden_size": 768,
        "num_layers": 12,
        "learning_rate": 0.0001
    }
}
EOF

import mlflow
import mlflow.pytorch

mlflow.set_experiment("chatbot-training")

with mlflow.start_run(run_name="baseline-v1"):
    # 记录参数
    mlflow.log_param("epochs", 100)
    mlflow.log_param("batch_size", 32)
    mlflow.log_param("learning_rate", 0.001)
    
    # 训练循环
    for epoch in range(100):
        train_loss = train_epoch(model, train_loader)
        val_loss, val_acc = evaluate(model, val_loader)
        
        # 记录指标
        mlflow.log_metrics({
            "train_loss": train_loss,
            "val_loss": val_loss,
            "val_accuracy": val_acc
        }, step=epoch)
        
        # 保存模型
        if val_acc > best_acc:
            mlflow.pytorch.log_model(model, "best_model")
            best_acc = val_acc

# 启动 MLflow UI 查看实验
# mlflow ui

docs/
├── README.md          # 项目说明
├── API.md             # API 文档
├── CHANGELOG.md       # 更新日志
├── CONTRIBUTING.md    # 贡献指南
└── tutorials/         # 教程文档
    ├── getting-started.md
    └── advanced-usage.md

# 1. 从 main 创建 develop 分支
git checkout main
git checkout -b develop

# 2. 从 develop 创建功能分支
git checkout -b feature/chat-interface develop

# 3. 完成功能后合并到 develop
git checkout develop
git merge --no-ff feature/chat-interface
git branch -d feature/chat-interface

# 4. 准备发布时创建 release 分支
git checkout -b release/v1.0.0 develop

# 5. 发布后合并到 main 和 develop
git checkout main
git merge --no-ff release/v1.0.0
git tag -a v1.0.0 -m "Release version 1.0.0"
git checkout develop
git merge --no-ff release/v1.0.0
git branch -d release/v1.0.0

# 6. 紧急修复
git checkout -b hotfix/critical-fix main
# 修复后
git checkout main
git merge --no-ff hotfix/critical-fix
git tag -a v1.0.1 -m "Hotfix version 1.0.1"
git checkout develop
git merge --no-ff hotfix/critical-fix
git branch -d hotfix/critical-fix

# 提交前检查
git diff --stat
git log --oneline -5

# 创建 PR 后，邀请至少 1-2 人审查
gh pr create \
    --title "feat: 实现聊天模型" \
    --body "## 变更内容
- 新增 ChatModel 类
- 添加单元测试
- 更新文档

## 测试
- [x] 本地测试通过
- [x] 单元测试覆盖率 95%

## 截图（如有 UI 变更）
..." \
    --reviewer @teammate1,@teammate2

# 审查代码
gh pr diff PR_NUMBER  # 查看变更
gh pr comment PR_NUMBER --body "LGTM!"  # 通过审查

# .github/ISSUE_TEMPLATE/feature_request.yml
name: Feature Request
description: 提出新功能建议
labels: [enhancement]
body:
  - type: markdown
    attributes:
      value: |
        ## 功能描述
        请详细描述您建议的功能。
        
  - type: textarea
    id: motivation
    attributes:
      label: 为什么需要这个功能？
    validations:
      required: true
      
  - type: textarea
    id: alternative
    attributes:
      label: 您考虑过哪些替代方案？

# .github/workflows/python-ci.yml
name: Python CI

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main, develop]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ['3.9', '3.10', '3.11']
    
    steps:
    - uses: actions/checkout@v4
    
    - name: Setup Python ${{ matrix.python-version }}
      uses: actions/setup-python@v5
      with:
        python-version: ${{ matrix.python-version }}
        
    - name: Cache pip packages
      uses: actions/cache@v3
      with:
        path: ~/.cache/pip
        key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}
        
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install pytest pytest-cov
        
    - name: Run tests
      run: |
        pytest tests/ --cov=src --cov-report=xml
        
    - name: Upload coverage
      uses: codecov/codecov-action@v3
      
  lint:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v4
    - uses: actions/setup-python@v5
      with:
        python-version: '3.10'
    - name: Run Black
      run: pip install black && black --check src/
    - name: Run Flake8
      run: pip install flake8 && flake8 src/

import numpy as np

# 多种数组创建方式
arr1 = np.array([1, 2, 3, 4, 5])                    # 从列表创建
arr2 = np.arange(0, 10, 2)                          # 类似 range：步长为 2
arr3 = np.linspace(0, 1, 5)                          # 线性等分：5 个点
arr4 = np.zeros((3, 4))                              # 全零矩阵
arr5 = np.ones((2, 3, 4))                           # 全一数组（3D）
arr6 = np.eye(4)                                     # 单位矩阵
arr7 = np.random.rand(3, 3)                         # 0-1 均匀分布
arr8 = np.random.randn(1000)                         # 标准正态分布
arr9 = np.random.randint(0, 10, (5, 5))            # 整数随机矩阵

# 高级索引
matrix = np.arange(25).reshape(5, 5)
print(matrix[[0, 2, 4]])              # 选择第 0, 2, 4 行
print(matrix[:, [1, 3]])             # 选择第 1, 3 列
print(matrix[matrix[:, 0] > 10])     # 布尔索引：选择第一列大于 10 的行

# 布尔索引应用：筛选满足条件的数据
data = np.random.randn(1000)
positive_data = data[data > 0]        # 筛选正数
threshold = np.percentile(data, 75)   # 75 百分位数
high_values = data[data > threshold]  # 筛选高值

# 基础广播
a = np.array([[1, 2, 3],
              [4, 5, 6]])
b = np.array([10, 20, 30])
print(a + b)  # [[11, 22, 33], [41, 52, 63]]

# 形状匹配规则
# (3, 3) + (3,) -> (3, 3) + (1, 3) -> (3, 3)
# (3, 3) + (3, 1) -> (3, 3) + (3, 1) -> (3, 3)
# (3, 1, 4) + (2, 1) -> (3, 2, 4)

# 实战示例：图像归一化
image = np.random.randint(0, 256, (224, 224, 3), dtype=np.uint8)
normalized = (image - image.mean()) / image.std()

# 批量数据处理
images_batch = np.random.randn(32, 224, 224, 3)
mean = images_batch.mean(axis=(1, 2), keepdims=True)  # 按样本计算通道均值
std = images_batch.std(axis=(1, 2), keepdims=True)
normalized_batch = (images_batch - mean) / (std + 1e-7)

import numpy as np

arr = np.array([[3, 1, 4], [1, 5, 9], [2, 6, 5]])

# 统计函数
np.sum(arr)                  # 求和
np.sum(arr, axis=0)          # 按列求和
np.sum(arr, axis=1)          # 按行求和
np.mean(arr)                 # 均值
np.std(arr)                  # 标准差
np.var(arr)                  # 方差
np.min(arr)                  # 最小值
np.max(arr)                  # 最大值
np.argmin(arr)               # 最小值索引
np.argmax(arr)               # 最大值索引
np.percentile(arr, 50)       # 百分位数

# 排序与搜索
np.sort(arr)                 # 排序
np.argsort(arr)             # 返回排序索引
np.where(arr > 5)           # 返回满足条件的索引
np.extract(arr > 5, arr)    # 提取满足条件的值
np.unique(arr)              # 去重

# 数学运算
np.sqrt(arr)                # 平方根
np.exp(arr)                 # 指数
np.log(arr)                 # 自然对数
np.log10(arr)               # 常用对数
np.power(arr, 2)            # 幂运算
np.dot(arr1, arr2)          # 点积
np.matmul(arr1, arr2)       # 矩阵乘法
np.linalg.inv(arr)          # 矩阵求逆
np.linalg.eig(arr)          # 特征值分解

# 数组操作
np.concatenate([arr1, arr2])    # 拼接
np.vstack([arr1, arr2])         # 垂直堆叠
np.hstack([arr1, arr2])         # 水平堆叠
np.split(arr, 3)                # 分割
np.tile(arr, (2, 3))            # 重复数组
np.repeat(arr, 3)               # 重复元素
np.reshape(arr, (1, 9))         # 改变形状
np.transpose(arr)              # 转置

import numpy as np

# 避免循环，使用向量化
# 低效
result = []
for i in range(1000):
    result.append(np.sin(i) + np.cos(i))

# 高效
x = np.arange(1000)
result = np.sin(x) + np.cos(x)

# 使用 out 参数避免内存分配
output = np.empty(1000)
np.multiply(arr1, arr2, out=output)

# 使用原地操作
arr = np.arange(10)
arr *= 2                    # 原地乘以 2
arr += 1                    # 原地加 1

# 预分配内存
size = 10000
result = np.empty(size)
for i in range(size):
    result[i] = compute_value(i)

# 使用 np.clip 替代循环
arr = np.random.randn(1000)
arr_clipped = np.clip(arr, -1, 1)  # 限制在 [-1, 1] 范围

# 使用 np.einsum 进行高效矩阵运算
a = np.random.randn(100, 200)
b = np.random.randn(200, 50)
# 计算 a*b 的迹（trace）
trace = np.einsum('ij,jk->ik', a, b)

import numpy as np
import matplotlib.pyplot as plt

def apply_gaussian_blur(image, kernel_size=5, sigma=1.0):
    """应用高斯模糊"""
    x = np.arange(kernel_size) - kernel_size // 2
    gaussian = np.exp(-x**2 / (2 * sigma**2))
    gaussian = gaussian / gaussian.sum()
    
    kernel_1d = gaussian.reshape(1, -1)
    kernel_2d = np.outer(kernel_1d, kernel_1d)
    
    from scipy.ndimage import convolve
    blurred = convolve(image, kernel_2d, mode='reflect')
    return blurred

def apply_edge_detection(image):
    """Sobel 边缘检测"""
    sobel_x = np.array([[-1, 0, 1],
                        [-2, 0, 2],
                        [-1, 0, 1]])
    sobel_y = sobel_x.T
    
    from scipy.ndimage import convolve
    gx = convolve(image.astype(float), sobel_x)
    gy = convolve(image.astype(float), sobel_y)
    
    magnitude = np.sqrt(gx**2 + gy**2)
    direction = np.arctan2(gy, gx)
    
    return magnitude, direction

def normalize_image(image):
    """图像归一化"""
    img_min = image.min()
    img_max = image.max()
    return (image - img_min) / (img_max - img_min)

def create_histogram(image, bins=256):
    """计算并绘制直方图"""
    hist, bin_edges = np.histogram(image.flatten(), bins=bins, range=(0, 255))
    return hist, bin_edges

# 图像处理示例
image = np.random.randint(0, 256, (100, 100), dtype=np.uint8)
blurred = apply_gaussian_blur(image)
edges, _ = apply_edge_detection(image)
normalized = normalize_image(image.astype(float))
hist, _ = create_histogram(image)

import pandas as pd
import numpy as np

# 创建 DataFrame
df = pd.DataFrame({
    'name': ['Alice', 'Bob', 'Charlie', 'David'],
    'age': [25, 30, 35, 40],
    'score': [85.5, 90.2, 78.8, 92.1],
    'city': ['Beijing', 'Shanghai', 'Beijing', 'Guangzhou']
})

# 基础查询
df.head(2)                          # 前两行
df.tail(2)                          # 后两行
df.shape                            # 形状 (4, 4)
df.info()                           # 数据信息
df.describe()                       # 统计描述

# 条件筛选
df[df['age'] > 30]                  # 年龄大于 30
df[(df['age'] > 25) & (df['score'] > 85)]  # 多条件
df[df['city'].isin(['Beijing', 'Shanghai'])]  # 列表匹配

# 列操作
df[['name', 'score']]               # 选择多列
df.drop('city', axis=1)             # 删除列
df.rename(columns={'name': 'username'})  # 重命名列

# 排序
df.sort_values('score', ascending=False)  # 按分数降序
df.sort_values(['city', 'age'])      # 多列排序

# 添加计算列
df['age_months'] = df['age'] * 12
df['score_normalized'] = (df['score'] - df['score'].mean()) / df['score'].std()

import pandas as pd
import numpy as np

# 处理缺失值
df = pd.DataFrame({
    'A': [1, 2, np.nan, 4],
    'B': [5, np.nan, np.nan, 8],
    'C': [9, 10, 11, 12]
})

df.isnull()                          # 检测缺失值
df.notnull()                         # 检测非缺失值
df.dropna()                          # 删除含缺失值的行
df.dropna(axis=1)                   # 删除含缺失值的列
df.fillna(0)                        # 用 0 填充缺失值
df.fillna(df.mean())                # 用均值填充
df.fillna(method='ffill')           # 前向填充
df.fillna(method='bfill')           # 后向填充

# 处理重复值
df.duplicated()                      # 检测重复行
df.drop_duplicates()                # 删除重复行
df.drop_duplicates(subset=['A', 'B'])  # 按指定列去重

# 数据类型转换
df['date'] = pd.to_datetime(df['date'])
df['price'] = df['price'].astype(int)
df['category'] = df['category'].astype('category')

# 字符串处理
df['name'] = df['name'].str.lower()      # 转小写
df['name'] = df['name'].str.strip()       # 去除空格
df['email'] = df['email'].str.contains('@')  # 包含检查
df['name'] = df['name'].str.replace('old', 'new')  # 替换

# 异常值处理
def remove_outliers(df, column, n=3):
    mean = df[column].mean()
    std = df[column].std()
    return df[(df[column] > mean - n*std) & (df[column] < mean + n*std)]

# 使用 IQR 方法
Q1 = df['score'].quantile(0.25)
Q3 = df['score'].quantile(0.75)
IQR = Q3 - Q1
lower = Q1 - 1.5 * IQR
upper = Q3 + 1.5 * IQR
df_cleaned = df[(df['score'] > lower) & (df['score'] < upper)]

import pandas as pd
import numpy as np

df = pd.DataFrame({
    'category': ['A', 'A', 'B', 'B', 'A', 'B'],
    'product': ['X', 'Y', 'X', 'Y', 'Z', 'Z'],
    'sales': [100, 150, 200, 180, 120, 220],
    'quantity': [10, 15, 20, 18, 12, 22]
})

# 基础分组聚合
df.groupby('category').sum()
df.groupby('category')['sales'].sum()
df.groupby(['category', 'product']).mean()

# 多指标聚合
df.groupby('category').agg({
    'sales': ['sum', 'mean', 'max'],
    'quantity': ['sum', 'mean']
})

# 自定义聚合函数
def weighted_mean(x):
    return np.average(x['sales'], weights=x['quantity'])

df.groupby('category').apply(weighted_mean)

# 命名聚合（pandas 0.25+）
df.groupby('category').agg(
    total_sales=('sales', 'sum'),
    avg_sales=('sales', 'mean'),
    total_qty=('quantity', 'sum')
)

# 分组后迭代
for name, group in df.groupby('category'):
    print(f"Category: {name}")
    print(f"Total sales: {group['sales'].sum()}\n")

# 变换操作
df['sales_rank'] = df.groupby('category')['sales'].rank(ascending=False)
df['sales_pct'] = df.groupby('category')['sales'].transform(lambda x: x / x.sum())

import pandas as pd
import numpy as np

# 合并操作
df1 = pd.DataFrame({'id': [1, 2, 3], 'name': ['Alice', 'Bob', 'Charlie']})
df2 = pd.DataFrame({'id': [1, 2, 4], 'score': [85, 90, 95]})

pd.merge(df1, df2, on='id')                   # 内连接
pd.merge(df1, df2, on='id', how='left')       # 左连接
pd.merge(df1, df2, on='id', how='right')     # 右连接
pd.merge(df1, df2, on='id', how='outer')     # 全连接
pd.merge(df1, df2, left_on='id', right_on='user_id')  # 不同列名

# 拼接操作
pd.concat([df1, df2])                         # 垂直拼接
pd.concat([df1, df2], axis=1)                 # 水平拼接
pd.concat([df1, df2], ignore_index=True)      # 重置索引

# 重塑操作
df = pd.DataFrame({'A': [1, 2, 3, 4],
                   'B': [5, 6, 7, 8],
                   'C': [9, 10, 11, 12]})

df.melt(id_vars=['A'], value_vars=['B', 'C'])       # 宽转长
df.pivot(index='date', columns='category', values='sales')  # 长转宽

# 透视表
df.pivot_table(values='sales', 
               index='category',
               columns='month',
               aggfunc='sum',
               fill_value=0)

# 交叉表
pd.crosstab(df['category'], df['month'])

import pandas as pd
import numpy as np

# 使用 categoricals 优化字符串列
df = pd.DataFrame({
    'category': np.random.choice(['A', 'B', 'C', 'D'], 1000000),
    'value': np.random.randn(1000000)
})

# 转换为 category 类型
df['category'] = df['category'].astype('category')

# category vs object 性能对比
%timeit df[df['category'] == 'A']          # category 类型更快

# 使用 query 加速查询
%timeit df.query('category == "A" and value > 0')

# 使用 eval 加速计算
df['new_col'] = pd.eval('value * 2 + value ** 2')

# 分块处理大文件
chunk_size = 100000
chunks = pd.read_csv('large_file.csv', chunksize=chunk_size)
result = pd.concat([process(chunk) for chunk in chunks])

# 使用 inplace 减少内存（慎用）
df.drop('column', axis=1, inplace=True)

# 使用 copy 避免 SettingWithCopyWarning
df_copy = df[df['condition']].copy()
df_copy['new_col'] = df_copy['col'] * 2

# 内存优化
df = pd.read_csv('data.csv')
df['int_col'] = pd.to_numeric(df['int_col'], downcast='integer')
df['float_col'] = pd.to_numeric(df['float_col'], downcast='float')
df['str_col'] = df['str_col'].astype('category')

import matplotlib.pyplot as plt
import numpy as np

# 根据数据类型选择图表
# 趋势数据 -> 折线图
x = np.arange(100)
y = np.cumsum(np.random.randn(100))
plt.plot(x, y)

# 分类对比 -> 柱状图
categories = ['A', 'B', 'C', 'D']
values = [30, 45, 25, 60]
plt.bar(categories, values)
plt.barh(categories, values)  # 水平柱状图

# 占比分布 -> 饼图
sizes = [15, 30, 45, 10]
labels = ['A', 'B', 'C', 'D']
plt.pie(sizes, labels=labels, autopct='%1.1f%%')

# 分布情况 -> 直方图/箱线图
data = np.random.randn(1000)
plt.hist(data, bins=30)
plt.boxplot([data[data > 0], data[data < 0]])

# 相关性 -> 散点图
x = np.random.randn(100)
y = 2*x + np.random.randn(100)*0.5
plt.scatter(x, y, alpha=0.5, c=range(100), cmap='viridis')

import matplotlib.pyplot as plt
import numpy as np

# 设置全局样式
plt.style.use('seaborn-v0_8-darkgrid')  # 或 'ggplot', 'fivethirtyeight'

# 自定义参数
plt.rcParams.update({
    'figure.figsize': (10, 6),
    'figure.dpi': 100,
    'font.size': 12,
    'font.family': 'sans-serif',
    'axes.titlesize': 16,
    'axes.labelsize': 12,
    'xtick.labelsize': 10,
    'ytick.labelsize': 10,
    'legend.fontsize': 10,
    'figure.titlesize': 18,
    'axes.spines.top': False,
    'axes.spines.right': False,
})

# 自定义颜色
colors = ['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728', '#9467bd']
plt.plot(x, y, color=colors[0], linewidth=2, linestyle='--')

# 使用 colormap
cmap = plt.cm.viridis
for i in range(5):
    plt.plot(x, y + i*10, color=cmap(i/4))

# 颜色映射
plt.scatter(x, y, c=z, cmap='coolwarm', alpha=0.7)
plt.colorbar(label='Z Value')

# 标记和线型
plt.plot(x, y, 'o-', markersize=8, linewidth=2)  # 圆形标记实线
plt.plot(x, y, 's--', markersize=6)             # 方形标记虚线

import matplotlib.pyplot as plt
import numpy as np

fig, axes = plt.subplots(2, 2, figsize=(12, 10))
axes = axes.flatten()

# 子图1
x = np.linspace(0, 10, 100)
axes[0].plot(x, np.sin(x), 'b-', label='sin(x)')
axes[0].set_title('Sine Wave')
axes[0].legend()
axes[0].grid(True, alpha=0.3)

# 子图2
axes[1].bar(['A', 'B', 'C', 'D'], [30, 45, 25, 60])
axes[1].set_title('Bar Chart')

# 子图3
data = np.random.randn(1000)
axes[2].hist(data, bins=30, color='green', alpha=0.7)
axes[2].set_title('Histogram')

# 子图4
theta = np.linspace(0, 2*np.pi, 100)
axes[3].plot(np.cos(theta), np.sin(theta))
axes[3].set_title('Circle')
axes[3].set_aspect('equal')

plt.tight_layout()
plt.savefig('combined_plot.png', dpi=150, bbox_inches='tight')
plt.show()

# GridSpec 高级布局
from matplotlib.gridspec import GridSpec
fig = plt.figure(figsize=(12, 8))
gs = GridSpec(3, 3, figure=fig)

ax1 = fig.add_subplot(gs[0, :])       # 第一行，跨所有列
ax2 = fig.add_subplot(gs[1, :2])      # 第二行，前两列
ax3 = fig.add_subplot(gs[1:, 2])      # 后两行，最后一列
ax4 = fig.add_subplot(gs[2, 0])       # 最后一行，第一列
ax5 = fig.add_subplot(gs[2, 1])       # 最后一行，第二列

import matplotlib.pyplot as plt
import numpy as np
from matplotlib.animation import FuncAnimation
from IPython.display import HTML

# 动画示例：动态更新折线图
fig, ax = plt.subplots()
x = np.arange(100)
line, = ax.plot([], [], 'b-', lw=2)
ax.set_xlim(0, 100)
ax.set_ylim(-3, 3)

def init():
    line.set_data([], [])
    return line,

def animate(i):
    y = np.sin(x + i*0.1)
    line.set_data(x, y)
    return line,

anim = FuncAnimation(fig, animate, init_func=init,
                     frames=100, interval=50, blit=True)
plt.show()

# 保存动画
anim.save('animation.gif', writer='pillow', fps=30)

# Jupyter 中显示
HTML(anim.to_jshtml())

import matplotlib.pyplot as plt
import numpy as np

# 保存不同格式
plt.savefig('plot.png', dpi=300, bbox_inches='tight')        # PNG 高清
plt.savefig('plot.pdf', bbox_inches='tight')                 # PDF 矢量图
plt.savefig('plot.svg', bbox_inches='tight')                  # SVG 矢量图
plt.savefig('plot.jpg', dpi=150, quality=95)                  # JPEG
plt.savefig('plot.eps', bbox_inches='tight')                  # EPS 矢量图

# 保存数据
x = np.arange(100)
y = np.sin(x)
np.savetxt('data.csv', np.column_stack([x, y]), delimiter=',')

# 设置透明背景
plt.savefig('plot.png', transparent=True, dpi=300)

# 添加水印
fig, ax = plt.subplots()
ax.plot([1, 2, 3], [1, 2, 3])
fig.text(0.95, 0.05, '版权信息', fontsize=10, 
         ha='right', va='bottom', alpha=0.5)

HTTP 方法对应 CRUD 操作：
- GET    -> 读取资源
- POST   -> 创建资源
- PUT    -> 更新资源（完整）
- PATCH  -> 部分更新资源
- DELETE -> 删除资源

常见状态码：
- 200 OK         成功
- 201 Created    创建成功
- 400 Bad Request 请求错误
- 401 Unauthorized 未授权
- 403 Forbidden   禁止访问
- 404 Not Found   资源不存在
- 500 Server Error 服务器错误

import requests
import json

# 基础 GET 请求
response = requests.get('https://api.example.com/data')
print(response.status_code)
print(response.json())
print(response.text)

# 带参数请求
params = {
    'page': 1,
    'per_page': 20,
    'category': 'tech'
}
response = requests.get('https://api.example.com/posts', params=params)

# POST 请求
data = {
    'title': 'New Post',
    'content': 'Hello World',
    'author': 'Alice'
}
response = requests.post('https://api.example.com/posts', json=data)

# 上传文件
files = {
    'file': open('image.jpg', 'rb')
}
response = requests.post('https://api.example.com/upload', files=files)

# 下载文件
response = requests.get('https://example.com/file.zip', stream=True)
with open('file.zip', 'wb') as f:
    for chunk in response.iter_content(chunk_size=8192):
        f.write(chunk)

import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth

# Basic Auth
response = requests.get(
    'https://api.example.com/protected',
    auth=HTTPBasicAuth('username', 'password')
)

# 或使用元组形式
response = requests.get(
    'https://api.example.com/protected',
    auth=('username', 'password')
)

# Token Auth
headers = {
    'Authorization': 'Bearer your_token_here'
}
response = requests.get('https://api.example.com/api', headers=headers)

# API Key
params = {'api_key': 'your_api_key'}
response = requests.get('https://api.example.com/data', params=params)

# OAuth 2.0
auth_url = 'https://oauth.example.com/token'
token_data = {
    'grant_type': 'client_credentials',
    'client_id': 'your_client_id',
    'client_secret': 'your_client_secret'
}
token_response = requests.post(auth_url, data=token_data)
access_token = token_response.json()['access_token']

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time

# 配置重试策略
session = requests.Session()
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["HEAD", "GET", "OPTIONS", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)

# 完整错误处理
def safe_request(url, method='GET', max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            response = requests.request(url, method=method, **kwargs)
            response.raise_for_status()
            return response
        except requests.exceptions.Timeout:
            print(f"请求超时 (尝试 {attempt + 1}/{max_retries})")
        except requests.exceptions.ConnectionError as e:
            print(f"连接错误 (尝试 {attempt + 1}/{max_retries}): {e}")
        except requests.exceptions.HTTPError as e:
            print(f"HTTP错误: {e}")
            if response.status_code < 500:
                raise  # 客户端错误不重试
        except requests.exceptions.RequestException as e:
            print(f"请求异常: {e}")
        
        if attempt < max_retries - 1:
            time.sleep(2 ** attempt)  # 指数退避
    
    raise Exception(f"请求失败，已重试 {max_retries} 次")

# 使用示例
try:
    response = safe_request(
        'https://api.example.com/data',
        params={'key': 'value'},
        timeout=10
    )
    data = response.json()
except Exception as e:
    print(f"最终失败: {e}")

import requests
import json
import os

class AIClient:
    def __init__(self, api_key, base_url="https://api.openai.com/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def chat_completion(self, model, messages, temperature=0.7, max_tokens=1000):
        """调用 ChatGPT API"""
        url = f"{self.base_url}/chat/completions"
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        response = requests.post(
            url,
            headers=self.headers,
            json=payload,
            timeout=60
        )
        response.raise_for_status()
        return response.json()
    
    def text_embedding(self, text, model="text-embedding-ada-002"):
        """获取文本嵌入向量"""
        url = f"{self.base_url}/embeddings"
        payload = {
            "input": text,
            "model": model
        }
        
        response = requests.post(
            url,
            headers=self.headers,
            json=payload,
            timeout=30
        )
        response.raise_for_status()
        return response.json()

# 使用示例
api_key = os.getenv("OPENAI_API_KEY")
client = AIClient(api_key)

# 对话示例
messages = [
    {"role": "system", "content": "你是一个专业的Python编程助手。"},
    {"role": "user", "content": "请解释Python中的装饰器是什么？"}
]

result = client.chat_completion("gpt-3.5-turbo", messages)
print(result['choices'][0]['message']['content'])

# 嵌入示例
embedding_result = client.text_embedding("Python is a great programming language")
vector = embedding_result['data'][0]['embedding']
print(f"嵌入向量维度: {len(vector)}")

import pprint

# 基础打印
def debug_example():
    data = {'key': 'value', 'list': [1, 2, 3]}
    print("DEBUG: data =", data)
    
    for i, item in enumerate(data['list']):
        print(f"DEBUG: index={i}, item={item}")

# 使用 pprint 格式化输出
pprint.pprint(complex_data_structure)

# 条件打印装饰器
import functools

def debug(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print(f"调用 {func.__name__}...")
        print(f"参数 args={args}")
        print(f"参数 kwargs={kwargs}")
        result = func(*args, **kwargs)
        print(f"{func.__name__} 返回: {result}")
        return result
    return wrapper

@debug
def calculate(x, y):
    return x ** y

calculate(2, 10)

# 方式一：在代码中插入断点
import pdb

def buggy_function(data):
    pdb.set_trace()  # 执行到此会暂停，进入调试模式
    result = []
    for i, item in enumerate(data):
        if item > 0:
            result.append(item ** 2)
        else:
            result.append(0)  # Bug: 应该处理负数
    return result

# Python 3.7+ 推荐写法
def buggy_function(data):
    breakpoint()  # 自动调用 pdb
    return [x**2 if x > 0 else 0 for x in data]

# pdb 常用命令
"""
h(help)    - 显示帮助
n(next)    - 执行下一行
s(step)    - 进入函数内部
c(cont)    - 继续执行直到断点
l(list)    - 显示当前代码上下文
p(print)   - 打印变量值
pp(pprint)  - 格式化打印
w(where)    - 显示调用栈
u(up)      - 往调用栈上层移动
d(down)    - 往调用栈下层移动
b(break)   - 设置断点
tbreak     - 设置临时断点
cl(clear)  - 清除断点
condition  - 设置断点条件
r(return)  - 继续执行直到函数返回
q(quit)    - 退出调试器
"""

# 命令行启动 pdb
# python -m pdb script.py
# python -pdb script.py  # 遇到错误自动进入 pdb

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python: Current File",
            "type": "python",
            "request": "launch",
            "program": "${file}",
            "console": "integratedTerminal",
            "justMyCode": false  // 包含库代码
        },
        {
            "name": "Python: Attach to Process",
            "type": "python",
            "request": "attach",
            "port": 5678,
            "host": "localhost"
        }
    ]
}

# 安装调试扩展
%pip install ipdb

# 在 Notebook 中使用
%autoload 2
import ipdb

def debug_function(data):
    ipdb.set_trace()
    return sum(data) / len(data)

# 使用魔术命令
%debug        # 事后调试
%tb            # 显示回溯
%pdb on        # 开启自动调试

# traceback 增强
import traceback
try:
    risky_operation()
except Exception as e:
    print("发生错误:")
    traceback.print_exc()
    traceback.print_tb(e.__traceback__)

import logging
import sys
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(f'app_{datetime.now():%Y%m%d}.log'),
        logging.StreamHandler(sys.stdout)
    ]
)

logger = logging.getLogger(__name__)

class AIProcessing:
    def __init__(self, config):
        self.logger = logging.getLogger(self.__class__.__name__)
        self.config = config
    
    def process(self, data):
        self.logger.info(f"开始处理数据，样本数: {len(data)}")
        try:
            result = self._transform(data)
            self.logger.info(f"处理完成，结果形状: {result.shape}")
            return result
        except Exception as e:
            self.logger.error(f"处理失败: {str(e)}", exc_info=True)
            raise
    
    def _transform(self, data):
        self.logger.debug(f"Transform 输入: {type(data)}")
        # 转换逻辑
        return data

# 不同级别的日志
logger.debug("调试信息")
logger.info("一般信息")
logger.warning("警告信息")
logger.error("错误信息")
logger.critical("严重错误")

import timeit

# 测量执行时间
result = timeit.timeit(
    '[x**2 for x in range(1000)]',
    number=10000
)
print(f"执行时间: {result:.4f} 秒")

# 测量函数执行时间
def list_comprehension():
    return [x**2 for x in range(1000)]

def for_loop():
    result = []
    for x in range(1000):
        result.append(x**2)
    return result

# 比较两种方法
t1 = timeit.timeit(list_comprehension, number=1000)
t2 = timeit.timeit(for_loop, number=1000)
print(f"列表推导式: {t1:.4f} 秒")
print(f"For 循环: {t2:.4f} 秒")

# 使用 repeat 获取多次测量
times = timeit.repeat(
    'math.sqrt(x)',
    'import math; x = 2.0',
    number=100000,
    repeat=5
)
print(f"最快时间: {min(times):.6f} 秒")

import cProfile
import pstats
from io import StringIO

# 基本使用
cProfile.run('sum([x**2 for x in range(100000)])')

# 详细分析并保存
profiler = cProfile.Profile()
profiler.enable()

# 被分析的代码
import numpy as np
data = np.random.randn(1000, 1000)
result = np.dot(data, data.T)
_ = np.linalg.svd(data)

profiler.disable()

# 输出统计信息
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(20)  # 前 20 条

# 保存到文件
stats.dump_stats('profile_results.prof')

# 使用 SnakeViz 可视化
# pip install snakeviz
# snakeviz profile_results.prof

# 命令行使用
# python -m cProfile -o output.prof my_script.py
# python -m cProfile -s cumtime my_script.py

import tracemalloc
import sys

# 启动内存追踪
tracemalloc.start()

# 执行代码
data = [x**2 for x in range(100000)]
more_data = [[i, i*2] for i in range(50000)]

# 获取内存快照
snapshot1 = tracemalloc.take_snapshot()
current, peak = tracemalloc.get_traced_memory()
print(f"当前内存: {current / 1024 / 1024:.2f} MB")
print(f"峰值内存: {peak / 1024 / 1024:.2f} MB")

# 比较内存变化
tracemalloc.clear_traces()
new_data = [x**3 for x in range(50000)]
snapshot2 = tracemalloc.take_snapshot()

# 找出内存增长最大的对象
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
print("\n内存增长最多的位置:")
for stat in top_stats[:5]:
    print(stat)

tracemalloc.stop()

# 使用 memory_profiler
# pip install memory_profiler
# from memory_profiler import profile
# @profile
# def memory_intensive_function():
#     data = [x**2 for x in range(1000000)]
#     return data

import numpy as np
import time

# 瓶颈1：Python 循环代替向量化
def slow_approach():
    result = 0
    for i in range(1000000):
        result += i ** 2
    return result

def fast_approach():
    arr = np.arange(1000000)
    return np.sum(arr ** 2)

# 瓶颈2：频繁的内存分配
def slow_append():
    result = []
    for i in range(10000):
        result.append(i ** 2)
    return result

def fast_append():
    return [i ** 2 for i in range(10000)]  # 预分配

# 瓶颈3：不必要的类型转换
def slow_type_convert():
    data = list(range(10000))
    return sum(float(x) for x in data)

def fast_type_convert():
    return sum(float(x) for x in range(10000))

# 瓶颈4：全局变量访问
GLOBAL_DATA = np.random.randn(1000)

def slow_global_access():
    for _ in range(1000):
        _ = GLOBAL_DATA.mean()

def fast_global_access():
    local_data = GLOBAL_DATA  # 本地缓存
    for _ in range(1000):
        _ = local_data.mean()

import numpy as np
from functools import lru_cache
import time

# 技巧1：使用局部变量
def optimize_local_vars():
    data = list(range(10000))
    append = data.append  # 缓存方法引用
    extend = data.extend
    for i in range(10000):
        append(i ** 2)
    return data

# 技巧2：善用缓存
@lru_cache(maxsize=128)
def cached_expensive_computation(n):
    time.sleep(0.1)  # 模拟耗时计算
    return n ** 2

# 技巧3：使用生成器代替列表
def generator_vs_list():
    # 生成器：按需生成，节省内存
    gen = (x**2 for x in range(1000000))
    return next(gen), next(gen)

# 技巧4：NumPy 批量操作
def batch_numpy():
    arr = np.random.randn(10000)
    # 一次性计算比分步计算快
    return np.sqrt(np.abs(arr)) * np.log(np.abs(arr) + 1)

# 技巧5：条件判断优化
def optimized_condition(value):
    # 使用字典/表驱动代替 if-elif
    handlers = {
        'A': lambda x: x * 2,
        'B': lambda x: x / 2,
        'C': lambda x: x ** 2
    }
    return handlers.get(value, lambda x: x)(value)

# 技巧6：使用 numba JIT 加速
# pip install numba
# from numba import jit
# @jit(nopython=True)
# def numba_accelerated():
#     result = 0.0
#     for i in range(1000000):
#         result += i ** 2
#     return result

# 常见错误：IndexError
data = [1, 2, 3]
try:
    print(data[5])  # IndexError: list index out of range
except IndexError as e:
    print(f"索引错误: {e}")
    print(f"列表长度为 {len(data)}，有效索引为 0-{len(data)-1}")

# 解决方案：安全索引访问
def safe_index(data, idx, default=None):
    """安全获取列表元素"""
    try:
        return data[idx]
    except IndexError:
        return default

# NumPy 数组索引错误
import numpy as np
arr = np.array([1, 2, 3])
try:
    arr[5]  # 触发 IndexError
except IndexError:
    print("数组索引越界")

# 解决方案：使用 np.where 安全处理
idx = 5
if idx < len(arr):
    value = arr[idx]
else:
    value = np.nan

# 常见错误：TypeError
def process_data(data):
    try:
        return sum(data)  # 要求可迭代对象
    except TypeError as e:
        print(f"类型错误: {e}")

process_data(123)  # TypeError: 'int' object is not iterable

# 解决方案：类型检查与转换
def safe_process(data):
    if isinstance(data, (int, float)):
        return [data]
    elif isinstance(data, str):
        return [float(data)] if data.replace('.', '').replace('-', '').isdigit() else []
    elif hasattr(data, '__iter__'):
        return list(data)
    else:
        return []

# 类型注解与检查
from typing import List, Union, Optional
def typed_function(numbers: List[float]) -> Optional[float]:
    """带类型注解的函数"""
    if not numbers:
        return None
    return sum(numbers) / len(numbers)

# 使用 isinstance 进行类型检查
def process_with_check(data):
    if isinstance(data, dict):
        return {k: v*2 for k, v in data.items()}
    elif isinstance(data, (list, tuple)):
        return [x*2 for x in data]
    elif isinstance(data, (int, float)):
        return data * 2
    else:
        raise TypeError(f"不支持的数据类型: {type(data)}")

import gc

# 常见内存溢出问题
def memory_leak_example():
    """累积引用导致内存泄漏"""
    objects = []
    for i in range(100000):
        obj = create_large_object()
        objects.append(obj)  # 所有对象被引用，无法释放
    return objects

def safe_memory_usage():
    """使用生成器分批处理"""
    def data_generator():
        for i in range(1000000):
            yield create_large_object()
    
    for obj in data_generator():
        process(obj)  # 处理后立即释放
        gc.collect()  # 主动垃圾回收

# 大数组分块处理
import numpy as np

def chunked_processing(file_path, chunk_size=10000):
    """分块读取大文件，避免内存溢出"""
    for chunk in pd.read_csv(file_path, chunksize=chunk_size):
        result = heavy_computation(chunk)
        save_result(result)

# 使用 np.memmap 处理超大数组
def memory_mapped_array(shape):
    """创建内存映射数组，处理超大数据集"""
    filename = 'large_array.dat'
    fp = np.memmap(filename, dtype='float32', mode='w+', shape=shape)
    return fp

# 清理大型中间变量
import numpy as np

def compute_with_cleanup(data):
    # 创建临时大数组
    temp = np.random.randn(10000, 10000)
    result = np.dot(temp, data)
    
    # 清理临时变量
    del temp
    gc.collect()
    
    return result

# 常见依赖冲突
# pip show package_name  # 查看包信息
# pip list --outdated     # 列出过时的包
# pip freeze > requirements.txt  # 导出依赖

# 使用虚拟环境隔离
import subprocess
import sys

def create_venv(venv_path):
    """创建独立的虚拟环境"""
    subprocess.check_call([sys.executable, '-m', 'venv', venv_path])

# 使用 requirements.txt 管理依赖
# 格式：package==version
# 常用命令：
# pip install -r requirements.txt
# pip freeze > requirements.txt

# 使用 pip-tools 锁定依赖
# pip install pip-tools
# pip-compile requirements.in  # 生成 requirements.txt
# pip-sync requirements.txt    # 同步环境

# pyproject.toml（现代方式）
# [project]
# name = "my-project"
# version = "0.1.0"
# dependencies = [
#     "numpy>=1.21.0",
#     "pandas>=1.3.0",
# ]

# conda 环境管理
# conda create -n ai-env python=3.10
# conda activate ai-env
# conda install numpy pandas pytorch
# conda env export > environment.yml
# conda env create -f environment.yml

import os
import sys

# Python 版本问题
print(f"Python版本: {sys.version}")
print(f"Python路径: {sys.executable}")

# 确保使用正确的 Python
# which python  # Unix
# where python  # Windows

# 路径问题
print(f"当前工作目录: {os.getcwd()}")
print(f"sys.path: {sys.path}")

# 添加路径
sys.path.insert(0, '/path/to/module')
from my_module import my_function

# 环境变量配置
os.environ['PYTHONPATH'] = '/path/to/project'
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'  # TensorFlow 日志级别

# 使用 dotenv 管理环境变量
# pip install python-dotenv
from dotenv import load_dotenv
load_dotenv()  # 加载 .env 文件

api_key = os.getenv('API_KEY')

# 跨平台路径处理
from pathlib import Path

def get_data_path():
    base_dir = Path(__file__).parent
    data_dir = base_dir / 'data' / 'raw'
    data_dir.mkdir(parents=True, exist_ok=True)
    return data_dir

my-ai-project/
├── .github/                    # GitHub 配置
│   ├── workflows/             # CI/CD 工作流
│   │   └── ci.yml
│   └── ISSUE_TEMPLATE/       # Issue 模板
├── configs/                   # 配置文件目录
│   ├── model_config.yaml
│   ├── training_config.yaml
│   └── inference_config.yaml
├── data/                      # 数据目录
│   ├── raw/                   # 原始数据
│   ├── processed/             # 处理后数据
│   └── external/             # 外部数据
├── docs/                      # 文档目录
│   ├── API.md
│   ├── CHANGELOG.md
│   └── tutorials/
├── logs/                      # 日志目录
├── models/                    # 模型保存目录
│   ├── checkpoints/
│   └── saved_models/
├── notebooks/                 # Jupyter Notebooks
│   ├── exploration/
│   └── tutorials/
├── reports/                   # 分析报告
│   └── figures/
├── src/                       # 源代码目录
│   ├── __init__.py
│   ├── data/                  # 数据处理模块
│   │   ├── __init__.py
│   │   ├── dataset.py
│   │   ├── transforms.py
│   │   └── loader.py
│   ├── models/               # 模型定义模块
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── transformer.py
│   │   └── classifier.py
│   ├── training/              # 训练模块
│   │   ├── __init__.py
│   │   ├── trainer.py
│   │   ├── optimizer.py
│   │   └── scheduler.py
│   ├── evaluation/            # 评估模块
│   │   ├── __init__.py
│   │   ├── metrics.py
│   │   └── evaluator.py
│   └── utils/                 # 工具模块
│       ├── __init__.py
│       ├── logger.py
│       └── helpers.py
├── tests/                     # 测试目录
│   ├── __init__.py
│   ├── test_data/
│   ├── test_models/
│   └── test_training/
├── .gitignore
├── .dockerignore
├── Dockerfile
├── docker-compose.yml
├── pyproject.toml
├── setup.py
├── requirements.txt
├── requirements-dev.txt
├── README.md
└── CHANGELOG.md

# src/data/__init__.py
from .dataset import BaseDataset, TextDataset, ImageDataset
from .transforms import TextTransform, ImageTransform
from .loader import DataLoader, create_data_loader

__all__ = [
    'BaseDataset',
    'TextDataset', 
    'ImageDataset',
    'TextTransform',
    'ImageTransform',
    'DataLoader',
    'create_data_loader'
]

# src/models/base.py
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
import torch
import torch.nn as nn

class BaseModel(ABC, nn.Module):
    """模型基类，定义通用接口"""
    
    def __init__(self, config: Dict[str, Any]):
        super().__init__()
        self.config = config
        self._build_model()
    
    @abstractmethod
    def _build_model(self):
        """子类实现模型构建逻辑"""
        pass
    
    @abstractmethod
    def forward(self, x):
        """前向传播"""
        pass
    
    def save(self, path: str):
        """保存模型"""
        torch.save(self.state_dict(), path)
    
    def load(self, path: str):
        """加载模型"""
        self.load_state_dict(torch.load(path))
    
    def freeze(self, layers: Optional[list] = None):
        """冻结指定层"""
        if layers is None:
            for param in self.parameters():
                param.requires_grad = False
        else:
            for name, param in self.named_parameters():
                if any(layer in name for layer in layers):
                    param.requires_grad = False
    
    def unfreeze(self, layers: Optional[list] = None):
        """解冻指定层"""
        if layers is None:
            for param in self.parameters():
                param.requires_grad = True
        else:
            for name, param in self.named_parameters():
                if any(layer in name for layer in layers):
                    param.requires_grad = True

# configs/model_config.yaml
model:
  name: "TransformerClassifier"
  hidden_size: 768
  num_layers: 12
  num_heads: 12
  dropout: 0.1
  vocab_size: 50000

training:
  batch_size: 32
  learning_rate: 0.0001
  epochs: 100
  warmup_steps: 1000
  gradient_clip: 1.0
  weight_decay: 0.01

optimizer:
  type: "AdamW"
  betas: [0.9, 0.999]
  eps: 1.0e-8

scheduler:
  type: "CosineAnnealingLR"
  T_max: 100000

# 配置加载器
import yaml
from pathlib import Path
from typing import Dict, Any
from dataclasses import dataclass, field

@dataclass
class ModelConfig:
    name: str = "BaseModel"
    hidden_size: int = 512
    num_layers: int = 6
    dropout: float = 0.1

@dataclass
class TrainingConfig:
    batch_size: int = 32
    learning_rate: float = 0.001
    epochs: int = 100
    warmup_steps: int = 1000
    gradient_clip: float = 1.0

class ConfigManager:
    def __init__(self, config_path: str):
        self.config_path = Path(config_path)
        self._raw_config = self._load_yaml()
    
    def _load_yaml(self) -> Dict[str, Any]:
        with open(self.config_path, 'r') as f:
            return yaml.safe_load(f)
    
    @property
    def model(self) -> ModelConfig:
        return ModelConfig(**self._raw_config.get('model', {}))
    
    @property
    def training(self) -> TrainingConfig:
        return TrainingConfig(**self._raw_config.get('training', {}))
    
    def get(self, key: str, default: Any = None) -> Any:
        keys = key.split('.')
        value = self._raw_config
        for k in keys:
            if isinstance(value, dict):
                value = value.get(k)
            else:
                return default
        return value if value is not None else default

# 使用示例
config = ConfigManager('configs/model_config.yaml')
model_config = config.model
training_config = config.training
print(f"训练配置: batch_size={training_config.batch_size}")

from pathlib import Path
from dataclasses import dataclass
from typing import Optional

@dataclass
class DataPaths:
    """数据路径管理器"""
    base_dir: Path
    
    @property
    def raw(self) -> Path:
        return self.base_dir / 'raw'
    
    @property
    def processed(self) -> Path:
        return self.base_dir / 'processed'
    
    @property
    def external(self) -> Path:
        return self.base_dir / 'external'
    
    def ensure_dirs(self):
        """确保所有目录存在"""
        for attr in ['raw', 'processed', 'external']:
            getattr(self, attr).mkdir(parents=True, exist_ok=True)

# 数据版本管理
class DataVersionManager:
    def __init__(self, data_dir: Path):
        self.data_dir = data_dir
        self.version_file = data_dir / 'version.json'
    
    def get_current_version(self) -> str:
        if self.version_file.exists():
            import json
            with open(self.version_file, 'r') as f:
                return json.load(f).get('version', 'v0.0.0')
        return 'v0.0.0'
    
    def save_version(self, version: str, metadata: dict):
        import json
        self.version_file.parent.mkdir(parents=True, exist_ok=True)
        with open(self.version_file, 'w') as f:
            json.dump({'version': version, 'metadata': metadata}, f, indent=2)

from typing import TypeVar, List, Callable, Any
from functools import wraps

T = TypeVar('T')

# 原则1：单一职责
def validate_data(data: List[T]) -> List[T]:
    """只做数据验证，不要混杂其他逻辑"""
    if not data:
        return []
    return [item for item in data if item is not None]

# 原则2：使用类型注解
def process_batch(
    data: List[T],
    transform: Callable[[T], T],
    batch_size: int = 32,
    parallel: bool = False
) -> List[T]:
    """类型注解使函数接口清晰"""
    if parallel:
        from concurrent.futures import ThreadPoolExecutor
        with ThreadPoolExecutor() as executor:
            return list(executor.map(transform, data))
    return [transform(item) for item in data]

# 原则3：错误处理
class ProcessingError(Exception):
    """自定义异常"""
    pass

def safe_process(data: Any, default: Any = None) -> Any:
    """提供安全的默认处理"""
    try:
        return process(data)
    except ValueError as e:
        print(f"处理失败: {e}")
        return default
    except ProcessingError as e:
        print(f"处理错误: {e}")
        raise

# 原则4：文档字符串
def calculate_metrics(
    predictions: List[float],
    labels: List[float],
    threshold: float = 0.5
) -> dict:
    """
    计算分类指标。

    Args:
        predictions: 预测概率列表
        labels: 真实标签列表
        threshold: 分类阈值

    Returns:
        包含 accuracy, precision, recall, f1 的字典

    Raises:
        ValueError: 当输入列表为空或不匹配时

    Example:
        >>> preds = [0.8, 0.3, 0.9]
        >>> labels = [1, 0, 1]
        >>> calculate_metrics(preds, labels)
        {'accuracy': 0.67, 'precision': 1.0, 'recall': 0.5, 'f1': 0.67}
    """
    if len(predictions) != len(labels):
        raise ValueError("predictions 和 labels 长度必须一致")
    if not predictions:
        raise ValueError("输入列表不能为空")

    pred_binary = [1 if p > threshold else 0 for p in predictions]
    
    tp = sum(1 for p, l in zip(pred_binary, labels) if p == 1 and l == 1)
    fp = sum(1 for p, l in zip(pred_binary, labels) if p == 1 and l == 0)
    tn = sum(1 for p, l in zip(pred_binary, labels) if p == 0 and l == 0)
    fn = sum(1 for p, l in zip(pred_binary, labels) if p == 0 and l == 1)
    
    accuracy = (tp + tn) / len(labels)
    precision = tp / (tp + fp) if (tp + fp) > 0 else 0
    recall = tp / (tp + fn) if (tp + fn) > 0 else 0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
    
    return {
        'accuracy': round(accuracy, 4),
        'precision': round(precision, 4),
        'recall': round(recall, 4),
        'f1': round(f1, 4)
    }

# 原则5：函数组合
def compose(*functions):
    """函数组合器"""
    def inner(x):
        result = x
        for func in reversed(functions):
            result = func(result)
        return result
    return inner

# 使用示例
pipeline = compose(
    lambda x: x.strip(),
    lambda x: x.lower(),
    lambda x: x.replace(' ', '_'),
    lambda x: f"processed_{x}"
)
result = pipeline("  Hello World")

from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, field
import numpy as np

@dataclass
class ModelOutput:
    """模型输出数据结构"""
    predictions: np.ndarray
    probabilities: Optional[np.ndarray] = None
    embeddings: Optional[np.ndarray] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

class BaseClassifier(ABC):
    """分类器基类"""
    
    def __init__(self, num_classes: int):
        self.num_classes = num_classes
        self.is_trained = False
    
    @abstractmethod
    def fit(self, X, y, **kwargs):
        """训练模型"""
        pass
    
    @abstractmethod
    def predict(self, X) -> np.ndarray:
        """预测类别"""
        pass
    
    def predict_proba(self, X) -> np.ndarray:
        """预测概率"""
        raise NotImplementedError("子类需要实现此方法")
    
    @abstractmethod
    def save(self, path: str):
        """保存模型"""
        pass
    
    @abstractmethod
    def load(self, path: str):
        """加载模型"""
        pass

class ModelRegistry:
    """模型注册表，支持多种模型动态切换"""
    
    _models: Dict[str, type] = {}
    
    @classmethod
    def register(cls, name: str):
        """装饰器注册模型"""
        def decorator(model_class):
            cls._models[name] = model_class
            return model_class
        return decorator
    
    @classmethod
    def create(cls, name: str, **kwargs) -> BaseClassifier:
        """创建模型实例"""
        if name not in cls._models:
            raise ValueError(f"未注册的模型: {name}，可用: {list(cls._models.keys())}")
        return cls._models[name](**kwargs)
    
    @classmethod
    def available_models(cls) -> List[str]:
        """列出可用模型"""
        return list(cls._models.keys())

# 使用示例
@ModelRegistry.register('naive_bayes')
class NaiveBayesClassifier(BaseClassifier):
    def __init__(self, alpha: float = 1.0):
        super().__init__(num_classes=2)
        self.alpha = alpha
    
    def fit(self, X, y, **kwargs):
        self.is_trained = True
        return self
    
    def predict(self, X) -> np.ndarray:
        return np.zeros(len(X))
    
    def save(self, path: str):
        pass
    
    def load(self, path: str):
        pass

# 创建已注册的模型
model = ModelRegistry.create('naive_bayes', alpha=0.5)

from typing import Callable, Any
import numpy as np

# 策略模式：封装算法族
class DataAugmentationStrategy(ABC):
    @abstractmethod
    def apply(self, image: np.ndarray) -> np.ndarray:
        pass

class RandomFlip(DataAugmentationStrategy):
    def apply(self, image: np.ndarray) -> np.ndarray:
        if np.random.rand() > 0.5:
            return np.fliplr(image)
        return image

class RandomRotate(DataAugmentationStrategy):
    def __init__(self, max_angle: float = 15):
        self.max_angle = max_angle
    
    def apply(self, image: np.ndarray) -> np.ndarray:
        from scipy.ndimage import rotate
        angle = np.random.uniform(-self.max_angle, self.max_angle)
        return rotate(image, angle, reshape=False)

class AugmentationPipeline:
    def __init__(self, strategies: List[DataAugmentationStrategy]):
        self.strategies = strategies
    
    def apply(self, image: np.ndarray) -> np.ndarray:
        for strategy in self.strategies:
            image = strategy.apply(image)
        return image

# 观察者模式：训练进度监控
class TrainingObserver:
    def on_epoch_start(self, epoch: int):
        pass
    
    def on_epoch_end(self, epoch: int, metrics: dict):
        pass
    
    def on_batch_end(self, batch: int, loss: float):
        pass

class TensorBoardObserver(TrainingObserver):
    def __init__(self, log_dir: str):
        from torch.utils.tensorboard import SummaryWriter
        self.writer = SummaryWriter(log_dir)
    
    def on_epoch_end(self, epoch: int, metrics: dict):
        for key, value in metrics.items():
            self.writer.add_scalar(key, value, epoch)

class MetricsLoggerObserver(TrainingObserver):
    def __init__(self, log_file: str):
        self.log_file = log_file
    
    def on_epoch_end(self, epoch: int, metrics: dict):
        with open(self.log_file, 'a') as f:
            f.write(f"Epoch {epoch}: {metrics}\n")

class Trainer:
    def __init__(self, model, observers: List[TrainingObserver] = None):
        self.model = model
        self.observers = observers or []
    
    def train(self, train_loader, epochs: int):
        for epoch in range(epochs):
            for observer in self.observers:
                observer.on_epoch_start(epoch)
            
            # 训练逻辑
            for batch_idx, (data, target) in enumerate(train_loader):
                # 前向传播、反向传播...
                loss = 0.1  # 示例
                for observer in self.observers:
                    observer.on_batch_end(batch_idx, loss)
            
            metrics = {'train_loss': 0.1, 'val_acc': 0.95}
            for observer in self.observers:
                observer.on_epoch_end(epoch, metrics)

from typing import TypeVar, Generic, List, Optional, Callable
from functools import wraps
import time

T = TypeVar('T')

# 泛型工具类
class BatchProcessor(Generic[T]):
    """通用批处理器"""
    
    def __init__(self, batch_size: int = 32):
        self.batch_size = batch_size
    
    def process(
        self, 
        items: List[T], 
        func: Callable[[T], Any]
    ) -> List[Any]:
        results = []
        for i in range(0, len(items), self.batch_size):
            batch = items[i:i + self.batch_size]
            results.extend([func(item) for item in batch])
        return results

# 装饰器复用
def timing_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"{func.__name__} 执行时间: {time.time() - start:.2f}秒")
        return result
    return wrapper

def retry(max_attempts: int = 3, delay: float = 1.0):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

def cache_result(func):
    """简单缓存装饰器"""
    cache = {}
    
    @wraps(func)
    def wrapper(*args, **kwargs):
        key = str(args) + str(kwargs)
        if key not in cache:
            cache[key] = func(*args, **kwargs)
        return cache[key]
    
    wrapper.cache = cache
    wrapper.clear_cache = lambda: cache.clear()
    return wrapper

# 混入类（Mixin）
class SerializableMixin:
    def to_dict(self):
        return {
            key: value 
            for key, value in self.__dict__.items() 
            if not key.startswith('_')
        }
    
    @classmethod
    def from_dict(cls, data: dict):
        return cls(**data)

class LoggedMixin:
    def log(self, message: str):
        print(f"[{self.__class__.__name__}] {message}")

class BaseService(SerializableMixin, LoggedMixin):
    pass

# 项目名称

简短的项目描述，说明项目的主要功能和目标。

[![CI](https://github.com/user/project/actions/workflows/ci.yml/badge.svg)](https://github.com/user/project/actions)
[![PyPI version](https://badge.fury.io/py/package-name.svg)](https://badge.fury.io/py/package-name)
[![License](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)

## ✨ 特性

- 特性1：详细描述
- 特性2：详细描述
- 特性3：详细描述

## 📦 安装

### 依赖要求

- Python >= 3.8
- PyTorch >= 1.9
- ...

### 安装方式

```bash
pip install package-name
# 或从源码安装
git clone https://github.com/user/project.git
cd project
pip install -e .

from package_name import main_function

# 基本使用
result = main_function(input_data)
print(result)

git clone https://github.com/user/project.git
cd project
python -m venv venv
source venv/bin/activate  # Linux/Mac
# 或 venv\Scripts\activate  # Windows
pip install -r requirements-dev.txt

pytest tests/
pytest --cov=src tests/  # 带覆盖率

### 代码注释规范

```python
"""
模块文档字符串

本模块提供 AI 模型的核心功能，包括：
- 模型定义与训练
- 推理与预测
- 模型保存与加载

Example:
    >>> from src.models import Transformer
    >>> model = Transformer(vocab_size=50000)
    >>> output = model(input_ids)
"""

from typing import Dict, List, Optional, Tuple
import torch
import torch.nn as nn

class Transformer(nn.Module):
    """
    Transformer 模型实现。
    
    基于 "Attention Is All You Need" 论文实现，
    支持自定义层数、注意力头数和隐藏层大小。
    
    Attributes:
        vocab_size: 词汇表大小
        hidden_size: 隐藏层维度
        num_layers: 编码器/解码器层数
    
    Example:
        >>> model = Transformer(vocab_size=1000, hidden_size=256, num_layers=4)
        >>> input_ids = torch.randint(0, 1000, (2, 50))
        >>> output = model(input_ids)
    """
    
    def __init__(
        self,
        vocab_size: int,
        hidden_size: int = 512,
        num_layers: int = 6,
        num_heads: int = 8,
        dropout: float = 0.1
    ) -> None:
        """
        初始化 Transformer 模型。
        
        Args:
            vocab_size: 输入词汇表大小
            hidden_size: 隐藏层维度，默认 512
            num_layers: Transformer 层数，默认 6
            num_heads: 注意力头数，默认 8
            dropout: Dropout 概率，默认 0.1
        
        Raises:
            ValueError: 当 hidden_size 不能被 num_heads 整除时
        """
        if hidden_size % num_heads != 0:
            raise ValueError(
                f"hidden_size ({hidden_size}) 必须能被 num_heads ({num_heads}) 整除"
            )
        
        super().__init__()
        self.vocab_size = vocab_size
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        
        # 嵌入层：将词索引映射为密集向量
        self.embedding = nn.Embedding(vocab_size, hidden_size)
        self.positional_encoding = PositionalEncoding(hidden_size, dropout)
        self.transformer = nn.TransformerEncoder(
            nn.TransformerEncoderLayer(
                d_model=hidden_size,
                nhead=num_heads,
                dropout=dropout,
                batch_first=True
            ),
            num_layers=num_layers
        )
        self.fc = nn.Linear(hidden_size, vocab_size)
    
    def forward(
        self,
        input_ids: torch.Tensor,
        attention_mask: Optional[torch.Tensor] = None
    ) -> torch.Tensor:
        """
        前向传播。
        
        Args:
            input_ids: 输入序列的词索引，形状 (batch_size, seq_len)
            attention_mask: 注意力掩码，可选
        
        Returns:
            logits: 模型输出，形状 (batch_size, seq_len, vocab_size)
        
        Note:
            此实现使用编码器架构，如需解码器请使用 TransformerModel 类。
        """
        # 嵌入 + 位置编码
        x = self.embedding(input_ids)
        x = self.positional_encoding(x)
        
        # Transformer 编码器
        x = self.transformer(x, src_key_padding_mask=attention_mask)
        
        # 输出层
        logits = self.fc(x)
        return logits
    
    def generate(
        self,
        input_ids: torch.Tensor,
        max_length: int = 50,
        temperature: float = 1.0,
        top_k: int = 0,
        top_p: float = 1.0
    ) -> torch.Tensor:
        """
        自回归生成文本。
        
        Args:
            input_ids: 初始输入序列
            max_length: 最大生成长度
            temperature: 采样温度，越高越随机
            top_k: Top-K 采样参数
            top_p: Nucleus 采样参数
        
        Returns:
            generated: 生成的序列
        """
        self.eval()
        with torch.no_grad():
            for _ in range(max_length):
                logits = self.forward(input_ids)
                logits = logits[:, -1, :] / temperature
                
                if top_k > 0:
                    indices_to_remove = logits < torch.topk(logits, top_k)[0][:, -1]
                    logits[indices_to_remove] = float('-inf')
                
                if top_p < 1.0:
                    sorted_logits, sorted_indices = torch.sort(logits, descending=True)
                    cumulative_probs = torch.cumsum(
                        torch.softmax(sorted_logits, dim=-1), dim=-1
                    )
                    sorted_indices_to_remove = cumulative_probs > top_p
                    sorted_indices_to_remove[:, 1:] = sorted_indices_to_remove[:, :-1].clone()
                    sorted_indices_to_remove[:, 0] = 0
                    indices_to_remove = sorted_indices_to_remove.scatter(
                        1, sorted_indices, sorted_indices_to_remove
                    )
                    logits[indices_to_remove] = float('-inf')
                
                probs = torch.softmax(logits, dim=-1)
                next_token = torch.multinomial(probs, num_samples=1)
                input_ids = torch.cat([input_ids, next_token], dim=1)
        
        return input_ids

# docs/conf.py
import os
import sys
sys.path.insert(0, os.path.abspath('..'))

project = 'My AI Project'
copyright = '2024, Author'
author = 'Author'

extensions = [
    'sphinx.ext.autodoc',
    'sphinx.ext.napoleon',
    'sphinx.ext.viewcode',
    'sphinx.ext.coverage',
    'sphinx.ext.intersphinx',
]

templates_path = ['_templates']
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']

html_theme = 'sphinx_book_theme'
html_static_path = ['_static']

intersphinx_mapping = {
    'python': ('https://docs.python.org/3', None),
    'torch': ('https://pytorch.org/docs/stable', None),
}

# docs/API.rst
"""
API 参考
========

.. contents::
   :local:
   :depth: 2

模型模块
--------

.. automodule:: src.models
   :members:
   :undoc-members:
   :show-inheritance:

数据处理
--------

.. automodule:: src.data
   :members:
   :undoc-members:
   :show-inheritance:
"""

# Changelog

所有重要的项目变更都将记录在此文件中。格式基于 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.0.0/)。

## [2.0.0] - 2024-01-15

### 新增
- 新增 Transformer 模型支持
- 添加分布式训练功能
- 新增 ONNX 模型导出
- 添加模型量化工具

### 优化
- 重构数据加载器，提升 30% 加载速度
- 优化 GPU 内存使用
- 改进梯度累积策略

### 修复
- 修复多GPU训练时的同步问题
- 修复模型保存时的内存泄漏

### 破坏性变更
- 移除旧版 CNN 模型（请使用 v1.x 分支）
- 配置文件格式已更新，请参考迁移指南

## [1.5.0] - 2023-12-01

### 新增
- 新增 BERT 模型微调示例
- 添加 wandb 日志集成
- 支持模型检查点自动保存

### 文档
- 更新快速开始教程
- 添加 API 文档
- 新增故障排除指南

---

## 提交信息规范

采用 Conventional Commits 格式：

类型（type）：
- feat: 新功能
- fix: Bug 修复
- docs: 文档更新
- style: 代码格式（不影响功能）
- refactor: 重构
- perf: 性能优化
- test: 测试相关
- chore: 构建或辅助工具

示例：