第2章现代 Python 高阶特性¶

学习目标¶

熟练运用 typing 模块、TypeVar、Generic 编写类型安全的代码
使用 dataclasses 与 Pydantic 模型构建数据载体
理解并实现装饰器与上下文管理器，封装横切逻辑
掌握生成器与协程在流式处理中的应用
理解描述符协议及其在属性管理中的作用

问题导入¶

AI Agent 系统中存在大量"数据流动"：用户输入、模型输出、工具调用、中间状态都需要被精确建模。如果用裸 dict 传递数据，字段拼写错误在运行时才会暴露，重构时 IDE 无法给出有效提示；如果用普通函数堆砌业务逻辑，日志、缓存、鉴权等横切关注点会污染核心代码；如果一次性把 LLM 流式输出的全部 token 收集再处理，又会丧失流式响应的低延迟体验。

现代 Python（3.10+）提供了类型提示、dataclasses、装饰器、上下文管理器、生成器、描述符等一整套"高阶武器"，它们让代码在保持运行时灵活性的同时获得静态类型保障，让横切逻辑可复用，让流式处理优雅自然。掌握这些特性，是从"能写出 Python"到"能写出工程级 Python"的关键跃迁。本章将逐一拆解这些特性在 AI Agent 场景中的实战用法。

核心概念¶

类型提示与泛型¶

类型提示（Type Hints）为 Python 这种动态类型语言引入了静态检查能力。typing 模块提供 List、Dict、Optional、Union 等构造；TypeVar 与 Generic 支持泛型编程，使容器与工具函数能保留元素类型信息。Python 3.9+ 允许直接用 list[int]、dict[str, int] 等内置泛型，3.10+ 提供 X | Y 联合类型语法，3.12 进一步引入 def f[T](x: T) -> T 的简洁泛型语法。配合 mypy / pyright，可在不运行代码的前提下发现大量类型错误。

dataclasses¶

dataclasses.dataclass 装饰器自动为类生成 __init__、__repr__、__eq__ 等方法，大幅减少样板代码。它支持 field(default_factory=...)、frozen=True 不可变实例、slots=True 节省内存等选项，是定义值对象与配置类的首选。与 Pydantic 相比，dataclasses 是标准库、零依赖、轻量，但不做数据校验；Pydantic 则在 dataclass 思想之上增加了运行时校验与序列化能力。

装饰器¶

装饰器（decorator）是一个接收函数并返回新函数的高阶函数，用 @decorator 语法糖应用。它在不修改原函数代码的前提下，为其增加日志、缓存、重试、鉴权、计时等横切能力。functools.wraps 用于保留原函数的元信息。装饰器也可接收参数（此时需要三层嵌套函数），或装饰类的方法。

上下文管理器¶

上下文管理器（context manager）通过 __enter__ / __exit__ 协议或 @contextlib.contextmanager 装饰器实现，配合 with 语句，保证资源（文件、连接、锁）在使用后被正确释放，即便发生异常也不例外。它是"获取-使用-释放"模式的优雅封装，在数据库会话、HTTP 客户端、临时环境切换等场景中不可或缺。

生成器与协程¶

生成器（generator）是包含 yield 的函数，调用后返回一个迭代器，按需产出值而非一次性构建整个列表，节省内存。yield 还可用于实现协程式的协作式多任务。在 AI Agent 场景中，生成器是处理 LLM 流式输出、分块读取大文件、管道式数据变换的利器。

描述符¶

描述符（descriptor）是实现了 __get__、__set__、__delete__ 中任一方法的类，用作另一个类的属性时能拦截属性的访问与赋值。property、classmethod、staticmethod 本质都是描述符。理解描述符有助于构建自定义 ORM 字段、带校验的属性、懒加载等高级模式。

原理图¶

flowchart TB
    subgraph 类型系统
        TypeHint[类型提示] --> Checker[mypy / pyright 静态检查]
        TypeHint --> Generic[泛型 TypeVar/Generic]
        Generic --> Container[泛型容器 Result T]
    end
    subgraph 数据建模
        DC[dataclasses] -->|零依赖| ValueObj[值对象/配置]
        Pyd[Pydantic] -->|运行时校验| Model[API 模型]
    end
    subgraph 横切关注点
        Dec[装饰器] --> Log[日志/重试/缓存]
        CM[上下文管理器] --> Res[资源管理]
    end
    subgraph 流式与状态
        Gen[生成器 yield] --> Stream[流式输出/管道]
        Desc[描述符] --> Prop[属性拦截/校验]
    end
    Checker --> Quality[工程级代码质量]
    Model --> Quality
    Log --> Quality
    Stream --> Quality

图：现代 Python 高阶特性分层，类型系统保障静态安全，数据建模与横切逻辑提升可维护性

示例代码¶

类型提示与泛型 Result 容器¶

