필사 모드: AI/ML 的 Python 高阶技巧：性能优化、并行处理、内存管理

import cProfile
import pstats
import io
import numpy as np

def slow_matrix_multiply(a, b):
    """低效的矩阵乘法"""
    rows_a = len(a)
    cols_a = len(a[0])
    cols_b = len(b[0])
    result = [[0] * cols_b for _ in range(rows_a)]
    for i in range(rows_a):
        for j in range(cols_b):
            for k in range(cols_a):
                result[i][j] += a[i][k] * b[k][j]
    return result

def profile_function():
    size = 100
    a = [[float(i + j) for j in range(size)] for i in range(size)]
    b = [[float(i * j + 1) for j in range(size)] for i in range(size)]
    slow_matrix_multiply(a, b)

# 执行性能剖析
pr = cProfile.Profile()
pr.enable()
profile_function()
pr.disable()

# 分析结果
stream = io.StringIO()
ps = pstats.Stats(pr, stream=stream)
ps.sort_stats('cumulative')
ps.print_stats(20)
print(stream.getvalue())

python -m cProfile -s cumulative your_script.py
python -m cProfile -o output.prof your_script.py

import pstats

p = pstats.Stats('output.prof')
p.sort_stats('cumulative')
p.print_stats(10)  # 排名前 10 的函数
p.print_callers('slow_function')  # 特定函数的调用者
p.print_callees('main')  # 特定函数所调用的函数

pip install line_profiler

# line_profiler 使用示例
from line_profiler import LineProfiler
import numpy as np

def process_batch(data):
    # 测量每一行的执行时间
    result = []
    for item in data:
        normalized = (item - item.mean()) / (item.std() + 1e-8)
        scaled = normalized * 2.0
        clipped = np.clip(scaled, -5, 5)
        result.append(clipped)
    return np.stack(result)

# 设置剖析器
lp = LineProfiler()
lp_wrapper = lp(process_batch)

# 用样本数据运行
data = [np.random.randn(1000) for _ in range(1000)]
lp_wrapper(data)

# 输出结果
lp.print_stats()

# 在 Jupyter 中使用 line_profiler
%load_ext line_profiler

%lprun -f process_batch process_batch(data)

pip install memory_profiler

from memory_profiler import profile
import numpy as np

@profile
def memory_intensive_function(n=1_000_000):
    # 追踪内存分配
    list_data = list(range(n))          # 创建列表
    arr = np.array(list_data)           # 转换为 NumPy 数组
    del list_data                       # 删除列表

    squared = arr ** 2                  # 创建新数组
    filtered = squared[squared > 100]   # 过滤

    return filtered.sum()

result = memory_intensive_function()
print(f"结果: {result}")

python -m memory_profiler your_script.py
mprof run your_script.py
mprof plot  # 可视化

pip install py-spy

# 对正在运行的进程做剖析
py-spy top --pid 12345

# 生成火焰图
py-spy record -o profile.svg --pid 12345

# 直接对脚本做剖析
py-spy record -o profile.svg -- python your_script.py

pip install snakeviz

python -m cProfile -o output.prof your_script.py
snakeviz output.prof

import numpy as np
import time

# 性能对比函数
def benchmark(func, *args, runs=5):
    times = []
    for _ in range(runs):
        start = time.perf_counter()
        result = func(*args)
        end = time.perf_counter()
        times.append(end - start)
    return np.mean(times), np.std(times), result

# 1. Python for 循环
def relu_loop(x):
    result = []
    for val in x:
        result.append(max(0, val))
    return result

# 2. NumPy 向量化
def relu_numpy(x):
    return np.maximum(0, x)

# 3. NumPy 条件索引
def relu_clip(x):
    return np.clip(x, 0, None)

# 对比
n = 1_000_000
data_list = [np.random.randn() for _ in range(n)]
data_np = np.array(data_list)

t_loop, _, _ = benchmark(relu_loop, data_list)
t_numpy, _, _ = benchmark(relu_numpy, data_np)
t_clip, _, _ = benchmark(relu_clip, data_np)

print(f"Python 循环:      {t_loop:.4f}s")
print(f"NumPy maximum:    {t_numpy:.4f}s  ({t_loop/t_numpy:.0f}x faster)")
print(f"NumPy clip:       {t_clip:.4f}s  ({t_loop/t_clip:.0f}x faster)")

import numpy as np

# 创建自定义 ufunc
def custom_activation_scalar(x, alpha=0.1):
    """ELU 激活函数"""
    if x >= 0:
        return x
    else:
        return alpha * (np.exp(x) - 1)

# 用 vectorize 创建 ufunc
elu_ufunc = np.vectorize(custom_activation_scalar)

# 完整的 NumPy ufunc(更快)
def elu_numpy(x, alpha=0.1):
    return np.where(x >= 0, x, alpha * (np.exp(x) - 1))

# 用 frompyfunc 创建 ufunc
elu_frompyfunc = np.frompyfunc(
    lambda x: x if x >= 0 else 0.1 * (np.exp(x) - 1),
    1, 1
)

# 性能测试
x = np.random.randn(1_000_000)
t_vectorize, _, _ = benchmark(elu_ufunc, x)
t_numpy, _, _ = benchmark(elu_numpy, x)

print(f"np.vectorize: {t_vectorize:.4f}s")
print(f"np.where:     {t_numpy:.4f}s")

import numpy as np

# 爱因斯坦求和记号示例
batch_size, seq_len, d_model = 32, 128, 512
heads, d_head = 8, 64

# 矩阵乘法
A = np.random.randn(batch_size, seq_len, d_model)
W = np.random.randn(d_model, d_model)

# 标准方法
result_matmul = A @ W  # (32, 128, 512)

# 用 einsum 实现同样的运算
result_einsum = np.einsum('bsi,ij->bsj', A, W)

