Claude Code Python开发子代理深度实战：构建你的Python全栈专家

# 场景对比：读取并处理一个大型CSV文件

# ❌ 通用Claude可能给你的代码
import csv
data = []
with open('large_file.csv', 'r') as f:
    reader = csv.reader(f)
    for row in reader:
        data.append(row)  # 内存爆炸！
# 问题：没考虑内存、没用pandas、没有错误处理、效率低

# ✅ Python子代理会给你的专业方案
import pandas as pd
from pathlib import Path
import logging
from typing import Generator, Dict, Any
from memory_profiler import profile

logger = logging.getLogger(__name__)

@profile
def process_large_csv(
    file_path: Path, 
    chunk_size: int = 10000
) -> Generator[pd.DataFrame, None, None]:
    """
    高效处理大型CSV文件，使用分块读取避免内存溢出
    
    Args:
        file_path: CSV文件路径
        chunk_size: 每次读取的行数
        
    Yields:
        pd.DataFrame: 处理后的数据块
    """
    try:
        # 使用iterator分块读取
        with pd.read_csv(
            file_path,
            chunksize=chunk_size,
            dtype_backend='pyarrow',  # 使用Arrow后端提升性能
            on_bad_lines='skip',
            low_memory=False
        ) as reader:
            for chunk in reader:
                # 数据清洗和转换
                chunk = chunk.dropna(subset=['critical_column'])
                chunk['processed_date'] = pd.to_datetime(
                    chunk['date'], 
                    errors='coerce'
                )
                
                yield chunk
                
    except FileNotFoundError:
        logger.error(f"文件不存在: {file_path}")
        raise
    except pd.errors.EmptyDataError:
        logger.warning(f"文件为空: {file_path}")
        return
    except Exception as e:
        logger.error(f"处理CSV时发生错误: {e}")
        raise

# 使用示例
if __name__ == "__main__":
    file_path = Path("large_data.csv")
    
    for chunk_df in process_large_csv(file_path):
        # 逐块处理，避免内存溢出
        result = chunk_df.groupby('category').agg({
            'amount': 'sum',
            'count': 'size'
        })
        # 保存或进一步处理结果
        result.to_parquet(f"output_{chunk_df.index[0]}.parquet")

---
name: python-developer
description: Write clean, efficient Python code following PEP standards. Specializes in Django/FastAPI web development, data processing, and automation. Use PROACTIVELY for Python-specific projects and performance optimization.
model: sonnet
---
You are a Python development expert focused on writing Pythonic, efficient, and maintainable code following community best practices.

## Python Mastery
- Modern Python 3.12+ features (pattern matching, type hints, async/await)
- Web frameworks (Django, FastAPI, Flask) with proper architecture
- Data processing libraries (pandas, NumPy, polars) for performance
- Async programming with asyncio and concurrent.futures
- Testing frameworks (pytest, unittest, hypothesis) with high coverage
- Package management (Poetry, pip-tools) and virtual environments
- Code quality tools (black, ruff, mypy, pre-commit hooks)
- Performance profiling and optimization techniques

## Development Standards
1. PEP 8 compliance with automated formatting
2. Comprehensive type annotations for better IDE support
3. Proper exception handling with custom exception classes
4. Context managers for resource management
5. Generator expressions for memory efficiency
6. Dataclasses and Pydantic models for data validation
7. Proper logging configuration with structured output
8. Virtual environment isolation and dependency pinning

## Code Quality Focus
- Clean, readable code following SOLID principles
- Comprehensive docstrings following Google/NumPy style
- Unit tests with >90% coverage using pytest
- Performance benchmarks and memory profiling
- Security scanning with bandit and safety
- Automated code formatting with black and isort
- Linting with ruff and type checking with mypy
- CI/CD integration with GitHub Actions or similar
- Package distribution following Python packaging standards

Write Python code that is not just functional but exemplary. Focus on readability, performance, and maintainability while leveraging Python's unique strengths and idioms.

---
name: python-developer
description: 编写符合PEP标准的简洁高效Python代码。专精Django/FastAPI Web开发、数据处理和自动化。在Python项目和性能优化时主动使用。
model: sonnet
---
你是一位Python开发专家，专注于编写Pythonic、高效、可维护的代码，遵循社区最佳实践。

