Python Concepts Every AI Engineer Must Master

Machine Learning Mastery

Machine Learning Mastery2026年6月12日

Python Concepts Every AI Engineer Must Master

8.5Score

TL;DR · AI 摘要

AI工程师必须掌握Python的生成器、上下文管理器、异步编程等核心概念，以构建高效、可扩展的AI系统。

核心要点

生成器和惰性求值可实现大规模数据流处理，内存使用保持恒定。
上下文管理器有助于管理硬件资源和清理操作，提升系统稳定性。
异步编程能提升LLM API查询和并发代理工具执行的效率。

结构提纲

按章节快速跳转。

§引言
AI工程师需要掌握Python核心概念，以构建可扩展的AI系统。
·生成器与惰性求值
生成器通过惰性求值实现大规模数据流处理，内存使用保持恒定。
·上下文管理器
上下文管理器用于管理硬件资源和清理操作，提升系统稳定性。
·异步编程
异步编程能提升LLM API查询和并发代理工具执行的效率。
·数据类与 Pydantic
数据类和Pydantic用于验证配置和构建结构化工具调用模式。
·魔法方法
魔法方法用于设计与深度学习框架兼容的ML抽象。

思维导图

用一张图看清主题之间的关系。

查看大纲文本（无障碍 / 无 JS 友好）

AI工程师必须掌握的Python概念
- 生成器与惰性求值
  - 内存高效的数据流处理
- 上下文管理器
  - 资源管理与清理
- 异步编程
  - 提升API查询与并发执行效率

金句 / Highlights

值得收藏与分享的关键句。

生成器通过惰性求值实现大规模数据流处理，内存使用保持恒定。
— 第 2 段
⬇︎ 下载 PNG 𝕏 分享到 X
异步编程能提升LLM API查询和并发代理工具执行的效率。
— 第 3 段
⬇︎ 下载 PNG 𝕏 分享到 X
数据类和Pydantic用于验证配置和构建结构化工具调用模式。
— 第 4 段
⬇︎ 下载 PNG 𝕏 分享到 X

#Python#AI工程#异步编程#生成器#上下文管理器

打开原文

每位 AI 工程师必须掌握的 Python 概念 - MachineLearningMastery.com

每位 AI 工程师必须掌握的 Python 概念

By

Matthew Mayo

on

June 12, 2026

in

Artificial Intelligence

0

Share

Post

在本文中，你将学习到每位 AI 工程师必须掌握的五个关键 Python 概念，以构建可扩展、生产级别的 AI 系统。

我们将涵盖的主题包括：

生成器和延迟评估如何让你以恒定的内存开销流式传输大型数据集。

上下文管理器、异步编程和 Pydantic 模型如何帮助你管理硬件资源、扩展 API 调用并安全地验证配置。

Python 魔术方法如何让你构建与 PyTorch 等深度学习框架兼容的自定义抽象。

每位 AI 工程师必须掌握的 Python 概念

AI 工程师需要了解的内容

从编写本地实验脚本过渡到构建可扩展、生产级别的 AI 系统，需要改变我们编写 Python 的方式。虽然动态类型、基本循环和列表推导式对于原型设计模型或探索数据是合理的，但它们无法满足实际 AI 应用的性能、内存和延迟限制。

AI 工程不仅仅是训练算法或加载预训练权重，它还涉及处理大型数据集、管理昂贵的硬件资源（如 GPU）、并发连接到外部 API，以及构建干净、类型安全的软件接口。要在此层次上操作，你必须掌握专业开发人员和深度学习框架所依赖的原生语言结构。

在本文中，我们将探讨你，作为 AI 工程师，必须掌握的五个关键 Python 概念：

生成器与延迟评估：用于以恒定的内存开销流式传输大型数据集

上下文管理器：用于管理宝贵的硬件状态和资源清理

异步编程：用于扩展 LLM API 查询和并发代理工具执行

数据类与 Pydantic：用于验证配置并构建用于工具调用的结构化模式

魔术方法：用于从头开始设计与框架兼容的 ML 抽象

1. 生成器与延迟评估（内存高效的流式数据传输）

在训练模型或对大型数据集进行批量推理时，一次性将所有数据加载到内存中会导致内存溢出错误。如果你的数据集包含数百万个文本文档、高分辨率图像或特征向量，标准列表会强制 Python 一次性为所有项目分配内存。

生成器通过延迟评估解决了这个问题。通过使用 yield 关键字，生成器返回一个迭代器，按需逐个计算并生成元素。无论你流式传输 100 个样本还是 1 亿个样本，这都会保持你的 RAM 使用量平稳。