# 计算注意力分数(批量矩阵乘法)
Q = np.random.randn(batch_size, heads, seq_len, d_head)
K = np.random.randn(batch_size, heads, seq_len, d_head)

# 标准方法
scores_manual = Q @ K.transpose(0, 1, 3, 2)  # (32, 8, 128, 128)

# 用 einsum 计算注意力分数
scores_einsum = np.einsum('bhid,bhjd->bhij', Q, K)

# 批量外积(outer product)
a = np.random.randn(batch_size, d_model)
b = np.random.randn(batch_size, d_model)
outer = np.einsum('bi,bj->bij', a, b)  # (32, 512, 512)

# 迹(对角线之和)
M = np.random.randn(batch_size, d_model, d_model)
trace = np.einsum('bii->b', M)  # (32,)

print(f"注意力分数 shape: {scores_einsum.shape}")
print(f"外积 shape: {outer.shape}")
print(f"迹 shape: {trace.shape}")

# 用 optimize 参数选择最优计算路径
result = np.einsum('bhid,bhjd->bhij', Q, K, optimize='optimal')

import numpy as np
from numpy.lib.stride_tricks import as_strided, sliding_window_view

# 滑动窗口(旧方法)
def sliding_window_strided(arr, window_size):
    """用步幅技巧生成滑动窗口"""
    shape = (arr.shape[0] - window_size + 1, window_size)
    strides = (arr.strides[0], arr.strides[0])
    return as_strided(arr, shape=shape, strides=strides)

# 新方法(NumPy 1.20+)
arr = np.arange(20, dtype=float)
window_size = 5

# 现代化的方法
windows = sliding_window_view(arr, window_size)
print(f"滑动窗口 shape: {windows.shape}")  # (16, 5)

# 计算移动平均
moving_avg = windows.mean(axis=-1)
print(f"移动平均: {moving_avg[:5]}")

# 2D 滑动窗口(卷积的预处理)
image = np.random.randn(64, 64)
kernel_size = (3, 3)
windows_2d = sliding_window_view(image, kernel_size)
print(f"2D 窗口 shape: {windows_2d.shape}")  # (62, 62, 3, 3)

# 批处理中的步幅技巧
batch = np.random.randn(1000, 10)  # (样本, 特征)
# 不重叠的分块
chunk_size = 50
n_chunks = len(batch) // chunk_size
chunks = batch[:n_chunks * chunk_size].reshape(n_chunks, chunk_size, -1)
print(f"分块 shape: {chunks.shape}")  # (20, 50, 10)

from numba import jit, njit, prange
import numpy as np
import time

# @jit: 支持回退模式(允许 Python 对象)
@jit(nopython=False)
def sum_with_jit(arr):
    total = 0.0
    for i in range(len(arr)):
        total += arr[i]
    return total

# @njit: 纯原生模式(更快，推荐使用)
@njit
def sum_with_njit(arr):
    total = 0.0
    for i in range(len(arr)):
        total += arr[i]
    return total

# 复杂函数示例
@njit
def compute_distances(points):
    """计算各点之间的欧氏距离矩阵"""
    n = len(points)
    distances = np.zeros((n, n))
    for i in range(n):
        for j in range(i + 1, n):
            diff = points[i] - points[j]
            dist = np.sqrt(np.sum(diff ** 2))
            distances[i, j] = dist
            distances[j, i] = dist
    return distances

# 性能对比
arr = np.random.randn(1_000_000)

# 首次调用包含编译时间
_ = sum_with_njit(arr)  # 预热

start = time.perf_counter()
result_numpy = np.sum(arr)
print(f"NumPy sum: {time.perf_counter() - start:.4f}s")

start = time.perf_counter()
result_njit = sum_with_njit(arr)
print(f"Numba njit: {time.perf_counter() - start:.4f}s")

# 计算点云距离
points = np.random.randn(500, 3).astype(np.float64)
_ = compute_distances(points)  # 预热

start = time.perf_counter()
dist_matrix = compute_distances(points)
print(f"Numba 距离矩阵 ({points.shape[0]}x{points.shape[0]}): {time.perf_counter() - start:.4f}s")

from numba import njit, float64, int64
from numba import types
import numpy as np

# 用显式类型编译(无首次调用延迟)
@njit(float64[:](float64[:], float64))
def scale_array_typed(arr, factor):
    return arr * factor

# 多个类型签名
from numba import float32, float64

@njit([
    float32[:](float32[:], float32),
    float64[:](float64[:], float64)
])
def scale_multi_type(arr, factor):
    return arr * factor

# 复杂的类型声明
@njit
def softmax_numba(x):
    """由 Numba 加速的 softmax"""
    max_val = np.max(x)
    exp_x = np.exp(x - max_val)
    return exp_x / np.sum(exp_x)

# 批量 softmax
@njit
def batch_softmax(X):
    """2D 数组的批量 softmax"""
    result = np.empty_like(X)
    for i in range(X.shape[0]):
        result[i] = softmax_numba(X[i])
    return result

# 测试
X = np.random.randn(1000, 512).astype(np.float64)
_ = batch_softmax(X)  # 预热

start = time.perf_counter()
result = batch_softmax(X)
print(f"Numba 批量 softmax: {time.perf_counter() - start:.4f}s")

from numba import njit, prange
import numpy as np

@njit(parallel=True)
def parallel_normalize(X):
    """按行归一化(并行处理)"""
    result = np.empty_like(X)
    for i in prange(X.shape[0]):  # 自动并行化
        row = X[i]
        mean = row.mean()
        std = row.std()
        result[i] = (row - mean) / (std + 1e-8)
    return result

@njit(parallel=True)
def parallel_matrix_multiply(A, B):
    """并行矩阵乘法"""
    m, k = A.shape
    k2, n = B.shape
    C = np.zeros((m, n))
    for i in prange(m):
        for j in range(n):
            for l in range(k):
                C[i, j] += A[i, l] * B[l, j]
    return C

