# 場(chǎng)景對(duì)比：讀取并處理一個(gè)大型CSV文件

# ? 通用Claude可能給你的代碼
import csv
data = []
with open('large_file.csv', 'r') as f:
    reader = csv.reader(f)
    for row in reader:
        data.append(row)  # 內(nèi)存爆炸！
# 問(wèn)題：沒(méi)考慮內(nèi)存、沒(méi)用pandas、沒(méi)有錯(cuò)誤處理、效率低

# ? Python子代理會(huì)給你的專業(yè)方案
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文件，使用分塊讀取避免內(nèi)存溢出
    
    Args:
        file_path: CSV文件路徑
        chunk_size: 每次讀取的行數(shù)
        
    Yields:
        pd.DataFrame: 處理后的數(shù)據(jù)塊
    """
    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:
                # 數(shù)據(jù)清洗和轉(zhuǎn)換
                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時(shí)發(fā)生錯(cuò)誤: {e}")
        raise

# 使用示例
if __name__ == "__main__":
    file_path = Path("large_data.csv")
    
    for chunk_df in process_large_csv(file_path):
        # 逐塊處理，避免內(nèi)存溢出
        result = chunk_df.groupby('category').agg({
            'amount': 'sum',
            'count': 'size'
        })
        # 保存或進(jìn)一步處理結(jié)果
        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: 編寫(xiě)符合PEP標(biāo)準(zhǔn)的簡(jiǎn)潔高效Python代碼。專精Django/FastAPI Web開(kāi)發(fā)、數(shù)據(jù)處理和自動(dòng)化。在Python項(xiàng)目和性能優(yōu)化時(shí)主動(dòng)使用。
model: sonnet
---
你是一位Python開(kāi)發(fā)專家，專注于編寫(xiě)Pythonic、高效、可維護(hù)的代碼，遵循社區(qū)最佳實(shí)踐。

## Python精通技能 / Python Mastery
- 現(xiàn)代Python 3.12+特性（模式匹配、類(lèi)型提示、async/await）
- Web框架（Django、FastAPI、Flask）與合理架構(gòu)
- 數(shù)據(jù)處理庫(kù)（pandas、NumPy、polars）性能優(yōu)化
- asyncio和concurrent.futures異步編程
- 測(cè)試框架（pytest、unittest、hypothesis）高覆蓋率
- 包管理（Poetry、pip-tools）和虛擬環(huán)境
- 代碼質(zhì)量工具（black、ruff、mypy、pre-commit鉤子）
- 性能分析和優(yōu)化技術(shù)

## 開(kāi)發(fā)標(biāo)準(zhǔn) / Development Standards
1. 遵循PEP 8規(guī)范，自動(dòng)化格式化
2. 完整的類(lèi)型注解，提升IDE支持
3. 合理的異常處理與自定義異常類(lèi)
4. 使用上下文管理器管理資源
5. 生成器表達(dá)式提升內(nèi)存效率
6. 使用dataclasses和Pydantic進(jìn)行數(shù)據(jù)驗(yàn)證
7. 配置結(jié)構(gòu)化日志輸出
8. 虛擬環(huán)境隔離和依賴版本鎖定

## 代碼質(zhì)量關(guān)注點(diǎn) / Code Quality Focus
- 遵循SOLID原則的清晰可讀代碼
- 遵循Google/NumPy風(fēng)格的完整文檔字符串
- 使用pytest實(shí)現(xiàn)>90%的測(cè)試覆蓋率
- 性能基準(zhǔn)測(cè)試和內(nèi)存分析
- 使用bandit和safety進(jìn)行安全掃描
- 使用black和isort自動(dòng)格式化代碼
- 使用ruff進(jìn)行代碼檢查，mypy進(jìn)行類(lèi)型檢查
- 集成GitHub Actions等CI/CD
- 遵循Python打包標(biāo)準(zhǔn)進(jìn)行包分發(fā)

編寫(xiě)的Python代碼不僅要能工作，更要成為典范。
專注于可讀性、性能和可維護(hù)性，同時(shí)充分利用Python的獨(dú)特優(yōu)勢(shì)和慣用法。

# 1. PEP 8規(guī)范是什么？
PEP 8 = Python Enhancement Proposal 8
就是Python的"交通規(guī)則"：
- 縮進(jìn)用4個(gè)空格
- 函數(shù)名用小寫(xiě)+下劃線
- 類(lèi)名用大駝峰
- 一行不超過(guò)79字符

# 2. Type Hints類(lèi)型提示
def add(a: int, b: int) -> int:  # 明確告訴別人輸入輸出類(lèi)型
    return a + b

# 3. 上下文管理器
with open('file.txt') as f:  # 自動(dòng)處理文件關(guān)閉
    content = f.read()

# 4. 生成器表達(dá)式
squares = (x**2for x in range(1000000))  # 不會(huì)占用大量?jī)?nèi)存

# 在Claude Code中輸入
/agents

創(chuàng)建一個(gè)用戶管理的FastAPI服務(wù)，包含CRUD操作和認(rèn)證

# app/main.py - FastAPI主應(yīng)用
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

# 創(chuàng)建數(shù)據(jù)庫(kù)表
models.Base.metadata.create_all(bind=engine)

# 初始化FastAPI應(yīng)用
app = FastAPI(
    title="用戶管理API",
    description="基于FastAPI的用戶管理微服務(wù)",
    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")

# 依賴注入：獲取當(dāng)前用戶
asyncdef get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Session = Depends(get_db)
) -> models.User:
    """
    驗(yàn)證JWT令牌并返回當(dāng)前用戶
    
    Args:
        token: JWT訪問(wèn)令牌
        db: 數(shù)據(jù)庫(kù)會(huì)話
        
    Returns:
        當(dāng)前認(rèn)證的用戶對(duì)象
        
    Raises:
        HTTPException: 令牌無(wú)效或過(guò)期
    """
    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)
):
    """用戶注冊(cè)接口"""
    # 檢查郵箱是否已存在
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(
            status_code=400,
            detail="Email already registered"
        )
    
    # 創(chuàng)建新用戶
    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"},
        )
    
    # 生成訪問(wèn)令牌
    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)]
):
    """獲取當(dāng)前用戶信息"""
    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)
):
    """獲取用戶列表（需要認(rèn)證）"""
    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)
):
    """更新用戶信息"""
    # 檢查權(quán)限（只能更新自己的信息或管理員）
    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 - 數(shù)據(jù)模型
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from .database import Base

class User(Base):
    """用戶數(shù)據(jù)模型"""
    __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 - 單元測(cè)試
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

# 測(cè)試數(shù)據(jù)庫(kù)配置
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():
    """覆蓋數(shù)據(jù)庫(kù)依賴"""
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

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

class TestUserAPI:
    """用戶API測(cè)試類(lèi)"""
    
    @pytest.fixture(autouse=True)
    def setup(self):
        """測(cè)試前置設(shè)置"""
        Base.metadata.create_all(bind=engine)
        yield
        Base.metadata.drop_all(bind=engine)
    
    def test_register_user(self):
        """測(cè)試用戶注冊(cè)"""
        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):
        """測(cè)試用戶登錄"""
        # 先注冊(cè)
        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):
        """測(cè)試注冊(cè)參數(shù)驗(yàn)證"""
        response = client.post(
            "/register",
            json={
                "email": email,
                "username": username,
                "password": password
            }
        )
        assert response.status_code == expected_status

優(yōu)化這個(gè)數(shù)據(jù)處理腳本的性能

# data_processor.py - 高性能數(shù)據(jù)處理
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:
    """高性能數(shù)據(jù)處理器"""
    
    def __init__(self, n_jobs: Optional[int] = None):
        """
        初始化數(shù)據(jù)處理器
        
        Args:
            n_jobs: 并行處理的進(jìn)程數(shù)，None表示使用所有CPU核心
        """
        self.n_jobs = n_jobs or mp.cpu_count()
        logger.info(f"初始化數(shù)據(jù)處理器，使用{self.n_jobs}個(gè)進(jìn)程")
    
    @lru_cache(maxsize=128)
    def _get_dtypes(self, file_path: str) -> Dict[str, Any]:
        """
        緩存文件的數(shù)據(jù)類(lèi)型，避免重復(fù)推斷
        
        Args:
            file_path: 文件路徑
            
        Returns:
            列名到數(shù)據(jù)類(lèi)型的映射
        """
        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"開(kāi)始處理文件: {file_path}")
        
        # 獲取數(shù)據(jù)類(lèi)型以優(yōu)化內(nèi)存使用
        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
            )):
                # 提交處理任務(wù)
                future = executor.submit(self._process_chunk, chunk, i)
                futures.append(future)
            
            # 收集結(jié)果
            results = []
            for future in futures:
                result = future.result()
                if result isnotNone:
                    results.append(result)
        
        # 合并結(jié)果并保存
        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"處理完成，結(jié)果保存到: {output_path}")
    
    @staticmethod
    def _process_chunk(chunk: pd.DataFrame, chunk_id: int) -> Optional[pd.DataFrame]:
        """
        處理單個(gè)數(shù)據(jù)塊
        
        Args:
            chunk: 數(shù)據(jù)塊
            chunk_id: 塊ID
            
        Returns:
            處理后的數(shù)據(jù)塊
        """
        try:
            # 數(shù)據(jù)清洗
            chunk = chunk.dropna(subset=['important_column'])
            
            # 向量化操作（比循環(huán)快100倍）
            chunk['new_column'] = np.where(
                chunk['value'] > 100,
                chunk['value'] * 1.1,
                chunk['value'] * 0.9
            )
            
            # 使用NumPy進(jìn)行數(shù)值計(jì)算
            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}時(shí)出錯(cuò): {e}")
            returnNone
    
    def optimize_dataframe(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        優(yōu)化DataFrame內(nèi)存使用
        
        Args:
            df: 輸入DataFrame
            
        Returns:
            優(yōu)化后的DataFrame
        """
        initial_memory = df.memory_usage(deep=True).sum() / 1024**2
        logger.info(f"初始內(nèi)存使用: {initial_memory:.2f} MB")
        
        # 優(yōu)化數(shù)值類(lèi)型
        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')
        
        # 轉(zhuǎn)換字符串為category類(lèi)型（如果重復(fù)值多）
        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"優(yōu)化后內(nèi)存使用: {final_memory:.2f} MB")
        logger.info(f"內(nèi)存減少: {(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
-科學(xué)計(jì)算：NumPy,SciPy,SymPy
-機(jī)器學(xué)習(xí)：scikit-learn,XGBoost,LightGBM
-深度學(xué)習(xí)：PyTorch,TensorFlow
-數(shù)據(jù)可視化：Matplotlib,Seaborn,Plotly
-特征工程：Feature-engine,CategoryEncoders

## Python Mastery
-Django5.0全棧開(kāi)發(fā)
-FastAPI+SQLAlchemy+Alembic
-Celery異步任務(wù)隊(duì)列
-Redis緩存策略
-GraphQLwithStrawberry

## Python Mastery
- Ansible自動(dòng)化部署
- Fabric遠(yuǎn)程執(zhí)行
- Paramiko SSH操作
- Schedule定時(shí)任務(wù)
- Click CLI工具開(kāi)發(fā)

## Company Standards
- 代碼風(fēng)格：使用公司的.pylintrc配置
- 測(cè)試要求：最低95%覆蓋率
- 文檔規(guī)范：使用Sphinx生成文檔
- Git提交：遵循conventional commits
- 部署流程：Docker + Kubernetes

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

# 檢查規(guī)范
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()