Python编程基础
课程介绍
本模块专注于AI开发中常用的Python高级特性和最佳实践。通过实际案例和hands-on练习，帮助你掌握AI开发必备的Python技能。
学习目标
完成本模块学习后，你将能够：
- 熟练运用Python高级特性进行AI开发
- 掌握异步编程在AI中的应用
- 理解并实践Python性能优化
- 搭建规范的AI开发环境
1. 开发环境配置
1.1 Python安装与版本选择
# 检查Python版本
python --version # 建议使用3.8+版本
# 安装Python（如果需要）
brew install python@3.8 # macOS1.2 虚拟环境管理
# 创建虚拟环境
python -m venv .venv
# 激活虚拟环境
source .venv/bin/activate # Unix/macOS
.venv\Scripts\activate # Windows
# 安装依赖
pip install -r requirements.txt2. Python高级特性
2.1 类型提示与静态检查
from typing import List, Dict, Optional
def process_data(data: List[float]) -> Dict[str, float]:
"""处理数值数据并返回统计结果"""
return {
'mean': sum(data) / len(data),
'max': max(data),
'min': min(data)
}
# 使用mypy进行静态类型检查
# mypy data_processor.py2.2 装饰器与元编程
import time
from functools import wraps
def timing_decorator(func):
"""测量函数执行时间的装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
print(f"{func.__name__} 执行时间: {end_time - start_time:.2f}秒")
return result
return wrapper
@timing_decorator
def train_model(epochs: int):
"""模型训练函数"""
time.sleep(1) # 模拟训练过程
print(f"完成{epochs}轮训练")3. 异步编程
3.1 协程与异步IO
import asyncio
import aiohttp
import aiofiles
async def download_data(url: str, filename: str):
"""异步下载数据"""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
data = await response.read()
async with aiofiles.open(filename, 'wb') as f:
await f.write(data)
async def main():
"""并发下载多个数据集"""
urls = [
'https://example.com/dataset1.csv',
'https://example.com/dataset2.csv'
]
tasks = [
download_data(url, f'dataset_{i}.csv')
for i, url in enumerate(urls)
]
await asyncio.gather(*tasks)
# 运行异步任务
asyncio.run(main())3.2 并发数据处理
import concurrent.futures
import numpy as np
def process_chunk(data: np.ndarray) -> np.ndarray:
"""处理数据块"""
return np.square(data) # 示例操作
def parallel_processing(data: np.ndarray, n_jobs: int = -1):
"""并行处理大型数据集"""
chunks = np.array_split(data, n_jobs)
with concurrent.futures.ProcessPoolExecutor() as executor:
results = list(executor.map(process_chunk, chunks))
return np.concatenate(results)4. 性能优化
4.1 代码优化技巧
# 使用列表推导式
squares = [x**2 for x in range(1000)] # 比for循环快
# 使用生成器节省内存
def data_generator(n: int):
for i in range(n):
yield i**2
# 使用NumPy向量化运算
import numpy as np
data = np.array([1, 2, 3, 4, 5])
result = np.square(data) # 比循环快得多4.2 内存管理
# 使用__slots__优化内存
class DataPoint:
__slots__ = ['x', 'y', 'label']
def __init__(self, x: float, y: float, label: str):
self.x = x
self.y = y
self.label = label
# 使用生成器处理大文件
def process_large_file(filename: str):
with open(filename) as f:
for line in f: # 逐行读取，避免一次性加载
yield process_line(line)5. Python高级特性
5.1 装饰器与元类
# 🎯 实战案例：创建模型性能监控装饰器
import time
from functools import wraps
def monitor_performance(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
print(f"函数 {func.__name__} 执行时间: {end_time - start_time:.2f}秒")
return result
return wrapper
@monitor_performance
def train_model(epochs):
# 模拟模型训练
time.sleep(2)
return "模型训练完成"5.2 生成器与迭代器
# 🔄 实战案例：批量数据生成器
def batch_generator(data, batch_size=32):
"""高效的批量数据生成器"""
for i in range(0, len(data), batch_size):
yield data[i:i + batch_size]
# 使用示例
training_data = list(range(1000))
for batch in batch_generator(training_data):
# 处理每个批次
pass6. 异步编程
6.1 协程基础
# ⚡️ 实战案例：异步数据加载
import asyncio
import aiohttp
async def fetch_data(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
async def main():
urls = [
"https://api.example.com/data1",
"https://api.example.com/data2"
]
tasks = [fetch_data(url) for url in urls]
results = await asyncio.gather(*tasks)
return results6.2 异步框架应用
# 🚀 实战案例：FastAPI服务端点
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class PredictionRequest(BaseModel):
text: str
@app.post("/predict")
async def predict(request: PredictionRequest):
# 异步预测处理
result = await process_prediction(request.text)
return {"prediction": result}7. 性能优化
7.1 代码优化技巧
# ⚡️ 实战案例：向量化运算
import numpy as np
# 优化前
def slow_distance(x1, y1, x2, y2):
result = []
for i in range(len(x1)):
result.append(((x1[i]-x2[i])**2 + (y1[i]-y2[i])**2)**0.5)
return result
# 优化后
def fast_distance(x1, y1, x2, y2):
return np.sqrt((x1-x2)**2 + (y1-y2)**2)7.2 并行计算
# 🔄 实战案例：多进程数据处理
from multiprocessing import Pool
def process_chunk(data_chunk):
# 数据处理逻辑
return processed_result
def parallel_processing(data, num_processes=4):
with Pool(num_processes) as pool:
results = pool.map(process_chunk, data)
return results8. 环境管理
8.1 虚拟环境
# 创建项目专用环境
python -m venv ai-env
# 激活环境
source ai-env/bin/activate # Unix
ai-env\Scripts\activate # Windows
# 安装依赖
pip install -r requirements.txt8.2 依赖管理
# 📦 requirements.txt 示例
numpy>=1.21.0
pandas>=1.3.0
torch>=1.9.0
transformers>=4.5.0实战项目：AI数据处理Pipeline
项目描述
构建一个高效的AI数据处理Pipeline，综合运用本模块所学的Python高级特性。
项目代码框架
class AIPipeline:
def __init__(self):
self.steps = []
@monitor_performance
def add_step(self, step_func):
self.steps.append(step_func)
async def process(self, data):
for step in self.steps:
data = await step(data)
return data
# 使用示例
pipeline = AIPipeline()
pipeline.add_step(preprocess_data)
pipeline.add_step(feature_extraction)
pipeline.add_step(model_prediction)练习与作业
- 实现一个异步数据加载器
- 优化现有代码性能
- 搭建完整的AI开发环境
扩展阅读
小测验
- 装饰器的主要用途是什么？
- 异步编程如何提高AI应用性能？
- 如何选择合适的并行处理方式？
下一步学习
- 数学基础
- 数据处理技术
- 机器学习算法
常见问题解答
Q: 为什么要使用异步编程？ A: 在AI开发中，异步编程可以显著提高数据加载和预处理的效率，特别是在处理大量I/O操作时。
Q: 如何选择合适的Python版本？ A: 建议使用Python 3.8+，这些版本提供了完整的类型提示支持和最新的异步特性。
Q: 为什么要使用类型提示？ A: 类型提示可以帮助我们在开发早期发现潜在问题，提高代码可维护性，并提供更好的IDE支持。