# 老写法：签名是黑盒
def train(model, data, epochs):
    ...

# 现代写法：意图一目了然
from torch import Tensor
from torch.nn import Module
from torch.utils.data import DataLoader

def train(
    model: Module,
    data: DataLoader,
    epochs: int,
    *,
    lr: float = 1e-3,        # ← 关键字参数强制写名，调用更安全
    device: str = "cuda",
) -> dict[str, float]:      # ← 返回类型也标注
    ...
    return {"final_loss": 0.23, "best_acc": 0.91}

# 方案 A：dataclass —— 轻量、纯 Python、无依赖
from dataclasses import dataclass, field

@dataclass(slots=True)            # 3.10+ slots 省内存，AI 场景里很关键
class GPTConfig:
    vocab_size: int = 50257
    n_layer: int = 12
    n_head: int = 12
    d_model: int = 768
    dropout: float = 0.0
    block_size: int = 1024

cfg = GPTConfig(d_model=1024)
print(cfg)                        # 自动生成 __repr__：GPTConfig(vocab_size=50257, ...)


# 方案 B：Pydantic —— 带"运行时校验 + 序列化"，FastAPI 默认引擎
from pydantic import BaseModel, Field, field_validator

class ChatRequest(BaseModel):
    model: str = "gpt-2"
    prompt: str = Field(..., min_length=1, max_length=8192)
    temperature: float = Field(0.8, ge=0.0, le=2.0)   # 范围约束
    max_tokens: int = 256

    @field_validator("prompt")
    @classmethod
    def no_ctrl_chars(cls, v: str) -> str:
        if any(ord(c) < 0x20 and c not in "\n\t" for c in v):
            raise ValueError("控制字符不允许")
        return v

req = ChatRequest(prompt="你好", temperature=3.0)
# → ValidationError: temperature must be ≤ 2.0   ← 在边界处就拦住

# 旧写法：if-elif 链
def handle(msg):
    if msg["type"] == "text":
        return process_text(msg["content"])
    elif msg["type"] == "image":
        return process_image(msg["url"], msg.get("caption", ""))
    elif msg["type"] == "tool_call":
        if msg["tool"] == "calculator":
            return run_calc(msg["args"])
        elif msg["tool"] == "search":
            return run_search(msg["args"])
    else:
        return None

# 现代写法：match/case 结构化解构
def handle(msg: dict) -> object:
    match msg:
        case {"type": "text", "content": c}:
            return process_text(c)
        case {"type": "image", "url": u, "caption": caption}:
            return process_image(u, caption)
        case {"type": "tool_call", "tool": "calculator", "args": a}:
            return run_calc(a)
        case {"type": "tool_call", "tool": "search", "args": a}:
            return run_search(a)
        case {"type": "error", "code": code, **rest}:
            log_error(code, rest)
            return None
        case _:
            return None   # ← 通配，等价于 default

# 还能匹配类结构
match response:
    case OpenAIResponse(choices=[Choice(text=t, finish_reason="stop")]):
        return t
    case OpenAIResponse(choices=[Choice(delta=Delta(content=c))]):
        yield c        # 流式 token

# 场景 1：避免重复调用昂贵的函数
# 老写法：调两次
if len(prompt_tokens) > 8192:
    print(f"超出 {len(prompt_tokens) - 8192} tokens")

# 海象：只算一次
if (n := len(prompt_tokens)) > 8192:
    print(f"超出 {n - 8192} tokens")


# 场景 2：在 while 条件里捕获值
# 老写法
line = f.readline()
while line:
    process(line)
    line = f.readline()

# 海象：经典的"读一行"
while (line := f.readline()):
    process(line)


# 场景 3：列表推导里复用计算
# 老写法：算两遍
results = [compute_loss(x) for x in batch if compute_loss(x) < 0.5]

# 海象：算一次
results = [loss for x in batch if (loss := compute_loss(x)) < 0.5]


# 场景 4：用 in 表达式同时绑定（替代 dict.get + if None 检查）
if (cap := msg.get("max_tokens")) is not None:
    use(cap)

import asyncio
from fastapi import FastAPI

app = FastAPI()

# 协程：用 async def 定义，await 挂起等待
async def generate_stream(prompt: str) -> str:
    # await 一个耗 IO 的协程（不阻塞整个事件循环）
    tokens = await model.generate_async(prompt)
    return "".join(tokens)

@app.post("/chat")
async def chat(req: ChatRequest):
    # 关键：把阻塞调用丢给线程池，别用 await 包装同步函数
    text = await asyncio.to_thread(sync_inference, req.prompt)
    return {"reply": text}

# 并发跑多个任务：asyncio.gather
async def batch_inference(prompts: list[str]) -> list[str]:
    tasks = [generate_stream(p) for p in prompts]
    return await asyncio.gather(*tasks)      # 并发，不串行

# 限流：Semaphore 控制并发数，保护 GPU 不被打爆
sem = asyncio.Semaphore(8)

async def safe_generate(prompt: str) -> str:
    async with sem:        # ← 同时最多 8 个进入
        return await generate_stream(prompt)

# f-string（3.6+）：比 % 和 .format 都快、都直观
name, n = "transformer", 12
print(f"模型 {name} 有 {n} 层，参数量 {n * 768 * 768 / 1e6:.1f}M")
# 模型 transformer 有 12 层，参数量 7.1M

