小菜鸟

import pandas as pd

# 模拟加载一份销售数据
df = pd.read_csv('sales_data.csv')

# .head() 查看前5行，快速了解数据结构
print(df.head())

# .tail() 查看后5行，检查数据末尾是否有异常
print(df.tail())

# .sample(5) 随机查看5行，避免数据排序带来的偏差
print(df.sample(5))

from sqlalchemy import select
# 2.0统一查询风格
stmt = select(User).where(User.age > 18)
users = session.execute(stmt).scalars().all()

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.asyncio import async_sessionmaker

async_engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

async with AsyncSessionLocal() as session:
    user = User(name="Alice")
    session.add(user)
    await session.commit()

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)

from datetime import datetime
from typing import List, Optional
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, Integer, DateTime, ForeignKey, Boolean

class Base(DeclarativeBase):
    """所有数据库模型的基类"""
    pass

class User(Base):
    """用户模型"""
    __tablename__ = "users"
    
    # 主键定义
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    
    # 必填字段
    username: Mapped[str] = mapped_column(
        String(50),
        unique=True,
        nullable=False,
        comment="用户名"
    )
    
    # 可选字段
    email: Mapped[Optional[str]] = mapped_column(
        String(100),
        unique=True,
        nullable=True
    )
    
    # 带默认值的字段
    is_active: Mapped[bool] = mapped_column(
        Boolean,
        default=True,
        comment="是否激活"
    )
    
    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        default=datetime.utcnow
    )
    
    # 关系定义
    posts: Mapped[List["Post"]] = relationship(
        back_populates="author",
        lazy="selectin"
    )
    
    def __repr__(self) -> str:
        return f"<User(id={self.id}, username='{self.username}')>"

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 创建数据库引擎
engine = create_engine("sqlite:///example.db", echo=True)

# 创建表结构
Base.metadata.create_all(engine)

# 创建会话工厂
SessionLocal = sessionmaker(bind=engine)

class Post(Base):
    __tablename__ = 'posts'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    
    # 定义与User模型的关系
    author: Mapped["User"] = relationship(back_populates="posts")

from sqlalchemy.orm import selectinload

# 使用selectinload优化查询
stmt = select(User).options(selectinload(User.posts))
users_with_posts = session.execute(stmt).scalars().all()

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.asyncio import async_sessionmaker

async_engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

async def create_user(name: str):
    async with AsyncSessionLocal() as session:
        user = User(name=name)
        session.add(user)
        await session.commit()
        return user

try:
    # 执行一系列操作
    session.add(order)
    session.delete(cart_item)
    await session.commit()
except Exception:
    await session.rollback()
    raise

import torch
import torch.nn as nn
import math