在这种天真的方法中，我们读取并预处理一个文本负载数据集，在可以迭代它们之前，将所有处理后的字典加载到内存中的一个巨大列表中：

import json import io

一个模拟的 JSONL 文件流，包含原始文本负载

def get_dataset_stream(): data = "\n".join([json.dumps({"id": i, "text": f"User query raw text payload {i}"}) for i in range(50000)]) return io.StringIO(data)

一个简单的列表函数，一次性处理所有记录

def load_all_records_naive(stream): records = [] for line in stream: payload = json.loads(line)

立即处理数据并添加到列表中

processed = { "id": payload["id"], "text": payload["text"].lower(), "length": len(payload["text"]) } records.append(processed) return records

运行此代码需要将所有 50,000 个处理后的字典加载到内存中

stream = get_dataset_stream() data = load_all_records_naive(stream) print(f"以简单方式加载了 {len(data)} 条记录。")

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

import

json

io

A mock JSONL file stream of raw text payloads

def

get_dataset_stream

(

)

:

data

=

"\n"

.

join

[

dumps

{

"id"

i

,

"text"

f

"User query raw text payload {i}"

}

for

range

50000

]

return

StringIO

Naive list function processing all records at once

load_all_records_naive

stream

records

line

payload

loads

Process data immediately and append to a list

processed

lower

"length"

len

append

Running this requires loading all 50,000 processed dictionaries into RAM

print

"Loaded {len(data)} records naive-style."

通过将我们的读取器转换为生成器，我们可以按需逐批流式传输预处理后的负载。让我们看看一个使用 Python 的 tracemalloc 库来测量峰值内存使用差异的脚本：

import json import io import tracemalloc

一个模拟的 JSONL 文件流，包含原始文本负载

def get_dataset_stream(): data = "\n".join([json.dumps({"id": i, "text": f"User query raw text payload {i}"}) for i in range(50000)]) return io.StringIO(data)

一个简单的列表函数，一次性处理所有记录

def load_all_records_naive(stream): records = [] for line in stream: payload = json.loads(line)

立即处理数据并添加到列表中

processed = { "id": payload["id"], "text": payload["text"].lower(), "length": len(payload["text"]) } records.append(processed) return records

生成器函数，逐个生成预处理后的记录

def stream_records_generator(stream): for line in stream: payload = json.loads(line) yield { "id": payload["id"], "text": payload["text"].lower(), "length": len(payload["text"]) }

测量简单实现的内存使用情况

tracemalloc.start() stream_naive = get_dataset_stream() records_list = load_all_records_naive(stream_naive) for r in records_list: pass # 模拟训练循环步骤 _, peak_naive = tracemalloc.get_traced_memory() tracemalloc.stop()

测量生成器实现的内存使用情况

tracemalloc.start() stream_gen = get_dataset_stream() records_generator = stream_records_generator(stream_gen) for r in records_generator: pass # 模拟训练循环步骤 _, peak_gen = tracemalloc.get_traced_memory() tracemalloc.stop()

输出结果

print(f"简单实现的峰值内存使用：{peak_naive / 1024 / 1024:.4f} MB") print(f"生成器实现的峰值内存使用：{peak_gen / 1024 / 1024:.4f} MB")

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

tracemalloc

Generator function yielding preprocessed records one-by-one

stream_records_generator

yield

Measure the naive implementation

start

stream_naive

records_list

r

pass

模拟训练循环步骤

_

peak_naive

get_traced_memory

stop

测量生成器实现

stream_gen

records_generator

peak_gen

输出结果

"Naive peak RAM: {peak_naive / 1024 / 1024:.4f} MB"

"Generator peak RAM: {peak_gen / 1024 / 1024:.4f} MB"

输出：

Naive peak RAM: 25.2114 MB Generator peak RAM: 13.9610 MB

Naive

peak

RAM

25.2114

MB

Generator

13.9610

通过使用生成器，峰值内存消耗减少到几乎一半。在处理大型语言模型的多千兆字节文本数据集或为视觉模型批量处理图像时，流式传输数据可以确保内存消耗保持平稳且可预测，避免在生产环境中担心内存不足的问题。

2. 上下文管理器（硬件状态与资源管理）

不，不是那个上下文！

AI 应用程序是物理资源和状态绑定资源的重度使用者。你需要打开和关闭与向量数据库的连接，管理 PyTorch 的梯度计算，或动态分析延迟块。