# 并行批处理
@njit(parallel=True)
def parallel_batch_process(data, weights):
    """批数据的加权求和(并行)"""
    n_samples, n_features = data.shape
    results = np.zeros(n_samples)
    for i in prange(n_samples):
        dot = 0.0
        for j in range(n_features):
            dot += data[i, j] * weights[j]
        results[i] = dot
    return results

# 性能测试
X = np.random.randn(10000, 512).astype(np.float64)

# 预热
_ = parallel_normalize(X)

start = time.perf_counter()
result_parallel = parallel_normalize(X)
t_parallel = time.perf_counter() - start

# 对比：顺序版本
@njit
def sequential_normalize(X):
    result = np.empty_like(X)
    for i in range(X.shape[0]):
        row = X[i]
        mean = row.mean()
        std = row.std()
        result[i] = (row - mean) / (std + 1e-8)
    return result

_ = sequential_normalize(X)
start = time.perf_counter()
result_seq = sequential_normalize(X)
t_seq = time.perf_counter() - start

print(f"顺序处理: {t_seq:.4f}s")
print(f"并行处理: {t_parallel:.4f}s ({t_seq/t_parallel:.1f}x 更快)")

from numba import njit
import numpy as np

# cache=True 保存编译好的代码
@njit(cache=True)
def cached_computation(x, y):
    """已缓存的 JIT 函数 —— 下次运行时立即加载"""
    result = np.empty(len(x))
    for i in range(len(x)):
        result[i] = x[i] ** 2 + y[i] ** 2
    return result

# 确认编译后的函数信息
arr1 = np.random.randn(1000)
arr2 = np.random.randn(1000)
result = cached_computation(arr1, arr2)
print(f"缓存文件位置: {cached_computation.stats}")

# setup.py —— Cython 构建配置
from setuptools import setup
from Cython.Build import cythonize
import numpy as np

setup(
    ext_modules=cythonize(
        "fast_ops.pyx",
        compiler_directives={
            'language_level': '3',
            'boundscheck': False,
            'wraparound': False,
            'nonecheck': False,
        }
    ),
    include_dirs=[np.get_include()]
)

# fast_ops.pyx
# cython: language_level=3
# cython: boundscheck=False
# cython: wraparound=False

import numpy as np
cimport numpy as cnp
from libc.math cimport exp, sqrt

# 用 C 类型声明达到最大性能
def relu_cython(cnp.ndarray[cnp.double_t, ndim=1] x):
    cdef int n = len(x)
    cdef cnp.ndarray[cnp.double_t, ndim=1] result = np.empty(n)
    cdef int i
    cdef double val

    for i in range(n):
        val = x[i]
        result[i] = val if val > 0 else 0.0

    return result

# 矩阵运算
def matrix_dot_cython(
    cnp.ndarray[cnp.double_t, ndim=2] A,
    cnp.ndarray[cnp.double_t, ndim=2] B
):
    cdef int m = A.shape[0]
    cdef int k = A.shape[1]
    cdef int n = B.shape[1]
    cdef cnp.ndarray[cnp.double_t, ndim=2] C = np.zeros((m, n))
    cdef int i, j, l
    cdef double sum_val

    for i in range(m):
        for j in range(n):
            sum_val = 0.0
            for l in range(k):
                sum_val += A[i, l] * B[l, j]
            C[i, j] = sum_val

    return C

# softmax(使用 C 数学函数)
def softmax_cython(cnp.ndarray[cnp.double_t, ndim=1] x):
    cdef int n = len(x)
    cdef cnp.ndarray[cnp.double_t, ndim=1] result = np.empty(n)
    cdef double max_val = x[0]
    cdef double sum_exp = 0.0
    cdef int i

    for i in range(n):
        if x[i] > max_val:
            max_val = x[i]

    for i in range(n):
        result[i] = exp(x[i] - max_val)
        sum_exp += result[i]

    for i in range(n):
        result[i] /= sum_exp

    return result

python setup.py build_ext --inplace

import numpy as np
import fast_ops  # 已编译的 Cython 模块

x = np.random.randn(1_000_000)
result = fast_ops.relu_cython(x)

# 在 Jupyter 中使用 Cython
%load_ext Cython

%%cython -a
# -a: 注解模式(黄色 = Python，白色 = C)
import numpy as np
cimport numpy as cnp

def fast_l2_norm(cnp.ndarray[cnp.double_t, ndim=1] v):
    cdef int n = len(v)
    cdef double total = 0.0
    cdef int i
    for i in range(n):
        total += v[i] * v[i]
    return total ** 0.5

import threading
import multiprocessing
import time
import numpy as np

def cpu_bound_task(n):
    """CPU 密集型任务"""
    total = 0
    for i in range(n):
        total += i ** 2
    return total