from __future__ import annotations
from dataclasses import dataclass
from typing import Generic, TypeVar

T = TypeVar("T")
E = TypeVar("E", covariant=True)

@dataclass(frozen=True)
class Ok(Generic[T]):
    value: T
    @property
    def is_ok(self) -> bool:
        return True

@dataclass(frozen=True)
class Err(Generic[E]):
    error: E
    @property
    def is_ok(self) -> bool:
        return False

Result = Ok[T] | Err[str]  # 类型别名

def divide(a: float, b: float) -> Result[float]:
    if b == 0.0:
        return Err("division by zero")
    return Ok(a / b)

if __name__ == "__main__":
    r1 = divide(10, 2)
    r2 = divide(1, 0)
    # 模式匹配提取结果
    match r1:
        case Ok(value=v):
            print(f"结果: {v}")  # 结果: 5.0
        case Err(error=e):
            print(f"错误: {e}")
    print(r2.is_ok)  # False

dataclasses 与 frozen 配置¶

from dataclasses import dataclass, field

@dataclass(frozen=True, slots=True)
class AgentConfig:
    model: str = "gpt-4o-mini"
    temperature: float = 0.7
    max_tokens: int = 1024
    tools: list[str] = field(default_factory=list)

    def with_overrides(self, **kwargs: object) -> "AgentConfig":
        # frozen 实例不可变，返回新实例
        from dataclasses import replace
        return replace(self, **kwargs)

cfg = AgentConfig(tools=["search", "calculator"])
cfg2 = cfg.with_overrides(temperature=0.2)
print(cfg.temperature)   # 0.7
print(cfg2.temperature)  # 0.2

装饰器：重试与计时¶

import functools
import logging
import time
from typing import Callable, ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")
logger = logging.getLogger(__name__)

def retry(times: int = 3, delay: float = 0.5):
    """带参数的重试装饰器"""
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            last_exc: Exception | None = None
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as exc:
                    last_exc = exc
                    logger.warning("第 %s 次调用失败: %s", attempt, exc)
                    if attempt < times:
                        time.sleep(delay)
            assert last_exc is not None
            raise last_exc
        return wrapper
    return decorator

def timed(func: Callable[P, R]) -> Callable[P, R]:
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        start = time.perf_counter()
        result = func(*args, **kwargs)
        logger.info("%s 耗时 %.3fs", func.__name__, time.perf_counter() - start)
        return result
    return wrapper

@timed
@retry(times=3, delay=0.1)
def flaky_call(n: int) -> int:
    if n < 2:
        raise ValueError("参数太小")
    return n * 10

print(flaky_call(5))  # 50

上下文管理器：计时与资源¶

import time
from contextlib import contextmanager
from typing import Iterator

@contextmanager
def timer(label: str) -> Iterator[None]:
    start = time.perf_counter()
    try:
        yield
    finally:
        print(f"[{label}] 耗时 {time.perf_counter() - start:.3f}s")

with timer("数据处理"):
    total = sum(i * i for i in range(100_000))
    print(f"求和结果: {total}")

生成器：LLM 流式输出模拟¶

from typing import Iterator

def stream_tokens(text: str, chunk_size: int = 4) -> Iterator[str]:
    """模拟 LLM 流式返回 token"""
    for i in range(0, len(text), chunk_size):
        yield text[i:i + chunk_size]

def echo_stream() -> None:
    full = ""
    for token in stream_tokens("你好，我是一个 AI Agent，可以帮你处理任务。"):
        full += token
        print(f"收到: {token!r}  累计: {full}")

if __name__ == "__main__":
    echo_stream()

运行结果示例：

收到: '你好，我'  累计: 你好，我
收到: '是一个 '  累计: 你好，我是一个
...

描述符：带校验的属性¶

class ValidatedField:
    def __init__(self, min_val: int, max_val: int) -> None:
        self.min = min_val
        self.max = max_val
        self.name = ""

    def __set_name__(self, owner: type, name: str) -> None:
        self.name = name

    def __get__(self, obj: object, owner: type) -> int:
        if obj is None:
            return self  # type: ignore
        return obj.__dict__[self.name]

    def __set__(self, obj: object, value: int) -> None:
        if not isinstance(value, int):
            raise TypeError(f"{self.name} 必须是 int")
        if not (self.min <= value <= self.max):
            raise ValueError(f"{self.name} 需在 [{self.min}, {self.max}]")
        obj.__dict__[self.name] = value

class AgentSettings:
    max_tokens = ValidatedField(1, 8192)
    top_k = ValidatedField(1, 100)

    def __init__(self, max_tokens: int, top_k: int) -> None:
        self.max_tokens = max_tokens
        self.top_k = top_k

s = AgentSettings(max_tokens=512, top_k=10)
# s.max_tokens = 99999  # 抛出 ValueError

企业案例¶