如果你未能清理资源，或在设置恢复之前发生异常，你可能会导致内存泄漏或让状态变量保持在错误的配置中。上下文管理器使用 with 语句来包装执行块，确保即使抛出错误，设置和清理逻辑也能干净地运行。

在这里，我们尝试使用 try-finally 块临时将一个模拟模型设置为评估模式，跟踪其推理延迟，并手动清除 GPU 缓存。这种方法是样板代码繁重的，并作为示例使用：

import time class MockPyTorchModel: def __init__(self): self.training = True def __call__(self, x): return [val * 1.5 for val in x] # Create model model = MockPyTorchModel() # Start manual setup and execution start_time = time.perf_counter() original_mode = model.training # Manually set model to evaluation mode model.training = False try: # Perform inference outputs = model([1.0, 2.0, 3.0]) print(f"Inference outputs: {outputs}") finally: # We must explicitly clean up and restore state model.training = original_mode elapsed = time.perf_counter() - start_time print(f"[Manual Profile] Inference took {elapsed:.6f}s") print("[Manual GPU] Simulating: torch.cuda.empty_cache()")

time

class

MockPyTorchModel

__init__

self

training

True

__call__

x

val *

1.5

val

Create model

model

Start manual setup and execution

start_time

perf_counter

original_mode

Manually set model to evaluation mode

False

try

Perform inference

outputs

1.0

2.0

3.0

"Inference outputs: {outputs}"

finally

We must explicitly clean up and restore state

elapsed

-

"[Manual Profile] Inference took {elapsed:.6f}s"

"[Manual GPU] Simulating: torch.cuda.empty_cache()"

我们可以使用标准的基于类的 __enter__ 和 __exit__ 方法，将这种行为封装到一个干净且可重用的上下文管理器中：

import time

class MockPyTorchModel: def __init__(self): self.training = True

def __call__(self, x): return [val * 1.5 for val in x]

class InferenceProfiler: def __init__(self, model): self.model = model

def __enter__(self): self.start_time = time.perf_counter() self.original_mode = self.model.training # 设置模型为评估模式 self.model.training = False print("[Enter] 切换模型为评估模式，开始计时。") return self

def __exit__(self, exc_type, exc_val, exc_tb):

恢复原始训练状态

self.model.training = self.original_mode elapsed = time.perf_counter() - self.start_time print(f"[Exit] 块延迟: {elapsed:.6f} 秒") print("[Exit] 恢复训练状态。模拟CUDA缓存清理。")

返回False确保任何发生的异常不会被抑制

return False

执行变得异常干净和稳健

model = MockPyTorchModel() with InferenceProfiler(model): res = model([1.0, 2.0, 3.0]) print(f"上下文内的预测结果: {res}")

InferenceProfiler

__enter__

设置模型为评估模式

"[Enter] 切换模型为评估模式，开始计时。"

__exit__

exc_type

exc_val

exc_tb

恢复原始训练状态

"[Exit] 块延迟: {elapsed:.6f} 秒"

"[Exit] 恢复训练状态。模拟CUDA缓存清理。"

返回False确保任何发生的异常不会被抑制

执行变得异常干净和稳健

with

res

"上下文内的预测结果: {res}"

[Enter] 切换模型为评估模式，开始计时。上下文内的预测结果: [1.5, 3.0, 4.5] [Exit] 块延迟: 0.000045 秒 [Exit] 恢复训练状态。模拟CUDA缓存清理。

Enter

切换

到

评估

模式

开始

计时

预测

在

上下文

内

4.5

Exit

块

延迟

0.000045

秒

恢复

状态

模拟

CUDA

缓存

清理

通过定义 InferenceProfiler，你将错误处理和清理逻辑抽象化。无论推理成功还是中途崩溃，上下文管理器都确保模型的原始训练状态被恢复，并安全地捕获执行信息。

3. 异步编程（扩展LLM API和代理工具调用）

由于LLM驱动的应用和代理工作流程，网络输入/输出（I/O）通常是主要的延迟瓶颈。如果代理需要使用云API评估50个用户提示，或者查询远程向量存储，按顺序发送这些请求会在每次网络调用时阻塞程序。

使用asyncio进行异步编程允许Python同时处理多个任务。Python不会在等待HTTP响应时空闲等待，而是暂停当前任务并执行其他操作，从而加快多代理循环和工具执行的速度。

在这里，我们遍历提示，对每个提示进行标准的同步网络调用。程序在模拟的HTTP等待时间期间完全空闲：

import time