## Python精通技能 / Python Mastery
- 现代Python 3.12+特性（模式匹配、类型提示、async/await）
- Web框架（Django、FastAPI、Flask）与合理架构
- 数据处理库（pandas、NumPy、polars）性能优化
- asyncio和concurrent.futures异步编程
- 测试框架（pytest、unittest、hypothesis）高覆盖率
- 包管理（Poetry、pip-tools）和虚拟环境
- 代码质量工具（black、ruff、mypy、pre-commit钩子）
- 性能分析和优化技术

## 开发标准 / Development Standards
1. 遵循PEP 8规范，自动化格式化
2. 完整的类型注解，提升IDE支持
3. 合理的异常处理与自定义异常类
4. 使用上下文管理器管理资源
5. 生成器表达式提升内存效率
6. 使用dataclasses和Pydantic进行数据验证
7. 配置结构化日志输出
8. 虚拟环境隔离和依赖版本锁定

## 代码质量关注点 / Code Quality Focus
- 遵循SOLID原则的清晰可读代码
- 遵循Google/NumPy风格的完整文档字符串
- 使用pytest实现>90%的测试覆盖率
- 性能基准测试和内存分析
- 使用bandit和safety进行安全扫描
- 使用black和isort自动格式化代码
- 使用ruff进行代码检查，mypy进行类型检查
- 集成GitHub Actions等CI/CD
- 遵循Python打包标准进行包分发

编写的Python代码不仅要能工作，更要成为典范。
专注于可读性、性能和可维护性，同时充分利用Python的独特优势和惯用法。

# 1. PEP 8规范是什么？
PEP 8 = Python Enhancement Proposal 8
就是Python的"交通规则"：
- 缩进用4个空格
- 函数名用小写+下划线
- 类名用大驼峰
- 一行不超过79字符

# 2. Type Hints类型提示
def add(a: int, b: int) -> int:  # 明确告诉别人输入输出类型
    return a + b

# 3. 上下文管理器
with open('file.txt') as f:  # 自动处理文件关闭
    content = f.read()

# 4. 生成器表达式
squares = (x**2for x in range(1000000))  # 不会占用大量内存

# 在Claude Code中输入
/agents

创建一个用户管理的FastAPI服务，包含CRUD操作和认证

# app/main.py - FastAPI主应用
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from typing import List, Optional, Annotated
from datetime import datetime, timedelta
import uvicorn

from . import models, schemas, crud, auth
from .database import engine, get_db
from .config import settings

# 创建数据库表
models.Base.metadata.create_all(bind=engine)