class ScaledDotProductAttention(nn.Module):
    def __init__(self, d_k):
        super().__init__()
        self.d_k = d_k # 键向量的维度
        # 将一组原始分数（logits）转换成概率分布。把任意数值变成 0 到 1 之间的概率值，并且保证这些概率值加起来等于 1
        self.softmax = nn.Softmax(dim=-1)

    def forward(self, Q, K, V):
        # 1. 计算Q和K的相似度 (矩阵乘法)
        scores = torch.matmul(Q, K.transpose(-1, -2))
        
        # 2. 缩放，防止点积过大导致Softmax梯度消失
        scores = scores / math.sqrt(self.d_k)
        
        # 3. 应用Softmax，得到注意力权重
        attention_weights = self.softmax(scores)
        
        # 4. 用权重对V加权求和，得到最终输出
        output = torch.matmul(attention_weights, V)
        return output

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super().__init__()
        self.d_model = d_model # 词向量的总维度，比如512
        self.num_heads = num_heads # “头”的数量，比如8个
        self.d_k = d_model // num_heads # 每个“头”的维度，512/8=64
        
        # 这三个线性层负责把输入 x 转换成 Q(查询), K(键), V(值)
        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        # 这个线性层负责把所有“头”的结果融合起来
        self.W_o = nn.Linear(d_model, d_model) 
        # 计算注意力的“工具”
        self.attention = ScaledDotProductAttention(self.d_k)

    def forward(self, x):
        batch_size, seq_len, _ = x.shape
        
        # 1. 线性投影并分割成多个头
        # 用线性层把输入 x 转换成查询向量 Q。此时 Q 的形状是 (batch_size, seq_len, d_model)
        # view().transpose(1, 2): 它把 Q 的最后一个维度（比如512）切分成 num_heads 个（比如8个）小向量，每个小向量的维度是 d_k（比如64）。然后通过转置，把“头”的维度放到前面。
        # 意义是：原来我们只有一个大的查询向量，现在我们把它分成了8个“小专家”（头）。每个“小专家”都从自己的视角去观察整个句子
        Q = self.W_q(x).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        K = self.W_k(x).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        V = self.W_v(x).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        
        # 2. 并行计算多头注意力
        head_outputs = []
        # 让每一个“小专家”（头）独立工作
        for i in range(self.num_heads):
            head_output = self.attention(Q[:, i], K[:, i], V[:, i])
            # 每个头都会输出一个结果，我们把所有头的结果都收集到 head_outputs 列表里
            head_outputs.append(head_output)
            
        # 3. 把他们的成果拼接起来。torch.cat(..., dim=-1) 会在最后一个维度上进行拼接，把8个维度为 d_k (64) 的向量，重新拼成一个维度为 d_model (8 * 64 = 512) 的大向量。
        concatenated = torch.cat(head_outputs, dim=-1)
        
        # 4. 最终线性变换。把所有“小专家”的观点综合起来，形成一个最终的、更全面的理解
        output = self.W_o(concatenated)
        return output

# 一个完整的编码器层还包括前馈网络和残差连接
class EncoderLayer(nn.Module):
    def __init__(self, d_model, ffn_hidden, num_heads):
        super().__init__()
        self.attention = MultiHeadAttention(d_model, num_heads)
        # 前馈网络，一个简单的两层全连接网络
        self.ffn = nn.Sequential(
            nn.Linear(d_model, ffn_hidden), # 先放大维度
            nn.ReLU(), # 加一个非线性激活函数
            nn.Linear(ffn_hidden, d_model) # 再变回原来的维度
        )
        # 两个层归一化，用来稳定训练
        self.layer_norm1 = nn.LayerNorm(d_model)
        self.layer_norm2 = nn.LayerNorm(d_model)

    def forward(self, x):
        # 残差连接 + 层归一化
        x = self.layer_norm1(x + self.attention(x))
        x = self.layer_norm2(x + self.ffn(x))
        return x

import os
# 使用国内镜像
os.environ['HF_ENDPOINT'] = 'https://hf-mirror.com'
from transformers import pipeline

# 一行代码加载一个情感分析模型
classifier = pipeline("sentiment-analysis")

# 直接使用
result = classifier("I love learning about Transformer with PyTorch!")
print(result)
# 输出: [{'label': 'POSITIVE', 'score': 0.9998}]

def my_decorator(func):
    def wrapper(*args, **kwargs):
        print("在函数执行之前...")
        result = func(*args, **kwargs)
        print("在函数执行之后...")
        return result
    return wrapper

@my_decorator
def say_hello():
    print("Hello!")

say_hello()

import torch

# 定义一个需要追踪梯度的张量
x = torch.tensor(2.0, requires_grad=True)

# 进行一系列运算：y = x^2 + 3x + 1
y = x**2 + 3*x + 1

# 此时，y 拥有一个 grad_fn 属性，指向创建它的函数
# grad_fn 属性：它指向创建该张量的函数对象。这个对象记录了前向传播的操作细节，以便在反向传播时能够利用链式法则计算梯度
print(f"计算图节点: {y.grad_fn}")