模拟对LLM的同步外部API调用

def query_llm_sync(prompt: str) -> str: time.sleep(0.1) # 模拟100毫秒网络延迟 return f"对 '{prompt}' 的响应"

def run_sequential(prompts): start = time.perf_counter() results = [] for p in prompts: results.append(query_llm_sync(p)) elapsed = time.perf_counter() - start print(f"顺序处理耗时 {elapsed:.4f} 秒。") return results

prompts = [f"解释主题 {i}" for i in range(20)] _ = run_sequential(prompts)

模拟对LLM的同步外部API调用

query_llm_sync

prompt

str

->

sleep

0.1

模拟 100 毫秒网络延迟

"对 '{prompt}' 的响应"

run_sequential

prompts

results

p

"顺序处理耗时 {elapsed:.4f} 秒。"

"解释主题 {i}"

顺序处理耗时 2.0864 秒。

顺序

处理

耗时

2.0864

通过使用 asyncio 和 await，我们可以同时执行所有 20 个网络任务。这与诸如 httpx 之类的生产库和如 AsyncOpenAI 这样的异步 SDK 完美匹配：

import asyncio import time # 模拟对 LLM 的异步外部 API 调用 async def query_llm_async(prompt: str) -> str: await asyncio.sleep(0.1) # 非阻塞睡眠模拟异步网络 I/O return f"对 '{prompt}' 的响应" async def run_concurrent(prompts): start = time.perf_counter() # 安排所有 LLM 调用同时执行 tasks = [query_llm_async(p) for p in prompts] results = await asyncio.gather(*tasks) elapsed = time.perf_counter() - start print(f"并发处理耗时 {elapsed:.4f} 秒。") return results # 执行异步运行器 prompts = [f"解释主题 {i}" for i in range(20)] _ = asyncio.run(run_concurrent(prompts))

asyncio

模拟对 LLM 的异步外部 API 调用

async

query_llm_async

await

非阻塞睡眠模拟异步网络 I/O

run_concurrent

安排所有 LLM 调用同时执行

tasks

gather

*

"并发处理耗时 {elapsed:.4f} 秒。"

执行异步运行器

run

并发处理耗时 0.1013 秒。

并发

0.1013

通过切换到 asyncio，我们实现了对 20 次 API 调用的约 20 倍加速。由于这些调用是并发执行的，总运行时间由最慢的单个请求决定，而不是所有请求的总和。

4. 数据类与 Pydantic（结构化配置与工具验证）

机器学习模型对配置非常敏感。超参数键中的一个拼写错误（如 learningrate 而不是 learning_rate）可能会静默地回退到默认值，使训练运行变得毫无意义。此外，现代 LLM API 使用结构化 JSON 模式来支持工具调用和结构化输出。

Python 的标准数据类提供了一种清晰的方式来定义结构化配置模板。为了运行时验证，Pydantic 扩展了这一概念，自动解析类型，强制执行约束（例如匹配范围限制），并直接导出 JSON 模式。

依赖原始字典进行超参数配置允许拼写错误和类型不匹配静默通过，从而导致数学错误或意外的训练行为：

def train_model(config: dict): # 无类型的提取，默认回退 learning_rate = config.get("learning_rate", 0.001) batch_size = config.get("batch_size", 32) optimizer = config.get("optimizer", "adam") # 类型错误：如果 batch_size 作为字符串 "64" 传递，这个数学运算会失败 num_steps = 1000 // batch_size print(f"使用 LR={learning_rate}, Batch Size={batch_size}, Steps={num_steps} 进行训练") # 拼写错误或类型错误不会立即发出警告 train_model({"learning_rate": -0.05, "batch_size": "64"})

train_model

config

dict

无类型的提取，默认回退

learning_rate

get

"learning_rate"

0.001

batch_size

"batch_size"

optimizer

"optimizer"

"adam"

类型错误：如果 batch_size 作为字符串 "64" 传递，这个数学运算会失败

num_steps

1000

// batch_size

"使用 LR={learning_rate}, Batch Size={batch_size}, Steps={num_steps} 进行训练"

拼写错误或类型错误不会立即发出警告

0.05

"64"

通过使用 Pydantic 定义配置，参数在实例化时会被解析并严格检查。这确保了在训练代码执行之前配置已经通过验证，并为 LLM 生成干净的 JSON 架构：

python

from pydantic import BaseModel, Field, ValidationError