# 多线程(由于 GIL，没有实质性的并行化)
def run_threaded(n_tasks=4, n_per_task=1_000_000):
    threads = []
    start = time.perf_counter()
    for _ in range(n_tasks):
        t = threading.Thread(target=cpu_bound_task, args=(n_per_task,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    return time.perf_counter() - start

# 多进程(真正的并行化)
def run_multiprocessing(n_tasks=4, n_per_task=1_000_000):
    start = time.perf_counter()
    with multiprocessing.Pool(processes=n_tasks) as pool:
        results = pool.map(cpu_bound_task, [n_per_task] * n_tasks)
    return time.perf_counter() - start

t_threaded = run_threaded()
t_mp = run_multiprocessing()

print(f"多线程: {t_threaded:.2f}s")
print(f"多进程: {t_mp:.2f}s ({t_threaded/t_mp:.1f}x 更快)")

import multiprocessing
import numpy as np
from functools import partial

def preprocess_sample(args):
    """单个样本的预处理"""
    idx, data, mean, std = args
    normalized = (data - mean) / (std + 1e-8)
    augmented = normalized + np.random.randn(*normalized.shape) * 0.01
    return idx, augmented

def preprocess_dataset_parallel(dataset, n_workers=None):
    """数据集的并行预处理"""
    if n_workers is None:
        n_workers = multiprocessing.cpu_count()

    mean = dataset.mean(axis=0)
    std = dataset.std(axis=0)

    # 打包参数
    args = [(i, dataset[i], mean, std) for i in range(len(dataset))]

    with multiprocessing.Pool(processes=n_workers) as pool:
        # map: 保证结果顺序
        results = pool.map(preprocess_sample, args, chunksize=100)

    # 结果排序和重组
    results.sort(key=lambda x: x[0])
    return np.stack([r[1] for r in results])

# 用 imap_unordered 做流式处理
def process_large_dataset(data_path_list, n_workers=4):
    """大文件的并行处理"""
    def load_and_process(path):
        data = np.load(path)
        return data.mean(), data.std()

    results = []
    with multiprocessing.Pool(processes=n_workers) as pool:
        # 无序接收结果，以提高内存效率
        for result in pool.imap_unordered(load_and_process, data_path_list):
            results.append(result)

    return results

# 测试
dataset = np.random.randn(10000, 128)
processed = preprocess_dataset_parallel(dataset, n_workers=4)
print(f"处理后的数据集 shape: {processed.shape}")

from concurrent.futures import ProcessPoolExecutor, as_completed
import numpy as np

def train_fold(fold_data):
    """交叉验证折的训练"""
    fold_idx, X_train, y_train, X_val, y_val = fold_data

    # 简单的逻辑回归(示例)
    from sklearn.linear_model import LogisticRegression
    from sklearn.metrics import accuracy_score

    model = LogisticRegression(max_iter=1000)
    model.fit(X_train, y_train)

    val_pred = model.predict(X_val)
    accuracy = accuracy_score(y_val, val_pred)

    return fold_idx, accuracy, model

def parallel_cross_validation(X, y, n_folds=5):
    """用 ProcessPoolExecutor 做并行交叉验证"""
    from sklearn.model_selection import KFold

    kf = KFold(n_splits=n_folds, shuffle=True, random_state=42)
    fold_data_list = []

    for fold_idx, (train_idx, val_idx) in enumerate(kf.split(X)):
        fold_data_list.append((
            fold_idx,
            X[train_idx], y[train_idx],
            X[val_idx], y[val_idx]
        ))

    results = {}

    with ProcessPoolExecutor(max_workers=n_folds) as executor:
        # 所有折同时执行
        future_to_fold = {
            executor.submit(train_fold, fd): fd[0]
            for fd in fold_data_list
        }

        for future in as_completed(future_to_fold):
            fold_idx, accuracy, model = future.result()
            results[fold_idx] = {
                'accuracy': accuracy,
                'model': model
            }
            print(f"折 {fold_idx}: 准确率 = {accuracy:.4f}")

    avg_accuracy = np.mean([r['accuracy'] for r in results.values()])
    print(f"\n平均准确率: {avg_accuracy:.4f}")
    return results

# 生成测试数据
from sklearn.datasets import make_classification
X, y = make_classification(n_samples=5000, n_features=20, random_state=42)
results = parallel_cross_validation(X, y, n_folds=5)

from multiprocessing import shared_memory
import numpy as np
import multiprocessing

def worker_shared_memory(shm_name, shape, dtype, start_idx, end_idx):
    """在共享内存中处理数据"""
    # 连接到已存在的共享内存
    existing_shm = shared_memory.SharedMemory(name=shm_name)
    arr = np.ndarray(shape, dtype=dtype, buffer=existing_shm.buf)

    # 切片处理
    arr[start_idx:end_idx] = arr[start_idx:end_idx] ** 2

    existing_shm.close()

def process_with_shared_memory(data):
    """使用共享内存的并行处理"""
    # 创建共享内存
    shm = shared_memory.SharedMemory(create=True, size=data.nbytes)
    shared_array = np.ndarray(data.shape, dtype=data.dtype, buffer=shm.buf)
    shared_array[:] = data[:]  # 复制数据

    n_workers = multiprocessing.cpu_count()
    chunk_size = len(data) // n_workers

    processes = []
    for i in range(n_workers):
        start = i * chunk_size
        end = start + chunk_size if i < n_workers - 1 else len(data)

        p = multiprocessing.Process(
            target=worker_shared_memory,
            args=(shm.name, data.shape, data.dtype, start, end)
        )
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    result = shared_array.copy()
    shm.close()
    shm.unlink()  # 释放共享内存

    return result

# 测试
data = np.random.randn(1_000_000).astype(np.float64)
result = process_with_shared_memory(data)
print(f"共享内存处理完成: {result[:5]}")

import asyncio
import time

# 基本的异步函数
async def fetch_embeddings(text: str, model: str = "text-embedding-3-small") -> list:
    """模拟异步嵌入请求"""
    await asyncio.sleep(0.1)  # 模拟 API 调用
    return [0.1, 0.2, 0.3]  # 占位嵌入

# 顺序处理(用于对比)
async def sequential_processing(texts: list) -> list:
    results = []
    for text in texts:
        embedding = await fetch_embeddings(text)
        results.append(embedding)
    return results

# 并行处理
async def parallel_processing(texts: list) -> list:
    tasks = [fetch_embeddings(text) for text in texts]
    results = await asyncio.gather(*tasks)
    return list(results)

# 性能对比
async def compare_approaches():
    texts = [f"文本样本 {i}" for i in range(20)]

    start = time.perf_counter()
    sequential_results = await sequential_processing(texts)
    t_seq = time.perf_counter() - start

    start = time.perf_counter()
    parallel_results = await parallel_processing(texts)
    t_par = time.perf_counter() - start

    print(f"顺序处理: {t_seq:.2f}s ({len(sequential_results)} 个结果)")
    print(f"并行处理: {t_par:.2f}s ({len(parallel_results)} 个结果)")
    print(f"速度提升: {t_seq/t_par:.1f}x")

asyncio.run(compare_approaches())

import asyncio
import aiohttp
import time
from typing import Optional

class AsyncAIClient:
    """异步 AI API 客户端"""

    def __init__(self, api_key: str, base_url: str, max_concurrent: int = 10):
        self.api_key = api_key
        self.base_url = base_url
        self.semaphore = asyncio.Semaphore(max_concurrent)  # 限制并发请求数
        self.session: Optional[aiohttp.ClientSession] = None

    async def __aenter__(self):
        connector = aiohttp.TCPConnector(limit=100)
        self.session = aiohttp.ClientSession(
            connector=connector,
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        return self

    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()

    async def get_embedding(self, text: str, retry: int = 3) -> list:
        """嵌入请求(含重试)"""
        async with self.semaphore:
            for attempt in range(retry):
                try:
                    async with self.session.post(
                        f"{self.base_url}/embeddings",
                        json={"input": text, "model": "text-embedding-3-small"}
                    ) as response:
                        if response.status == 200:
                            data = await response.json()
                            return data["data"][0]["embedding"]
                        elif response.status == 429:
                            # 限流：指数退避
                            await asyncio.sleep(2 ** attempt)
                        else:
                            response.raise_for_status()
                except aiohttp.ClientError as e:
                    if attempt == retry - 1:
                        raise
                    await asyncio.sleep(1)
        return []

    async def batch_embeddings(self, texts: list, batch_size: int = 100) -> list:
        """批量嵌入处理"""
        all_results = []

        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            tasks = [self.get_embedding(text) for text in batch]
            batch_results = await asyncio.gather(*tasks, return_exceptions=True)

            for j, result in enumerate(batch_results):
                if isinstance(result, Exception):
                    print(f"文本 {i+j} 出错: {result}")
                    all_results.append(None)
                else:
                    all_results.append(result)

        return all_results

# 使用示例
async def main():
    texts = [f"AI/ML 文本样本 {i}" for i in range(50)]

    async with AsyncAIClient(
        api_key="your-api-key",
        base_url="https://api.openai.com/v1",
        max_concurrent=20
    ) as client:
        # 实际会调用 API，这里只做模拟
        print("开始批量嵌入处理...")
        # embeddings = await client.batch_embeddings(texts)
        # print(f"嵌入数量: {len(embeddings)}")

# asyncio.run(main())

import asyncio
from dataclasses import dataclass, field
from typing import Any
import time

@dataclass
class BatchProcessor:
    """动态批处理器"""
    max_batch_size: int = 20
    max_wait_time: float = 0.05  # 50ms
    queue: asyncio.Queue = field(default_factory=asyncio.Queue)
    results: dict = field(default_factory=dict)

    async def process_request(self, request_id: str, data: Any) -> Any:
        """把请求加入队列并等待结果"""
        future = asyncio.get_event_loop().create_future()
        await self.queue.put((request_id, data, future))
        return await future

    async def batch_worker(self):
        """批处理工作协程"""
        while True:
            batch = []
            deadline = time.perf_counter() + self.max_wait_time

            # 收集批次
            while len(batch) < self.max_batch_size:
                timeout = deadline - time.perf_counter()
                if timeout <= 0:
                    break

                try:
                    item = await asyncio.wait_for(
                        self.queue.get(),
                        timeout=timeout
                    )
                    batch.append(item)
                except asyncio.TimeoutError:
                    break

            if not batch:
                await asyncio.sleep(0.001)
                continue

            # 批处理(模拟 API 调用)
            request_ids, data_list, futures = zip(*batch)

            try:
                # 这里是实际的批量 API 调用
                results = await self._call_api_batch(list(data_list))

                for future, result in zip(futures, results):
                    future.set_result(result)
            except Exception as e:
                for future in futures:
                    future.set_exception(e)

    async def _call_api_batch(self, data_list: list) -> list:
        """模拟批量 API 调用"""
        await asyncio.sleep(0.1)  # 模拟 API 延迟
        return [f"结果_{d}" for d in data_list]

# 测试
async def test_batch_processor():
    processor = BatchProcessor(max_batch_size=10, max_wait_time=0.05)

    # 启动后台工作协程
    worker_task = asyncio.create_task(processor.batch_worker())

    # 模拟并发请求
    async def make_request(i):
        result = await processor.process_request(f"req_{i}", f"data_{i}")
        return result

    start = time.perf_counter()
    tasks = [make_request(i) for i in range(50)]
    results = await asyncio.gather(*tasks)
    elapsed = time.perf_counter() - start

    print(f"处理 50 个请求: {elapsed:.3f}s")
    print(f"前 5 个结果: {results[:5]}")

    worker_task.cancel()

asyncio.run(test_batch_processor())

import sys
import gc

# 确认对象大小
data_types = {
    'int': 42,
    'float': 3.14,
    'str (small)': "hello",
    'str (large)': "hello" * 1000,
    'list (empty)': [],
    'dict (empty)': {},
    'tuple (empty)': (),
}

print("对象大小:")
for name, obj in data_types.items():
    print(f"  {name}: {sys.getsizeof(obj)} bytes")

# 嵌套结构的实际内存使用量
def deep_size(obj, seen=None):
    """计算对象及其全部引用对象的总内存"""
    if seen is None:
        seen = set()
    obj_id = id(obj)
    if obj_id in seen:
        return 0
    seen.add(obj_id)
    size = sys.getsizeof(obj)
    if isinstance(obj, dict):
        size += sum(deep_size(k, seen) + deep_size(v, seen)
                   for k, v in obj.items())
    elif hasattr(obj, '__iter__') and not isinstance(obj, (str, bytes)):
        size += sum(deep_size(item, seen) for item in obj)
    return size

nested = {'data': [i for i in range(1000)], 'meta': {'name': 'test'}}
print(f"\n嵌套字典大小: {deep_size(nested):,} bytes")

import numpy as np
from typing import Generator, Iterator

# 大规模数据集生成器
def dataset_generator(
    file_paths: list,
    batch_size: int = 32
) -> Generator:
    """内存高效的数据加载器"""
    for path in file_paths:
        # 实际中应从文件加载
        data = np.random.randn(1000, 128)
        labels = np.random.randint(0, 10, 1000)

        for start in range(0, len(data), batch_size):
            end = min(start + batch_size, len(data))
            yield data[start:end], labels[start:end]

# 内存效率对比
def load_all_to_memory(n_samples: int = 100_000) -> list:
    """把全部数据加载到内存"""
    return [np.random.randn(128) for _ in range(n_samples)]

def streaming_process(n_samples: int = 100_000) -> float:
    """流式处理"""
    total = 0.0
    for i in range(n_samples):
        sample = np.random.randn(128)
        total += sample.sum()
    return total

# 生成器流水线
def augment_generator(gen: Iterator, noise_std: float = 0.01):
    """生成器链式组合"""
    for X, y in gen:
        X_aug = X + np.random.randn(*X.shape) * noise_std
        yield X_aug, y

def normalize_generator(gen: Iterator):
    """归一化生成器"""
    for X, y in gen:
        X_norm = (X - X.mean(axis=1, keepdims=True)) / (X.std(axis=1, keepdims=True) + 1e-8)
        yield X_norm, y

# 构建流水线
fake_paths = [f"data_{i}.npy" for i in range(5)]
base_gen = dataset_generator(fake_paths, batch_size=32)
aug_gen = augment_generator(base_gen)
norm_gen = normalize_generator(aug_gen)

# 每次只处理一个批次
for i, (X_batch, y_batch) in enumerate(norm_gen):
    if i >= 3:
        break
    print(f"批次 {i}: shape={X_batch.shape}, 均值={X_batch.mean():.4f}")

import sys

# 普通类
class RegularEmbedding:
    def __init__(self, vector, label, metadata):
        self.vector = vector
        self.label = label
        self.metadata = metadata

# __slots__ 类
class SlottedEmbedding:
    __slots__ = ['vector', 'label', 'metadata']

    def __init__(self, vector, label, metadata):
        self.vector = vector
        self.label = label
        self.metadata = metadata

import numpy as np

# 内存对比
n = 100_000
vector_size = 128

regular_objects = [
    RegularEmbedding(
        np.random.randn(vector_size),
        i % 10,
        {'source': 'test'}
    ) for i in range(n)
]

slotted_objects = [
    SlottedEmbedding(
        np.random.randn(vector_size),
        i % 10,
        {'source': 'test'}
    ) for i in range(n)
]

regular_size = sys.getsizeof(regular_objects[0]) + sys.getsizeof(regular_objects[0].__dict__)
slotted_size = sys.getsizeof(slotted_objects[0])

print(f"普通类实例: {regular_size} bytes")
print(f"__slots__ 类实例: {slotted_size} bytes")
print(f"节省的内存: {(regular_size - slotted_size) / regular_size * 100:.1f}%")

import weakref
import gc

class ModelCache:
    """使用弱引用的模型缓存"""

    def __init__(self):
        self._cache = weakref.WeakValueDictionary()

    def get_or_load(self, model_name: str):
        model = self._cache.get(model_name)
        if model is None:
            print(f"正在加载模型: {model_name}")
            model = self._load_model(model_name)
            self._cache[model_name] = model
        else:
            print(f"从缓存返回: {model_name}")
        return model

    def _load_model(self, model_name: str):
        # 实际中应加载模型文件
        import numpy as np
        return {'name': model_name, 'weights': np.random.randn(100, 100)}

cache = ModelCache()

# 首次加载
model = cache.get_or_load('resnet50')
print(f"缓存大小: {len(cache._cache)}")

# 再次请求 —— 从缓存返回
model_again = cache.get_or_load('resnet50')

# 移除强引用后做 GC
del model
del model_again
gc.collect()

# 此时缓存已自动清除
print(f"GC 后的缓存大小: {len(cache._cache)}")

# 重新加载
new_model = cache.get_or_load('resnet50')

pip install mypy
mypy your_script.py --strict

from typing import Optional, Union, List, Dict, Tuple, Callable
import numpy as np
from numpy.typing import NDArray

# 基本类型提示
def compute_loss(
    predictions: NDArray[np.float32],
    targets: NDArray[np.float32],
    reduction: str = 'mean'
) -> float:
    diff = predictions - targets
    loss = np.sum(diff ** 2)
    if reduction == 'mean':
        return float(loss / len(predictions))
    return float(loss)

# Optional 与 Union
def load_checkpoint(
    path: str,
    device: Optional[str] = None,
    strict: bool = True
) -> Optional[Dict[str, NDArray]]:
    """加载检查点(失败时返回 None)"""
    try:
        # 实际中应加载文件
        return {'weights': np.random.randn(100)}
    except FileNotFoundError:
        return None

# 复杂的返回类型
def split_dataset(
    X: NDArray,
    y: NDArray,
    test_size: float = 0.2,
    random_state: Optional[int] = None
) -> Tuple[NDArray, NDArray, NDArray, NDArray]:
    if random_state is not None:
        np.random.seed(random_state)

    n = len(X)
    n_test = int(n * test_size)
    indices = np.random.permutation(n)

    test_idx = indices[:n_test]
    train_idx = indices[n_test:]

    return X[train_idx], X[test_idx], y[train_idx], y[test_idx]

from typing import TypeVar, Generic, Protocol, runtime_checkable
import numpy as np

T = TypeVar('T')
T_co = TypeVar('T_co', covariant=True)

# 泛型类
class DataLoader(Generic[T]):
    def __init__(self, dataset: List[T], batch_size: int = 32):
        self.dataset = dataset
        self.batch_size = batch_size
        self._idx = 0

    def __iter__(self):
        self._idx = 0
        return self

    def __next__(self) -> List[T]:
        if self._idx >= len(self.dataset):
            raise StopIteration

        batch = self.dataset[self._idx:self._idx + self.batch_size]
        self._idx += self.batch_size
        return batch

# Protocol(结构化子类型)
@runtime_checkable
class Optimizer(Protocol):
    """优化器 Protocol"""
    def step(self) -> None: ...
    def zero_grad(self) -> None: ...
    def state_dict(self) -> Dict: ...

@runtime_checkable
class Model(Protocol):
    def forward(self, x: NDArray) -> NDArray: ...
    def parameters(self) -> List[NDArray]: ...
    def train(self) -> None: ...
    def eval(self) -> None: ...

# 使用 Protocol 的函数
def train_epoch(
    model: Model,
    optimizer: Optimizer,
    dataloader: DataLoader,
    loss_fn: Callable[[NDArray, NDArray], float]
) -> float:
    model.train()
    total_loss = 0.0
    n_batches = 0

    for batch in dataloader:
        optimizer.zero_grad()
        # 实际的训练逻辑
        n_batches += 1

    return total_loss / max(n_batches, 1)

from dataclasses import dataclass, field
from typing import TypedDict, Optional
import numpy as np

# TypedDict: 字典类型提示
class ModelConfig(TypedDict):
    hidden_size: int
    num_layers: int
    dropout: float
    activation: str

class TrainingConfig(TypedDict, total=False):
    learning_rate: float
    batch_size: int
    max_epochs: int
    early_stopping: bool

# dataclass
@dataclass
class ExperimentConfig:
    # 必填字段
    experiment_name: str
    model_type: str

    # 带默认值的字段
    learning_rate: float = 1e-3
    batch_size: int = 32
    max_epochs: int = 100
    seed: int = 42

    # 复杂的默认值(使用 field)
    tags: List[str] = field(default_factory=list)
    hyperparams: Dict[str, float] = field(default_factory=dict)

    # 后处理方法
    def __post_init__(self):
        if self.learning_rate <= 0:
            raise ValueError(f"学习率必须为正数: {self.learning_rate}")
        if not self.experiment_name:
            raise ValueError("需要提供实验名称")

    def to_dict(self) -> Dict:
        from dataclasses import asdict
        return asdict(self)

# 使用示例
config = ExperimentConfig(
    experiment_name="bert-finetuning-v1",
    model_type="bert-base",
    learning_rate=2e-5,
    batch_size=16,
    tags=["nlp", "classification"]
)

print(f"实验: {config.experiment_name}")
print(f"配置: {config.to_dict()}")

from pydantic import BaseModel, Field, validator, root_validator
from typing import Optional, List
import numpy as np

class DataConfig(BaseModel):
    """用 Pydantic 校验数据配置"""
    data_path: str
    batch_size: int = Field(ge=1, le=1024, default=32)
    shuffle: bool = True
    num_workers: int = Field(ge=0, le=16, default=4)

    @validator('data_path')
    def path_must_exist(cls, v):
        import os
        if not os.path.exists(v) and v != "test_path":
            raise ValueError(f"数据路径不存在: {v}")
        return v

class ModelConfig(BaseModel):
    """模型配置"""
    architecture: str
    hidden_sizes: List[int] = Field(min_items=1)
    dropout_rate: float = Field(ge=0.0, le=0.9, default=0.1)
    activation: str = "relu"

    @validator('activation')
    def valid_activation(cls, v):
        valid = ['relu', 'gelu', 'silu', 'tanh', 'sigmoid']
        if v not in valid:
            raise ValueError(f"激活函数必须是 {valid} 之一")
        return v

    @validator('hidden_sizes')
    def sizes_must_be_positive(cls, v):
        if any(s <= 0 for s in v):
            raise ValueError("隐藏层大小必须为正数")
        return v

class TrainingConfig(BaseModel):
    """训练配置(嵌套模型)"""
    data: DataConfig
    model: ModelConfig
    learning_rate: float = Field(gt=0, le=1.0, default=1e-3)
    weight_decay: float = Field(ge=0, default=1e-4)
    max_epochs: int = Field(ge=1, le=10000, default=100)

    @root_validator
    def check_consistency(cls, values):
        model = values.get('model')
        if model and model.dropout_rate > 0.5:
            print("警告: 较高的 dropout 比例可能导致欠拟合")
        return values

    class Config:
        # 允许生成 JSON schema
        schema_extra = {
            "example": {
                "data": {
                    "data_path": "/data/train",
                    "batch_size": 32
                },
                "model": {
                    "architecture": "transformer",
                    "hidden_sizes": [512, 256]
                },
                "learning_rate": 1e-4
            }
        }

# 使用与校验
try:
    config = TrainingConfig(
        data=DataConfig(data_path="test_path", batch_size=64),
        model=ModelConfig(
            architecture="transformer",
            hidden_sizes=[512, 256, 128],
            dropout_rate=0.1
        ),
        learning_rate=2e-5
    )
    print(f"配置校验成功: {config.model.architecture}")
    print(f"JSON: {config.json(indent=2)}")
except Exception as e:
    print(f"校验错误: {e}")

# 安装 Poetry 并初始化项目
curl -sSL https://install.python-poetry.org | python3 -
poetry new my-ml-project
cd my-ml-project

# 添加依赖
poetry add numpy torch torchvision
poetry add --group dev pytest black ruff mypy

# 激活虚拟环境
poetry shell

# 安装依赖包
poetry install

# 导出 requirements.txt
poetry export -f requirements.txt --output requirements.txt

[tool.poetry]
name = "my-ml-project"
version = "0.1.0"
description = "AI/ML 项目"
authors = ["开发者姓名 <email@example.com>"]
license = "MIT"
readme = "README.md"
packages = [{include = "my_ml_project"}]

[tool.poetry.dependencies]
python = "^3.10"
numpy = "^1.24"
torch = "^2.1"
torchvision = "^0.16"
transformers = "^4.35"
pydantic = "^2.0"
aiohttp = "^3.9"

[tool.poetry.group.dev.dependencies]
pytest = "^7.4"
pytest-cov = "^4.1"
black = "^23.0"
ruff = "^0.1"
mypy = "^1.6"
pre-commit = "^3.5"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

[tool.black]
line-length = 88
target-version = ['py310', 'py311']

[tool.ruff]
select = ["E", "F", "I", "N", "W"]
ignore = ["E501"]
line-length = 88

[tool.mypy]
python_version = "3.10"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true

[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = "-v --cov=my_ml_project --cov-report=html"

# 安装 uv(基于 Rust，速度非常快)
curl -LsSf https://astral.sh/uv/install.sh | sh

# 创建项目
uv init my-project
cd my-project

# 添加依赖(比 pip 快 10-100 倍)
uv add numpy torch transformers
uv add --dev pytest ruff mypy

# 创建虚拟环境并安装
uv sync

# 管理 Python 版本
uv python install 3.11
uv python pin 3.11

# 快速运行脚本
uv run python train.py

from abc import ABC, abstractmethod
from typing import Dict, Type, Optional
import numpy as np

# Strategy 模式 —— 损失函数
class LossFunction(ABC):
    @abstractmethod
    def __call__(self, predictions: np.ndarray, targets: np.ndarray) -> float:
        pass

    @abstractmethod
    def gradient(self, predictions: np.ndarray, targets: np.ndarray) -> np.ndarray:
        pass

class MSELoss(LossFunction):
    def __call__(self, predictions, targets):
        return float(np.mean((predictions - targets) ** 2))

    def gradient(self, predictions, targets):
        return 2 * (predictions - targets) / len(predictions)

class CrossEntropyLoss(LossFunction):
    def __call__(self, predictions, targets):
        eps = 1e-8
        return float(-np.mean(targets * np.log(predictions + eps)))

    def gradient(self, predictions, targets):
        return -targets / (predictions + 1e-8)

class HuberLoss(LossFunction):
    def __init__(self, delta: float = 1.0):
        self.delta = delta

    def __call__(self, predictions, targets):
        diff = predictions - targets
        is_small = np.abs(diff) <= self.delta
        loss = np.where(is_small, 0.5 * diff**2,
                       self.delta * np.abs(diff) - 0.5 * self.delta**2)
        return float(np.mean(loss))

    def gradient(self, predictions, targets):
        diff = predictions - targets
        return np.where(np.abs(diff) <= self.delta, diff,
                       self.delta * np.sign(diff))

# Factory 模式 —— 模型创建
class ModelFactory:
    _registry: Dict[str, Type] = {}

    @classmethod
    def register(cls, name: str):
        def decorator(model_class):
            cls._registry[name] = model_class
            return model_class
        return decorator

    @classmethod
    def create(cls, name: str, **kwargs):
        if name not in cls._registry:
            available = list(cls._registry.keys())
            raise ValueError(f"未知的模型: {name}. 可用: {available}")
        return cls._registry[name](**kwargs)

@ModelFactory.register("linear")
class LinearModel:
    def __init__(self, input_size: int, output_size: int):
        self.W = np.random.randn(input_size, output_size) * 0.01
        self.b = np.zeros(output_size)

    def forward(self, X: np.ndarray) -> np.ndarray:
        return X @ self.W + self.b

@ModelFactory.register("mlp")
class MLPModel:
    def __init__(self, layer_sizes: list):
        self.layers = []
        for i in range(len(layer_sizes) - 1):
            W = np.random.randn(layer_sizes[i], layer_sizes[i+1]) * 0.01
            b = np.zeros(layer_sizes[i+1])
            self.layers.append((W, b))

    def forward(self, X: np.ndarray) -> np.ndarray:
        out = X
        for i, (W, b) in enumerate(self.layers):
            out = out @ W + b
            if i < len(self.layers) - 1:
                out = np.maximum(0, out)  # ReLU
        return out

# 使用示例
model = ModelFactory.create("mlp", layer_sizes=[128, 256, 10])
loss_fn = CrossEntropyLoss()

X = np.random.randn(32, 128)
y = np.zeros((32, 10))
for i in range(32):
    y[i, np.random.randint(10)] = 1

predictions = model.forward(X)
loss = loss_fn(predictions, y)
print(f"损失: {loss:.4f}")

pip install hydra-core omegaconf

# config/train.yaml 的内容(以注释形式展示)
# model:
#   type: transformer
#   hidden_size: 512
#   num_layers: 6
# training:
#   lr: 1e-4
#   batch_size: 32
#   max_epochs: 100

import hydra
from omegaconf import DictConfig, OmegaConf

@hydra.main(version_base=None, config_path="config", config_name="train")
def train(cfg: DictConfig):
    print(OmegaConf.to_yaml(cfg))

    # 访问配置
    model_type = cfg.model.type
    lr = cfg.training.lr

    print(f"模型: {model_type}, 学习率: {lr}")

# 命令行覆盖
# python train.py training.lr=1e-3 model.hidden_size=256