案例 1：类型安全的大型 AI 平台¶

某 LLM 平台后端代码量超 20 万行，早期全用 dict 传递请求，重构时频繁出现 KeyError。引入 pyproject.toml 中 mypy 严格模式后，将所有接口入参/出参改为 dataclass 或 Pydantic 模型，并在 CI 中强制 mypy --strict 通过。半年后，运行时 AttributeError / KeyError 类缺陷下降 65%，IDE 跳转与重构能力显著提升。

案例 2：装饰器统一 LLM 调用治理¶

团队有数十个调用 LLM 的函数，各自实现重试、超时、计费日志，逻辑重复且不一致。他们抽象出 @llm_call(provider="openai", model="gpt-4o") 装饰器，统一封装重试（指数退避）、超时、Token 计费、链路追踪。业务函数只关心 prompt 构造与结果解析，治理逻辑集中可演进。

案例 3：生成器处理海量语料¶

一个数据团队需要清洗 TB 级语料训练 Agent。用列表一次性加载会 OOM，改用生成器管道：read_chunks() → clean() → tokenize() → batch()，每一步都是 yield，整体内存占用稳定在百 MB 级，且可方便地插入 map/filter 式变换。

常见问题¶

Q1：dataclass 和 Pydantic 该选哪个？ 内部值对象、配置、不需要校验的数据结构用 dataclass（零依赖、更快）；API 边界、需要从 JSON 反序列化并校验的模型用 Pydantic（自带校验、序列化、OpenAPI）。两者也可混用，Pydantic v2 支持 dataclass 桥接。

Q2：装饰器堆叠时执行顺序是怎样的？ @A @B def f 等价于 A(B(f))。调用时最外层装饰器先执行其"前置"逻辑，最内层最后；return 后的"后置"逻辑则相反。即 @timed 放最外层才能正确计时包含重试的整个过程。

Q3：生成器和 async generator 有什么区别？ 普通生成器用 yield，迭代用 for；异步生成器用 async yield，迭代用 async for。前者是同步阻塞，后者允许在 yield 处让出事件循环，适合异步流式场景（如异步读取 LLM 流）。

Q4：frozen=True 的 dataclass 还能修改字段吗？ 不能直接赋值，会抛 FrozenInstanceError。但可用 dataclasses.replace(obj, field=new) 返回一个新实例，实现"不可变更新"，类似函数式编程中的 copy-on-write。

面试题¶

题1：解释 TypeVar、Generic、ParamSpec 三者的区别。 参考答案：TypeVar 声明一个泛型类型变量，用于保留单个函数参数与返回值的类型关联；Generic 用于将类声明为泛型类，使其能参数化；ParamSpec 用于捕获一个可调用对象的"参数签名"（含参数名与类型），常用于装饰器类型签名，使被装饰函数保留原有参数类型。

题2：写一个 @cached 装饰器，缓存函数结果并支持设置最大缓存数。 参考答案：使用 functools.lru_cache(maxsize=N) 即可；手写实现可用 dict 存储 args -> result，并在容量超限时按 LRU 策略淘汰。注意可变参数不可哈希需特殊处理。

题3：with 语句的执行流程是什么？异常如何被处理？ 参考答案：进入 with 时调用上下文管理器的 __enter__，其返回值赋给 as 变量；块内执行业务逻辑；退出时无论是否异常都调用 __exit__(exc_type, exc_val, exc_tb)，若 __exit__ 返回真值则吞掉异常，否则继续向上抛出。

题4：描述符的 __set_name__ 有什么用？ 参考答案：__set_name__(owner, name) 在描述符被赋值给类的属性时由解释器自动调用，告知描述符它在宿主类中使用的属性名，便于描述符在 obj.__dict__ 中用正确的 key 存储实际值，无需在 __init__ 中显式传入字段名。

练习题¶

上机：实现一个泛型 Cache[K, V] 类，使用 dict 存储，提供 get、set、has 方法，并保证 cache.get("k") 的返回类型为 V | None。用 mypy 验证类型正确。
思考：如果要用装饰器实现一个"限流"（每秒最多调用 N 次）功能，你会如何设计？考虑线程安全与异步两种情况。
上机：用 @contextmanager 实现一个 temporary_env(key, value) 上下文管理器，在 with 块内将环境变量临时设为某值，退出后恢复原值，即使块内抛异常也需恢复。

小结¶

本章覆盖了现代 Python 工程化最核心的高阶特性：类型系统让代码"可信"，dataclasses/Pydantic 让数据"可塑"，装饰器与上下文管理器让横切逻辑"可复用"，生成器与描述符让流式与属性管理"可控"。这些特性在后续章节（异步编程、FastAPI、测试）中会被反复使用。延伸阅读：typing 模块文档、PEP 695 类型参数语法、Real Python: Decorators。

第2章 现代 Python 高阶特性¶