class ModelConfig(BaseModel):
    learning_rate: float = Field(gt=0.0, lt=1.0, description="Learning rate must be between 0 and 1")
    batch_size: int = Field(gt=0, description="Batch size must be a positive integer")
    optimizer: str = Field(default="adam")  # Pydantic performs runtime type coercion (coercing string "64" to int 64)

try:
    valid_config = ModelConfig(learning_rate=0.001, batch_size="64")
    print(f"Valid configuration initialized: {valid_config}")
except ValidationError as e:
    print(f"Unexpected error: {e}")  # Catching invalid parameters instantly

try:
    invalid_config = ModelConfig(learning_rate=-0.05, batch_size=0)
except ValidationError as e:
    print("\nValidation Errors Caught:")
    print(e)  # Export schema directly for LLM Tool / Function Calling schemas

print("\nJSON Schema for LLM Tool Definition:")
print(ModelConfig.model_json_schema())

Valid configuration initialized: learning_rate=0.001 batch_size=64 optimizer='adam' Validation Errors Caught: 2 validation errors for ModelConfig learning_rate: Input should be greater than 0 [type=greater_than, input_value=-0.05, input_type=float] For further information visit https://errors.pydantic.dev/2.12/v/greater_than batch_size: Input should be greater than 0 [type=greater_than, input_value=0, input_type=int] For further information visit https://errors.pydantic.dev/2.12/v/greater_than JSON Schema for LLM Tool Definition: {'properties': {'learning_rate': {'description': 'Learning rate must be between 0 and 1', 'exclusiveMaximum': 1.0, 'exclusiveMinimum': 0.0, 'title': 'Learning Rate', 'type': 'number'}, 'batch_size': {'description': 'Batch size must be a positive integer', 'exclusiveMinimum': 0, 'title': 'Batch Size', 'type': 'integer'}, 'optimizer': {'default': 'adam', 'title': 'Optimizer', 'type': 'string'}}, 'required': ['learning_rate', 'batch_size'], 'title': 'ModelConfig', 'type': 'object'}

使用 Pydantic 可以保护运行时环境免受配置错误的影响，安全地解析原始输入，并为代理函数自动生成架构定义。

5. 魔法方法（构建自定义抽象）

自定义的训练流程和推理引擎必须能够与外部库生态系统顺畅地交互。例如，如果你构建了一个自定义的文本加载器，PyTorch 的 DataLoader 应该能够自然地对其进行索引和采样。

Python 使用双下划线（“dunder”）魔法方法来实现对象接口。通过为 __len__、__getitem__ 和 __call__ 等方法编写自定义逻辑，你可以让你的自定义 Python 类像内置列表或可执行函数一样工作。

让我们编写一个带有任意方法名称的自定义类。这个数据集不能直接传递给期望标准 Python 协议的外部库：

python

class CustomDataset:
    def __init__(self, data_list):
        self.data_list = data_list

    def fetch_index(self, i):
        return self.data_list[i]

    def count_items(self):
        return len(self.data_list)

dataset = CustomDataset(["Sample A", "Sample B", "Sample C"])

# 客户端代码被迫学习自定义 API
print(f"Items: {dataset.count_items()}, First item: {dataset.fetch_index(0)}")

# 尝试 len(dataset) 或 dataset[0] 会触发 TypeError
print(f"Dataset length: {len(dataset)}")

CustomDataset data_list fetch_index count_items dataset "Sample A" "Sample B" "Sample C"

客户端代码被迫学习自定义 API

"Items: {dataset.count_items()}, First item: {dataset.fetch_index(0)}"

尝试 len(dataset) 或 dataset[0] 会触发 TypeError

"Dataset length: {len(dataset)}" Items: 3, First item: Sample A Traceback (most recent call last): File "./testing.py", line 15, in <module> print(f"Dataset length: {len(dataset)}") ^^^^^^^^^^^ TypeError: object of type 'CustomDataset' has no len()

Items First item Sample A Traceback most recent call last File "./testing.py" < module > ^ TypeError object of 'CustomDataset' has no

通过实现 __len__ 和 __getitem__，我们让类像原生序列一样工作。通过实现 __call__，我们让自定义推理流程实例像函数一样行为：

python

class CustomDatasetPythonic:
    def __init__(self, data_list):
        self.data = data_list

    def __len__(self) -> int:
        return len(self.data)

    def __getitem__(self, idx: int):
        return self.data[idx]

class PredictionPipeline:
    def __init__(self, step_value: float):
        self.step_value = step_value

    def __call__(self, x: float) -> float:
        # 实现 __call__ 让实例像函数一样可调用
        return x * self.step_value