# 初始化FastAPI应用
app = FastAPI(
    title="用户管理API",
    description="基于FastAPI的用户管理微服务",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

# 配置CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# OAuth2配置
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 依赖注入：获取当前用户
asyncdef get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Session = Depends(get_db)
) -> models.User:
    """
    验证JWT令牌并返回当前用户
    
    Args:
        token: JWT访问令牌
        db: 数据库会话
        
    Returns:
        当前认证的用户对象
        
    Raises:
        HTTPException: 令牌无效或过期
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    payload = auth.verify_token(token)
    if payload isNone:
        raise credentials_exception
        
    user = crud.get_user_by_email(db, email=payload.get("sub"))
    if user isNone:
        raise credentials_exception
        
    return user

# API路由
@app.post("/register", response_model=schemas.UserResponse, status_code=201)
asyncdef register(
    user: schemas.UserCreate,
    db: Session = Depends(get_db)
):
    """用户注册接口"""
    # 检查邮箱是否已存在
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(
            status_code=400,
            detail="Email already registered"
        )
    
    # 创建新用户
    return crud.create_user(db=db, user=user)

@app.post("/token", response_model=schemas.Token)
asyncdef login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: Session = Depends(get_db)
):
    """用户登录获取JWT令牌"""
    user = auth.authenticate_user(
        db, 
        form_data.username, 
        form_data.password
    )
    
    ifnot user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    # 生成访问令牌
    access_token = auth.create_access_token(
        data={"sub": user.email}
    )
    
    return {
        "access_token": access_token,
        "token_type": "bearer"
    }

@app.get("/users/me", response_model=schemas.UserResponse)
asyncdef read_users_me(
    current_user: Annotated[models.User, Depends(get_current_user)]
):
    """获取当前用户信息"""
    return current_user

@app.get("/users", response_model=List[schemas.UserResponse])
asyncdef read_users(
    skip: int = 0,
    limit: int = 100,
    current_user: Annotated[models.User, Depends(get_current_user)],
    db: Session = Depends(get_db)
):
    """获取用户列表（需要认证）"""
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

@app.put("/users/{user_id}", response_model=schemas.UserResponse)
asyncdef update_user(
    user_id: int,
    user_update: schemas.UserUpdate,
    current_user: Annotated[models.User, Depends(get_current_user)],
    db: Session = Depends(get_db)
):
    """更新用户信息"""
    # 检查权限（只能更新自己的信息或管理员）
    if current_user.id != user_id andnot current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Not enough permissions"
        )
    
    user = crud.update_user(db, user_id, user_update)
    ifnot user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    
    return user

# 健康检查
@app.get("/health")
asyncdef health_check():
    """健康检查接口"""
    return {
        "status": "healthy",
        "timestamp": datetime.utcnow().isoformat()
    }

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

# app/models.py - 数据模型
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from .database import Base

class User(Base):
    """用户数据模型"""
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    username = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)
    is_admin = Column(Boolean, default=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
    
    def __repr__(self) -> str:
        returnf"<User(id={self.id}, email='{self.email}')>"

# tests/test_api.py - 单元测试
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.main import app, get_db
from app.database import Base

# 测试数据库配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base.metadata.create_all(bind=engine)

def override_get_db():
    """覆盖数据库依赖"""
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)

class TestUserAPI:
    """用户API测试类"""
    
    @pytest.fixture(autouse=True)
    def setup(self):
        """测试前置设置"""
        Base.metadata.create_all(bind=engine)
        yield
        Base.metadata.drop_all(bind=engine)
    
    def test_register_user(self):
        """测试用户注册"""
        response = client.post(
            "/register",
            json={
                "email": "test@example.com",
                "username": "testuser",
                "password": "testpass123"
            }
        )
        assert response.status_code == 201
        data = response.json()
        assert data["email"] == "test@example.com"
        assert"id"in data
    
    def test_login_user(self):
        """测试用户登录"""
        # 先注册
        client.post(
            "/register",
            json={
                "email": "test@example.com",
                "username": "testuser",
                "password": "testpass123"
            }
        )
        
        # 再登录
        response = client.post(
            "/token",
            data={
                "username": "test@example.com",
                "password": "testpass123"
            }
        )
        assert response.status_code == 200
        data = response.json()
        assert"access_token"in data
        assert data["token_type"] == "bearer"
    
    @pytest.mark.parametrize("email,username,password,expected_status", [
        ("", "user", "pass", 422),
        ("invalid", "user", "pass", 422),
        ("test@test.com", "", "pass", 422),
        ("test@test.com", "user", "", 422),
    ])
    def test_register_validation(self, email, username, password, expected_status):
        """测试注册参数验证"""
        response = client.post(
            "/register",
            json={
                "email": email,
                "username": username,
                "password": password
            }
        )
        assert response.status_code == expected_status

优化这个数据处理脚本的性能

# data_processor.py - 高性能数据处理
import pandas as pd
import numpy as np
from typing import Optional, List, Dict, Any
from pathlib import Path
import logging
from functools import lru_cache
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp
from memory_profiler import profile
import pyarrow.parquet as pq

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class DataProcessor:
    """高性能数据处理器"""
    
    def __init__(self, n_jobs: Optional[int] = None):
        """
        初始化数据处理器
        
        Args:
            n_jobs: 并行处理的进程数，None表示使用所有CPU核心
        """
        self.n_jobs = n_jobs or mp.cpu_count()
        logger.info(f"初始化数据处理器，使用{self.n_jobs}个进程")
    
    @lru_cache(maxsize=128)
    def _get_dtypes(self, file_path: str) -> Dict[str, Any]:
        """
        缓存文件的数据类型，避免重复推断
        
        Args:
            file_path: 文件路径
            
        Returns:
            列名到数据类型的映射
        """
        sample = pd.read_csv(file_path, nrows=1000)
        return sample.dtypes.to_dict()
    
    @profile
    def process_large_file(
        self, 
        file_path: Path,
        output_path: Path,
        chunk_size: int = 50000
    ) -> None:
        """
        处理大型文件，使用分块和并行处理
        
        Args:
            file_path: 输入文件路径
            output_path: 输出文件路径
            chunk_size: 分块大小
        """
        logger.info(f"开始处理文件: {file_path}")
        
        # 获取数据类型以优化内存使用
        dtypes = self._get_dtypes(str(file_path))
        
        # 使用并行处理
        with ProcessPoolExecutor(max_workers=self.n_jobs) as executor:
            futures = []
            
            # 分块读取
            for i, chunk in enumerate(pd.read_csv(
                file_path,
                chunksize=chunk_size,
                dtype=dtypes,
                low_memory=False
            )):
                # 提交处理任务
                future = executor.submit(self._process_chunk, chunk, i)
                futures.append(future)
            
            # 收集结果
            results = []
            for future in futures:
                result = future.result()
                if result isnotNone:
                    results.append(result)
        
        # 合并结果并保存
        if results:
            final_df = pd.concat(results, ignore_index=True)
            
            # 使用Parquet格式保存（更高效）
            final_df.to_parquet(
                output_path,
                engine='pyarrow',
                compression='snappy',
                index=False
            )
            logger.info(f"处理完成，结果保存到: {output_path}")
    
    @staticmethod
    def _process_chunk(chunk: pd.DataFrame, chunk_id: int) -> Optional[pd.DataFrame]:
        """
        处理单个数据块
        
        Args:
            chunk: 数据块
            chunk_id: 块ID
            
        Returns:
            处理后的数据块
        """
        try:
            # 数据清洗
            chunk = chunk.dropna(subset=['important_column'])
            
            # 向量化操作（比循环快100倍）
            chunk['new_column'] = np.where(
                chunk['value'] > 100,
                chunk['value'] * 1.1,
                chunk['value'] * 0.9
            )
            
            # 使用NumPy进行数值计算
            numeric_cols = chunk.select_dtypes(include=[np.number]).columns
            chunk[numeric_cols] = chunk[numeric_cols].apply(
                lambda x: np.log1p(x) if x.dtype != 'object'else x
            )
            
            return chunk
            
        except Exception as e:
            logger.error(f"处理块{chunk_id}时出错: {e}")
            returnNone
    
    def optimize_dataframe(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        优化DataFrame内存使用
        
        Args:
            df: 输入DataFrame
            
        Returns:
            优化后的DataFrame
        """
        initial_memory = df.memory_usage(deep=True).sum() / 1024**2
        logger.info(f"初始内存使用: {initial_memory:.2f} MB")
        
        # 优化数值类型
        for col in df.select_dtypes(include=['int']).columns:
            df[col] = pd.to_numeric(df[col], downcast='integer')
        
        for col in df.select_dtypes(include=['float']).columns:
            df[col] = pd.to_numeric(df[col], downcast='float')
        
        # 转换字符串为category类型（如果重复值多）
        for col in df.select_dtypes(include=['object']).columns:
            num_unique_values = len(df[col].unique())
            num_total_values = len(df[col])
            if num_unique_values / num_total_values < 0.5:
                df[col] = df[col].astype('category')
        
        final_memory = df.memory_usage(deep=True).sum() / 1024**2
        logger.info(f"优化后内存使用: {final_memory:.2f} MB")
        logger.info(f"内存减少: {(1 - final_memory/initial_memory)*100:.1f}%")
        
        return df