# 3.8+ 调试语法：在表达式后加 = 直接打印
x = compute_loss(batch)
print(f"{x=}")        # → x=0.234

# 多行 f-string + 表达式
template = f"""
[Prompt]
{prompt}

[Config]
temperature={temp:.2f}
max_tokens={max_tok}
"""


# pathlib（取代 os.path）：面向对象、跨平台、可链式
from pathlib import Path

# 老写法
import os
ckpt = os.path.join(os.path.dirname(__file__), "checkpoints", "gpt.pt")

# 现代写法：/ 运算符重载，语义清晰
ckpt = Path(__file__).parent / "checkpoints" / "gpt.pt"

# 常用操作一气呵成
for f in Path("data/").rglob("*.jsonl"):
    if f.stat().st_size > 0:
        process(f.read_text(encoding="utf-8"))

# 写文件：一行
Path("output.txt").write_text("done", encoding="utf-8")

# 创建目录（已存在不报错）
Path("runs/exp1").mkdir(parents=True, exist_ok=True)

# 推导式：列表 / 字典 / 集合，一行表达映射+过滤
tokens = [tok for line in corpus for tok in line.split() if tok.isalpha()]

vocab = {tok: i for i, tok in enumerate(sorted(set(tokens)))}

stop = {"the", "a", "an"}
filtered = {w for w in vocab if w not in stop and len(w) > 2}


# 生成器表达式：惰性求值，省内存，适合大数据流
# 用 () 而非 []，不一次实例化全部
total = sum(len(t) for t in token_stream)   # 不会先建 list


# yield：把函数变成迭代器，流式产出
def stream_tokens(path: Path):
    for line in path.open(encoding="utf-8"):
        for tok in line.split():
            yield tok

# yield from：委托给子生成器
def stream_all(paths: list[Path]):
    for p in paths:
        yield from stream_tokens(p)


# Protocol（PEP 544）：结构化子类型，比 ABC 更轻
from typing import Protocol

class Tokenizer(Protocol):
    def encode(self, text: str) -> list[int]: ...
    def decode(self, ids: list[int]) -> str: ...

# 任何有这两个方法的对象都自动满足 Tokenizer，无需继承
class MyTok:
    def encode(self, text): return [ord(c) for c in text]
    def decode(self, ids):  return "".join(chr(i) for i in ids)

def train(tok: Tokenizer, data):   # ← 鸭子类型 + 静态检查
    ...

train(MyTok(), ...)   # ✅ 不报错

# 老写法（3.11 及之前）：样板代码多
from typing import TypeVar, Generic, TypeAlias

T = TypeVar("T")

class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: list[T] = []
    def push(self, x: T) -> None:
        self._items.append(x)
    def pop(self) -> T:
        return self._items.pop()

Vector: TypeAlias = list[float]


# 新写法（3.12+，PEP 695）：语法糖，干净
class Stack[T]:
    def __init__(self) -> None:
        self._items: list[T] = []
    def push(self, x: T) -> None:
        self._items.append(x)
    def pop(self) -> T:
        return self._items.pop()

type Vector = list[float]      # ← type 语句，3.12 新

def first(xs: list[T]) -> T:   # 函数级泛型，自动推断
    return xs[0]


# 更实战的例子：一个泛型缓存
class LRUCache[K, V]:
    def __init__(self, cap: int): ...
    def get(self, k: K) -> V | None: ...
    def put(self, k: K, v: V) -> None: ...

cache: LRUCache[str, Tensor] = LRUCache(1024)

from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol
import asyncio, json
from fastapi import FastAPI
from pydantic import BaseModel, Field

# 1. 用 Protocol 定义抽象，不绑死实现
class Tokenizer(Protocol):
    def encode(self, text: str) -> list[int]: ...
    def decode(self, ids: list[int]) -> str: ...

# 2. dataclass 做配置，slots 省 GPU 服务器内存
@dataclass(slots=True)
class GenConfig:
    max_tokens: int = 256
    temperature: float = 0.8
    top_p: float = 0.95

# 3. Pydantic 做请求 DTO，边界校验
class ChatReq(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=8192)
    stream: bool = False

app = FastAPI()
sem = asyncio.Semaphore(4)        # 4. 并发限流，保护 GPU

# 5. async + 生成器：流式产出 token
async def stream_tokens(tok: Tokenizer, ids: list[int], cfg: GenConfig):
    async with sem:
        for i in range(cfg.max_tokens):
            nxt = await model.next_token(ids)   # 假设的异步推理
            if nxt == EOS:
                break
            ids.append(nxt)
            yield tok.decode([nxt])             # 流式吐出

# 6. match/case 分流：流式 vs 一次性
@app.post("/chat")
async def chat(req: ChatReq):
    ids = tokenizer.encode(req.prompt)
    cfg = GenConfig()

    match req:
        case ChatReq(stream=True):
            async def gen():
                async for piece in stream_tokens(tokenizer, ids, cfg):
                    yield f"data: {json.dumps({'text': piece})}\n\n"
            return StreamingResponse(gen(), media_type="text/event-stream")

        case _:
            pieces = []
            async for p in stream_tokens(tokenizer, ids, cfg):
                pieces.append(p)
            return {"reply": "".join(pieces)}

# 7. 海象 + pathlib：启动时加载权重
if (ckpt := Path("models/gpt2.pt")).exists():
    model.load(ckpt.read_bytes())