# 实例化协议兼容的数据集
dataset = CustomDatasetPythonic(["Sample A", "Sample B", "Sample C"])
print(f"Dataset length: {len(dataset)}")
print(f"Index access [1]: {dataset[1]}")

# 实例化可调用的流程
pipeline = PredictionPipeline(step_value=2.5)

# 直接调用对象
result = pipeline(10.0)
print(f"Pipeline call execution result: {result}")

CustomDatasetPythonic __len__ __getitem__ idx PredictionPipeline step_value

实现 call 让实例像函数一样可调用

x * step value

实例化协议兼容的数据集

"Index access [1]: {dataset[1]}"

实例化可调用的流程

pipeline 2.5

直接调用对象

result 10.0 "Pipeline call execution result: {result}" Dataset length: 3 Index access [1]: Sample B Pipeline call execution result: 25.0

length Index access B execution 25.0

在深度学习库中，养成使用调用语法（model(x)）来执行层或模型的习惯，而不是显式调用 forward 方法（model.forward(x)）。PyTorch 的基础 nn.Module 类重写了 __call__ 方法，以便在调用 forward() 之前注册并运行反向/正向钩子。直接执行 .forward() 会绕过这些钩子，导致梯度断裂或跟踪错误。

总结

从简单的笔记本转向强大的 AI 应用程序，需要使用 Python 的原生工程机制来编写高性能、可读性强且整洁的代码。

以下是关键要点：

使用生成器流式传输数据，以在处理大型数据集时保持内存使用平稳

使用上下文管理器干净地管理系统和硬件状态，以保护你的 GPU 边界

通过利用并发的 asyncio 管道来解决查询外部 API 时的网络瓶颈

使用 Pydantic 验证模型来保护配置并为 LLM 工具自动生成模式

通过实现魔术方法，将自定义抽象干净地集成到框架包中

通过以软件工程的严谨态度对待你的代码管道，你可以确保 AI 系统运行快速、安全失败，并与生产基础设施干净集成。

Python Concepts Every AI Engineer Must Master

TL;DR · AI 摘要

核心要点

结构提纲

思维导图

金句 / Highlights

每位 AI 工程师必须掌握的 Python 概念 - MachineLearningMastery.com

每位 AI 工程师必须掌握的 Python 概念

AI 工程师需要了解的内容

1. 生成器与延迟评估（内存高效的流式数据传输）

一个模拟的 JSONL 文件流，包含原始文本负载

一个简单的列表函数，一次性处理所有记录

立即处理数据并添加到列表中

运行此代码需要将所有 50,000 个处理后的字典加载到内存中

A mock JSONL file stream of raw text payloads

Naive list function processing all records at once

Process data immediately and append to a list

Running this requires loading all 50,000 processed dictionaries into RAM

一个模拟的 JSONL 文件流，包含原始文本负载

一个简单的列表函数，一次性处理所有记录

立即处理数据并添加到列表中

生成器函数，逐个生成预处理后的记录

测量简单实现的内存使用情况

测量生成器实现的内存使用情况

输出结果

Generator function yielding preprocessed records one-by-one

Measure the naive implementation

模拟训练循环步骤

测量生成器实现

输出结果

2. 上下文管理器（硬件状态与资源管理）

Create model

Start manual setup and execution

Manually set model to evaluation mode

Perform inference

We must explicitly clean up and restore state

恢复原始训练状态

返回False确保任何发生的异常不会被抑制

执行变得异常干净和稳健

设置模型为评估模式

恢复原始训练状态

返回False确保任何发生的异常不会被抑制

执行变得异常干净和稳健

3. 异步编程（扩展LLM API和代理工具调用）

模拟对LLM的同步外部API调用

模拟对LLM的同步外部API调用

模拟 100 毫秒网络延迟

模拟对 LLM 的异步外部 API 调用

非阻塞睡眠模拟异步网络 I/O

安排所有 LLM 调用同时执行

执行异步运行器

4. 数据类与 Pydantic（结构化配置与工具验证）

无类型的提取，默认回退

类型错误：如果 batch_size 作为字符串 "64" 传递，这个数学运算会失败

拼写错误或类型错误不会立即发出警告

5. 魔法方法（构建自定义抽象）

客户端代码被迫学习自定义 API

尝试 len(dataset) 或 dataset[0] 会触发 TypeError

实现 __call__ 让实例像函数一样可调用

实例化协议兼容的数据集

实例化可调用的流程

直接调用对象

总结

更多相关内容

实现 call 让实例像函数一样可调用