if __name__ == "__main__":
    # 使用示例
    processor = DataProcessor()
    
    # 处理大文件
    processor.process_large_file(
        file_path=Path("large_data.csv"),
        output_path=Path("processed_data.parquet")
    )

## Python Mastery
-科学计算：NumPy,SciPy,SymPy
-机器学习：scikit-learn,XGBoost,LightGBM
-深度学习：PyTorch,TensorFlow
-数据可视化：Matplotlib,Seaborn,Plotly
-特征工程：Feature-engine,CategoryEncoders

## Python Mastery
-Django5.0全栈开发
-FastAPI+SQLAlchemy+Alembic
-Celery异步任务队列
-Redis缓存策略
-GraphQLwithStrawberry

## Python Mastery
- Ansible自动化部署
- Fabric远程执行
- Paramiko SSH操作
- Schedule定时任务
- Click CLI工具开发

## Company Standards
- 代码风格：使用公司的.pylintrc配置
- 测试要求：最低95%覆盖率
- 文档规范：使用Sphinx生成文档
- Git提交：遵循conventional commits
- 部署流程：Docker + Kubernetes

# 格式化代码
black your_code.py
isort your_code.py

# 检查规范
ruff check your_code.py
mypy your_code.py

# Python 3.10+
match command:
    case "start":
        start_server()
    case "stop":
        stop_server()

# 兼容旧版本
if command == "start":
    start_server()
elif command == "stop":
    stop_server()