Appearance
Day21 Python 高阶基础（AI 工程必备）：GIL、可变对象、装饰器、生成器、asyncio 协程
📖 基础知识讲解【基础原理 + 高级深挖知识点】
基础原理
1. Day21 的定位：Python 不是「换个语法的 Java」，而是另一套运行模型
Java 技术栈切到 Python，最大的认知陷阱是把 Python 当成「语法更短的 Java」。这会导致两类线上事故：
- 以为
Thread能利用多核 → 写了一个多线程 CPU 密集任务，8 核机器只跑满 1 核； - 以为
async def加上就是异步 → 里面调了一个同步的requests.get()，整个事件循环被一个请求卡死，并发从 100 掉到 1。
这两个坑的根源是同一件事：Python 的并发模型由 GIL 决定，而 GIL 是解释器层面的设计，不是语言语法层面的。
text
Java 的心智模型 Python（CPython）的心智模型
变量 = 盒子，装着值 变量 = 便利贴，贴在对象上
int a = 1; a 这块内存里就是 1 a = 1 → a 这个名字指向 int 对象 1
多线程 = 真并行，JVM 直接映射 OS 线程 多线程 = OS 线程是真的，但要抢 GIL
8 核能跑满 8 核 CPU 运算同一时刻只有 1 个线程在跑
并发靠线程池 + synchronized IO 并发靠 asyncio（单线程事件循环）
CPU 并发靠 multiprocessing（多进程）| 维度 | Java | Python（CPython） | 切换时最容易错的点 |
|---|---|---|---|
| 变量语义 | 基本类型值语义、对象引用语义 | 全部是对象引用语义 | 以为 list 传进函数是副本 |
| 多线程并行 | 真并行，可跑满多核 | 受 GIL 限制，CPU 运算串行 | 多线程做 CPU 密集任务 |
| 异步 | CompletableFuture + 线程池 | asyncio 单线程事件循环 | 异步函数里混同步阻塞调用 |
| 默认参数 | 不支持默认参数（靠重载） | 支持，但只求值一次 | 默认参数写 [] 导致状态复用 |
| 资源释放 | try-with-resources | with 上下文管理器 | 手动 open() 忘记 close() |
| 类型 | 编译期强类型 | 运行期动态类型 + 可选注解 | 以为注解会做运行时校验 |
⚠️ AI 工程为什么必须吃透这一天：LLM 应用的典型负载是「大量 IO 等待 + 少量 CPU 计算」——一次 LLM 调用要等 2~30 秒，而本地做的事只是拼 prompt 和解析 JSON。这种负载的最优解是 asyncio 协程，不是多线程、更不是多进程。而 RAG 里的向量计算、文本切分、本地 embedding 又是 CPU 密集的，这部分要丢到进程池。一个 AI 服务里两种模型会同时存在，选错一边就是吞吐差 10 倍。
面试加分表述：「我们的 AI 网关是 FastAPI + asyncio，LLM 调用、向量库查询、Redis 这些纯 IO 走协程；文档解析和本地 reranker 这类 CPU 密集的活用
run_in_executor丢进程池，不让它们占住事件循环。单机从原来 Flask 多线程的 30 QPS 提到 400 QPS，机器数反而少了。」
2. 对象模型：变量是「名字」，不是「盒子」
这是理解后面所有坑（可变对象、默认参数、深浅拷贝、闭包）的唯一前提。Python 里 a = 1 的语义不是「把 1 放进 a 这块内存」，而是「让名字 a 指向对象 1」。
text
a = [1, 2, 3]
b = a # 不是拷贝！只是多贴一张便利贴
名字空间 堆内存里的对象
┌───────┐
│ a │ ──────┐
└───────┘ ├──────▶ ┌──────────────┐
┌───────┐ │ │ list [1,2,3] │ id=0x7f...a0
│ b │ ──────┘ │ 引用计数 = 2 │
└───────┘ └──────────────┘
b.append(4) → 对象被原地改了，a 看到的也是 [1,2,3,4]
b = [9] → 只是把 b 这张便利贴撕下来贴到新对象上，a 不变三个必须分清的操作：
| 写法 | 语义 | 对原对象的影响 |
|---|---|---|
b = a | 名字绑定（rebind） | 无影响，两个名字指同一对象 |
b.append(4) | 原地变更（in-place mutate） | 原对象被改，所有指向它的名字都受影响 |
b = b + [4] | 新建对象后重新绑定 | 原对象不变，b 指向新 list |
b += [4] | 对 list 等价于 extend，是原地变更 | 原对象被改（和上一行结果完全不同） |
python
# ✅ 正确理解：+= 对可变对象是原地变更，对不可变对象是重新绑定
a = [1, 2]
b = a
b += [3] # 原地 extend
print(a) # [1, 2, 3] ← a 也变了
a = [1, 2]
b = a
b = b + [3] # 新建 list 再绑定
print(a) # [1, 2] ← a 没变
t = (1, 2)
s = t
s += (3,) # tuple 不可变，只能新建对象
print(t) # (1, 2) ← t 没变判断「是不是同一个对象」用 is（比较 id），判断「值是否相等」用 ==：
python
a = [1, 2]
b = [1, 2]
print(a == b) # True 值相等
print(a is b) # False 不是同一个对象
print(id(a), id(b))
# ⚠️ 小整数池与字符串驻留会让 is 的结果「看起来很诡异」，别依赖它
x = 256; y = 256
print(x is y) # True 小整数 -5~256 被缓存复用
x = 257; y = 257
print(x is y) # 交互式下可能 False，脚本里因常量折叠可能 True⚠️ 永远不要用
is比较数值和字符串。is只用于和None、True、False这三个单例比较（if x is None）。小整数池、字符串驻留、常量折叠这些优化会让is的结果随 Python 版本和执行方式变化，写进业务逻辑就是定时炸弹。
3. 可变对象与不可变对象：到底「不可变」的是什么
很多人把「不可变」理解成「值不能改」，这是错的。不可变指的是「对象创建后，它内部的数据不能被原地修改」；而「名字能不能换指向」是另一回事，任何名字都能随时重新绑定。
| 类别 | 类型 | 修改时发生什么 | 能否作为 dict 的 key |
|---|---|---|---|
| 不可变 | int、float、bool、str、bytes、tuple、frozenset | 新建对象，id 变化 | ✅ 可以（__hash__ 稳定） |
| 可变 | list、dict、set、bytearray、自定义类实例（默认） | 原地修改，id 不变 | ❌ 不行（list / dict / set 不可哈希） |
python
# ✅ 不可变对象：str 的「修改」其实是新建
s = "abc"
print(id(s))
s += "d" # 等价于 s = s + "d"，新建了一个 str
print(id(s)) # id 变了，原来的 "abc" 等着被回收
# ✅ 可变对象：list 的修改是原地的
lst = [1, 2]
print(id(lst))
lst.append(3)
print(id(lst)) # id 没变关键边界：不可变容器里可以装可变对象。tuple 不可变，指的是「它的槽位里存的那几个引用不能换」，不是「它装的东西不能变」：
python
t = (1, [2, 3])
# t[1] = [9] # ❌ TypeError：不能换槽位里的引用
t[1].append(4) # ✅ 合法！list 自己是可变的
print(t) # (1, [2, 3, 4])
# 所以这个 tuple 不可哈希，不能当 dict 的 key
# hash(t) # ❌ TypeError: unhashable type: 'list'
hash((1, (2, 3))) # ✅ 全是不可变元素才能哈希最高频的线上坑：函数默认参数用可变对象。默认参数在函数定义时求值一次，之后所有调用共用同一个对象：
python
# ❌ 错误写法：所有调用共用同一个 list / dict
def add_msg(msg, history=[]):
history.append(msg)
return history
print(add_msg("第一轮")) # ['第一轮']
print(add_msg("第二轮")) # ['第一轮', '第二轮'] ← 上一次的脏数据还在！
def build_req(prompt, extra={}):
extra["prompt"] = prompt
return extra
print(build_req("A")) # {'prompt': 'A'}
print(build_req("B")) # {'prompt': 'B'} ← extra 始终是同一个 dict
# 上一次写入的 user_id 会被带到这一次python
# ✅ 正确写法：默认值写 None，函数内部新建
def add_msg(msg, history=None):
if history is None:
history = []
history.append(msg)
return history
print(add_msg("第一轮")) # ['第一轮']
print(add_msg("第二轮")) # ['第二轮'] ← 互不干扰
# 验证一下默认参数确实只求值一次
def f(x=[]):
pass
print(f.__defaults__) # ([],) ← 这个 list 挂在函数对象上，终身只有一个⚠️ 这个坑在 AI 工程里特别危险。多轮对话的
messages=[]、工具调用的tools=[]、请求头headers={}全是高危位置——一旦用了可变默认值，A 用户的对话历史会漏到 B 用户的 prompt 里，这是数据越权，不只是 bug。Day21 的 Code Review 第一条就是搜def .*=[]和def .*={}这两种写法。
面试加分表述：「默认参数是在函数定义时求值并存到
func.__defaults__上的，所以可变默认值相当于一个函数级别的全局状态。我们在 CI 里加了ruff的B006规则（flake8-bugbear），这类写法直接卡住不让合。」
4. 深浅拷贝：三层语义必须分清
拷贝问题之所以难，是因为它有三层而不是两层，面试时只答「浅拷贝拷一层、深拷贝拷全部」会被追问到卡住。
text
原数据： data = {"name": "张三", "tags": ["AI", "后端"]}
① b = data（直接赋值 / 名字绑定）
b ──┐
├──▶ 同一个 dict ──▶ 同一个 list
data┘
改 b["name"]、b["tags"].append() → data 全都变
② b = copy.copy(data)（浅拷贝）
b ──▶ 新 dict ──┐
├──▶ 同一个 list（只拷了引用）
data ──▶ 旧 dict ──┘
改 b["name"] → data 不变 ✅
改 b["tags"].append() → data 也变 ❌
③ b = copy.deepcopy(data)（深拷贝）
b ──▶ 新 dict ──▶ 新 list
data ──▶ 旧 dict ──▶ 旧 list
改任何层级 → data 都不变 ✅python
import copy
data = {"name": "张三", "tags": ["AI", "后端"], "age": 28}
# ① 赋值：同一个对象
b1 = data
b1["name"] = "李四"
print(data["name"]) # 李四 ← 被改了
data = {"name": "张三", "tags": ["AI", "后端"], "age": 28}
# ② 浅拷贝：第一层独立，嵌套层共享
b2 = copy.copy(data) # 等价写法：data.copy() / dict(data) / {**data}
b2["name"] = "李四"
b2["tags"].append("算法")
print(data["name"]) # 张三 ← 第一层没被影响 ✅
print(data["tags"]) # ['AI', '后端', '算法'] ← 嵌套层被影响 ❌
data = {"name": "张三", "tags": ["AI", "后端"], "age": 28}
# ③ 深拷贝：递归全部独立
b3 = copy.deepcopy(data)
b3["tags"].append("算法")
print(data["tags"]) # ['AI', '后端'] ← 完全隔离 ✅常见的「伪浅拷贝」写法，全都等价于 copy.copy()：
| 对象 | 浅拷贝写法 | 说明 |
|---|---|---|
list | lst[:]、list(lst)、lst.copy()、[*lst] | 四种完全等价 |
dict | d.copy()、dict(d)、{**d} | {**d} 只展开一层 |
set | s.copy()、set(s) | 元素必须不可变，所以浅拷贝通常够用 |
| 嵌套结构 | 只能 copy.deepcopy() | 手写递归容易漏掉循环引用 |
深拷贝的三个必须知道的细节（面试追问重灾区）：
- 循环引用能正确处理。
deepcopy内部维护一个memo字典（id(原对象) → 新对象），遇到已拷过的对象直接复用，所以a.append(a)这种自引用结构不会死循环。 - 不可变对象不会真拷。
deepcopy(1)、deepcopy("abc")、deepcopy((1, 2))会直接返回原对象（全不可变的tuple也走这个优化），因为没有被改坏的风险。 - 可以用
__deepcopy__/__copy__自定义行为。数据库连接、线程锁、文件句柄这些不该被拷贝的资源必须自定义，否则deepcopy会抛TypeError: cannot pickle ...或拷出一个废掉的连接。
python
import copy
# ① 循环引用：deepcopy 靠 memo 表安全处理
a = [1, 2]
a.append(a) # 自引用
b = copy.deepcopy(a)
print(b[2] is b) # True ← 新结构里也是自引用，没死循环
# ② 不可变对象直接返回原对象
t = (1, 2, 3)
print(copy.deepcopy(t) is t) # True
# ③ 自定义：不该被拷的资源要显式排除
class LLMClient:
def __init__(self, api_key, session):
self.api_key = api_key
self.session = session # 假设是 aiohttp 连接池，不能拷
def __deepcopy__(self, memo):
new = LLMClient(copy.deepcopy(self.api_key, memo), self.session)
return new # session 共享，不深拷⚠️ AI 工程里
deepcopy是性能杀手。RAG 场景一个 chunk 列表可能有几万个 dict，deepcopy要递归遍历所有对象、维护memo表，比浅拷贝慢 1~2 个数量级。实践原则：能不改原数据就别拷；必须隔离时优先用「不可变数据 + 重建」而不是deepcopy（比如new_msgs = [*msgs, new_one]）。
面试加分表述：「我们规定对话上下文
messages在链路里只读，任何需要改写的地方都返回新列表，不做深拷贝也不原地改。这样既避免了deepcopy的开销，又天然规避了多轮对话串数据的问题。」
5. GIL 全局解释器锁：它到底锁住了什么
GIL（Global Interpreter Lock）是CPython 解释器的一把进程级互斥锁。一句话定义：同一个进程内，任意时刻只有一个线程能持有 GIL 并执行 Python 字节码。
先把容易混淆的边界拆清楚，这是面试区分度最高的地方：
| 说法 | 对不对 | 真实情况 |
|---|---|---|
| 「GIL 是 Python 语言的特性」 | ❌ | 是 CPython 实现的特性。Jython、IronPython 没有 GIL；PyPy 有；3.13 起 CPython 提供可选的 no-GIL 构建 |
| 「有 GIL 所以 Python 没有真线程」 | ❌ | 线程是真的 OS 线程，能被操作系统调度，只是执行字节码要抢 GIL |
| 「有 GIL 所以多线程完全没用」 | ❌ | IO 密集场景非常有用，等 IO 时 GIL 会被释放 |
| 「有 GIL 所以多线程天然线程安全」 | ❌ | 最危险的误解。GIL 保证的是字节码级别的原子性，不是业务操作级别的 |
| 「GIL 让 CPU 密集型多线程跑不满多核」 | ✅ | 这才是 GIL 真正的限制 |
text
GIL 的实际执行时序（2 个线程，单核视角）
CPU 密集任务（纯计算，GIL 争抢激烈）
线程A │████──────│████──────│████──────│ 实际只有一个在算
线程B │──────████│──────████│──────████│ 总耗时 ≈ 串行，还多了切换开销
└─ 持有 GIL ─┘ 释放 ─┘
IO 密集任务（GIL 在等待时主动释放）
线程A │██░░░░░░░░░░░░░░██│ ██=执行字节码 ░░=等 IO（已释放 GIL）
线程B │──██░░░░░░░░░░░░░░│
线程C │────██░░░░░░░░░░░░│ 三个线程的 IO 等待时间完全重叠 ✅
总耗时 ≈ max(单个 IO 耗时)，不是 sumGIL 在什么时候会被释放（这是「IO 密集型适合多线程」的技术根据）：
- 执行阻塞式 IO 时：
socket.recv、file.read、time.sleep、数据库驱动等待响应——CPython 在调用系统调用前后包了一对Py_BEGIN_ALLOW_THREADS/Py_END_ALLOW_THREADS； - 执行时间超过切换间隔（默认 5ms，
sys.getswitchinterval()）时被强制让出； - 进入释放了 GIL 的 C 扩展时：NumPy 的大矩阵运算、
hashlib、部分压缩/加解密库，底层循环里会主动放开 GIL； - 线程结束时。
python
import sys, time, threading
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
print(sys.getswitchinterval()) # 0.005 ← 5ms 的 GIL 切换间隔
def cpu_task(n=10_000_000): # CPU 密集：纯累加
s = 0
for i in range(n):
s += i
return s
def io_task(): # IO 密集：模拟网络等待
time.sleep(1)
return "ok"
# ❌ CPU 密集用多线程：几乎没有加速，甚至更慢（多了 GIL 争抢）
t0 = time.perf_counter()
with ThreadPoolExecutor(4) as ex:
list(ex.map(cpu_task, [10_000_000] * 4))
print(f"CPU 密集 + 4 线程：{time.perf_counter() - t0:.2f}s") # ≈ 串行耗时
# ✅ CPU 密集用多进程：接近线性加速（每个进程有独立 GIL）
t0 = time.perf_counter()
with ProcessPoolExecutor(4) as ex:
list(ex.map(cpu_task, [10_000_000] * 4))
print(f"CPU 密集 + 4 进程：{time.perf_counter() - t0:.2f}s") # ≈ 串行 / 核数
# ✅ IO 密集用多线程：4 个 1 秒的任务总共约 1 秒
t0 = time.perf_counter()
with ThreadPoolExecutor(4) as ex:
list(ex.map(lambda _: io_task(), range(4)))
print(f"IO 密集 + 4 线程：{time.perf_counter() - t0:.2f}s") # ≈ 1.0sGIL 不等于线程安全，这一点必须单独强调。GIL 的粒度是字节码指令，而一个业务操作往往要好几条字节码，中间可以被切走：
python
import threading, dis
counter = 0
def unsafe_incr():
global counter
for _ in range(100_000):
counter += 1 # 看起来一行，实际是 4 条字节码
# 看一下它被拆成了什么
dis.dis(lambda: counter + 1)
# LOAD_GLOBAL counter ← 取值
# LOAD_CONST 1
# BINARY_OP + ← 相加（这里可能被切走！）
# ... ← 回写
threads = [threading.Thread(target=unsafe_incr) for _ in range(8)]
[t.start() for t in threads]
[t.join() for t in threads]
print(counter) # ❌ 远小于 800000，发生了丢失更新python
import threading
counter = 0
lock = threading.Lock()
def safe_incr():
global counter
for _ in range(100_000):
with lock: # ✅ 业务级别的互斥，自己加锁
counter += 1
threads = [threading.Thread(target=safe_incr) for _ in range(8)]
[t.start() for t in threads]
[t.join() for t in threads]
print(counter) # 800000 ✅⚠️ 「GIL 能不能去掉」的标准答法不是「不能」，而是「代价在哪」。阻碍主要有三点：① 海量 C 扩展（NumPy、Pandas、各种数据库驱动）的 API 假设了 GIL 存在，去掉要全部改造；② 去掉 GIL 要给每个对象加细粒度锁或改引用计数为原子操作，单线程性能会下降（早期尝试慢了 30%~40%）；③ 引用计数本身是非线程安全的，要换成带偏向/原子的方案。CPython 3.13 通过 PEP 703 提供了
--disable-gil的自由线程（free-threaded）构建，3.13/3.14 仍是实验/逐步成熟阶段，生态适配还在进行中。
面试加分表述：「我回答 GIL 会分三层：锁的是什么（同一进程内执行字节码的权限）、什么时候释放（阻塞 IO、5ms 切换间隔、放开 GIL 的 C 扩展）、对我有什么影响（CPU 密集走多进程，IO 密集走协程，并且 GIL 不代替业务加锁）。最后补一句 3.13 的 PEP 703 free-threaded 构建，说明我在跟进这块的演进。」
6. 进程 / 线程 / 协程：三选一决策表
这是 Day21 最实用的一张表，面试几乎必问，项目里也天天用。
| 维度 | 多进程 multiprocessing | 多线程 threading | 协程 asyncio |
|---|---|---|---|
| 调度者 | 操作系统 | 操作系统 | 事件循环（用户态） |
| 切换开销 | 最大（上下文 + 页表） | 中（内核态上下文切换，约 μs 级） | 最小（函数栈帧切换，约 ns~百 ns 级） |
| 内存 | 每进程独立，占用大 | 共享进程内存，几 MB 栈 | 共享，单个协程仅 KB 级 |
| 能否跑满多核 | ✅ 能 | ❌ 受 GIL 限制 | ❌ 单线程 |
| 数据共享 | 需 IPC（Queue、Pipe、共享内存） | 直接共享，需加锁 | 直接共享，单线程内无需锁 |
| 并发规模 | 几个~几十个（≈ 核数） | 几十~几百（线程栈吃内存） | 几千~几万 |
| 崩溃隔离 | ✅ 一个进程挂了不影响别的 | ❌ 段错误会带走整个进程 | ❌ 同上 |
| 调试难度 | 中（跨进程栈难追） | 高（竞态、死锁） | 中（栈是完整的，但要防阻塞） |
| 适用场景 | CPU 密集：向量计算、文本切分、图像处理、模型推理 | 阻塞式 IO 且库不支持 async（老 SDK、pymysql） | 高并发 IO：LLM 调用、HTTP、向量库、Redis |
text
决策流程图（照着问三个问题就能定）
① 这个任务是在「算」还是在「等」？
├─ 在算（CPU 占满）──────▶ 多进程（ProcessPoolExecutor）
│ 进程数 ≈ CPU 核数，不要更多
└─ 在等（IO 阻塞）
│
② 这个 IO 操作有 async 版本的库吗？
├─ 有（httpx / aiomysql / redis.asyncio）──▶ asyncio 协程 ✅ 首选
└─ 没有（只能用同步 SDK）
│
③ 我的主体代码是不是异步的？
├─ 是 ──▶ asyncio + run_in_executor(线程池) 包一层
└─ 不是 ─▶ 多线程 ThreadPoolExecutorpython
import asyncio, time
from concurrent.futures import ProcessPoolExecutor
# 场景：AI 服务里 CPU 密集 + IO 密集混合，一个请求要做两件事
def split_document(text: str) -> list[str]:
"""CPU 密集：文本切分 + 本地分词，纯计算"""
time.sleep(0.5) # 用 sleep 模拟 CPU 耗时
return [text[i:i + 100] for i in range(0, len(text), 100)]
async def call_embedding(chunk: str) -> list[float]:
"""IO 密集：调远端 embedding 接口"""
await asyncio.sleep(0.3)
return [0.1] * 1536
# ✅ 正确组合：CPU 的活丢进程池，IO 的活用协程并发
async def ingest(text: str, pool: ProcessPoolExecutor):
loop = asyncio.get_running_loop()
chunks = await loop.run_in_executor(pool, split_document, text) # 不占事件循环
vectors = await asyncio.gather(*(call_embedding(c) for c in chunks))
return len(vectors)
async def main():
with ProcessPoolExecutor(max_workers=4) as pool:
texts = ["x" * 1000 for _ in range(4)]
t0 = time.perf_counter()
res = await asyncio.gather(*(ingest(t, pool) for t in texts))
print(res, f"{time.perf_counter() - t0:.2f}s")
asyncio.run(main())⚠️ 进程池的
max_workers不要照搬线程池的经验值。线程池可以开几十上百，进程池超过 CPU 核数后只会增加上下文切换和内存，吞吐不升反降。而且每次run_in_executor传给进程池的参数都要序列化（pickle），大对象（比如几十 MB 的 DataFrame、模型对象）走进程池反而比串行慢——这种情况要么用共享内存，要么改成在子进程里重新加载。
面试加分表述：「我选型只问三句：在算还是在等；有没有 async 版本的库；主体代码是不是异步的。按这三句走下来结论是唯一的，不会出现『该用协程的地方开了 200 个线程』这种事。」
7. 函数是一等公民：闭包与 LEGB 作用域
装饰器能存在的前提是「函数可以被当成值传递和返回」。这一节是装饰器的地基。
python
def outer(prefix):
count = 0 # 自由变量，被内层函数捕获
def inner(msg):
nonlocal count # 声明要修改外层函数的变量
count += 1
return f"[{prefix}#{count}] {msg}"
return inner # 返回函数对象本身，不加括号
log = outer("LLM")
print(log("第一次调用")) # [LLM#1] 第一次调用
print(log("第二次调用")) # [LLM#2] 第二次调用 ← 状态被闭包保住了
# 闭包捕获的变量存在 __closure__ 里
print(log.__closure__) # (cell, cell)
print([c.cell_contents for c in log.__closure__]) # [2, 'LLM']
print(log.__code__.co_freevars) # ('count', 'prefix')LEGB 查找顺序：Python 查一个名字时按 Local → Enclosing → Global → Builtin 依次找，找到就停。
text
Builtin 内置（len、print、Exception……）
▲
Global 模块级（文件最外层定义的变量）
▲
Enclosing 外层函数的局部变量（闭包捕获的来源）
▲
Local 当前函数的局部变量 ← 从这里开始找
写操作的默认行为：写入 Local，除非用 global / nonlocal 声明| 关键字 | 作用 | 典型场景 |
|---|---|---|
| 不声明 | 读走 LEGB；写会创建新的 Local 变量 | 大多数情况 |
global x | 读写都指向模块级变量 | 修改模块级配置/计数器（尽量避免） |
nonlocal x | 读写都指向最近的外层函数的变量 | 闭包里累加计数、缓存状态 |
python
# ❌ 典型错误：只读不声明没问题，一写就报 UnboundLocalError
total = 0
def bad():
total += 1 # ❌ UnboundLocalError
# 因为有赋值，total 被判定为 Local，
# 但读的时候它还没被赋值
# bad()
# ✅ 写法一：显式 global（能用但不推荐，隐式全局状态难测）
def ok_global():
global total
total += 1
# ✅ 写法二：闭包 + nonlocal（状态被封在函数内，推荐）
def make_counter():
n = 0
def incr():
nonlocal n
n += 1
return n
return incr闭包最经典的坑：延迟绑定（late binding）。闭包捕获的是变量本身，不是定义时刻的值：
python
# ❌ 错误：三个函数共享同一个 i，循环结束后 i == 2
funcs = []
for i in range(3):
funcs.append(lambda: i)
print([f() for f in funcs]) # [2, 2, 2] ← 不是 [0, 1, 2]！
# ✅ 解法一：用默认参数在定义时「快照」当前值
funcs = []
for i in range(3):
funcs.append(lambda i=i: i)
print([f() for f in funcs]) # [0, 1, 2]
# ✅ 解法二：用 functools.partial 绑定
from functools import partial
funcs = [partial(lambda x: x, i) for i in range(3)]
print([f() for f in funcs]) # [0, 1, 2]
# ✅ 解法三：多包一层工厂函数，让每个 i 落在独立作用域
def make(i):
return lambda: i
funcs = [make(i) for i in range(3)]
print([f() for f in funcs]) # [0, 1, 2]⚠️ 延迟绑定在 AI 工程里的真实事故：批量注册 Function Calling 工具时，用
for tool in tools: register(name, lambda args: call(tool, args))注册了 20 个工具，结果每个工具调用的都是最后那一个。排查了半天才发现是闭包捕获了循环变量。标准写法是lambda args, tool=tool: call(tool, args)。
面试加分表述：「闭包的本质是函数对象上挂了
__closure__这组 cell，捕获的是变量的 cell 而不是值。所以循环里造闭包必须用默认参数或partial做值快照——我们注册 LLM 工具函数时就踩过这个坑，现在 code review 看到循环体里出现lambda都会多看一眼。」
8. 装饰器 Decorator：原理、执行时机与 wraps
装饰器就是一个接收函数、返回函数的高阶函数，@ 只是语法糖。它解决的是 Java 里 AOP 解决的问题：在不改原函数源码的前提下横向加能力（计时、日志、重试、限流、鉴权、缓存）。
python
from functools import wraps
import time
def timer(func): # ① 装饰器本体，入参是被装饰的函数
@wraps(func) # ③ 拷贝元信息
def wrapper(*args, **kwargs): # ② 包装函数，签名要能接住任意参数
start = time.perf_counter()
try:
return func(*args, **kwargs)
finally:
cost = time.perf_counter() - start
print(f"{func.__qualname__} 耗时 {cost * 1000:.1f}ms")
return wrapper
@timer
def llm_call(prompt: str) -> str:
"""调用大模型"""
time.sleep(1)
return f"回答：{prompt}"
# @timer 完全等价于： llm_call = timer(llm_call)
print(llm_call("你好"))执行时机是最高频的追问点。装饰器在模块导入、函数定义的那一刻就执行一次，而不是调用时：
python
def deco(func):
print(f"① 装饰器执行了：{func.__name__}") # 定义时打印
def wrapper(*args, **kwargs):
print("② 包装逻辑执行了") # 每次调用时打印
return func(*args, **kwargs)
return wrapper
@deco
def foo():
print("③ 原函数执行了")
print("--- 以上是 import 阶段 ---")
foo()
foo()
# 输出：
# ① 装饰器执行了：foo
# --- 以上是 import 阶段 ---
# ② 包装逻辑执行了
# ③ 原函数执行了
# ② 包装逻辑执行了
# ③ 原函数执行了text
时间轴
模块 import
│
├─ 执行 def deco(...) 定义装饰器
├─ 执行 def foo(...) 先创建原函数对象
├─ 执行 foo = deco(foo) ★ 装饰器在这里跑，只跑一次
│ 此时「① 装饰器执行了」被打印
▼
运行期
├─ foo() → 实际调用的是 wrapper → 内部再调原 foo
└─ foo() → 同上，wrapper 每次都跑，deco 不再跑functools.wraps 到底做了什么：装饰后 foo 这个名字指向的是 wrapper，所以 __name__、__doc__ 等元信息全变成 wrapper 的。wraps 就是把原函数的元信息拷到 wrapper 上，并把原函数挂在 __wrapped__。
python
from functools import wraps
def no_wraps(func):
def wrapper(*a, **kw):
return func(*a, **kw)
return wrapper
def with_wraps(func):
@wraps(func)
def wrapper(*a, **kw):
return func(*a, **kw)
return wrapper
@no_wraps
def a(x: int) -> str:
"""原始文档"""
@with_wraps
def b(x: int) -> str:
"""原始文档"""
# ❌ 不加 wraps：元信息全丢
print(a.__name__, a.__doc__) # wrapper None
# ✅ 加了 wraps：元信息保留，还能通过 __wrapped__ 拿到原函数
print(b.__name__, b.__doc__) # b 原始文档
print(b.__wrapped__) # <function b at 0x...>
import inspect
print(inspect.signature(a)) # (*a, **kw) ← 签名坏了
print(inspect.signature(b)) # (x: int) -> str ← 签名正确wraps 具体拷贝这些属性：
| 属性 | 作用 | 丢了会怎样 |
|---|---|---|
__name__ / __qualname__ | 函数名 | 日志、监控埋点里全是 wrapper，无法定位 |
__doc__ | 文档字符串 | Swagger / FastAPI 自动文档变空 |
__module__ | 所属模块 | 异常栈、反射定位失效 |
__dict__ | 自定义属性 | 其他框架挂在函数上的标记丢失 |
__annotations__ | 类型注解 | FastAPI / Pydantic 无法解析参数，请求直接报错 |
__wrapped__ | 指回原函数 | inspect.signature 拿不到真实签名 |
⚠️ 不加
wraps在 FastAPI 项目里是致命的。FastAPI 靠__annotations__和签名做依赖注入与参数校验，装饰器一旦吞掉注解，接口会表现为「参数永远收不到」或启动直接报错。同理，Celery、Click、pytest 的 fixture 都依赖函数元信息。写装饰器时@wraps不是可选项，是必写项。
面试加分表述：「我答装饰器会强调两个点：第一，它在定义时执行一次，所以可以在装饰器里做注册、校验这类一次性的事，比如把工具函数注册进 Function Calling 的 schema 表；第二，
wraps不只是为了好看，FastAPI 这类框架靠签名和注解工作，不拷元信息直接起不来。」
9. 生成器 Generator：yield 的惰性求值
只要函数体里出现 yield，这个函数就不再是普通函数，调用它不执行任何代码，而是返回一个生成器对象。真正的执行发生在每次 next() 时，执行到 yield 就暂停并交出一个值。
python
def gen():
print("开始")
yield 1
print("中间")
yield 2
print("结束")
g = gen() # ← 什么都没打印！只创建了生成器对象
print(type(g)) # <class 'generator'>
print(next(g)) # 打印「开始」，返回 1，函数暂停在第一个 yield
print(next(g)) # 打印「中间」，返回 2
# print(next(g)) # 打印「结束」，然后抛 StopIterationtext
生成器的状态机
created ──next()──▶ running ──遇到 yield──▶ suspended
▲ │
└──────── next() ────────┘
│
函数 return / 执行完 ─────▶ closed
│
抛 StopIteration惰性求值的价值：内存从 O(n) 降到 O(1)。这是读取超大文件、流式处理的唯一正确方式：
python
import sys
# ❌ 一次性加载：10GB 文件直接把内存打爆
def read_all_bad(path):
with open(path, encoding="utf-8") as f:
return f.readlines() # 整个文件进内存
# ✅ 生成器逐行：内存只占一行
def read_lines(path):
with open(path, encoding="utf-8") as f:
for line in f: # 文件对象本身就是迭代器
line = line.strip()
if line: # 顺手过滤空行
yield line
# 内存对比：列表推导 vs 生成器表达式
lst = [i * i for i in range(1_000_000)] # 列表推导：立刻算完
genexp = (i * i for i in range(1_000_000)) # 生成器表达式：只存状态
print(sys.getsizeof(lst)) # 约 8 MB
print(sys.getsizeof(genexp)) # 约 200 字节 ← 差 4 万倍生成器的三个高阶能力，面试区分度很高：
| 方法 | 作用 | 典型用途 |
|---|---|---|
next(g) / g.send(None) | 推进到下一个 yield | 普通迭代 |
g.send(value) | 推进的同时把值送回 yield 表达式 | 协程式的双向通信（旧式协程基础） |
g.throw(exc) | 在 yield 暂停处抛异常 | 外部通知生成器出错 |
g.close() | 在暂停处抛 GeneratorExit，触发清理 | 提前终止流式输出，释放连接 |
python
# send：yield 既是出口也是入口
def accumulator():
total = 0
while True:
x = yield total # yield 出 total，同时接收 send 进来的 x
if x is None:
break
total += x
acc = accumulator()
print(next(acc)) # 0 ← 必须先启动到第一个 yield
print(acc.send(10)) # 10
print(acc.send(5)) # 15
# close + finally：流式输出提前中断时的资源清理
def stream_tokens(prompt):
conn = f"连接已建立({prompt})"
print(conn)
try:
for i in range(1000):
yield f"token{i}"
except GeneratorExit:
print("消费方提前断开，走清理分支")
raise
finally:
print("连接已关闭") # ✅ 无论正常结束还是提前中断都会执行
s = stream_tokens("你好")
print(next(s), next(s))
s.close() # 打印「消费方提前断开」+「连接已关闭」return 在生成器里的含义：return value 不是返回值，而是结束迭代并把 value 塞进 StopIteration.value：
python
def gen_with_return():
yield 1
yield 2
return "统计结果" # 不会被 for 循环看到
g = gen_with_return()
try:
while True:
print(next(g))
except StopIteration as e:
print("return 的值在这里：", e.value) # 统计结果
# for 循环会吞掉 StopIteration，所以拿不到 return 值
for x in gen_with_return():
print(x) # 只打印 1、2⚠️ 生成器只能消费一次。遍历完就是
closed状态，再for一遍什么都没有。这个坑在 RAG 流水线里非常常见：chunks = split(doc)返回生成器，第一处用来算 token 数、第二处用来存库，结果存进去的是空的。判断准则：如果同一份数据要用两次，就在拿到的地方立刻list(...)物化，或者改成每次重新调用生成器函数。
面试加分表述：「我们 LLM 流式输出整条链路都是生成器/异步生成器：模型 SSE → 解析 → 敏感词过滤 → 下发前端，每层都是
yield，内存里只有当前这个 token。而且在finally里关连接，用户中途关页面时GeneratorExit会一路传上去，模型侧的连接能及时释放，不会泄漏。」
10. 迭代器协议：可迭代对象 ≠ 迭代器
这组概念是「生成器为什么能用在 for 里」的答案，也是面试里用来区分「背过」和「懂了」的分水岭。
| 概念 | 判定标准 | 能否 for 循环 | 能否 next() | 能否重复遍历 |
|---|---|---|---|---|
| 可迭代对象 Iterable | 实现 __iter__ | ✅ | ❌ | 看实现（list 可以） |
| 迭代器 Iterator | 同时实现 __iter__ 和 __next__ | ✅ | ✅ | ❌ 一次性 |
| 生成器 Generator | 迭代器的一种，由 yield 或生成器表达式产生 | ✅ | ✅ | ❌ 一次性 |
text
for x in obj 实际做了什么
it = iter(obj) ← 调 obj.__iter__()，拿到迭代器
while True:
try:
x = next(it) ← 调 it.__next__()
except StopIteration:
break
...循环体...
list 是可迭代对象但不是迭代器：
每次 iter(list) 都返回一个「新的」list_iterator，所以能重复 for
生成器自己就是迭代器：
iter(g) is g → True，所以只能遍历一次python
lst = [1, 2, 3]
print(hasattr(lst, "__iter__"), hasattr(lst, "__next__")) # True False
# next(lst) # ❌ TypeError: not an iterator
it = iter(lst)
print(hasattr(it, "__iter__"), hasattr(it, "__next__")) # True True
print(next(it)) # 1
g = (i for i in range(3))
print(iter(g) is g) # True ← 生成器的 __iter__ 返回自己
# ✅ list 可重复遍历
print(sum(lst), sum(lst)) # 6 6
# ❌ 生成器只能遍历一次
print(sum(g), sum(g)) # 3 0 ← 第二次已经空了手写一个迭代器类，用来理解协议（实际项目优先用生成器，代码少 80%）：
python
# 写法一：实现迭代器协议（啰嗦，但能看清协议要求）
class Countdown:
def __init__(self, n):
self.n = n
def __iter__(self):
return self # 自己就是迭代器
def __next__(self):
if self.n <= 0:
raise StopIteration # ✅ 必须抛这个异常来终止
self.n -= 1
return self.n + 1
print(list(Countdown(3))) # [3, 2, 1]
# 写法二：可迭代对象 + 独立迭代器（支持重复遍历，推荐给需要多次遍历的数据源）
class ChunkStore:
def __init__(self, path):
self.path = path
def __iter__(self): # 每次 for 都新开一个生成器
with open(self.path, encoding="utf-8") as f:
for line in f:
yield line.strip()
# store = ChunkStore("big.txt")
# print(sum(1 for _ in store)) # ✅ 可以遍历多次
# print(sum(1 for _ in store)) # ✅ 每次都重新打开文件⚠️ 「生成器只能遍历一次」的最佳工程解法不是
list()，而是把它包成可迭代对象（上面的写法二）。list()会把惰性优势全部抵消；而实现__iter__返回生成器的类，既能重复遍历，又每次都是惰性的。RAG 的 chunk 数据源、日志回放、训练样本加载都适合这个模式。
面试加分表述：「我会用一句话区分：可迭代对象是能产出迭代器的东西，迭代器是能自己往前走并且会耗尽的东西。
iter(g) is g返回 True 就说明生成器是迭代器，这也直接解释了它为什么不能遍历两次。」
11. asyncio 异步协程：事件循环是怎么跑起来的
asyncio 的全部能力建立在一个简单机制上：单线程里有一个事件循环，它不停地从就绪队列里取任务执行；任务一旦要等 IO，就把自己挂起并把控制权交回循环，循环立刻去跑下一个任务。
text
事件循环（Event Loop）的一轮迭代
┌───────────────────────────────────────────────────────────┐
│ ① 取出 ready 队列里的所有回调，依次执行 │
│ 协程跑到 await 且要等 IO → 注册到 selector，挂起 │
│ 协程跑到 await 但结果已就绪 → 直接继续往下跑 │
│ ② 计算最近的定时器到期时间 timeout │
│ ③ selector.select(timeout) ← 唯一「真正阻塞」的地方 │
│ 操作系统通知哪些 fd 可读/可写 │
│ ④ 把就绪 fd 对应的回调塞进 ready 队列 │
│ ⑤ 把到期的 call_later 定时器回调塞进 ready 队列 │
└───────────────────────────────────────────────────────────┘
▲ │
└───────────────── 下一轮 ───────────────┘
★ 关键推论：如果 ① 里某个协程在做同步阻塞操作（time.sleep、requests.get），
循环永远走不到 ③，所有其他任务一起卡死。这就是「异步里混同步」的灾难。三个必须分清的概念：
| 概念 | 是什么 | 创建方式 | 会不会立刻开始执行 |
|---|---|---|---|
| 协程函数 coroutine function | 用 async def 定义的函数 | async def f(): ... | 不涉及 |
| 协程对象 coroutine | 调用协程函数得到的对象 | c = f() | ❌ 不会，必须被 await 或包成 Task |
| Task | 被事件循环接管调度的协程 | asyncio.create_task(c) | ✅ 会，在下一轮循环开始跑 |
| Future | 表示「将来会有结果」的底层占位对象 | loop.create_future() | 由别人 set_result |
python
import asyncio, time
async def work(name, sec):
print(f"{name} 开始")
await asyncio.sleep(sec) # ✅ 挂起点，交出控制权
print(f"{name} 完成")
return name
# ❌ 误区一：调用协程函数不会执行，只是拿到一个协程对象
async def wrong1():
c = work("A", 1) # 什么都不会发生
print(type(c)) # <class 'coroutine'>
# 如果到这里就 return，Python 会警告：coroutine was never awaited
await c # 必须 await
# ❌ 误区二：挨个 await，等于串行，完全没有并发
async def wrong2():
t0 = time.perf_counter()
await work("A", 1)
await work("B", 1)
print(f"串行耗时 {time.perf_counter() - t0:.2f}s") # ≈ 2.00s
# ✅ 正确：先把协程变成 Task 让循环接管，再一起等
async def right1():
t0 = time.perf_counter()
ta = asyncio.create_task(work("A", 1)) # 立刻排进事件循环
tb = asyncio.create_task(work("B", 1))
await ta
await tb
print(f"并发耗时 {time.perf_counter() - t0:.2f}s") # ≈ 1.00s
# ✅ 更常用：gather 自动把协程包成 Task
async def right2():
t0 = time.perf_counter()
res = await asyncio.gather(work("A", 1), work("B", 1))
print(res, f"{time.perf_counter() - t0:.2f}s") # ≈ 1.00s
asyncio.run(right2())协程为什么比线程轻：
| 对比项 | 线程切换 | 协程切换 |
|---|---|---|
| 切换主体 | 操作系统内核 | 事件循环（纯 Python/C 层） |
| 需要进内核态 | ✅ 需要 | ❌ 不需要 |
| 保存的现场 | 寄存器 + 内核栈 + 用户栈 | 一个栈帧对象 |
| 单次开销 | 约 1~10 μs | 约 0.1~1 μs |
| 单位内存 | 默认 8MB 虚拟栈（实际数十 KB~MB） | 几百 B ~ 几 KB |
| 并发上限 | 几百（再多就被切换开销吃掉） | 几万 |
| 要不要加锁 | ✅ 要，随时可能被抢占 | ❌ 单线程内，两个 await 之间是原子的 |
⚠️ 「协程不用加锁」只在「两个
await之间」成立。只要你的临界区里包含await，就可能被切走，照样有竞态：
python
import asyncio
balance = 100
# ❌ 临界区里有 await，依然会出现竞态
async def bad_withdraw(n):
global balance
if balance >= n: # 检查
await asyncio.sleep(0) # ← 这里被切走了，别人也通过了检查
balance -= n # 扣减
# ✅ 用 asyncio.Lock 保护跨 await 的临界区
lock = asyncio.Lock()
async def good_withdraw(n):
global balance
async with lock: # 注意是 async with，不是 with
if balance >= n:
await asyncio.sleep(0)
balance -= n面试加分表述：「我讲 asyncio 一定从事件循环的那一轮迭代讲起：取回调、算超时、
select等 fd、把就绪的塞回队列。这样能顺带解释三件事——为什么同步阻塞会卡死全局、为什么协程切换比线程便宜、为什么只有跨await的临界区才需要asyncio.Lock。」
12. await 到底在等什么：可等待对象的三种形态
await 后面只能跟可等待对象（awaitable），一共三类。搞不清这点就会写出 await requests.get(...) 这种必然报错的代码。
| 可等待对象 | 说明 | 常见来源 |
|---|---|---|
| 协程对象 | async def 函数的调用结果 | 自己写的 async def |
| Task | 被循环调度的协程封装 | asyncio.create_task() |
| Future | 底层结果占位符 | loop.run_in_executor()、第三方库内部 |
| 自定义对象 | 实现了 __await__ 的类 | 框架内部用，业务少见 |
python
import asyncio, requests
async def demo():
# ✅ 合法：协程对象
await asyncio.sleep(0.1)
# ✅ 合法：Task
t = asyncio.create_task(asyncio.sleep(0.1))
await t
# ✅ 合法：run_in_executor 返回 Future
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, lambda: sum(range(100)))
# ❌ 非法：普通函数的返回值不是 awaitable
# await requests.get("https://example.com")
# TypeError: object Response can't be used in 'await' expression
# ❌ 更隐蔽的错误：函数本身是同步的，await 一个同步函数的返回值
def sync_fn():
return 1
# await sync_fn() # TypeError: object int can't be used in 'await'同步阻塞混进异步代码的后果与救法，这是 AI 工程里最常见的性能事故：
python
import asyncio, time
# ❌ 事故现场：async def 的壳子 + 同步阻塞的芯
async def fake_async(i):
time.sleep(1) # 同步阻塞！事件循环被整个卡住
return i
async def bad():
t0 = time.perf_counter()
await asyncio.gather(*(fake_async(i) for i in range(5)))
print(f"假异步：{time.perf_counter() - t0:.2f}s") # ≈ 5.00s 完全串行
# ✅ 救法一：换用原生异步库（首选）
# requests → httpx.AsyncClient / aiohttp
# pymysql → aiomysql / asyncpg
# redis → redis.asyncio
# time.sleep → asyncio.sleep
async def good1(i):
await asyncio.sleep(1)
return i
# ✅ 救法二：实在只有同步 SDK，用 to_thread 丢线程池（IO 密集）
async def good2(i):
return await asyncio.to_thread(blocking_sdk_call, i) # Python 3.9+
def blocking_sdk_call(i):
time.sleep(1)
return i
# ✅ 救法三：CPU 密集的同步函数，丢进程池
from concurrent.futures import ProcessPoolExecutor
async def good3(data, pool):
loop = asyncio.get_running_loop()
return await loop.run_in_executor(pool, cpu_heavy, data)
def cpu_heavy(data):
return sum(i * i for i in range(10_000_000))
async def main():
t0 = time.perf_counter()
await asyncio.gather(*(good1(i) for i in range(5)))
print(f"真异步：{time.perf_counter() - t0:.2f}s") # ≈ 1.00s
t0 = time.perf_counter()
await asyncio.gather(*(good2(i) for i in range(5)))
print(f"to_thread：{time.perf_counter() - t0:.2f}s") # ≈ 1.00s
asyncio.run(main())怎么把「谁在卡事件循环」查出来：开启 debug 模式，asyncio 会对执行超过阈值的回调打警告。
python
import asyncio, logging
logging.basicConfig(level=logging.WARNING)
async def main():
loop = asyncio.get_running_loop()
loop.set_debug(True)
loop.slow_callback_duration = 0.1 # 超过 100ms 就告警
import time
time.sleep(0.5) # 故意阻塞
await asyncio.sleep(0)
asyncio.run(main(), debug=True)
# WARNING:asyncio:Executing <Task ...> took 0.501 seconds
# ← 直接告诉你哪个 Task、卡了多久| 排查手段 | 命令 / 配置 | 能看到什么 |
|---|---|---|
| asyncio debug 模式 | asyncio.run(main(), debug=True) | 慢回调告警、未 await 的协程、Task 异常被吞 |
| 慢回调阈值 | loop.slow_callback_duration = 0.1 | 把阈值调低，更早暴露阻塞点 |
| 环境变量 | PYTHONASYNCIODEBUG=1 | 不改代码也能开 debug |
| 协程栈快照 | asyncio.all_tasks() + task.get_stack() | 当前有多少 Task、各自卡在哪一行 |
| 第三方 | aiomonitor、py-spy dump | 线上进程实时看协程栈 |
⚠️ 判断一个第三方库能不能在 asyncio 里用，唯一标准是「它的 IO 操作有没有 await 版本」，不是「它有没有 async 的 API」。有些 SDK 提供了
async def包装但内部还是同步 socket，等于骗人。验证方法：起 10 个并发，如果总耗时 ≈ 单次 × 10，就说明是假异步。
面试加分表述：「我们接入新的模型供应商 SDK 时有个固定动作：起 20 个并发打一次，看总耗时是接近单次耗时还是 20 倍。只要是后者，就用
asyncio.to_thread包起来并单独配线程池上限，不让它拖垮主事件循环。」
13. 并发原语：gather / as_completed / wait / Semaphore / TaskGroup
并发怎么组织，直接决定了 AI 服务的吞吐和故障表现。这是 Day21 在项目里用得最多的一节。
| API | 返回顺序 | 异常行为 | 典型场景 |
|---|---|---|---|
asyncio.gather(*aws) | 与入参顺序一致 | 默认第一个异常立刻抛出，其余任务继续跑但结果丢弃 | 结果要按序拼装（多路召回合并） |
gather(*aws, return_exceptions=True) | 同上 | 异常作为结果值返回，不中断 | 部分失败可降级（多模型并发，谁活用谁） |
asyncio.as_completed(aws) | 谁先完成谁先出 | 每个 await 单独抛 | 要边出边处理（首个可用答案、流式聚合） |
asyncio.wait(tasks, return_when=...) | 返回 (done, pending) 两个集合 | 不抛，异常留在 Task 里 | 需要精细控制（FIRST_COMPLETED 抢答） |
asyncio.wait_for(aw, timeout) | 单个 | 超时抛 TimeoutError 并取消内部任务 | 给单次 LLM 调用加超时 |
asyncio.timeout(sec)（3.11+） | 上下文管理器 | 同上，可包一段代码 | 给「检索 + 生成」整段加预算 |
asyncio.TaskGroup（3.11+） | 无返回，靠变量收集 | 任一失败则取消全部兄弟任务，抛 ExceptionGroup | 要求「要么都成功，要么都别干」 |
asyncio.Semaphore(n) | 无 | 无 | 并发度限流，保护下游不被打爆 |
python
import asyncio, random
async def call_model(name, delay, fail=False):
await asyncio.sleep(delay)
if fail:
raise RuntimeError(f"{name} 调用失败")
return f"{name} 的回答"
# ① gather：结果按入参顺序，一个失败整体抛
async def demo_gather():
try:
res = await asyncio.gather(
call_model("GPT", 0.3),
call_model("Claude", 0.1, fail=True),
call_model("Qwen", 0.2),
)
except RuntimeError as e:
print("gather 直接抛：", e) # Claude 调用失败
# ② gather + return_exceptions：部分失败可降级（AI 多模型兜底最常用）
async def demo_gather_tolerant():
res = await asyncio.gather(
call_model("GPT", 0.3),
call_model("Claude", 0.1, fail=True),
call_model("Qwen", 0.2),
return_exceptions=True,
)
ok = [r for r in res if not isinstance(r, Exception)]
bad = [r for r in res if isinstance(r, Exception)]
print(f"成功 {len(ok)} 个，失败 {len(bad)} 个：{ok}")
# ③ as_completed：谁先回谁先处理，适合「要首个可用答案」
async def demo_as_completed():
aws = [call_model("GPT", 0.3), call_model("Claude", 0.1), call_model("Qwen", 0.2)]
for fut in asyncio.as_completed(aws):
r = await fut
print("先到先处理：", r) # Claude → Qwen → GPT
# ④ wait + FIRST_COMPLETED：抢答模式，拿到第一个就取消其余
async def demo_race():
tasks = {asyncio.create_task(call_model(n, random.uniform(0.1, 0.5)))
for n in ("GPT", "Claude", "Qwen")}
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
for t in pending:
t.cancel() # ✅ 必须显式取消，否则任务继续烧 token
await asyncio.gather(*pending, return_exceptions=True) # 等取消落地
print("抢答结果：", done.pop().result())
asyncio.run(demo_gather_tolerant())Semaphore 限并发：没有它，gather 一万个任务会瞬间把下游打死（LLM 接口会直接 429）。
python
import asyncio
sem = asyncio.Semaphore(10) # 最多 10 个在飞
async def call_with_limit(prompt):
async with sem: # 超过 10 个的在这里排队等
await asyncio.sleep(1)
return f"回答：{prompt}"
async def main():
prompts = [f"问题{i}" for i in range(100)]
# ✅ 100 个任务，但同时只有 10 个在跑，总耗时 ≈ 10 秒
res = await asyncio.gather(*(call_with_limit(p) for p in prompts))
print(len(res))
asyncio.run(main())TaskGroup（3.11+）：结构化并发，推荐替代 gather。它保证「离开 async with 时所有子任务一定已经结束」，不会出现游离的 Task。
python
import asyncio
async def demo_taskgroup():
results = {}
try:
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(call_model("检索", 0.2))
t2 = tg.create_task(call_model("重排", 0.1))
# 任一任务抛异常 → 其余任务自动被 cancel
results["检索"] = t1.result() # 离开 with 时任务一定完成了
results["重排"] = t2.result()
except* RuntimeError as eg: # except* 处理 ExceptionGroup
print("子任务异常：", eg.exceptions)
return results| 选择准则 | 用哪个 |
|---|---|
| 要全部结果、按序、允许部分失败 | gather(return_exceptions=True) |
| 要全部结果、任一失败就整体放弃 | TaskGroup（3.11+）或 gather 默认行为 |
| 只要最快的那个答案 | wait(FIRST_COMPLETED) + 取消 pending |
| 结果要边出边流给前端 | as_completed 或异步生成器 |
| 任何时候下游有 QPS/并发上限 | 一定叠加 Semaphore |
⚠️
gather默认行为的坑：第一个异常抛出后，其余任务并没有被取消，它们还在后台跑、还在消耗 token 和下游配额，只是结果没人要了。如果要「失败就全停」，必须用TaskGroup，或者自己except之后遍历task.cancel()。
面试加分表述：「我们的多路召回用
gather(return_exceptions=True)+Semaphore(20)：向量库、ES、知识图谱三路并发，任一路挂了就降级成两路，不整体失败；并发度卡在 20 是因为向量库压测出来的拐点在 25 左右。单路超时用asyncio.timeout包，整段检索有 800ms 的总预算，超了就只用已经回来的那几路。」
14. 上下文管理器 with：__enter__ / __exit__ 与 contextlib
with 是 Python 版的 try-with-resources，解决的是「无论正常结束还是抛异常，收尾代码一定要执行」。
text
with expr as var:
body
实际等价于：
mgr = expr
var = mgr.__enter__() ← 进入：拿到资源
try:
body
except BaseException as e:
if not mgr.__exit__(type(e), e, e.__traceback__):
raise ← __exit__ 返回假值 → 异常继续往外抛
else:
mgr.__exit__(None, None, None) ← 正常退出，三个参数都是 Nonepython
import time
class LLMSession:
def __init__(self, model):
self.model = model
def __enter__(self):
self.t0 = time.perf_counter()
print(f"建立 {self.model} 连接")
return self # ✅ 返回值赋给 as 后面的变量
def __exit__(self, exc_type, exc_val, exc_tb):
cost = (time.perf_counter() - self.t0) * 1000
print(f"释放连接，耗时 {cost:.1f}ms")
if exc_type is not None:
print(f"捕获到异常：{exc_type.__name__}: {exc_val}")
return False # ✅ 返回 False：异常继续往外传播
# ⚠️ 返回 True：异常被「吞掉」，调用方感知不到
with LLMSession("gpt-4") as s:
print(f"用 {s.model} 回答问题")__exit__ 返回值的含义是面试高频追问点：
__exit__ 返回值 | 行为 | 什么时候该这样 |
|---|---|---|
False / None（默认） | 异常继续抛出 | 绝大多数场景，资源释放但不掩盖错误 |
True | 异常被抑制，with 之后的代码正常继续 | 只在明确要忽略某类异常时（如 contextlib.suppress） |
python
# ❌ 危险写法：无脑 return True，所有异常都被吞掉
class SwallowAll:
def __enter__(self): return self
def __exit__(self, *a): return True # 吞掉一切
with SwallowAll():
1 / 0 # ZeroDivisionError 被吞，没人知道
print("程序照常往下走，但数据已经是错的了")
# ✅ 要抑制异常就用标准库，意图明确且只抑制指定类型
from contextlib import suppress
with suppress(FileNotFoundError): # 只忽略文件不存在
open("not_exist.txt")contextlib.contextmanager：用生成器写上下文管理器，比写类少一半代码，是实际项目里的主流写法。
python
from contextlib import contextmanager, asynccontextmanager
import time
@contextmanager
def timed(name: str):
t0 = time.perf_counter()
try:
yield # ← yield 之前是 __enter__，之后是 __exit__
finally: # ✅ 必须 try/finally，否则异常时收尾不执行
print(f"{name} 耗时 {(time.perf_counter() - t0) * 1000:.1f}ms")
with timed("向量检索"):
time.sleep(0.2)
# 异步版本：async with
@asynccontextmanager
async def async_timed(name: str):
t0 = time.perf_counter()
try:
yield
finally:
print(f"{name} 耗时 {(time.perf_counter() - t0) * 1000:.1f}ms")
import asyncio
async def main():
async with async_timed("LLM 调用"):
await asyncio.sleep(0.3)
asyncio.run(main())其他常用工具：
| 工具 | 作用 |
|---|---|
contextlib.suppress(*excs) | 只抑制指定异常 |
contextlib.closing(obj) | 退出时调 obj.close()，适配没实现协议的旧对象 |
contextlib.ExitStack() | 动态数量的资源统一管理（打开 N 个文件） |
contextlib.AsyncExitStack() | 上一项的异步版本 |
contextlib.nullcontext(x) | 占位用的空上下文，写「可选加锁/可选 trace」时很好用 |
python
from contextlib import ExitStack, nullcontext
import threading
# 动态资源：文件数量运行时才知道
paths = ["a.txt", "b.txt", "c.txt"]
# with ExitStack() as stack:
# files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths]
# # 离开 with 时三个文件按相反顺序全部关闭
# nullcontext：让「要不要加锁」变成一个参数，避免写两份代码
lock = threading.Lock()
def write(data, thread_safe=True):
with (lock if thread_safe else nullcontext()):
print("写入", data)
write("x", thread_safe=False)⚠️
@contextmanager里不写try/finally等于白写。如果只是yield后面跟清理代码，with体里抛异常时清理代码不会执行（异常会在yield处被抛回生成器）。这是 code review 的固定检查项。
面试加分表述：「我们把『耗时埋点 + traceId 透传 + 异常兜底』封成了一个
@asynccontextmanager，RAG 每个阶段（改写、召回、重排、生成）都套一层，出问题能直接在日志里看到是哪一段超了预算。__exit__一律返回 False，绝不吞异常——吞掉的异常会变成第二天更难查的数据问题。」
15. 类型注解与 Pydantic：AI 工程里的数据契约
Python 的类型注解在运行时不做校验，它只是元信息。但在 AI 工程里它是刚需，因为 LLM 返回的是不可信的自由文本，必须有一层强约束把它变成结构化数据。
python
from typing import Optional, Literal, TypedDict
# 注解只是提示，运行时完全不管
def add(a: int, b: int) -> int:
return a + b
print(add("x", "y")) # 'xy' ← 不报错！注解不生效
print(add.__annotations__) # {'a': <class 'int'>, 'b': ..., 'return': ...}| 手段 | 校验时机 | 适用位置 |
|---|---|---|
类型注解 + mypy / pyright | 静态检查期（CI 里跑） | 内部函数签名 |
pydantic.BaseModel | 运行时，自动转换 + 报错 | 接口入参、LLM 输出解析、配置加载 |
dataclass | 不校验，只生成样板代码 | 纯内部数据结构 |
TypedDict | 只做静态检查 | 已有 dict 结构不想改造时 |
python
from pydantic import BaseModel, Field, field_validator
from typing import Literal
# ✅ 用 Pydantic 给 LLM 的输出上契约
class Intent(BaseModel):
intent: Literal["查询订单", "退货", "咨询", "其他"]
confidence: float = Field(ge=0.0, le=1.0)
order_id: str | None = None
@field_validator("order_id")
@classmethod
def check_order_id(cls, v):
if v is not None and not v.isdigit():
raise ValueError("order_id 必须是纯数字")
return v
# 模型返回的 JSON 字符串直接解析 + 校验
raw = '{"intent": "查询订单", "confidence": 0.93, "order_id": "100231"}'
obj = Intent.model_validate_json(raw)
print(obj.intent, obj.confidence)
# 模型胡说八道时，立刻在边界上失败，而不是流到下游
bad = '{"intent": "订机票", "confidence": 1.5}'
try:
Intent.model_validate_json(bad)
except Exception as e:
print("校验失败，可触发重试或降级：", type(e).__name__)⚠️ AI 工程的铁律：LLM 的输出永远当成不可信输入。模型会多输出一段解释、会把 JSON 包在 markdown 代码块里、会把数字写成字符串、会编造枚举值。所以解析层必须做三件事：① 先抽取 JSON（容忍前后多余文本）；②
model_validate做结构校验；③ 校验失败走带错误信息的重试（把报错原文回灌给模型让它改）。少任何一步，脏数据都会直接落库。
面试加分表述：「我们所有 LLM 结构化输出都过 Pydantic，并且把校验失败率当成一个核心指标埋点。某个 prompt 的校验失败率从 2% 涨到 15%，基本就是上游模型版本变了，比用户投诉早两天发现。」
✨ 高级深挖知识点
16. 协程的本质：生成器 + 事件循环的两层拼装
async def 不是黑魔法。协程对象和生成器共用同一套底层机制——可暂停、可恢复的栈帧。CPython 里 coroutine 和 generator 的实现几乎一致，只是前者多了 __await__、少了 __iter__。
text
三层结构，从下往上
第三层 事件循环 EventLoop
├─ ready 队列：马上能跑的回调
├─ selector：epoll/kqueue/IOCP，监听 fd 就绪
└─ 定时器堆：call_later / sleep
第二层 Task（Future 的子类）
负责「不停地 send 推进协程」，并在协程挂起时
把自己注册成「IO 就绪后的回调」
第一层 coroutine（可暂停的栈帧）
每次 send 从上次 yield 的位置继续执行
★ 一次 await asyncio.sleep(1) 的完整路径
协程执行到 await → 向上抛出一个 Future
→ Task 收到 Future，给它 add_done_callback(自己继续跑)
→ 事件循环把定时器塞进堆，转去跑别的任务
→ 1 秒后定时器到期 → Future.set_result() → 回调入 ready 队列
→ 下一轮循环 → Task.__step() → coro.send(None) → 协程从 await 后面继续python
# 验证：协程底层和生成器同源
import types, inspect, asyncio
async def c():
await asyncio.sleep(0)
return 1
coro = c()
print(type(coro)) # <class 'coroutine'>
print(hasattr(coro, "send")) # True ← 和生成器一样有 send
print(hasattr(coro, "throw")) # True
print(inspect.getcoroutinestate(coro)) # CORO_CREATED
# 手动驱动协程（这就是 Task 在做的事）
try:
f = coro.send(None) # 推进到第一个挂起点
print("协程交出来的东西：", type(f))
except StopIteration as e:
print("直接结束，返回值：", e.value)
coro.close()| 维度 | 生成器 | 原生协程 |
|---|---|---|
| 定义 | 含 yield 的函数 | async def 定义 |
| 驱动方式 | next() / send() | await / 被 Task 调度 |
| 实现的协议 | __iter__ + __next__ | __await__ |
能否 for 循环 | ✅ | ❌（会报 TypeError） |
| 暂停语义 | 「产出一个值给消费者」 | 「我要等一个 IO，先跑别人」 |
面试加分表述：「协程的本质是可暂停的栈帧 + 一个调度器。栈帧的暂停恢复能力 Python 在生成器时代就有了，asyncio 只是在上面加了 Task 调度和 selector 这层 IO 通知。理解了这点就知道，
await并不是『等待』，而是把自己挂起并把恢复条件告诉事件循环。」
17. yield from 旧式协程 vs async/await 原生协程
Python 的协程演进了三代，面试里问「yield from 和 await 什么关系」就是在考这段历史。
| 阶段 | 版本 | 写法 | 问题 |
|---|---|---|---|
| 第一代 | 2.5+ | 生成器 + send() 手写调度 | 全靠手工，没有标准库支撑 |
| 第二代 | 3.3~3.4 | yield from + @asyncio.coroutine | 协程和生成器写法混淆，for 能迭代一个「协程」，语义混乱 |
| 第三代 | 3.5+ | async def + await | ✅ 语法层面彻底区分，是现在的唯一选择 |
python
import asyncio
# 第二代写法（已废弃，3.11 移除了 @asyncio.coroutine）
# @asyncio.coroutine
# def old_style():
# yield from asyncio.sleep(1)
# return "old"
# 第三代写法（现在的标准）
async def new_style():
await asyncio.sleep(1)
return "new"yield from 并没有被淘汰，它在「生成器委托」这个场景里仍然是最简洁的写法：
python
# yield from 的真实用途：把子生成器的所有产出透传出去
def read_file(path):
yield f"--- {path} ---"
# yield from 等价于 for line in inner(): yield line，但还透传 send/throw/返回值
yield from ("第 %d 行" % i for i in range(3))
def read_all(paths):
for p in paths:
yield from read_file(p) # 委托给子生成器
print(list(read_all(["a.txt", "b.txt"])))| 关键词 | 用在哪 | 语义 |
|---|---|---|
yield | 生成器函数内 | 产出一个值并暂停 |
yield from | 生成器函数内 | 把子生成器完全代理（含 send / throw / return 值） |
await | 协程函数内 | 挂起当前协程，等待可等待对象完成 |
⚠️ 别在
async def里写yield from，这是语法错误。async def里的yield表示异步生成器（见 #24），语义完全不同。另外，async def里既有yield又有return value也是语法错误。
面试加分表述：「我会直接点明：
await在字节码层面做的事和yield from高度相似（都是把子任务的暂停透传给上层驱动者），但async def把『协程』和『迭代器』在语法层面分开了，这才是 3.5 那次改动的核心价值——此前一个对象既能for又能当协程，可读性和出错率都很糟。」
18. GIL 的切换机制：从字节码计数到时间片
GIL 的切换策略在 Python 3.2 发生过一次重要变更，这是深挖 GIL 时的加分点。
| 版本 | 切换触发 | 问题 |
|---|---|---|
| ≤ 3.1 | 每执行 100 条字节码检查一次 | 一条字节码耗时差异巨大（a+b 和 big_list.sort() 都算一条），导致切换时机极不均匀；IO 线程响应延迟高 |
| ≥ 3.2 | 时间片，默认 5ms（sys.setswitchinterval） | 更公平，但「GIL 护航」问题（convoy effect）仍存在 |
python
import sys, threading, time
print(sys.getswitchinterval()) # 0.005 → 5ms
# 调大切换间隔：减少切换开销，但 IO 线程响应变慢
sys.setswitchinterval(0.05)
# 调小：IO 响应更及时，但切换开销上升
sys.setswitchinterval(0.001)
sys.setswitchinterval(0.005) # 还原默认text
GIL 的「护航效应」（convoy effect）
IO 线程被唤醒后想拿 GIL，但 CPU 线程正好持有且没到 5ms：
CPU线程 │█████████│ █████████│
IO线程 │ ↑想拿 ░░░░░░░░░ │ ↑拿到 │
被迫等到时间片用完，延迟被放大
★ 结论：一个进程里同时跑「CPU 密集线程」和「IO 密集线程」是最差组合。
CPU 线程会持续抢占 GIL，把 IO 线程的响应延迟拖到几十毫秒。
正确做法：CPU 任务放到独立进程，IO 任务留在主进程的事件循环里。⚠️ 这解释了一个真实现象：FastAPI 服务里只要有一个接口在做同步的 CPU 密集活（比如用
pandas处理大表），整个服务所有接口的 P99 都会恶化，不只是那个接口慢。因为它既占着 GIL，又占着事件循环。监控上的表现是「P50 正常、P99 爆炸」。
面试加分表述：「GIL 的切换从 3.2 开始由字节码计数改成 5ms 时间片，可以用
sys.getswitchinterval()看。更值得说的是『护航效应』——CPU 线程和 IO 线程混在一个进程里时，IO 线程的延迟会被放大到时间片级别。我们线上就因为一个接口里用 pandas 处理大表，把整个服务的 P99 从 200ms 拖到 3 秒，后来把它整体挪到进程池解决。」
19. 异步代码里混入同步阻塞：三种救法与线程池配置
#12 讲了现象，这里讲工程落地的细节，包括线程池怎么配、上下文怎么透传。
| 救法 | API | 适用 | 注意 |
|---|---|---|---|
| 换异步库 | httpx / aiomysql / redis.asyncio | 首选 | 要确认是真异步，不是假包装 |
| 丢线程池 | asyncio.to_thread(fn, *a)（3.9+） | 阻塞 IO | 用的是默认 executor，线程数有限且共享 |
| 丢自定义线程池 | loop.run_in_executor(my_pool, fn, *a) | 需要隔离/自定义大小 | 不同下游用不同池，防止互相挤占 |
| 丢进程池 | loop.run_in_executor(process_pool, fn, *a) | CPU 密集 | 参数要可 pickle，大对象慎用 |
python
import asyncio, functools
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
# ⚠️ 默认 executor 的线程数：min(32, cpu_count + 4)，所有 to_thread 共享
# 一旦被某个慢下游占满，其他 to_thread 全部排队
async def demo_default():
return await asyncio.to_thread(legacy_sdk_call, "参数")
# ✅ 按下游隔离线程池，互不影响（舱壁模式）
OCR_POOL = ThreadPoolExecutor(max_workers=8, thread_name_prefix="ocr")
DB_POOL = ThreadPoolExecutor(max_workers=16, thread_name_prefix="legacy-db")
CPU_POOL = ProcessPoolExecutor(max_workers=4)
async def call_ocr(img_bytes: bytes):
loop = asyncio.get_running_loop()
return await loop.run_in_executor(OCR_POOL, legacy_sdk_call, img_bytes)
# 带关键字参数时用 functools.partial 包一层（run_in_executor 不支持 kwargs）
async def call_with_kwargs(x):
loop = asyncio.get_running_loop()
fn = functools.partial(legacy_sdk_call, x, timeout=3, retry=2)
return await loop.run_in_executor(DB_POOL, fn)
def legacy_sdk_call(x, timeout=5, retry=0):
import time
time.sleep(0.1)
return f"result({x})"上下文透传问题：contextvars（traceId、用户身份）在 run_in_executor 的子线程里默认会丢。Python 3.9+ 的 asyncio.to_thread 自动带上了 context，但 run_in_executor 不会。
python
import asyncio, contextvars
from concurrent.futures import ThreadPoolExecutor
trace_id = contextvars.ContextVar("trace_id", default="-")
def sync_work():
return f"子线程看到的 traceId = {trace_id.get()}"
POOL = ThreadPoolExecutor(4)
async def main():
trace_id.set("abc-123")
# ✅ to_thread 自动复制 context（3.9+）
print(await asyncio.to_thread(sync_work)) # abc-123
# ❌ run_in_executor 默认丢 context
loop = asyncio.get_running_loop()
print(await loop.run_in_executor(POOL, sync_work)) # -
# ✅ 手动复制 context 再跑
ctx = contextvars.copy_context()
print(await loop.run_in_executor(POOL, lambda: ctx.run(sync_work))) # abc-123
asyncio.run(main())⚠️ 线程池大小不是越大越好。线程池是用来「兜住同步阻塞」的，不是用来提升吞吐的。配置准则：
线程数 ≈ 目标并发 × 单次阻塞时长 / 可接受延迟，并且一定要配上队列上限和拒绝策略——否则下游变慢时任务会在队列里无限堆积，表现为「接口全部超时但 CPU 和内存都正常」。
面试加分表述：「我们按下游划分线程池：OCR 一个池、老数据库 SDK 一个池，谁慢谁自己排队，不会互相传染。而且 traceId 这类
contextvars在run_in_executor里会丢，我们统一封了一个run_sync()帮助函数，内部用copy_context().run()透传，保证日志链路不断。」
20. 多进程的真实开销：fork vs spawn 与 pickle 代价
多进程绕开了 GIL，但代价常被低估。面试里能说清「为什么多进程不是免费的」很加分。
| 启动方式 | 平台默认 | 机制 | 代价 |
|---|---|---|---|
fork | Linux/macOS（3.14 起 macOS 默认 spawn） | 复制父进程内存（COW 写时复制） | 启动快；但会继承锁、线程、文件描述符，容易死锁 |
spawn | Windows、macOS 新版 | 全新解释器 + 重新 import 主模块 + pickle 传参 | 启动慢（几十到几百毫秒）；主模块必须可安全重复导入 |
forkserver | 可选（Unix） | 先 fork 一个干净的服务进程，再由它 fork | 兼顾安全与速度 |
python
import multiprocessing as mp, os, time
def worker(n: int) -> int:
return sum(i * i for i in range(n))
# ⚠️ Windows / spawn 下必须有这个保护，否则子进程重新 import 主模块时
# 会再次执行建池代码 → 无限递归创建进程 → 机器被打满
if __name__ == "__main__":
mp.set_start_method("spawn", force=True) # 显式声明，跨平台行为一致
with mp.Pool(processes=4) as pool:
print(pool.map(worker, [10**6] * 4))text
pickle 开销实测直觉（传 500MB DataFrame 到子进程）
父进程 ──pickle 序列化──> 管道/共享内存 ──unpickle──> 子进程
(CPU 1~3s, 内存翻倍) (CPU 1~3s, 内存再一份)
★ 真实算力可能全耗在搬数据上，计算本身只花 0.5s。
解法：① 子进程自己从磁盘/对象存储读；
② multiprocessing.shared_memory 零拷贝共享 numpy 数组；
③ 只传「路径/索引区间」这类小参数，不传大对象。python
# ✅ 用 shared_memory 共享大数组，避免 pickle 整块数据
import numpy as np
from multiprocessing import shared_memory, Pool
def worker(args):
name, shape, dtype = args
shm = shared_memory.SharedMemory(name=name)
arr = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
result = float(arr.mean()) # 直接读共享内存，无需拷贝
shm.close()
return result
if __name__ == "__main__":
data = np.random.rand(10_000_000)
shm = shared_memory.SharedMemory(create=True, size=data.nbytes)
buf = np.ndarray(data.shape, dtype=data.dtype, buffer=shm.buf)
buf[:] = data[:]
try:
with Pool(4) as p:
print(p.map(worker, [(shm.name, data.shape, data.dtype)] * 4))
finally:
shm.close()
shm.unlink() # 必须 unlink，否则共享内存泄漏fork 的经典死锁：父进程里某个线程正持有 logging 的内部锁时 fork，子进程继承了「已被锁住但持有者线程不存在」的锁 → 子进程第一次打日志就永久挂住。这就是为什么不要在已经启动线程池、连接池、日志异步 handler 之后再 fork。
| 常见坑 | 表现 | 解决 |
|---|---|---|
Windows 下没写 if __name__ == "__main__" | 进程无限繁殖，机器卡死 | 必须加保护 |
把闭包/lambda 传给 Pool.map | PicklingError | 用模块级函数，或换 loky/dill |
| 在 fork 前已创建 DB 连接池 | 多进程共用同一 socket，数据串包 | 在子进程 initializer 里重建连接 |
| 子进程异常被静默吞掉 | 任务永远不返回 | 用 concurrent.futures，future.result() 会重抛 |
没加 maxtasksperchild | 子进程内存只增不减（内存泄漏累积） | Pool(4, maxtasksperchild=100) 定期回收 |
⚠️ 进程池不要动态创建。每次请求
ProcessPoolExecutor(...)再销毁，启动开销（spawn 下几百毫秒）会吃掉所有收益。应在应用启动时建一次全局池，关闭时shutdown(wait=True)。
面试加分表述：「多进程确实绕开了 GIL，但要算清三笔账：启动开销、pickle 序列化开销、内存翻倍。我们做向量化批处理时，一开始把整个 DataFrame 传进子进程，结果 80% 时间花在序列化上；改成只传分片索引、让子进程自己读 parquet，整体快了 4 倍。另外 Windows 是 spawn，主模块一定要有
__main__保护，不然会进程爆炸。」
21. asyncio 的异常处理：为什么异常会「凭空消失」
这是 asyncio 最容易踩的坑，也是区分「会用」和「懂」的分水岭。
核心机制：Task 捕获协程抛出的异常后存在自己身上，不会往上冒泡。只有当有人 await task 或 task.result() 时才会重新抛出。没人取 → 异常被静默吞掉，直到 Task 被 GC 时打一条 Task exception was never retrieved 日志。
python
import asyncio
async def boom():
raise ValueError("出错了")
# ❌ 错误写法：fire-and-forget，异常被吞
async def bad():
asyncio.create_task(boom()) # 没 await，没回调
await asyncio.sleep(1)
print("主流程以为一切正常") # 照常打印，故障被隐藏
# ✅ 正确写法 1：显式 await
async def good_1():
task = asyncio.create_task(boom())
try:
await task
except ValueError as e:
print("捕获到：", e)
# ✅ 正确写法 2：真的要后台跑，就挂 done_callback + 持引用
BACKGROUND: set[asyncio.Task] = set()
def _on_done(t: asyncio.Task):
BACKGROUND.discard(t)
if t.cancelled():
return
if exc := t.exception():
print("后台任务异常：", repr(exc)) # 这里应该打日志 / 上报告警
def spawn(coro) -> asyncio.Task:
t = asyncio.create_task(coro)
BACKGROUND.add(t) # ⚠️ 必须持强引用，否则 Task 可能被 GC
t.add_done_callback(_on_done)
return t⚠️
create_task返回的 Task 必须被引用住。事件循环只持弱引用，如果你不存起来，Task 可能在执行到一半时被垃圾回收掉，表现为「任务莫名其妙不执行了」且没有任何报错。官方文档对此有明确警告。
gather 的两种异常语义（极高频考点）：
python
import asyncio
async def ok(i):
await asyncio.sleep(0.1)
return i
async def fail(i):
await asyncio.sleep(0.05)
raise RuntimeError(f"task {i} failed")
async def main():
# ① 默认：第一个异常立刻向上抛，但其他任务「仍在后台继续跑」且不会被取消
try:
await asyncio.gather(ok(1), fail(2), ok(3))
except RuntimeError as e:
print("gather 抛出：", e) # 只拿到第一个异常，ok(1)/ok(3) 的结果丢了
# ② return_exceptions=True：异常作为「结果」返回，不中断
res = await asyncio.gather(ok(1), fail(2), ok(3), return_exceptions=True)
print(res) # [1, RuntimeError('task 2 failed'), 3]
for i, r in enumerate(res):
if isinstance(r, BaseException):
print(f"第 {i} 个失败：{r!r}")
asyncio.run(main())| 场景 | 推荐 | 理由 |
|---|---|---|
| 批量调 LLM，允许部分失败 | gather(..., return_exceptions=True) | 拿到全部成功结果 + 失败清单，可做降级 |
| 多个依赖必须全成功 | asyncio.TaskGroup（3.11+） | 任一失败自动取消兄弟任务，无泄漏 |
| 后台常驻任务 | create_task + done_callback + 强引用 | 异常必须上报，不能静默 |
| 需要「谁先完成先处理」 | as_completed | 每个 await 都要单独 try |
TaskGroup 与 ExceptionGroup（3.11+）：
python
import asyncio
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(ok(1))
tg.create_task(fail(2))
tg.create_task(fail(3))
# 退出 async with 时等待全部完成；有失败则自动取消其余任务
except* RuntimeError as eg: # except* 专门处理 ExceptionGroup
for e in eg.exceptions:
print("子任务异常：", e)面试加分表述：「asyncio 里异常不会自动冒泡，Task 会把异常存在自己身上，没人
await就只在 GC 时打一条Task exception was never retrieved。我们线上出过一次：后台刷新缓存的任务其实天天在失败，但监控全绿，因为异常被吞了。后来统一封了一个spawn()工具——强引用 +done_callback里打日志上报，再加上gather(return_exceptions=True)显式区分成功失败，才把这类静默故障暴露出来。」
22. 取消与超时：CancelledError 为什么不能被 except Exception 吞掉
取消的本质：task.cancel() 并不会「杀死」协程，它只是在协程下一次 await 挂起点往里面扔一个 CancelledError。如果协程永远不 await（比如在跑一个死循环的纯计算），取消就永远不生效。
text
task.cancel() 的真实流程
task.cancel()
│ 打标记 _must_cancel = True
▼
协程跑到下一个 await 挂起点
│ 事件循环在这里 throw CancelledError 进去
▼
协程栈逐层向上抛 → finally / __aexit__ 依次执行（清理机会）
▼
抛到最外层 → Task 状态变 cancelled
★ 如果协程里没有 await（纯 CPU 死循环），取消信号永远送不进去。
★ 如果协程 except 掉了 CancelledError 又不重新抛，取消就「失败」了。python
import asyncio
# ❌ 错误写法：宽泛捕获把取消信号吞了
async def bad_worker():
try:
await asyncio.sleep(10)
except Exception as e: # 3.8+ CancelledError 继承 BaseException，
print("忽略异常", e) # 这里其实抓不到；但 3.7 及以下会被抓到
except BaseException as e: # ⚠️ 这才是真正危险的写法
print("吞掉了取消：", repr(e)) # 任务无法被取消，优雅退出失效
# ✅ 正确写法：捕获是为了清理，清理完必须重新抛出
async def good_worker(conn):
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
await conn.rollback() # 做必要的回滚/释放
raise # ⚠️ 一定要 raise，让取消语义继续传播
finally:
await conn.close() # finally 在取消时同样会执行⚠️ Python 3.8 起
asyncio.CancelledError从Exception改为继承BaseException，目的就是避免被业务代码的except Exception误吞。但except BaseException、裸except:、以及gather(return_exceptions=True)仍然可能把它拦下来，写通用重试/兜底逻辑时必须显式放行。
超时的三种写法：
| 写法 | 版本 | 行为 | 适用 |
|---|---|---|---|
asyncio.wait_for(coro, timeout=3) | 全版本 | 超时则取消内部任务并抛 TimeoutError | 单个协程 |
async with asyncio.timeout(3): | 3.11+ | 作用于整个代码块，可包多个 await | 一段流程的总预算 |
async with asyncio.timeout_at(deadline): | 3.11+ | 绝对截止时间 | 跨多层透传的「剩余预算」 |
python
import asyncio
async def call_llm(prompt: str) -> str:
await asyncio.sleep(5)
return "answer"
async def main():
# ① 单协程超时
try:
await asyncio.wait_for(call_llm("hi"), timeout=2)
except asyncio.TimeoutError:
print("LLM 超时，走降级")
# ② 整段流程共享一个总预算（3.11+），更贴近真实链路
try:
async with asyncio.timeout(3):
ctx = await retrieve_context("hi") # 检索
ans = await call_llm(ctx) # 生成
await save_log(ans) # 落库
except TimeoutError:
print("整条 RAG 链路超过 3s，整体降级")
# ③ 保护关键片段不被外层取消
async def must_finish():
await asyncio.sleep(1)
print("计费扣减完成，不能半途而废")
try:
async with asyncio.timeout(0.1):
await asyncio.shield(must_finish()) # shield 内部不会被取消
except TimeoutError:
print("外层超时返回，但 shield 的任务仍在后台跑完")
await asyncio.sleep(1.2)
async def retrieve_context(q): await asyncio.sleep(0.5); return q
async def save_log(a): await asyncio.sleep(0.1)
asyncio.run(main())| 坑 | 表现 | 解决 |
|---|---|---|
except Exception 包住整个协程体 | 任务取消不掉，服务无法优雅关闭 | 单独 except asyncio.CancelledError: raise |
| 协程里纯 CPU 死循环 | cancel() 毫无反应 | 循环里插 await asyncio.sleep(0) 让出控制权 |
wait_for 超时后以为任务停了 | 内部任务仍在清理阶段 | 超时后不要立刻复用连接等资源 |
shield 滥用 | 超时返回了，资源还在被后台占用 | 只保护「必须完成的副作用」，如扣费、幂等写入 |
| 每层各写一个 3s 超时 | 总耗时累加到 9s | 用 timeout_at 传递统一 deadline |
面试加分表述：「取消不是杀线程，是在下一个 await 点往协程里扔
CancelledError，所以纯 CPU 死循环的协程根本取消不掉。3.8 之后它继承BaseException，就是为了防止被业务的except Exception误吞——我们写通用重试装饰器时特意先except asyncio.CancelledError: raise再兜底。超时上我们用 3.11 的asyncio.timeout_at给整条 RAG 链路一个绝对 deadline，检索、生成、落库共享预算，避免每层各写 3s 最后累加成 9s。」
23. 装饰器进阶：带参、类装饰器、叠放顺序、同步异步双兼容
① 带参装饰器 = 三层函数
python
import functools, time
def retry(times: int = 3, delay: float = 0.5, exceptions=(Exception,)):
def decorator(fn): # 第二层：真正的装饰器
@functools.wraps(fn)
def wrapper(*args, **kwargs): # 第三层：替换函数
last = None
for i in range(times):
try:
return fn(*args, **kwargs)
except exceptions as e:
last = e
if i < times - 1:
time.sleep(delay * (2 ** i)) # 指数退避
raise last
return wrapper
return decorator # 第一层：返回装饰器
@retry(times=3, delay=0.2, exceptions=(TimeoutError,))
def fetch(): ...text
三层调用的展开过程
@retry(times=3) -> tmp = retry(times=3) # 第一层执行，返回 decorator
def fetch(): ... fetch = tmp(fetch) # 第二层执行，返回 wrapper
fetch() -> wrapper() # 第三层在调用时执行
★ 「带参」和「不带参」最大区别：带参的 @xxx(...) 先被调用一次。
★ 所以 @retry 和 @retry() 不是一回事——前者会把 fn 当成 times 传进去。② 兼容「带参 / 不带参」两种用法
python
import functools
def timeit(fn=None, *, label: str = ""):
if fn is None: # 用法：@timeit(label="x")
return functools.partial(timeit, label=label)
@functools.wraps(fn) # 用法：@timeit
def wrapper(*a, **kw):
import time
t = time.perf_counter()
try:
return fn(*a, **kw)
finally:
print(f"[{label or fn.__name__}] {time.perf_counter() - t:.3f}s")
return wrapper
@timeit # ✅ 可以
def a(): ...
@timeit(label="批量推理") # ✅ 也可以
def b(): ...③ 叠放顺序：由下往上包装，由上往下执行
python
def deco_a(fn):
def w(*x, **y):
print("A 前"); r = fn(*x, **y); print("A 后"); return r
return w
def deco_b(fn):
def w(*x, **y):
print("B 前"); r = fn(*x, **y); print("B 后"); return r
return w
@deco_a # 外层
@deco_b # 内层
def hello():
print("hello")
hello()
# A 前 → B 前 → hello → B 后 → A 后
# 等价于 hello = deco_a(deco_b(hello))⚠️ 顺序决定正确性。
@app.get("/x")这类注册型装饰器必须在最外层（最上面），否则注册进去的是未经包装的原函数；而缓存装饰器要放在重试装饰器内层，否则重试会穿透缓存反复打下游。经验口诀：「注册在最上，限流次之，重试再次，缓存最靠近业务函数」。
④ 同步 / 异步双兼容（AI 项目里几乎必写）
python
import asyncio, functools, inspect, time
def timing(fn):
if inspect.iscoroutinefunction(fn): # ⚠️ 必须分支判断
@functools.wraps(fn)
async def async_wrapper(*a, **kw):
t = time.perf_counter()
try:
return await fn(*a, **kw)
finally:
print(f"[async {fn.__name__}] {time.perf_counter()-t:.3f}s")
return async_wrapper
@functools.wraps(fn)
def sync_wrapper(*a, **kw):
t = time.perf_counter()
try:
return fn(*a, **kw)
finally:
print(f"[sync {fn.__name__}] {time.perf_counter()-t:.3f}s")
return sync_wrapper
# ❌ 错误写法：不判断，直接用同步 wrapper 包异步函数
# 结果 wrapper 立刻返回一个未被 await 的协程对象，计时恒为 0s，
# 并抛 RuntimeWarning: coroutine was never awaited⑤ 类装饰器与装饰类
python
import functools
class CallCounter: # 用类做装饰器：天然带状态
def __init__(self, fn):
functools.update_wrapper(self, fn) # 相当于 wraps 的类版本
self.fn, self.count = fn, 0
def __call__(self, *a, **kw):
self.count += 1
return self.fn(*a, **kw)
@CallCounter
def ping(): return "pong"
ping(); ping()
print(ping.count) # 2
def singleton(cls): # 装饰「类」：返回同一实例
inst = {}
@functools.wraps(cls)
def getter(*a, **kw):
if cls not in inst:
inst[cls] = cls(*a, **kw)
return inst[cls]
return getter
@singleton
class ModelClient:
def __init__(self): print("加载模型（只应出现一次）")
ModelClient(); ModelClient()⚠️ 用类做装饰器装饰实例方法时会丢
self绑定（因为CallCounter实例不是描述符）。需要实现__get__返回functools.partial(self.__call__, obj)，或者干脆用函数式装饰器 +contextvars存状态。
面试加分表述：「我们项目里所有装饰器都写成同步/异步双兼容——用
inspect.iscoroutinefunction分支返回两个 wrapper。之前没判断，同步 wrapper 包了 async 函数，返回的是协程对象没被 await，计时全是 0ms，日志里还一堆coroutine was never awaited。另外叠放顺序我们定了规范：路由注册最外、限流、重试、缓存最里，否则重试会穿透缓存反复打 LLM，成本直接翻倍。」
24. 异步生成器与 async for / async with：LLM 流式输出的基础设施
LLM 的 SSE 流式返回（打字机效果）在 Python 侧的标准形态就是异步生成器。
| 概念 | 关键字 | 协议方法 | 消费方式 |
|---|---|---|---|
| 生成器 | def + yield | __iter__ / __next__ | for |
| 异步生成器 | async def + yield | __aiter__ / __anext__ | async for |
| 上下文管理器 | — | __enter__ / __exit__ | with |
| 异步上下文管理器 | — | __aenter__ / __aexit__ | async with |
python
import asyncio
from typing import AsyncIterator
# ⚠️ async def 函数体里出现 yield → 它是「异步生成器函数」，不是协程函数
# 调用它得到异步生成器对象，不能 await，只能 async for
async def stream_llm(prompt: str) -> AsyncIterator[str]:
chunks = ["你", "好", "，", "我是", "助手"]
for c in chunks:
await asyncio.sleep(0.1) # 模拟网络等待
yield c # 吐一个 token 就让出控制权
async def main():
buf = []
async for token in stream_llm("hi"): # 边收边处理，首字延迟极低
print(token, end="", flush=True)
buf.append(token)
print("\n完整答案：", "".join(buf))
asyncio.run(main())真实场景：FastAPI 把 LLM 流式转成 SSE
python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json, asyncio
app = FastAPI()
async def sse_wrap(prompt: str):
try:
async for token in stream_llm(prompt):
yield f"data: {json.dumps({'delta': token}, ensure_ascii=False)}\n\n"
yield "data: [DONE]\n\n"
except asyncio.CancelledError:
# ⚠️ 客户端断开时这里会被取消，必须释放上游连接 & 记录已产生的 token 用于计费
raise
finally:
pass # 这里做清理：关上游流、落库、扣费
@app.get("/chat")
async def chat(q: str):
return StreamingResponse(sse_wrap(q), media_type="text/event-stream")异步生成器的三个专属坑：
python
import asyncio
# ① 异步生成器里不能跨 yield 持有外部事件循环之外的资源，
# 也不要在 yield 前后错配 async with 的生命周期
async def bad_gen(session):
async with session.get(url) as resp: # 伪代码
async for line in resp.content:
yield line
# ⚠️ 消费方提前 break 时，生成器不会立刻执行 __aexit__，
# 而是等 aclose()/GC 时才清理 → 连接可能长时间不释放
# ✅ 用 contextlib.aclosing 保证提前退出也能清理（3.10+）
from contextlib import aclosing
async def consume(session):
async with aclosing(bad_gen(session)) as gen:
async for line in gen:
if line == b"STOP":
break # 退出 async with 时会 await gen.aclose()| 坑 | 原因 | 解决 |
|---|---|---|
break 后上游连接不释放 | 异步生成器清理依赖 aclose()，GC 时机不可控 | contextlib.aclosing 或手动 await gen.aclose() |
程序退出时报 asyncgen ... was never awaited | 循环关闭前未关停挂起的异步生成器 | asyncio.run() 已自动调 shutdown_asyncgens()，别手写 loop.run_until_complete 绕过 |
在异步生成器里 return value | 异步生成器不允许带值 return | 只能裸 return，结果通过最后一个 yield 传出 |
想对它用 gather | 它不是协程，无法 await | 先用 [x async for x in gen] 收集，或用队列扇出 |
| 流式输出里混同步阻塞 | 阻塞点卡住事件循环，所有连接的打字机一起卡顿 | 阻塞段 asyncio.to_thread |
异步推导式（3.6+）：
python
tokens = [t async for t in stream_llm("hi")] # 异步列表推导
lengths = {t: len(t) async for t in stream_llm("hi")} # 异步字典推导
first3 = [t async for i, t in aenumerate(stream_llm("hi")) if i < 3]面试加分表述：「我们 LLM 流式接口是『异步生成器 + StreamingResponse』，首字延迟从 3 秒降到 300 毫秒。两个实战细节：一是客户端断开会给生成器抛
CancelledError，必须在finally里关上游流并按已产出 token 计费，否则会漏计费也会泄漏连接；二是消费方break时异步生成器不会立刻清理，我们统一用contextlib.aclosing包一层，解决了连接池被慢慢耗尽的问题。」
25. 内存与引用计数：循环引用、gc、__slots__、弱引用
AI 服务常驻内存大、对象多，内存模型是绕不开的一课。
CPython 的两套回收机制：
text
① 引用计数（主力，实时）
a = [1,2,3] refcount = 1
b = a refcount = 2
del a refcount = 1
del b refcount = 0 → 立刻释放
② 分代 GC（兜底，解决循环引用）
gen0（新对象）──存活──> gen1 ──存活──> gen2（老对象）
阈值默认 (700, 10, 10)：
gen0 分配数 - 释放数 > 700 时扫 gen0
gen0 扫了 10 次扫一次 gen1；gen1 扫 10 次扫一次 gen2
★ 引用计数无法回收循环引用，必须靠分代 GC。
★ 分代 GC 只管「容器对象」（list/dict/set/自定义类实例），
不管 int/str 这类无法形成环的类型。python
import gc, sys
class Node:
def __init__(self): self.peer = None
a, b = Node(), Node()
a.peer, b.peer = b, a # 形成循环引用
print(sys.getrefcount(a)) # > 1（getrefcount 本身会 +1）
del a, b # 引用计数不为 0，内存没释放
print(gc.collect()) # 手动触发 → 回收掉这些不可达的环
print(gc.get_threshold()) # (700, 10, 10)
print(gc.get_count()) # 当前各代待扫描数
# 调试内存泄漏：找出谁还引用着某个对象
import objgraph # pip install objgraph
# objgraph.show_backrefs([obj], max_depth=5, filename="refs.png")__slots__：批量小对象的内存优化
python
import sys
class WithDict: # 默认每个实例带一个 __dict__
def __init__(self, x, y): self.x, self.y = x, y
class WithSlots:
__slots__ = ("x", "y") # 固定字段，不生成 __dict__
def __init__(self, x, y): self.x, self.y = x, y
a, b = WithDict(1, 2), WithSlots(1, 2)
print(sys.getsizeof(a) + sys.getsizeof(a.__dict__)) # ≈ 48 + 104 = 152 字节
print(sys.getsizeof(b)) # ≈ 48 字节
# 100 万个对象：约 152MB vs 48MB，差 3 倍以上__slots__ 的代价 | 说明 |
|---|---|
| 不能动态加属性 | b.z = 1 直接 AttributeError |
子类未声明 __slots__ 会重新获得 __dict__ | 优化失效，继承链上每层都要声明 |
和 @cached_property 冲突 | 后者需要写 __dict__ |
| 弱引用默认失效 | 需显式加 __slots__ = ("x", "__weakref__") |
弱引用：打破缓存导致的内存泄漏
python
import weakref
class Model:
def __init__(self, name): self.name = name
# ❌ 普通 dict 缓存：只要缓存不清，对象永远不会被回收
STRONG_CACHE = {}
# ✅ 弱引用缓存：对象没有其他引用时自动从缓存消失
WEAK_CACHE: "weakref.WeakValueDictionary[str, Model]" = weakref.WeakValueDictionary()
m = Model("bert")
WEAK_CACHE["bert"] = m
print(list(WEAK_CACHE.keys())) # ['bert']
del m
import gc; gc.collect()
print(list(WEAK_CACHE.keys())) # [] → 自动清理
# 常见搭配：WeakSet 存观察者、weakref.ref(obj, callback) 做清理回调| 常见内存问题 | 典型原因 | 排查 / 解决 |
|---|---|---|
| 内存只增不减 | 全局 dict/list 当缓存无上限 | 换 functools.lru_cache(maxsize=N) 或 WeakValueDictionary |
lru_cache 装在实例方法上 | 缓存键含 self，实例永不释放 | 改用 cachetools + 显式 key，或 @cached_property |
| 闭包/装饰器意外持引用 | wrapper 捕获了大对象 | 只捕获必要的小参数 |
| 大量小对象占内存 | 每实例一个 __dict__ | __slots__ 或 dataclass(slots=True)（3.10+） |
deepcopy 配置对象 | 复制了整棵对象树 | 换不可变配置 / copy.replace 语义 |
| 子进程内存累积 | 长寿命 worker 持有缓存 | Pool(maxtasksperchild=N) 定期回收 |
⚠️
tracemalloc是定位内存增长的首选工具，比猜测靠谱得多：
python
import tracemalloc
tracemalloc.start(25)
snap1 = tracemalloc.take_snapshot()
# ... 跑一轮业务 ...
snap2 = tracemalloc.take_snapshot()
for stat in snap2.compare_to(snap1, "lineno")[:10]:
print(stat) # 按「哪一行新增了多少内存」排序面试加分表述：「引用计数是主力、分代 GC 兜底循环引用，这是 CPython 的基本盘。我们排查过一次常驻内存从 2G 涨到 8G 的问题：原因是
@lru_cache装在了实例方法上，缓存键里带着self，导致每个请求创建的 service 实例都被缓存永久持有。定位靠的是tracemalloc快照对比加objgraph看反向引用链。后来把高频小对象改成dataclass(slots=True)，向量元数据那块内存又降了 60%。」
26. AI 工程化：依赖管理、项目结构、配置与可观测
前端同学切到 Python 最容易「水土不服」的地方不是语法，是工程化约定。这里给一份对照表直接迁移心智模型。
| 前端 | Python 等价 | 说明 |
|---|---|---|
package.json | pyproject.toml | 现代标准（PEP 621），取代 setup.py / requirements.txt |
package-lock.json | uv.lock / poetry.lock | 锁定精确版本与哈希 |
node_modules | .venv（虚拟环境） | Python 没有项目级自动隔离，必须手动建 venv |
npm i / pnpm i | uv sync / poetry install | pip install 不写锁文件，不适合生产 |
npx | uvx / pipx run | 临时运行 CLI 工具 |
nvm | uv python install 3.12 / pyenv | 多版本管理 |
| ESLint + Prettier | ruff（二合一，Rust 写的，极快） | ruff check + ruff format |
TypeScript tsc | mypy / pyright | 注解只在这一步校验，运行时不管 |
| Jest / Vitest | pytest + pytest-asyncio | 异步测试要装 pytest-asyncio |
| husky + lint-staged | pre-commit | 同样是 git hook 框架 |
bash
# 推荐工具链：uv（2024 年后的事实标准，比 pip/poetry 快 10~100 倍）
uv init my-ai-service && cd my-ai-service
uv python install 3.12
uv add fastapi uvicorn openai pydantic-settings httpx
uv add --dev pytest pytest-asyncio ruff mypy
uv run uvicorn app.main:app --reload # 自动用项目 venv，无需 activate
uv sync --frozen # CI / 生产：严格按锁文件安装toml
# pyproject.toml
[project]
name = "my-ai-service"
requires-python = ">=3.11"
dependencies = ["fastapi>=0.115", "openai>=1.40", "pydantic-settings>=2.4"]
[tool.ruff]
line-length = 100
[tool.ruff.lint]
# B006 = 可变默认参数；ASYNC = 异步代码里的阻塞调用检测（强烈建议开）
select = ["E", "F", "I", "B", "UP", "ASYNC", "RUF"]
[tool.mypy]
python_version = "3.11"
strict = true
warn_unused_awaits = true
[tool.pytest.ini_options]
asyncio_mode = "auto" # 省掉每个异步测试上的 @pytest.mark.asyncio⚠️
ruff的ASYNC规则组值得单独开启：它能静态查出「异步函数里调用了time.sleep/requests.get/open()」这类阻塞事件循环的代码（ASYNC101、ASYNC210等），正好覆盖本文反复强调的最大坑。配进 CI 比靠人眼 review 可靠得多。
配置管理：pydantic-settings 取代到处散落的 os.getenv
python
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_prefix="APP_")
openai_api_key: str # 缺失直接启动失败
llm_timeout: float = Field(default=30.0, gt=0)
max_concurrency: int = Field(default=10, ge=1, le=200)
env: str = "dev"
settings = Settings() # ⚠️ 启动时校验，比运行到一半才发现配置错强太多
# ❌ 错误写法：散落各处，拼写错了只会拿到 None，排查半天
# key = os.getenv("OPENAI_APIKEY") # 少了下划线，静默为 None最小可用项目结构：
text
my-ai-service/
├── pyproject.toml / uv.lock
├── .env # 本地配置，必须 gitignore
├── .env.example # 提交到仓库，列出所有需要的 key
├── app/
│ ├── main.py # FastAPI 入口，lifespan 里建/销毁连接池
│ ├── settings.py # pydantic-settings 配置
│ ├── api/ # 路由层：只做参数校验和编排
│ ├── services/ # 业务层：RAG 编排、Agent 流程
│ ├── clients/ # 外部依赖：LLM、向量库、DB（统一超时重试）
│ ├── schemas/ # pydantic 模型（请求/响应/LLM 结构化输出）
│ └── core/ # 装饰器、日志、异常、并发工具
└── tests/python
# app/main.py —— 用 lifespan 管理全局资源（替代已废弃的 on_event）
from contextlib import asynccontextmanager
from fastapi import FastAPI
import httpx
@asynccontextmanager
async def lifespan(app: FastAPI):
# ✅ 连接池、线程池、模型只在启动时建一次
app.state.http = httpx.AsyncClient(timeout=30, limits=httpx.Limits(max_connections=100))
yield
await app.state.http.aclose() # 关闭时优雅释放
app = FastAPI(lifespan=lifespan)⚠️ 每次请求都
httpx.AsyncClient()是高频性能事故：连接无法复用，每次都重新 TLS 握手，并且 socket 会堆积到TIME_WAIT。客户端必须全局单例，在lifespan里创建和关闭。
面试加分表述：「我们用 uv 统一依赖和 Python 版本，CI 里
uv sync --frozen保证线上和本地完全一致；代码规范用 ruff 一个工具搞定 lint + format，而且开了 ASYNC 规则组，能静态拦住『异步函数里写 time.sleep』这种最常见的事故。配置全部收到pydantic-settings，启动时校验，少一个环境变量服务直接起不来，而不是跑到半夜报 NoneType。」
27. LLM 并发调用工程化：限流、重试、超时、熔断、Token 预算
这一节把前面所有知识点拼成生产可用的形态，也是面试里最容易出亮点的部分。
text
一次 LLM 调用要穿过的五道闸门
请求
│
├─[1] 并发闸门 Semaphore / 连接池上限 ── 保护下游 & 自己的内存
├─[2] 限流闸门 RPM/TPM 令牌桶 ── 遵守厂商配额，避免 429 风暴
├─[3] 超时闸门 timeout_at 统一 deadline ── 防止被慢请求拖死
├─[4] 重试闸门 指数退避 + 抖动 + 只重试可重试错误
└─[5] 熔断闸门 连续失败直接快速失败 ── 避免雪崩
│
下游 LLM| 闸门 | 工具 | 关键参数 | 不做会怎样 |
|---|---|---|---|
| 并发 | asyncio.Semaphore | 一般 5~20 | 瞬间几千并发打满下游，自己也 OOM |
| 限流 | 令牌桶（RPM + TPM 双维度） | 按厂商配额 | 429 后无脑重试 → 自我 DDoS |
| 超时 | asyncio.timeout_at | 总预算 + 单跳预算 | 一个慢请求占着连接不放 |
| 重试 | tenacity 或手写退避 | 退避基数 + 抖动 + 最大次数 | 要么不重试丢可用性，要么重试放大故障 |
| 熔断 | 状态机（闭合/打开/半开） | 失败率阈值 + 冷却时间 | 下游挂了还在猛打，恢复更慢 |
python
import asyncio, random, time
from dataclasses import dataclass, field
# ---------- ① 并发 + ② 限流（RPM 令牌桶） ----------
class RateLimiter:
"""按「每分钟请求数」平滑放行，不是粗暴的整分钟重置。"""
def __init__(self, rpm: int):
self.interval = 60.0 / rpm
self._lock = asyncio.Lock() # ⚠️ 跨 await 的临界区必须加锁
self._next = 0.0
async def acquire(self) -> None:
async with self._lock:
now = time.monotonic()
wait = max(0.0, self._next - now)
self._next = max(now, self._next) + self.interval
if wait:
await asyncio.sleep(wait) # 锁外 sleep，避免持锁等待
# ---------- ⑤ 熔断 ----------
@dataclass
class CircuitBreaker:
fail_threshold: int = 5
cooldown: float = 30.0
_fails: int = 0
_open_until: float = 0.0
def allow(self) -> bool:
return time.monotonic() >= self._open_until
def on_success(self) -> None:
self._fails = 0
def on_failure(self) -> None:
self._fails += 1
if self._fails >= self.fail_threshold:
self._open_until = time.monotonic() + self.cooldown
self._fails = 0
class CircuitOpen(RuntimeError): ...
# ---------- 组装 ----------
RETRYABLE = (asyncio.TimeoutError, ConnectionError)
class LLMClient:
def __init__(self, rpm: int = 60, concurrency: int = 10):
self.sem = asyncio.Semaphore(concurrency)
self.limiter = RateLimiter(rpm)
self.breaker = CircuitBreaker()
async def chat(self, prompt: str, *, deadline: float | None = None,
retries: int = 3) -> str:
if not self.breaker.allow():
raise CircuitOpen("LLM 熔断中，直接走降级")
last: BaseException | None = None
for attempt in range(retries):
try:
async with self.sem: # ① 并发闸门
await self.limiter.acquire() # ② 限流闸门
budget = 10.0 if deadline is None else deadline - time.monotonic()
if budget <= 0:
raise asyncio.TimeoutError("预算已耗尽")
async with asyncio.timeout(budget): # ③ 超时闸门
res = await self._raw_call(prompt)
self.breaker.on_success()
return res
except asyncio.CancelledError:
raise # ⚠️ 取消必须放行，不能当失败重试
except RETRYABLE as e: # ④ 只重试可重试错误
last = e
self.breaker.on_failure()
if attempt < retries - 1:
# 指数退避 + 抖动：避免所有失败请求同一时刻重试造成尖峰
backoff = min(8.0, 0.5 * 2 ** attempt) * (0.5 + random.random())
await asyncio.sleep(backoff)
except Exception as e: # 4xx 这类不可重试，直接失败
self.breaker.on_failure()
raise
raise last # type: ignore[misc]
async def _raw_call(self, prompt: str) -> str:
await asyncio.sleep(0.2)
return f"answer({prompt})"python
# ---------- 批量调用：保持顺序 + 部分失败可降级 ----------
async def batch_chat(client: LLMClient, prompts: list[str]) -> list[str | None]:
deadline = time.monotonic() + 30.0 # 整批共享总预算
results = await asyncio.gather(
*(client.chat(p, deadline=deadline) for p in prompts),
return_exceptions=True, # ⚠️ 不让单个失败炸掉整批
)
out: list[str | None] = []
for p, r in zip(prompts, results):
if isinstance(r, BaseException):
print(f"降级：{p[:20]} → {r!r}")
out.append(None) # 这里可以换成缓存结果 / 小模型兜底
else:
out.append(r)
return out # ✅ gather 保证顺序与输入一致| 实战经验 | 说明 |
|---|---|
| 并发不是越高越好 | LLM 厂商侧有配额，超了只会换成 429 + 退避，净吞吐反而下降 |
| 重试必须带抖动 | 否则 N 个失败请求同时退避、同时重试，形成周期性尖峰 |
| 区分可重试错误 | 429/5xx/超时可重试；400 参数错、401 鉴权错重试纯属浪费钱 |
| 流式请求重试要谨慎 | 已输出部分 token 再重试会重复内容，需按「是否已输出」决定 |
| Token 预算要前置估算 | 用 tiktoken 预估，超限先截断上下文，而不是等厂商报错 |
| 必须埋点 | 每次调用记录：模型、输入/输出 token、耗时、重试次数、是否命中缓存 |
⚠️ 幂等性：重试写操作前必须确认下游幂等。LLM 调用本身无副作用可以放心重试，但「调完就扣费 / 写库 / 发消息」的链路一定要带幂等键，否则重试会导致重复扣费。
面试加分表述：「我们的 LLM 客户端是五层闸门：Semaphore 控并发、令牌桶控 RPM/TPM、
timeout_at给整批共享 deadline、按错误类型做指数退避加抖动的重试、外面再套熔断。几个细节是踩坑换来的：重试不带抖动会出现周期性尖峰；CancelledError必须显式放行不能当失败重试，否则服务关不掉；批量用gather(return_exceptions=True)拿到失败清单做小模型降级，可用性从 97% 提到 99.9%。」
28. 版本演进：3.11 性能、3.12 特性、3.13 no-GIL 与子解释器
面试里被问「你用哪个版本、为什么」时，能说清版本收益就是加分项。
| 版本 | 关键变化 | 对 AI 工程的意义 |
|---|---|---|
| 3.11 | 性能提升 10~60%（Faster CPython）；asyncio.TaskGroup、asyncio.timeout；ExceptionGroup / except*；异常回溯带列号 | 升级性价比最高的一版，纯升级就能提速 |
| 3.12 | f-string 语法放宽（可嵌套引号）；type 语句定义泛型别名；@override 装饰器；per-interpreter GIL 基础（PEP 684） | 类型表达更顺手 |
| 3.13 | 实验性 free-threaded 构建（PEP 703，可关 GIL）；JIT 雏形；REPL 大幅改进 | 多核 CPU 并行的长期解法 |
| 3.14 | free-threaded 转为正式支持；子解释器进标准库（concurrent.interpreters） | 「多线程真并行」开始可用 |
python
# ---- 3.11：TaskGroup 替代裸 gather，异常不再泄漏 ----
import asyncio
async def main():
try:
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(fetch_a())
t2 = tg.create_task(fetch_b())
# 出 with 块时全部完成；任一失败 → 自动取消兄弟任务 + 抛 ExceptionGroup
print(t1.result(), t2.result())
except* ValueError as eg:
for e in eg.exceptions:
print("参数类错误：", e)
except* ConnectionError as eg:
print("网络类错误：", len(eg.exceptions), "个")
# ---- 3.12：type 语句 + @override ----
type Embedding = list[float] # 泛型别名，比 TypeAlias 更简洁
type Result[T] = tuple[T, float]
from typing import override
class Child(Base): # 伪代码
@override # 父类没这个方法时静态检查会报错
def run(self) -> None: ...free-threaded（no-GIL）到底意味着什么：
text
传统 CPython free-threaded CPython (3.13+)
┌──────────────────┐ ┌──────────────────┐
│ 线程1 ──┐ │ │ 线程1 ── CPU核1 │
│ 线程2 ──┼─ GIL ─│─ 1 核 │ 线程2 ── CPU核2 │ 真并行
│ 线程3 ──┘ │ │ 线程3 ── CPU核3 │
└──────────────────┘ └──────────────────┘
代价：
· 单线程性能下降（早期约 5~10%，用偏向引用计数等手段在收敛）
· 大量 C 扩展需要重新适配（numpy/pandas 等已陆续支持）
· 原先「靠 GIL 偷来的原子性」全部失效 → 必须显式加锁
★ 重点：GIL 消失后，data race 会真正出现。
之前 list.append 多线程还算安全（靠 GIL），no-GIL 下必须自己保证。bash
# 3.13+ 需要专门的 free-threaded 构建
uv python install 3.13t # 't' 后缀 = free-threaded
python3.13t -c "import sys; print(sys._is_gil_enabled())" # False
PYTHON_GIL=0 python3.13t app.py # 运行时控制子解释器（PEP 684 / 734）：第三条路
| 方案 | 隔离级别 | 启动开销 | 数据共享 | 真并行 |
|---|---|---|---|---|
| 多线程（有 GIL） | 共享一切 | 极低 | 直接共享 | ❌ |
| 子解释器（3.12+ per-interpreter GIL） | 独立解释器状态 | 低（毫秒级） | 需通过 channel 传递 | ✅ |
| 多进程 | 完全隔离 | 高（百毫秒级） | pickle / 共享内存 | ✅ |
⚠️ 当前（2026 年）的工程建议：生产环境主用 3.11 或 3.12——3.11 的性能提升和
TaskGroup/asyncio.timeout是实打实的收益，生态也完全成熟。free-threaded 构建可以在内部 CPU 密集服务上试点，但要先确认所有 C 扩展依赖都支持，否则会直接 import 失败或性能倒退。
面试加分表述：「我们生产用 3.12，升级到 3.11+ 主要是三个收益：Faster CPython 带来的纯性能提升、
TaskGroup让并发任务不会泄漏、ExceptionGroup+except*可以按错误类型分别处理一批子任务的异常。3.13 的 free-threaded 我们在离线特征计算的服务上试过，确实吃满了多核，但要注意两点：单线程性能有损耗，而且原来靠 GIL 偷来的原子性全失效了，list.append、计数器这类都必须显式加锁。更长远看，子解释器可能是比多进程更轻的真并行方案。」
29. 异常模型与错误分类：重试策略的地基
前面的重试装饰器、gather(return_exceptions=True)、CancelledError 不能吞，全都建立在「异常体系」这一层上。Java 转 Python 的人最容易在这里出事：Python 没有 checked exception，编译器不会提醒你漏了什么，所有错误处理的正确性都靠你自己对继承树的理解。
text
BaseException ← 捕获它 = 连 Ctrl+C 和超时取消都吞掉
├── SystemExit sys.exit() 触发，不该被业务代码捕获
├── KeyboardInterrupt Ctrl+C，不该被业务代码捕获
├── GeneratorExit 生成器 close() 时注入，不是「错误」
└── Exception ← 业务代码应该只捕获这一层及以下
├── asyncio.CancelledError ⚠️ 3.8+ 已移出 Exception，挂在 BaseException 下
├── ArithmeticError / LookupError / OSError ...
│ ├── TimeoutError 3.11+ asyncio.TimeoutError 是它的别名
│ └── ConnectionError → ConnectionResetError / ConnectionRefusedError
├── ValueError / TypeError / KeyError / AttributeError
└── ExceptionGroup（3.11+） 一次抛出多个异常，配 except* 使用
★ 记住两条线：
① CancelledError 在 BaseException 下 → except Exception 抓不到（这是好事）
② 但 except BaseException 和裸 except: 会把它一起吞掉（这是事故）四种捕获写法的实际差别
python
# ❌ 最危险：裸 except，连 Ctrl+C、SystemExit、CancelledError 一起吞
try:
await call_llm(prompt)
except: # noqa: E722
pass # 服务杀不掉、请求取消不掉、错误还被静默
# ❌ 次危险：except BaseException，同样吞掉取消信号
try:
await call_llm(prompt)
except BaseException as e:
log.error(e) # 超时取消后这里还会继续往下跑
# ⚠️ 可用但粗糙：范围对了，但分不清「能不能重试」
try:
await call_llm(prompt)
except Exception as e:
log.exception("llm failed")
return fallback() # 参数写错（ValueError）也被当成下游故障重试
# ✅ 推荐：先放行取消，再按错误类型分流
try:
await call_llm(prompt)
except asyncio.CancelledError:
raise # 协作式取消，必须原样向上
except (TimeoutError, ConnectionError) as e:
raise RetryableError(str(e)) from e # 下游抖动 → 可重试
except (ValueError, TypeError) as e:
raise FatalError(str(e)) from e # 自己的 bug → 重试没意义异常链：raise ... from 不是语法糖
python
# ❌ 丢上下文：只知道「解析失败」，不知道原始报文和底层报错
try:
data = json.loads(raw)
except json.JSONDecodeError:
raise LLMOutputError("模型返回不是合法 JSON")
# traceback 里只有 "During handling of the above exception, another occurred"
# —— 隐式链（__context__），表达的是「处理旧异常时又出错了」
# ✅ 显式声明因果关系（__cause__），traceback 变成 "The above was the direct cause of"
try:
data = json.loads(raw)
except json.JSONDecodeError as e:
raise LLMOutputError(f"模型返回不是合法 JSON: {raw[:200]}") from e
# ✅ 需要刻意屏蔽底层细节（比如对外暴露的 API）时，用 from None 断开链
try:
user = db.get(uid)
except DBConnectionError:
raise HTTPException(503, "service unavailable") from None| 写法 | 设置的属性 | traceback 文案 | 什么时候用 |
|---|---|---|---|
raise New(...) | __context__（自动） | During handling of... | 临时写代码可以，正式代码不推荐 |
raise New(...) from e | __cause__ | The above was the direct cause of... | 默认选这个，因果关系明确 |
raise New(...) from None | 抑制链 | 只显示新异常 | 对外 API 不想泄漏内部栈 |
裸 raise | 不变 | 原样 | 只想记日志/做清理，不改变异常 |
ExceptionGroup 与 except*（3.11+）：TaskGroup 的配套设施
TaskGroup 里可能同时有多个子任务失败，单个异常装不下，所以 3.11 引入了 ExceptionGroup。except* 的语义和 except 完全不同：它会把组里匹配的异常挑出来处理，剩下的继续往上抛，而且多个 except* 分支可以同时命中。
python
import asyncio
async def main(prompts: list[str]):
try:
async with asyncio.TaskGroup() as tg:
for p in prompts:
tg.create_task(call_llm(p))
except* TimeoutError as eg:
# eg 是 ExceptionGroup，eg.exceptions 是本次匹配到的所有超时异常
log.warning("超时 %d 个，准备降级", len(eg.exceptions))
except* ValueError as eg:
log.error("参数错误 %d 个，不重试", len(eg.exceptions))
# 没被任何 except* 匹配的异常会原样继续抛出，不会被静默
# 想按条件切分一个 ExceptionGroup，用 .split() / .subgroup()
def is_retryable(exc: BaseException) -> bool:
return isinstance(exc, (TimeoutError, ConnectionError))
retryable, fatal = eg.split(is_retryable) # 返回两个 ExceptionGroup 或 None⚠️
except和except*不能在同一个try里混用，语法层面就禁止。另外except*的分支里不能用break/continue/return跳出。
finally 的三个隐藏陷阱
python
# ❌ 陷阱一：finally 里 return，会把正在传播的异常「吃掉」
def bad():
try:
raise ValueError("真正的错误")
finally:
return "ok" # 异常凭空消失，调用方拿到 "ok"
# ❌ 陷阱二：finally 里又抛异常，会覆盖原异常（原异常降级成 __context__）
def bad2():
try:
raise ValueError("真正的错误")
finally:
conn.close() # 如果 close 自己抛了，排查方向就被带偏
# ✅ 清理代码要自己兜住，别让它遮住真正的错误
def good():
try:
do_work()
finally:
try:
conn.close()
except Exception:
log.warning("close failed", exc_info=True)AI 工程落地：把下游错误分成三类，驱动重试策略
python
class AppError(Exception):
"""所有自定义异常的基类，便于中间件统一兜底。"""
retryable = False
http_status = 500
class RetryableError(AppError): # 下游抖动：超时、429、5xx、连接重置
retryable = True
http_status = 503
class FatalError(AppError): # 自己的 bug：参数非法、prompt 模板错
retryable = False
http_status = 400
class DegradableError(AppError): # 可降级：rerank 挂了就用原始召回顺序
retryable = False
http_status = 200
def classify(exc: BaseException) -> type[AppError]:
"""把第三方 SDK 五花八门的异常，归一成自己的三类语义。"""
if isinstance(exc, (TimeoutError, ConnectionError)):
return RetryableError
status = getattr(exc, "status_code", None)
if status == 429 or (isinstance(status, int) and status >= 500):
return RetryableError
if status in (400, 401, 403):
return FatalError # 重试只会浪费配额和时间
return FatalError⚠️ 重试前必须先分类。LLM 网关最常见的浪费是「把 400 参数错误也重试 3 次」——不仅不会成功，还把错误放大 3 倍打到下游，429 的时候更会直接雪崩。重试的前提是「这次失败是瞬时的」。
面试加分表述：「我们在
clients/层把各家 LLM SDK 的异常统一翻译成RetryableError/FatalError/DegradableError三类，重试装饰器只认RetryableError，所以 400 这种参数错误会立刻失败并告警，而 429 和 5xx 走退避重试。异常全部用raise ... from e保留原始栈，对外响应时再from None掉内部细节。用了TaskGroup之后批量任务的异常是ExceptionGroup，我们用except*分别处理超时和参数错误——这比以前gather拿一个 list 回来自己isinstance判断清楚太多了。」
30. 异步代码怎么测：pytest-asyncio、AsyncMock、假时钟与并发断言
本文的 5 道手撕题每题都给了「边界 Case 自测清单」，但这些 Case 怎么真的跑起来是另一回事。异步代码的测试有三个独有难点：① 测试函数本身必须被事件循环驱动；② 下游是协程，普通 Mock 不能 await；③ 超时、退避、TTL 这类逻辑如果真等就会让单测跑几分钟。
① 基础设施：pytest-asyncio 的 auto 模式
toml
# pyproject.toml —— 配好这两行，异步测试函数不用逐个加 marker
[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function" # 每个用例独立 loop，避免互相污染python
# ❌ 错误写法一：忘了 asyncio_mode，测试函数被当成普通函数
def test_call():
result = call_llm("hi") # 拿到的是协程对象，断言永远「通过」
assert result # 协程对象永远为真 → 假绿灯，还会报 coroutine never awaited
# ❌ 错误写法二：在测试里自己 asyncio.run，和 pytest-asyncio 的 loop 打架
def test_call2():
assert asyncio.run(call_llm("hi")) # fixture 里建的连接池绑在另一个 loop 上
# ✅ 正确写法：直接写 async def，由插件注入事件循环
async def test_call():
assert await call_llm("hi") == "pong"⚠️ 「协程没被 await」是异步测试里最隐蔽的假绿灯。把
-W error::RuntimeWarning配进 pytest，让coroutine was never awaited直接变成测试失败，而不是一行容易被忽略的警告。
② mock 异步下游：AsyncMock 而不是 Mock
python
from unittest.mock import AsyncMock, MagicMock, patch
# ❌ Mock() 返回的是普通对象，await 它会报 TypeError: object Mock can't be used in await
client = MagicMock()
client.acreate.return_value = "fake"
# ✅ AsyncMock 的调用结果是协程，可以被 await
client = AsyncMock()
client.acreate.return_value = "fake"
assert await client.acreate("hi") == "fake"
# ✅ 用 side_effect 模拟「前两次超时，第三次成功」——专门验证重试逻辑
client.acreate.side_effect = [TimeoutError(), TimeoutError(), "ok"]
async def test_retry_until_success():
with patch("app.clients.llm", client):
assert await call_llm("hi") == "ok"
assert client.acreate.await_count == 3 # 注意是 await_count 不是 call_count| 断言 | 含义 | 常见误用 |
|---|---|---|
call_count | 被调用次数（拿到协程就算一次） | 异步场景下它不代表真的执行完了 |
await_count | 被 await 次数 | 验证重试次数应该用这个 |
assert_awaited_with(...) | 最后一次 await 的参数 | 用 assert_called_with 不校验 await |
await_args_list | 所有 await 的参数列表 | 验证批量调用的入参顺序 |
③ 不要真等：超时、退避、TTL 的时间控制
python
# ❌ 错误写法：单测里真的等 20 秒超时 + 两次指数退避，一个用例跑 30 秒
async def test_timeout_slow():
with pytest.raises(TimeoutError):
await call_llm("hi") # 用的是生产默认 timeout=20
# ✅ 做法一：把超时/退避做成可注入参数，测试时传极小值
async def test_timeout_fast():
with pytest.raises(TimeoutError):
await call_llm("hi", timeout=0.01, base_delay=0.001)
# ✅ 做法二：把时间源抽成可替换的 clock，用假时钟零等待验证 TTL
class FakeClock:
def __init__(self) -> None:
self.now = 1000.0
def __call__(self) -> float:
return self.now
def advance(self, sec: float) -> None:
self.now += sec
def test_ttl_expire():
clock = FakeClock()
calls = []
@ttl_lru_cache(maxsize=8, ttl=300, clock=clock) # 题 5 的变种 1
def f(x):
calls.append(x)
return x * 2
assert f(1) == 2 and f(1) == 2
assert calls == [1] # 第二次命中缓存
clock.advance(301) # 不真等 5 分钟
assert f(1) == 2
assert calls == [1, 1] # TTL 过期后重新执行
# ✅ 做法三：验证「退避确实在睡」而不验证睡了多久——patch 掉 sleep 并记录入参
async def test_backoff_grows():
delays = []
async def fake_sleep(d):
delays.append(d)
with patch("asyncio.sleep", fake_sleep):
with pytest.raises(TimeoutError):
await flaky_call(retries=3, base_delay=1, jitter=False)
assert delays == [1, 2, 4] # 指数退避，且一次都没真的等⚠️
freezegun冻结的是datetime.now()/time.time()，管不到time.monotonic()和asyncio的 loop 时钟。缓存和超时逻辑应该用monotonic（不受系统时间回拨影响），所以测它们要靠「可注入 clock」或patchasyncio.sleep，而不是freezegun。
④ 并发度怎么断言：计数器比耗时更可靠
题 3 的信号量限流只看总耗时是测不准的（CI 机器负载会让耗时忽上忽下）。正确做法是直接数同时在飞的任务数：
python
async def test_concurrency_never_exceeds_limit():
inflight = 0
peak = 0
async def fake_call(prompt: str) -> str:
nonlocal inflight, peak
inflight += 1
peak = max(peak, inflight)
await asyncio.sleep(0) # 交出控制权，制造真实的交错时机
inflight -= 1
return prompt.upper()
with patch("app.llm.call_one", fake_call):
results = await batch_call([f"p{i}" for i in range(50)], limit=5)
assert peak <= 5 # 限流生效
assert len(results) == 50 # 不丢任务
assert results[0] == "P0" and results[-1] == "P49" # ✅ 顺序与入参一致⚠️
await asyncio.sleep(0)是制造并发交错的关键。如果 fake 函数里一个 await 都没有，它会同步跑完，peak永远是 1，限流逻辑写错了也测不出来。
⑤ 测生成器与异步生成器：重点是「清理有没有执行」
题 2 的流式读、题 4 的双协议上下文管理器，核心风险都在「提前退出时资源有没有释放」，这正是单测最该盯的地方：
python
# ✅ 用列表记录生命周期事件，断言 close 真的被调用
def test_generator_cleanup_on_early_break():
events = []
def reader():
events.append("open")
try:
for i in range(1000):
yield i
finally:
events.append("close")
gen = reader()
for i in gen:
if i == 2:
break
gen.close() # 显式关闭，不依赖 GC 时机
assert events == ["open", "close"]
# ✅ 异步生成器必须显式 aclose，否则清理代码可能在 loop 关闭后才跑
async def test_async_generator_cleanup():
events = []
async def areader():
try:
for i in range(1000):
yield i
finally:
events.append("close")
agen = areader()
async for i in agen:
if i == 2:
break
await agen.aclose()
assert events == ["close"]
# ✅ 上下文管理器：异常路径下也要保证退出钩子执行
async def test_cm_releases_on_error():
cm = DualProtocolCM()
with pytest.raises(ValueError):
async with cm:
raise ValueError("boom")
assert cm.released is True⑥ 取消与 ExceptionGroup 的断言写法
python
# ✅ 断言任务真的被取消（而不是被普通异常打断）
async def test_cancel_propagates():
task = asyncio.create_task(long_job())
await asyncio.sleep(0) # 让 task 跑到第一个 await
task.cancel()
with pytest.raises(asyncio.CancelledError):
await task
assert task.cancelled() is True
# ✅ TaskGroup 抛的是 ExceptionGroup，不能直接 pytest.raises(ValueError)
async def test_task_group_collects_errors():
with pytest.raises(BaseExceptionGroup) as ei:
async with asyncio.TaskGroup() as tg:
tg.create_task(fail_with(ValueError("a")))
tg.create_task(fail_with(KeyError("b")))
types = {type(e) for e in ei.value.exceptions}
assert types == {ValueError, KeyError}
# ✅ Python 3.11+ 也可以用 pytest 内置的 group 匹配参数
# with pytest.raises(ExceptionGroup, match="unhandled errors"): ...⚠️
pytest.raises(ValueError)接不住ExceptionGroup，哪怕组里确实有 ValueError。用TaskGroup的代码，测试侧也要同步改成断言 group 再展开.exceptions，否则升级到 3.11 后一片测试会红。
面试加分表述：「异步代码的单测我有三条铁律。第一，所有超时、退避、TTL 的时间参数都必须可注入，测试里传极小值或者用假时钟，CI 上一个用例不该跑超过几十毫秒，更不能靠
time.sleep等真实时间。第二，mock 下游一律用AsyncMock，而且验证重试次数要看await_count不是call_count——Mock返回的协程没被 await，call_count会骗你。第三，限流和并发度不用耗时断言，用 inflight 计数器记峰值，再在 fake 函数里塞一个await asyncio.sleep(0)制造交错，这样限流写错一定会被测出来。另外迁到TaskGroup之后要记得测试侧同步改成pytest.raises(BaseExceptionGroup)再展开.exceptions，不然原来按单个异常类型写的断言会全挂。」
❓ 高频面试题【基础简答 + 连环追问链 + 精简口述答案 + 项目关联话术】
题 1：GIL 是什么？为什么有它？能不能去掉？
精简口述答案（90 秒版）
「GIL 是 CPython 解释器级别的一把互斥锁，保证同一时刻只有一个线程在执行 Python 字节码。它存在的根本原因是 CPython 的内存管理基于引用计数，而引用计数的加减不是原子操作——没有 GIL 的话，两个线程同时改一个对象的 refcount 会导致计数错乱，对象被提前释放或永久泄漏。GIL 是最简单有效的保护方式，代价是多线程无法利用多核做 CPU 密集计算。
要特别区分三点：第一，GIL 是 CPython 的实现细节，不是 Python 语言规范，Jython、IronPython 就没有；第二，GIL 不影响 IO 密集场景，因为线程在等 IO 时会主动释放 GIL，所以多线程爬虫、多线程调接口依然有效；第三，GIL 不等于线程安全，count += 1 会被编译成多条字节码，中间可能被切换，所以共享变量依然要加锁。
能不能去掉？Python 3.13 通过 PEP 703 提供了实验性的 free-threaded 构建，可以真正关掉 GIL，3.14 转为正式支持。代价是单线程性能有几个点的损耗、大量 C 扩展需要重新适配，而且原先靠 GIL『偷来』的原子性全部失效，并发代码必须显式加锁。」
连环追问链
| 追问 | 答题要点 |
|---|---|
| GIL 什么时候会被释放？ | ① 执行 IO 时主动释放；② 时间片到（默认 5ms，sys.getswitchinterval()）；③ 调用 time.sleep；④ C 扩展显式 Py_BEGIN_ALLOW_THREADS（numpy 大矩阵运算就是这么做的） |
| 有 GIL 为什么还要加锁？ | GIL 只保证「单条字节码」不被打断。count += 1 是 LOAD/ADD/STORE 多条字节码，可以用 dis.dis 展示，中间切换就会丢更新 |
| 多线程在 Python 里还有用吗？ | 有。IO 密集完全有效；另外调用释放了 GIL 的 C 扩展（numpy、部分加解密库）时也能真并行 |
| GIL 切换机制有变化过吗？ | 3.2 之前是每 100 条字节码检查一次，之后改为 5ms 时间片。更值得说的是「护航效应」：CPU 线程和 IO 线程混在一个进程时，IO 线程延迟被放大 |
| 去掉 GIL 后有什么新风险？ | 真实的 data race 出现。以前多线程 list.append 还算安全，no-GIL 下必须加锁；单线程性能有损耗；C 扩展兼容性是最大阻力 |
| 实际项目中怎么绕开 GIL？ | CPU 密集走多进程 / 进程池；向量化计算交给 numpy 等会释放 GIL 的库；更重的任务直接下沉到 C++/Rust 扩展或独立服务 |
✅ 项目关联话术：「我们的 AI 服务一开始是用多线程做批量文档解析，8 核机器 CPU 只跑到 110%，这就是典型的 GIL 卡点。改成『解析走
ProcessPoolExecutor、IO 走 asyncio』的混合模型后，吞吐提升了大约 5 倍。还有个更隐蔽的问题：有个接口在同步代码里用 pandas 处理大表，既占 GIL 又堵事件循环，导致整个服务的 P99 从 200ms 恶化到 3 秒，监控上表现为 P50 正常、P99 爆炸——定位到之后我们把这类计算全部挪进进程池隔离。」
题 2：进程、线程、协程在 Python 里怎么选？
精简口述答案（90 秒版）
「一句话决策：CPU 密集用多进程，IO 密集用协程，必须兼容老同步库时用线程池兜底。
原因是三者的开销和并行能力完全不同。进程有独立的解释器和 GIL，能真正吃多核，但启动开销在百毫秒级，参数要 pickle 序列化，内存也会翻倍。线程共享内存、切换成本中等，但受 GIL 限制，只在 IO 等待或调用释放 GIL 的 C 扩展时才有收益。协程是单线程内的用户态切换，一个协程只占几 KB 栈空间，切换开销是纳秒级，单机轻松撑几万并发，但它有个硬前提——全链路必须是非阻塞的，一旦中间有一个同步阻塞调用，整个事件循环就被卡死。
实际项目基本都是混合模型：外层 asyncio 做 IO 编排（调 LLM、查向量库、读 Redis），CPU 密集部分丢给 run_in_executor + 进程池，遗留的同步 SDK 丢给 asyncio.to_thread 的线程池。这样既吃满多核，又保持高并发。
我自己的判断流程是三问：这个任务在等 CPU 还是等 IO？用的库有没有异步版本？并发量级是几十还是几万？」
连环追问链
| 追问 | 答题要点 |
|---|---|
| 协程为什么比线程轻？ | 线程栈默认 MB 级且由内核调度（切换需陷入内核态、保存寄存器上下文），协程只保存少量帧对象在堆上，用户态切换，无系统调用 |
| 一个 asyncio 服务能开多少并发？ | 受限于 fd 上限、内存、下游承载能力，而非协程本身。几万协程很常见，但一定要用 Semaphore 限制对下游的实际并发 |
| 既然协程好，为什么还要线程池？ | 生态里大量库只有同步版本（老 ORM、厂商 SDK、OCR 库），没法重写时只能用线程池隔离阻塞 |
| 多进程为什么不是免费的？ | 三笔账：启动开销（spawn 百毫秒级）、pickle 序列化开销、内存翻倍。传大 DataFrame 时可能 80% 时间花在搬数据上 |
| 线程池大小怎么定？ | 不是越大越好。公式思路：线程数 ≈ 目标并发 × 阻塞时长 / 可接受延迟；且必须按下游隔离成多个池 + 配队列上限和拒绝策略 |
| 怎么验证自己选对了？ | 压测看指标：CPU 跑不满 → GIL 或阻塞问题；CPU 满但 QPS 低 → 真 CPU 瓶颈；内存暴涨 → 并发无限流 |
✅ 项目关联话术：「我们 RAG 服务的最终形态是三层：最外层 asyncio 并发调用检索和 LLM；文档切分、向量归一化这类 CPU 活走全局
ProcessPoolExecutor，启动时建一次、关闭时 shutdown，绝不每次请求新建；老的 OCR SDK 只有同步版本，我们给它单独配了一个 8 线程的池，和老数据库 SDK 的池完全隔离，谁慢谁自己排队、不会互相传染。另外traceId这类contextvars在run_in_executor里会丢，我们封了一个run_sync()内部用copy_context().run()透传，保证日志链路不断。」
题 3：Python 的可变对象和不可变对象有什么区别？函数默认参数为什么不能用可变对象？
精简口述答案（90 秒版）
「首先要纠正一个心智模型：Python 里变量是贴在对象上的名字，不是装东西的盒子。赋值永远是「让名字指向另一个对象」，不是「把值搬进变量」。
在这个前提下，可变对象（list、dict、set、自定义类实例）支持原地修改，id 不变；不可变对象（int、str、tuple、frozenset）任何「修改」都是创建新对象。最典型的对比是 b = b + [4] 和 b += [4]：前者创建新 list 并让 b 指向它，原对象不受影响；后者等价于 b.extend([4])，是原地修改，所有指向这个 list 的名字都会看到变化。
默认参数的坑在于：默认值对象在函数定义时只创建一次，之后所有调用共享它。所以 def f(items=[]) 里的那个空列表是函数对象的一部分，存在 f.__defaults__ 里，第一次调用往里 append，第二次调用就不是空列表了。正确写法是默认值用 None，在函数体里判空再创建。ruff 的 B006 规则可以静态拦住这类代码，建议直接配进 CI。
补充一个容易混的点：不可变容器装可变对象，内部依然可变。t = (1, [2, 3]) 这个 tuple 不可变指的是「它的元素引用不能换」，但 t[1].append(4) 完全合法——所以这种 tuple 不可哈希，不能做字典键。」
连环追问链
| 追问 | 答题要点 |
|---|---|
is 和 == 的区别？ | is 比 id（是否同一对象），== 比值（调 __eq__）。小整数缓存（-5~256）和字符串驻留会让 is 偶然为 True，所以只有和 None/单例比较才用 is |
a += b 和 a = a + b 对 list 有区别吗？ | 有。+= 调 __iadd__ 原地扩展，共享该对象的其他名字都会看到；a + b 产生新对象。对 tuple/str 则没区别（都只能新建） |
| 为什么可变默认参数的问题这么隐蔽？ | 单次调用、单元测试都正常，只有在「被多次调用」时才暴露，表现为数据莫名叠加。而且常见于「缓存历史消息」这类 AI 场景 |
| 哪些对象可以做字典键？ | 可哈希的（实现 __hash__ 且不可变）。装了 list 的 tuple 不可哈希；自定义类默认按 id 哈希，重写 __eq__ 必须同时重写 __hash__ |
| 函数参数传递是值传递还是引用传递？ | 都不是，是「对象引用的传值」（call by object reference）。重新赋值不影响调用方，原地修改会影响 |
| 工程上怎么系统性避免？ | ① 默认值一律 None；② ruff 开 B006/B008；③ 配置类用 frozen=True 的 dataclass 或 pydantic 的不可变模型；④ 跨边界传数据时明确「谁拥有这份数据」 |
✅ 项目关联话术：「这个坑我们在 Agent 的会话管理上真踩过：
def chat(msg, history=[])想让 history 默认空列表，结果所有用户共用了同一个 list——A 用户的对话被追加到了 B 用户的上下文里，既串了数据又把 prompt 撑爆导致 token 超限报错。改成history=None+ 函数内history = history or []之后立刻就好了，同时在 CI 的 ruff 配置里打开了 B006，彻底杜绝同类问题。另外我们所有配置对象都用不可变模型，避免某个模块偷偷改了全局配置导致别的模块行为异常。」
题 4：深拷贝和浅拷贝的区别？循环引用怎么处理？
精简口述答案（90 秒版）
「有三个层次，必须区分清楚：
赋值 b = a 完全不拷贝，只是多一个名字指向同一对象，b is a 为 True。
浅拷贝 copy.copy(a) 或 a[:] / list(a) / dict(a)，创建新的外层容器，但内部元素仍是原来的引用。所以改嵌套层会互相影响。
深拷贝 copy.deepcopy(a) 递归复制整棵对象树，新旧完全独立。
循环引用是深拷贝必须解决的问题：如果 a 里引用了 b，b 里又引用 a，朴素递归会无限栈溢出。deepcopy 内部维护一个 memo 字典，键是原对象的 id，值是已经创建的副本——每次递归前先查 memo，命中就直接返回已有副本，这样环就被正确地复制成环而不是无限展开。
工程上要特别注意 deepcopy 是性能杀手：它要遍历整棵对象树，还要维护 memo，对大配置对象、大 DataFrame、带模型句柄的对象可能慢几个数量级，甚至因为复制了不该复制的东西（socket、锁、线程池）直接报错。更好的做法是让对象不可变、或者只拷贝真正需要隔离的那一层。需要自定义拷贝行为时可以实现 __deepcopy__ 和 __copy__。」
连环追问链
| 追问 | 答题要点 |
|---|---|
| 浅拷贝有哪几种写法？ | copy.copy(x)、切片 x[:]、list(x)/dict(x)/set(x)、x.copy()。效果一致，都只复制一层 |
| 不可变对象 deepcopy 会怎样？ | 直接返回原对象（没必要复制），这是 deepcopy 的优化之一。所以 deepcopy(("a", 1)) is ("a", 1) 可能为 True |
| 怎么自定义拷贝行为？ | 实现 __copy__(self) 和 __deepcopy__(self, memo)。典型用途：让对象在 deepcopy 时复用模型句柄/连接池而不是复制 |
deepcopy 会调 __init__ 吗？ | 不会。默认走 __reduce_ex__，直接创建对象并填充 __dict__，所以 __init__ 里的副作用不会重放 |
| 什么时候必须深拷贝？ | 要把一份数据交给可能修改它的下游，且不能影响原始数据时。比如把 base prompt 模板交给多个 Agent 各自改写 |
| 有没有更轻的替代方案？ | ① 用不可变数据结构（frozen dataclass、tuple）；② 只手动拷贝需要隔离的字段；③ 用 model_copy(deep=True) 这类框架提供的精确拷贝；④ 用 JSON 序列化反序列化（仅限纯数据） |
✅ 项目关联话术：「我们做多 Agent 编排时，每个 Agent 要基于同一份基础上下文做自己的改写。最初直接传引用，结果 A Agent 改了消息列表，B Agent 看到的上下文就被污染了，输出完全乱掉。改成深拷贝后正确了，但性能又出问题——上下文里挂着检索到的文档全文，一次 deepcopy 要几十毫秒，QPS 上来之后 CPU 直接打满。最终方案是把上下文拆成『不可变的共享部分（文档、系统 prompt）』和『每个 Agent 独立的可变部分（消息列表）』，只拷贝后者，耗时降到微秒级。」
题 5：装饰器的原理是什么？为什么要用 functools.wraps？
精简口述答案（90 秒版）
「装饰器的本质就一行等价代码：@deco 加在 def f 上，等于 f = deco(f)。它利用了 Python 里函数是一等公民这个特性——函数可以作为参数传入、作为返回值返回、还能通过闭包携带状态。
有三个关键点：
第一，装饰器在函数定义时就执行了，不是调用时。模块一 import，装饰器外层代码就跑了一遍。所以装饰器里不要放耗时操作或有副作用的初始化。这也是 @app.get("/x") 这类注册型装饰器能工作的原理——import 阶段就把路由注册进去了。
第二，不加 functools.wraps 会丢元信息。包装后的函数实际是 wrapper，它的 __name__ 是 'wrapper'、__doc__ 是 None、__annotations__ 是空的、__wrapped__ 也没有。日志里全是 wrapper 很难排查，更致命的是像 FastAPI、pydantic、Celery 这些依赖类型注解做运行时反射的框架会直接失效——FastAPI 拿不到注解就没法做参数校验和生成 OpenAPI 文档。
第三，带参装饰器是三层函数：最外层接装饰器参数返回真正的装饰器，中间层接被装饰函数返回 wrapper，最内层才是运行时逻辑。所以 @retry 和 @retry() 不是一回事，前者会把函数本身当成第一个参数传进去。」
连环追问链
| 追问 | 答题要点 |
|---|---|
functools.wraps 具体拷了什么？ | 拷 __module__、__name__、__qualname__、__doc__、__annotations__，更新 __dict__，并设置 __wrapped__ 指向原函数（inspect.signature 靠它拿到真实签名） |
| 多个装饰器叠放的顺序？ | 由下往上包装，由上往下执行。@a @b def f 等于 f = a(b(f))。工程规范：注册在最上，限流次之，重试再次，缓存最靠近业务函数——否则重试会穿透缓存反复打下游 |
| 怎么写同时支持同步和异步的装饰器？ | 用 inspect.iscoroutinefunction(fn) 分支，返回 async def 或 def 两个不同的 wrapper。不判断的话同步 wrapper 包异步函数会返回未 await 的协程对象，计时恒为 0 并抛 coroutine was never awaited |
| 类装饰器和装饰类有什么区别？ | 类装饰器是用类的 __call__ 当装饰器（天然带状态，用 functools.update_wrapper）；装饰类是装饰器参数是 class，常用于单例、自动注册 |
| 怎么给装饰器加「可开关」能力？ | wrapper 里读配置/环境变量决定是否走增强逻辑；或在 __wrapped__ 上暴露原函数，让测试可以绕过装饰器 |
| 装饰器会影响性能吗？ | 每层增加一次函数调用开销（几百纳秒级）。热点路径上叠 5 层以上要注意；更大的风险是装饰器里误放了同步 IO 或 deepcopy |
✅ 项目关联话术：「我们所有 LLM 调用都通过一个
@llm_call装饰器统一接管：计时、token 统计、重试退避、异常上报、traceId 透传。两个踩坑换来的经验：一是最初没加functools.wraps，FastAPI 拿不到被装饰函数的类型注解，接口参数校验直接失效、OpenAPI 文档里参数全空，排查了小半天；二是装饰器栈顺序排错了——缓存装在重试外面，导致重试直接穿透缓存反复打 LLM，当天的 token 成本翻了一倍。现在我们把顺序写进了规范：路由注册最外、限流、重试、缓存最里。」
题 6：生成器和迭代器有什么区别？yield 的作用是什么？
精简口述答案（90 秒版）
「先理清三个概念的包含关系：可迭代对象 ⊃ 迭代器 ⊃ 生成器。
可迭代对象只需要实现 __iter__，比如 list；迭代器要同时实现 __iter__ 和 __next__，并且耗尽后不能重来；生成器是 Python 帮你自动生成的迭代器——函数体里只要出现 yield，调用它就不执行函数体，而是返回一个生成器对象。
yield 的核心作用是暂停并保存执行状态。执行到 yield 时，把值交出去，同时把局部变量、指令指针完整保留在生成器的栈帧里；下次 next() 从暂停处继续。这就是协程的底层基础。
生成器最大的价值是惰性求值：内存占用是 O(1) 而不是 O(n)。读 10GB 日志文件，用 readlines() 会直接 OOM，用生成器逐行 yield 则常驻内存只有几 KB。
还有个容易被忽略的双向通信能力：gen.send(value) 可以把值送进生成器，让 yield 表达式求值为这个值；gen.throw(exc) 往里面抛异常；gen.close() 在暂停点注入 GeneratorExit。生成器里 return value 不会作为迭代项产出，而是塞进 StopIteration.value，yield from 会把它作为表达式的值。
最后一个常见坑：生成器只能消费一次，第二次遍历得到空。需要多次遍历就得用 list 物化，或者写成一个每次返回新生成器的函数。」
连环追问链
| 追问 | 答题要点 |
|---|---|
| 可迭代对象和迭代器怎么区分？ | iter(x) is x 为 True 就是迭代器。list 是可迭代但不是迭代器（每次 iter() 给一个新迭代器，所以能重复遍历） |
for 循环底层做了什么？ | it = iter(obj) → 反复 next(it) → 捕获 StopIteration 退出。这就是为什么在业务代码里不要吞掉 StopIteration |
| 生成器表达式和列表推导的区别？ | (x for x in ...) 惰性、内存 O(1)、只能消费一次；[x for x in ...] 立刻物化。处理大数据流一律用前者 |
| 生成器怎么做管道？ | 多个生成器串联：read() → parse() → filter() → batch()，每层只做一件事，整条链保持惰性，内存恒定。这是 ETL / 向量化入库的标准写法 |
yield from 有什么用？ | 委托给子生成器，并透传 send/throw/返回值。扁平化嵌套结构和「旧式协程」都靠它，3.5 之后协程场景已被 async/await 取代 |
| 生成器怎么保证资源释放？ | 用 try/finally 包住，close() 或 GC 时 GeneratorExit 会触发 finally；更稳妥的是消费方用 contextlib.closing/aclosing，防止提前 break 导致迟迟不清理 |
✅ 项目关联话术：「我们做知识库入库时，单个 PDF 解析出来有几十万个文本块。最初一次性 load 成 list 再切分，8G 内存的容器直接 OOM。重构成生成器管道：
读取 → 清洗 → 切块 → 按 batch 聚合 → 调 Embedding → 写向量库，每层都是生成器，常驻内存稳定在 200MB 以内，还能边读边入库，首批数据可用时间从 20 分钟降到 30 秒。有个细节值得一提：中间加了一个batched()生成器把单条聚成 64 条一批，因为 Embedding 接口按批调用吞吐高得多——这个函数 3.12 的itertools.batched已经内置了。」
题 7：asyncio 的事件循环是怎么跑起来的？协程为什么比线程轻？
精简口述答案（90 秒版）
「事件循环本质是一个 while True 的调度器，一轮迭代做四件事：
第一，从 ready 队列取出所有就绪的回调并执行到下一个挂起点；第二，计算最近的定时器到期时间，作为 IO 多路复用的超时参数；第三，调用 select/epoll/kqueue 等待 IO 事件，有事件就把对应回调放回 ready 队列；第四，处理到期的定时器。然后进入下一轮。
关键在于：协程执行到 await 一个未完成的 awaitable 时会把控制权交回事件循环，并登记「我在等什么」。事件循环趁这个空档去跑别的协程，等 IO 就绪再回来恢复它。所以并发的来源是「等待时间重叠」，而不是「同时执行」——它始终是单线程。
协程比线程轻主要是三点：第一，栈空间，线程栈默认 MB 级别，协程只是堆上的几个帧对象，几 KB；第二，切换成本，线程切换由内核调度，要陷入内核态、保存恢复寄存器上下文，微秒级，协程是用户态的函数调用返回，纳秒级；第三，无锁，协程切换点是确定的（只在 await 处），同一个事件循环内不会出现线程那种任意指令间的抢占，大部分共享状态读写天然安全。
但要注意最后这点有边界：跨 await 的临界区仍然需要 asyncio.Lock，因为中间会让出控制权，别的协程可能插进来改状态。」
连环追问链
| 追问 | 答题要点 |
|---|---|
| 协程函数、协程对象、Task 有什么区别？ | async def f 是协程函数；f() 返回协程对象（此时还没执行）；create_task(f()) 包装成 Task 才被事件循环调度。只创建不 await 会报 coroutine was never awaited |
await 和 create_task 的区别？ | await coro 是串行等待；create_task 立刻登记调度，之后再 await 才是并发。想并发必须先全部 create_task 或用 gather |
| 协程里还需要加锁吗？ | 单个 await 之间的代码是原子的，但跨 await 的「读-改-写」必须用 asyncio.Lock。注意不能用 threading.Lock，它会阻塞整个事件循环 |
| 一个进程能跑多个事件循环吗？ | 一个线程同时只能跑一个。可以每个线程各跑一个（asyncio.run 在各自线程里），跨循环调度要用 run_coroutine_threadsafe |
| 怎么发现事件循环被阻塞了？ | 开 debug 模式（asyncio.run(main(), debug=True) 或 PYTHONASYNCIODEBUG=1），超过 100ms 的回调会打 warning；也可以用「心跳协程测 sleep(0.1) 的实际延迟」做线上监控 |
| uvloop 是什么？ | 基于 libuv 的事件循环替代实现，用 Cython 写，比标准库快 2~4 倍。一行 uvloop.install() 即可，生产常用 |
✅ 项目关联话术：「我们线上加了一个『事件循环延迟』指标：起一个协程每 100ms
await asyncio.sleep(0.1)，记录实际间隔与 100ms 的差值，这个值超过 50ms 就说明循环被什么东西卡住了。靠这个指标我们抓到过一次同步json.loads解析 30MB 响应体导致的卡顿，也抓到过第三方 SDK 偷偷用requests的问题。另外我们asyncio.Lock用在了『读取 token 余额 → 判断 → 扣减』这段跨 await 的逻辑上，之前没加锁时高并发下出现过超卖。」
题 8：在协程里调用一个同步阻塞函数会发生什么？怎么解决？
精简口述答案（90 秒版）
「会卡死整个事件循环。因为事件循环是单线程的，同步阻塞函数不会把控制权交回去，这段时间内所有协程都停摆——不只是当前这个请求慢，是整个服务的所有请求一起卡住。
最典型的就是在 async def 里写了 time.sleep(1)。看起来只是慢了 1 秒，但实际上这 1 秒内服务处理不了任何其他请求。换成 await asyncio.sleep(1) 才是真正的让出控制权。同类的还有 requests.get、同步的 pymysql、open().read() 读大文件、同步 Redis 客户端，以及 CPU 密集的计算。
解决方案有三个层次，优先级从高到低：
第一，换原生异步库——requests 换 httpx/aiohttp，pymysql 换 asyncmy/asyncpg，redis 换 redis.asyncio。这是最彻底的。
第二，丢线程池——await asyncio.to_thread(fn, *args)，适合没有异步版本的阻塞 IO。注意默认 executor 的线程数是 min(32, cpu_count+4) 且全局共享，关键路径最好用 run_in_executor 配独立线程池做隔离。
第三，丢进程池——loop.run_in_executor(process_pool, fn, ...)，这是 CPU 密集任务的唯一正解，线程池对 CPU 密集没用，因为还是被 GIL 卡着。
另外推荐在 CI 里开 ruff 的 ASYNC 规则组，它能静态查出异步函数里的 time.sleep、requests.get、open() 这类调用。」
连环追问链
| 追问 | 答题要点 |
|---|---|
| 怎么判断一个库是不是真异步？ | 看它的 IO 层实现，而不是看接口有没有 async。有些库只是把同步调用包进线程池（这还算能用），有些干脆 async def 里直接同步调用（假异步）。压测时看并发上去后延迟是否线性恶化 |
to_thread 和 run_in_executor 怎么选？ | to_thread 更简洁且自动透传 contextvars，但用共享的默认池；需要隔离、自定义池大小、或用进程池时必须 run_in_executor |
run_in_executor 怎么传关键字参数？ | 它只接受位置参数，用 functools.partial(fn, a, key=v) 包一层 |
contextvars 为什么会丢？ | run_in_executor 不会复制 context。解法是 ctx = contextvars.copy_context() 然后 ctx.run(fn)。asyncio.to_thread（3.9+）已内置这个处理 |
| 线程池满了会怎样？ | 任务在队列里排队，表现为「接口全部超时但 CPU 和内存都正常」。必须配队列上限 + 拒绝策略 + 监控队列长度 |
| 第三方框架内部阻塞怎么办？ | 比如 FastAPI 里把路由写成 def 而非 async def，框架会自动丢线程池（这是安全的）；最怕的是写成 async def 却在里面做同步 IO |
✅ 项目关联话术：「我们出过一次严重线上事故：有个同事在
async def的接口里用了同步的requests调下游，单接口 QPS 不高没人注意，但下游抖动变慢到 2 秒后，整个服务的事件循环被卡死，所有接口一起超时，表现为服务整体不可用。复盘后做了三件事：一是全量替换成httpx.AsyncClient并做成全局单例在 lifespan 里管理；二是 CI 里开了 ruff 的 ASYNC 规则组做静态拦截；三是加了事件循环延迟监控。现在只有历史的 OCR SDK 还留在线程池里，并且单独隔离了一个池。」
题 9：gather、as_completed、wait、TaskGroup 有什么区别？异常怎么处理？
精简口述答案（90 秒版）
「按『结果顺序』和『异常语义』两个维度区分：
gather 最常用，返回结果列表且顺序与传入顺序一致（不是完成顺序）。默认第一个异常会立刻向上抛，但其余任务不会被取消，仍在后台跑，它们的结果就丢了。加 return_exceptions=True 后异常会作为结果项返回，整批不中断——批量调 LLM 这种允许部分失败的场景基本都用这个。
as_completed 返回一个迭代器，谁先完成先 yield，适合流式处理、首个结果优先返回的场景。但它丢失了与输入的对应关系，想知道「这是哪个任务的结果」得自己包装；而且每个 await 都要单独 try。
wait 更底层，返回 (done, pending) 两个 Task 集合，支持 FIRST_COMPLETED / FIRST_EXCEPTION / ALL_COMPLETED 三种返回条件。它不会抛异常，异常留在 Task 里要自己取；pending 的任务也要自己决定取消还是继续。「谁先回用谁、其余取消」的竞速场景用它最合适。
TaskGroup（3.11+）是现在推荐的默认选择：async with 结构化并发，退出时等全部完成，任一失败自动取消所有兄弟任务并抛 ExceptionGroup，用 except* 按类型分别处理。它从根本上杜绝了任务泄漏。
另外两个配套工具：asyncio.Semaphore 控并发上限，asyncio.timeout / wait_for 控超时。」
连环追问链
| 追问 | 答题要点 |
|---|---|
gather 默认抛异常后，其他任务怎么样了？ | 仍在后台运行，不会被取消，结果被丢弃。这是常见的资源泄漏点。需要取消就改用 TaskGroup |
gather 的结果顺序是完成顺序吗？ | 不是，是传入顺序。这点很关键，批量处理时可以直接 zip(inputs, results) 对应 |
return_exceptions=True 有什么副作用？ | CancelledError 也会被当成结果返回，可能掩盖取消语义。处理结果时要用 isinstance(r, BaseException) 判断并单独识别取消 |
as_completed 怎么知道结果属于哪个输入？ | 把任务包成 async def wrap(i, coro): return i, await coro，或直接 for fut in as_completed(tasks) 配合 Task 到输入的映射字典 |
| 怎么实现「取最快的一个，其余取消」？ | done, pending = await asyncio.wait(tasks, return_when=FIRST_COMPLETED)，然后对 pending 逐个 cancel() 并 await gather(*pending, return_exceptions=True) 等清理完成 |
为什么推荐 TaskGroup？ | 结构化并发：作用域结束前所有子任务必定终结，没有「忘了 await 的孤儿 Task」；失败自动扇出取消；ExceptionGroup 保留所有异常而非只有第一个 |
✅ 项目关联话术：「我们批量调 LLM 用的是
gather(..., return_exceptions=True)加Semaphore(10)限流：拿到完整的成功结果加失败清单，失败的走小模型兜底降级，整体可用性从 97% 提到 99.9%，而且 gather 保证顺序和输入一致，后续 zip 对齐很方便。检索侧用的是as_completed，向量库和 ES 并发查，谁先回就先把结果推给前端，首屏时间减少约 40%。多路模型竞速的场景用wait(FIRST_COMPLETED)，拿到最快的那个之后记得把 pending 全部 cancel 掉，不然会白白烧 token。新代码我们统一改用TaskGroup，因为它能保证不漏任务。」
题 10：协程的取消是怎么实现的？CancelledError 有什么特殊性？
精简口述答案（90 秒版）
「task.cancel() 不会立刻杀死协程，它只是打一个标记，等协程执行到下一个 await 挂起点时，事件循环往里面抛一个 CancelledError。异常逐层向上传播，沿途的 finally 和 __aexit__ 会依次执行，给你清理资源的机会，最后 Task 状态变为 cancelled。
由此推出两个重要结论：
第一，纯 CPU 死循环的协程取消不掉，因为它永远到不了 await 点。需要在循环里插 await asyncio.sleep(0) 主动让出。
第二，如果协程把 CancelledError 吞掉且不重新抛出，取消就失败了。这会导致服务无法优雅关闭、超时控制失效。所以 Python 3.8 把 asyncio.CancelledError 的基类从 Exception 改成了 BaseException，专门防止被业务代码的 except Exception 误伤。但 except BaseException、裸 except:、以及 gather(return_exceptions=True) 仍然能拦住它，写通用重试/兜底逻辑时必须显式 except asyncio.CancelledError: raise 放行。
正确姿势是：捕获它只为了做清理（回滚事务、关连接、记录已消耗的 token），清理完一定要 raise。
超时本质上就是「定时取消」。3.11 之前用 wait_for，3.11 之后推荐 async with asyncio.timeout(n)，能包住一整段流程；timeout_at 接受绝对截止时间，适合跨多层透传同一个 deadline，避免每层各写 3 秒最后累加成 9 秒。如果有一段副作用绝对不能半途而废（比如扣费），用 asyncio.shield 保护起来。」
连环追问链
| 追问 | 答题要点 |
|---|---|
为什么 3.8 要把它改成 BaseException？ | 避免业务的 except Exception 误吞取消信号，导致服务无法优雅关闭、超时失效 |
finally 里能不能再 await？ | 可以，但要小心：清理期间如果再次被取消会更麻烦。关键清理用 asyncio.shield 包住，或者给清理设独立的短超时 |
shield 的真实语义？ | 外层取消时，被 shield 的内部任务继续运行，只是外层的 await 立刻抛 CancelledError。适合扣费、幂等写入这类必须完成的副作用，滥用会导致资源迟迟不释放 |
| 服务优雅关闭怎么做？ | 捕获 SIGTERM → 停止接收新请求 → 给在途任务一个宽限期 → 超时后统一 cancel() 并 gather(..., return_exceptions=True) 等清理 → 关闭连接池 |
wait_for 超时后任务一定停了吗？ | 不一定，它只是发起取消，协程可能还在跑 finally 清理。超时后不要立刻复用它持有的连接等资源 |
| 取消和超时在链路上怎么统一？ | 入口算出绝对 deadline，用 timeout_at 向下透传；每层只检查剩余预算，不再各自设固定超时 |
✅ 项目关联话术：「我们 LLM 流式接口里，客户端断开会让 FastAPI 取消这个协程，异步生成器收到
CancelledError。我们必须在这里关掉上游的流式连接、并且按已经产出的 token 做计费，所以在finally里补了这两段逻辑——之前没做，既漏计费又泄漏了上游连接。另一个坑是我们的通用重试装饰器最初写的是except Exception，Python 3.8 之后虽然抓不到取消了，但gather(return_exceptions=True)那层还是把它当普通失败重试，导致服务 SIGTERM 之后迟迟退不掉、K8s 直接 SIGKILL。现在统一在重试逻辑最前面显式except asyncio.CancelledError: raise。」
题 11：上下文管理器的原理是什么？__exit__ 返回值有什么讲究？
精简口述答案（90 秒版）
「with 语句依赖的是上下文管理器协议：__enter__ 和 __exit__。
with obj as x: 等价于：调用 obj.__enter__() 把返回值赋给 x，然后在 try/finally 里执行代码块，无论正常结束还是抛异常，都会调用 obj.__exit__(exc_type, exc_val, tb)。它的核心价值是把「获取-释放」成对的逻辑收敛到一处，彻底避免忘记释放。
__exit__ 的返回值是个高频考点：返回 True 表示异常已被处理、吞掉不再向外抛；返回 False 或 None（默认）表示异常继续传播。注意 Python 里不写 return 就是返回 None，所以默认行为是传播异常——这通常是对的。不小心写成 return True 会把所有异常静默吞掉，是非常危险的 bug。
写法上有两种：实现类的双下方法，或者用 @contextlib.contextmanager 装饰一个生成器——yield 之前是 __enter__，yield 之后是 __exit__，要捕获异常就用 try/except/finally 包住 yield。后者更简洁，但每次只能用一次（生成器会耗尽）。
异步版本是 __aenter__ / __aexit__，配 async with，用 @asynccontextmanager 写生成器版本。FastAPI 的 lifespan 就是这么管理连接池的。
标准库里还有几个很实用的：contextlib.suppress 忽略指定异常、closing 给只有 close() 的对象补协议、ExitStack 动态管理数量不定的资源、nullcontext 做「可选的 with」。」
连环追问链
| 追问 | 答题要点 |
|---|---|
__exit__ 的三个参数是什么？ | exc_type、exc_val、exc_tb。正常退出时都是 None。可以据此区分「成功提交」还是「异常回滚」 |
@contextmanager 里怎么捕获异常？ | try: yield resource except Exception: rollback(); raise finally: close()。注意不 raise 就相当于返回 True 吞掉异常 |
with 里 return / break 会执行 __exit__ 吗？ | 会。它等价于 try/finally，任何退出路径都触发 |
| 管理数量不定的资源怎么办？ | with ExitStack() as stack: files = [stack.enter_context(open(f)) for f in paths]，退出时逆序全部关闭 |
| 同一个上下文管理器能嵌套/重入吗？ | 类实现的可以设计成可重入；@contextmanager 生成的不可重用，因为生成器只能跑一次。需要复用就把它包成工厂函数 |
| 异步版本要注意什么？ | __aexit__ 里的清理如果再被取消会很麻烦，关键清理要 shield；异步生成器配 contextlib.aclosing，防止消费方 break 后迟迟不释放 |
✅ 项目关联话术：「我们用
@asynccontextmanager封了一个llm_session()：进入时申请 Semaphore 配额、记录开始时间、绑定 traceId，退出时统计 token 消耗、上报耗时指标、释放配额。这样所有调用点都是三行代码，不会有人忘记释放配额——之前手写 acquire/release 时，异常路径上漏了 release，跑几个小时后 Semaphore 配额被耗尽，服务直接全部卡死在等信号量。还有个坑是有同事在__exit__里写了return True，把下游的超时异常全吞了，表现为接口返回空结果但没有任何错误日志，排查了很久——现在 code review 会专门盯这一行。」
题 12：什么是闭包？LEGB 规则是什么？循环里创建闭包为什么会出问题？
精简口述答案（90 秒版）
「闭包是「函数 + 它捕获的外层作用域变量」。内层函数引用了外层函数的局部变量，外层函数返回后这些变量依然存活，被保存在函数对象的 __closure__ 里。
名字查找遵循 LEGB 顺序：Local 局部 → Enclosing 外层函数 → Global 模块全局 → Builtin 内置。要注意赋值会创建局部变量，所以想改外层函数的变量要用 nonlocal，想改模块全局变量要用 global，否则会报 UnboundLocalError——这个报错特别反直觉：明明外面有这个变量，但因为函数体里有赋值语句，整个函数里它就被认定为局部变量了。
循环里创建闭包的经典坑是延迟绑定：闭包捕获的是变量本身，不是创建时的值。所以 [lambda: i for i in range(3)] 里三个 lambda 全都返回 2，因为它们共享同一个 i，而循环结束时 i 已经是 2 了。
三种解法：第一，用默认参数在定义时固定值 lambda i=i: i；第二，用 functools.partial(fn, i)；第三，外面再包一层函数制造独立作用域。
这个坑在 AI 工程里有个很典型的触发场景：批量注册 Function Calling 的工具函数时用循环创建 handler，结果所有工具名都指向最后一个，调用任何工具都执行同一个函数。」
连环追问链
| 追问 | 答题要点 |
|---|---|
| 怎么看一个函数捕获了什么？ | fn.__closure__ 是 cell 对象元组，fn.__code__.co_freevars 是被捕获的变量名。配合看最直观 |
nonlocal 和 global 的区别？ | nonlocal 绑定到最近的外层函数作用域（不能是全局）；global 绑定到模块级。都只是声明「这个名字不是局部的」 |
为什么会 UnboundLocalError？ | 函数体内任何位置有赋值，该名字整个函数内都是局部的。读在赋值之前就会报错。解法是 nonlocal/global，或换个名字 |
| 类里的推导式为什么访问不到类属性？ | 推导式有自己的作用域且不包含类作用域（3.12 对此有改动但原则仍在）。解法是在推导式外先存成局部变量 |
| 闭包会造成内存泄漏吗？ | 会。闭包持有外层变量的强引用，如果捕获了大对象（整个 DataFrame、模型实例），这些对象就一直不释放。只捕获真正需要的小参数 |
| 闭包和类谁更合适？ | 状态少、逻辑单一用闭包更轻；状态多、需要多个方法操作同一状态用类更清晰。装饰器通常用闭包，带复杂状态的装饰器可以用类 |
✅ 项目关联话术：「我们注册 Agent 工具时踩过延迟绑定的坑：用
for cfg in tool_configs循环创建 handler 闭包，结果所有工具注册上去都指向最后一个配置，模型调『查天气』实际执行的是『查库存』，而且没有任何报错，纯靠用户反馈才发现。改成lambda cfg=cfg: ...固定默认参数后解决。后来我们干脆规范成用functools.partial显式传参，可读性更好也不容易被 review 漏掉。还有一次是闭包捕获了整个检索结果列表，几百个注册的 handler 把几百份文档全留在内存里，常驻内存涨了 2G——定位之后只捕获文档 ID，内存立刻降下来了。」
✍️ 手撕代码【核心实现 + 边界 Case 自测清单 + 变种拓展题】
题 1：实现「计时 + 日志 + 重试」三合一带参装饰器（同步 / 异步双兼容）
这是 AI 工程里最高频的一道手撕题。调 LLM、调向量库、调下游 HTTP，全都需要「超时重试 + 耗时埋点」，而业务里同步函数和协程函数混用，装饰器必须两种都能包。
核心实现
python
import asyncio
import functools
import inspect
import logging
import random
import time
from typing import Callable, TypeVar, Any
logger = logging.getLogger(__name__)
F = TypeVar("F", bound=Callable[..., Any])
def traced_retry(
*,
retries: int = 3,
base_delay: float = 0.5,
max_delay: float = 8.0,
jitter: bool = True,
exceptions: tuple[type[BaseException], ...] = (Exception,),
name: str | None = None,
) -> Callable[[F], F]:
"""带参装饰器：三层结构 = 配置层 -> 装饰层 -> 执行层。
- 自动识别被装饰对象是协程函数还是普通函数，分别返回 async / sync 包装
- 指数退避 + 随机抖动，避免重试风暴
- 只重试 exceptions 里声明的异常；CancelledError 永不重试
"""
def _sleep_seconds(attempt: int) -> float:
delay = min(base_delay * (2 ** attempt), max_delay)
if jitter:
# full jitter：在 [0, delay] 之间随机，彻底打散同时重试的请求
delay = random.uniform(0, delay)
return delay
def decorator(func: F) -> F:
label = name or func.__qualname__
if inspect.iscoroutinefunction(func):
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
started = time.perf_counter()
last_exc: BaseException | None = None
for attempt in range(retries + 1):
try:
result = await func(*args, **kwargs)
except asyncio.CancelledError:
# 取消是协作信号，不是失败，必须原样向上抛
logger.info("%s cancelled after %.3fs", label,
time.perf_counter() - started)
raise
except exceptions as exc:
last_exc = exc
if attempt >= retries:
break
wait = _sleep_seconds(attempt)
logger.warning(
"%s failed (%s), retry %d/%d in %.2fs",
label, exc.__class__.__name__,
attempt + 1, retries, wait,
)
await asyncio.sleep(wait)
else:
logger.info("%s ok in %.3fs (attempt=%d)", label,
time.perf_counter() - started, attempt + 1)
return result
logger.error("%s give up after %d attempts, %.3fs", label,
retries + 1, time.perf_counter() - started)
raise last_exc # type: ignore[misc]
return async_wrapper # type: ignore[return-value]
@functools.wraps(func)
def sync_wrapper(*args, **kwargs):
started = time.perf_counter()
last_exc: BaseException | None = None
for attempt in range(retries + 1):
try:
result = func(*args, **kwargs)
except exceptions as exc:
last_exc = exc
if attempt >= retries:
break
wait = _sleep_seconds(attempt)
logger.warning(
"%s failed (%s), retry %d/%d in %.2fs",
label, exc.__class__.__name__,
attempt + 1, retries, wait,
)
time.sleep(wait)
else:
logger.info("%s ok in %.3fs (attempt=%d)", label,
time.perf_counter() - started, attempt + 1)
return result
logger.error("%s give up after %d attempts, %.3fs", label,
retries + 1, time.perf_counter() - started)
raise last_exc # type: ignore[misc]
return sync_wrapper # type: ignore[return-value]
return decorator使用方式：
python
@traced_retry(retries=2, base_delay=0.2, exceptions=(TimeoutError, ConnectionError))
async def call_llm(prompt: str) -> str:
...
@traced_retry(retries=5, name="pg.write_embedding")
def save_embedding(doc_id: str, vec: list[float]) -> None:
...⚠️ 三个最容易被扣分的点：① 忘记
functools.wraps，__name__变成async_wrapper，FastAPI 的路由注册和 pytest 的用例收集都会出问题；② 用except Exception吞掉asyncio.CancelledError（3.8+ 它继承BaseException不会被吞，但如果写成except BaseException就会），导致超时取消后函数还在重试；③ 异步分支里用time.sleep退避，直接把整个事件循环卡死。
边界 Case 自测清单
| # | Case | 期望行为 |
|---|---|---|
| 1 | 第一次就成功 | 不进入重试分支，只打一条 ok 日志，返回值原样透传 |
| 2 | 前两次失败第三次成功 | 共调用 3 次，退避 2 次，最终返回成功结果，不抛异常 |
| 3 | 全部失败 | 抛出最后一次的异常对象（保留原始 traceback），不被包装成自定义异常 |
| 4 | 抛出的异常不在 exceptions 里 | 立刻向上抛，一次都不重试 |
| 5 | 被装饰对象是 async def | 返回的 wrapper 仍是协程函数，inspect.iscoroutinefunction 为 True |
| 6 | 异步执行中被外部 task.cancel() | CancelledError 原样抛出，不触发重试，日志记录 cancelled |
| 7 | retries=0 | 只执行一次，失败直接抛，不进 asyncio.sleep |
| 8 | 检查元信息 | fn.__name__ / __doc__ / __module__ / 类型注解与原函数一致 |
| 9 | 装饰实例方法 | self 作为第一个位置参数正常透传，不需要额外处理 |
| 10 | 并发 100 个任务同时失败重试 | 因为有 full jitter，退避时刻被打散，不出现整齐的重试尖峰 |
🚀 变种拓展题
- 把退避策略抽成可注入的策略对象（
FixedBackoff/ExponentialBackoff/DecorrelatedJitter），装饰器只依赖接口。 - 增加
on_retry/on_giveup回调钩子，方便接 Prometheus 打点和告警。 - 支持「整体 deadline」：不管重试几次，总耗时超过
total_timeout就立刻放弃（提示：用asyncio.timeout包住整个循环）。 - 用
contextvars把trace_id透传进日志，使得并发 100 个请求的重试日志可以按请求聚合。 - 把装饰器改写成类装饰器（实现
__call__+__get__），并说明__get__为什么是装饰实例方法时必须的。
题 2：用生成器流式读取超大文件，并按 token 预算分批喂给 Embedding
场景：一个 4GB 的 JSONL 知识库要灌进向量库。不能
read()全量进内存，也不能一条一条调 Embedding（QPS 打满、成本爆炸），必须「流式读 → 清洗 → 按预算攒批 → 批量调用」，全程内存恒定。
核心实现
python
import json
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
@dataclass(slots=True)
class Chunk:
doc_id: str
text: str
tokens: int
def read_jsonl(path: str, encoding: str = "utf-8") -> Iterator[dict]:
"""逐行惰性读取：文件句柄由 with 托管，内存里始终只有一行。"""
with open(path, "r", encoding=encoding) as f:
for lineno, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
try:
yield json.loads(line)
except json.JSONDecodeError:
# 脏数据跳过而不是中断整个灌库任务
print(f"[skip] bad json at line {lineno}")
continue
def to_chunks(rows: Iterable[dict], max_tokens: int = 500) -> Iterator[Chunk]:
"""清洗 + 切片：生成器管道的中间一级，不缓存任何中间列表。"""
for row in rows:
doc_id = row.get("id")
text = (row.get("content") or "").strip()
if not doc_id or not text:
continue
# 这里用「4 字符 ≈ 1 token」粗估，生产应换成 tiktoken
for i in range(0, len(text), max_tokens * 4):
piece = text[i: i + max_tokens * 4]
yield Chunk(doc_id=doc_id, text=piece, tokens=len(piece) // 4 + 1)
def batch_by_budget(
chunks: Iterable[Chunk],
max_items: int = 64,
max_tokens: int = 8000,
) -> Iterator[list[Chunk]]:
"""双预算攒批：条数和 token 数任一达标就出批。
关键点：单条就超预算时也必须单独成批，否则会死循环或永远攒不出来。
"""
buf: list[Chunk] = []
buf_tokens = 0
for c in chunks:
if buf and (len(buf) >= max_items or buf_tokens + c.tokens > max_tokens):
yield buf
buf, buf_tokens = [], 0
buf.append(c)
buf_tokens += c.tokens
if buf_tokens >= max_tokens:
yield buf
buf, buf_tokens = [], 0
if buf: # 收尾：最后一批一定不能漏
yield buf
def ingest(path: str) -> int:
total = 0
pipeline = batch_by_budget(to_chunks(read_jsonl(path)))
for batch in pipeline:
vectors = embed([c.text for c in batch]) # 一次网络调用处理一批
save_vectors(batch, vectors)
total += len(batch)
return total面试加分表述：「
read_jsonl → to_chunks → batch_by_budget是三级生成器管道，上游不会因为下游慢而堆积数据，整个链路的内存占用是 O(一批) 而不是 O(文件)。我们灌 4GB 知识库时常驻内存只有 200MB 左右，而且因为是惰性的，中途失败可以记录lineno断点续传。」
边界 Case 自测清单
| # | Case | 期望行为 |
|---|---|---|
| 1 | 空文件 | 管道不产出任何批次，ingest 返回 0，不抛异常 |
| 2 | 文件最后一行没有换行符 | 仍被正常解析，不丢最后一条数据 |
| 3 | 中间夹杂空行 / 非法 JSON | 跳过并告警，后续行继续处理，不中断任务 |
| 4 | 总量不足一批 | 收尾逻辑把残批 yield 出来，数据不丢 |
| 5 | 单条文本 token 数就超过 max_tokens | 该条独立成批，不死循环、不被丢弃 |
| 6 | 恰好整除（条数正好是 max_items 的倍数） | 不产出空批次 [] |
| 7 | 生成器被提前 break | with open 在生成器被 GC / close() 时触发关闭，文件句柄不泄漏 |
| 8 | 文件编码含 BOM 或非 UTF-8 字符 | 用 encoding="utf-8-sig" 或 errors="replace" 显式处理，不抛 UnicodeDecodeError 终止 |
| 9 | 内存观测 | 用 tracemalloc 对比全量 readlines()，峰值内存应为常数级而非线性增长 |
| 10 | 批内复用 buf 列表 | 必须 buf = [] 新建而不是 buf.clear()，否则已 yield 出去的批会被清空 |
🚀 变种拓展题
- 把同步管道改成
async def+async for的异步生成器版本，并用aiofiles读文件。 - 实现断点续传：把已处理的
(file, offset)落盘，重启后用f.seek(offset)继续。 - 给管道加一级「去重」：用布隆过滤器或
hash(text)集合过滤重复 chunk，并分析内存代价。 - 把
batch_by_budget改成同时限制「批内最长单条长度」，超长的先切分再入批。 - 用
itertools.islice实现一个通用batched(iterable, n)（Python 3.12 已内置itertools.batched），并说明它与本题实现的差异。
题 3：asyncio 并发调用 LLM —— 限流 + 超时 + 重试 + 结果顺序保持 + 部分失败可降级
这题几乎是 AI 工程岗的必考题。要求：输入 200 个 prompt，最多 8 并发，单请求 20s 超时，失败重试 2 次，返回结果顺序必须和输入一致，个别失败不能让整批崩掉，并且要能被外部整体取消。
核心实现
python
import asyncio
from dataclasses import dataclass
from typing import Sequence
@dataclass(slots=True)
class LLMResult:
index: int
prompt: str
text: str | None = None
error: str | None = None
@property
def ok(self) -> bool:
return self.error is None
async def _call_one(
index: int,
prompt: str,
sem: asyncio.Semaphore,
*,
timeout: float = 20.0,
retries: int = 2,
) -> LLMResult:
"""单个请求：信号量在最内层，确保「排队等额度」不占用并发额度。"""
last_err = "unknown"
for attempt in range(retries + 1):
try:
async with sem: # 闸门一：并发上限
async with asyncio.timeout(timeout): # 闸门二：单请求超时
text = await llm.acreate(prompt)
return LLMResult(index=index, prompt=prompt, text=text)
except asyncio.CancelledError:
# 外部整体取消：立刻放弃，绝不重试
raise
except TimeoutError:
last_err = f"timeout>{timeout}s"
except Exception as exc: # noqa: BLE001
last_err = f"{exc.__class__.__name__}: {exc}"
if attempt < retries:
await asyncio.sleep(min(0.5 * 2 ** attempt, 4.0))
return LLMResult(index=index, prompt=prompt, error=last_err)
async def batch_call(
prompts: Sequence[str],
concurrency: int = 8,
timeout: float = 20.0,
retries: int = 2,
) -> list[LLMResult]:
"""并发调用并保持顺序。
顺序保持的三种做法：
1. gather 天然按入参顺序返回（本实现采用，最简单）
2. 自己带 index，as_completed 拿到后按 index 排序（可做流式进度）
3. 预分配结果数组，task 完成后写回 results[index]
"""
if not prompts:
return []
sem = asyncio.Semaphore(concurrency)
tasks = [
asyncio.create_task(
_call_one(i, p, sem, timeout=timeout, retries=retries),
name=f"llm-{i}",
)
for i, p in enumerate(prompts)
]
try:
# return_exceptions=True：个别任务抛异常不会让整批立刻失败
raw = await asyncio.gather(*tasks, return_exceptions=True)
except asyncio.CancelledError:
for t in tasks:
t.cancel()
# 等取消真正落地，避免 "Task was destroyed but it is pending"
await asyncio.gather(*tasks, return_exceptions=True)
raise
results: list[LLMResult] = []
for i, item in enumerate(raw):
if isinstance(item, BaseException):
results.append(LLMResult(index=i, prompt=prompts[i],
error=repr(item)))
else:
results.append(item)
return results配套的「流式进度」版本（边跑边吐结果，用于前端进度条）：
python
async def batch_call_stream(prompts, concurrency=8):
sem = asyncio.Semaphore(concurrency)
tasks = [asyncio.create_task(_call_one(i, p, sem))
for i, p in enumerate(prompts)]
done_count = 0
try:
for fut in asyncio.as_completed(tasks):
res = await fut
done_count += 1
yield done_count, res # 谁先完成先吐，顺序由调用方按 index 还原
finally:
for t in tasks:
t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)⚠️ 四个高频踩坑：①
async with sem写在for attempt循环外面，重试期间一直占着并发额度，实际并发远低于设定值；② 只await asyncio.sleep不加抖动，8 个请求同时失败后又同时重试，形成周期性尖峰；③gather不传return_exceptions=True，一个请求挂掉剩下 199 个的结果全部丢失（任务其实还在后台跑完才被 GC）；④ 不保存tasks强引用，create_task的返回值被 GC 后任务「神秘消失」。
边界 Case 自测清单
| # | Case | 期望行为 |
|---|---|---|
| 1 | prompts 为空列表 | 直接返回 []，不创建任何 task，不触发网络调用 |
| 2 | 200 个 prompt、concurrency=8 | 任意时刻在途请求数 ≤ 8（可用计数器断言峰值） |
| 3 | 全部成功 | 返回长度 200，且 results[i].prompt == prompts[i]，顺序严格一致 |
| 4 | 第 37 个请求一直超时 | 重试 2 次后返回 ok=False 的结果，其余 199 个不受影响 |
| 5 | 下游整体不可用 | 所有结果都是失败态，函数本身不抛异常，由上层决定是否降级 |
| 6 | 外部 task.cancel() / 用户断开连接 | 所有子任务被取消并 await 回收，CancelledError 向上抛，无 pending 警告 |
| 7 | 重试期间的额度占用 | 退避 sleep 时信号量已释放，其他请求可以顶上 |
| 8 | 单请求超时与整体 deadline 并存 | 外层 asyncio.timeout 到点时内层重试立即中断，不会「超时后还在重试」 |
| 9 | 结果里混入 BaseException | 被兜底转成失败结果，不会把异常对象泄漏给业务层 |
| 10 | 重复 prompt | 不做隐式去重（除非显式开启缓存），每个 index 都有独立结果 |
🚀 变种拓展题
- 用
asyncio.TaskGroup（3.11+）重写batch_call，并说明它与gather(return_exceptions=True)在失败语义上的根本区别。 - 在信号量之外再加一层令牌桶，同时限制 QPS 和 TPM（每分钟 token 数）。
- 加入熔断器：连续失败 N 次后直接短路快速失败，半开状态放少量探测流量。
- 把
_call_one的结果接上题 5 的 TTL 缓存，命中缓存的请求不占用并发额度。 - 实现「优先级队列」版本：VIP 请求插队，用
asyncio.PriorityQueue+ 固定 worker 池替代信号量。
题 4：手写同步 / 异步双版本上下文管理器（资源保障 + 异常分类 + 耗时埋点）
要求：实现一个
span上下文管理器，进入时记录开始时间与 trace 上下文，退出时打点；异常时标记失败但不吞异常；同时提供with和async with两种用法；再实现一个真正会「吞掉特定异常」的suppress_and_log，讲清__exit__返回值的含义。
核心实现（类写法，把协议写全）
python
import time
import types
import contextvars
from contextlib import contextmanager, asynccontextmanager
current_span = contextvars.ContextVar[str]("current_span", default="root")
class Span:
"""同时实现同步与异步上下文管理器协议的类。"""
__slots__ = ("name", "_t0", "_token", "elapsed", "failed")
def __init__(self, name: str) -> None:
self.name = name
self._t0 = 0.0
self._token: contextvars.Token | None = None
self.elapsed = 0.0
self.failed = False
# ---------- 公共逻辑 ----------
def _enter(self) -> "Span":
self._t0 = time.perf_counter()
parent = current_span.get()
self._token = current_span.set(f"{parent}/{self.name}")
return self
def _exit(self, exc_type) -> bool:
self.elapsed = time.perf_counter() - self._t0
self.failed = exc_type is not None
if self._token is not None:
current_span.reset(self._token) # 必须 reset，否则污染后续调用
status = "ERR" if self.failed else "OK"
print(f"[span] {self.name} {status} {self.elapsed * 1000:.1f}ms")
return False # 返回 False/None = 不吞异常，原样向上抛
# ---------- 同步协议 ----------
def __enter__(self) -> "Span":
return self._enter()
def __exit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: types.TracebackType | None,
) -> bool:
return self._exit(exc_type)
# ---------- 异步协议 ----------
async def __aenter__(self) -> "Span":
return self._enter()
async def __aexit__(self, exc_type, exc, tb) -> bool:
return self._exit(exc_type)装饰器写法（更常用，但要注意它不能复用）
python
@contextmanager
def db_session(dsn: str):
conn = connect(dsn) # try 之前：获取失败就不需要释放
try:
yield conn
except Exception:
conn.rollback() # 业务异常：回滚后继续抛
raise
else:
conn.commit() # 只有正常走完才提交
finally:
conn.close() # 无论如何都归还连接
@asynccontextmanager
async def http_client(base_url: str):
client = httpx.AsyncClient(base_url=base_url, timeout=10)
try:
yield client
finally:
await client.aclose() # 异步清理必须 await，否则连接池泄漏真正「吞异常」的版本 —— 讲清 __exit__ 返回值
python
class suppress_and_log:
"""只吞掉指定异常并记日志，其余原样抛出。"""
def __init__(self, *exceptions: type[BaseException], tag: str = "") -> None:
self.exceptions = exceptions or (Exception,)
self.tag = tag
def __enter__(self) -> "suppress_and_log":
return self
def __exit__(self, exc_type, exc, tb) -> bool:
if exc_type is None:
return False
if issubclass(exc_type, self.exceptions):
print(f"[suppressed]{self.tag} {exc_type.__name__}: {exc}")
return True # ← 返回 True 才会吞掉异常，with 之后的代码继续跑
return Falsepython
# ❌ 错误写法：误以为 return 任意值都代表「处理完了」
def __exit__(self, exc_type, exc, tb):
log(exc)
return True # 所有异常都被静默吞掉，包括 KeyboardInterrupt / CancelledError
# ✅ 正确写法：默认不吞，只在明确声明的异常类型上返回 True
def __exit__(self, exc_type, exc, tb):
log(exc)
return exc_type is not None and issubclass(exc_type, self.exceptions)面试加分表述：「
__exit__的返回值是布尔语义：返回真值表示『异常我处理了，别再往上抛』，返回假值（False/None，也就是函数默认返回）表示『我只是做清理，异常继续冒泡』。最危险的写法是无条件return True，它会把CancelledError、KeyboardInterrupt一起吞掉——我们线上出过一次 FastAPI 请求取消后协程还在跑的问题，根因就是一个自定义事务上下文无脑返回了 True。」
边界 Case 自测清单
| # | Case | 期望行为 |
|---|---|---|
| 1 | 正常执行完 | __exit__ 被调用，failed=False，埋点打 OK，contextvars 被 reset |
| 2 | 块内抛业务异常 | 清理照常执行，异常原样向上抛（traceback 不丢） |
| 3 | 块内 return / break | __exit__ 仍被调用（这是上下文管理器的核心保障） |
| 4 | __enter__ 内部抛异常 | __exit__ 不会被调用，所以资源获取必须放在 try 之前 |
| 5 | 嵌套使用 | contextvars 的 span 路径正确形成 root/a/b，退出时逐层 reset |
| 6 | async with 版本中块内 await 抛异常 | __aexit__ 被 await 调用，清理完成后异常继续抛 |
| 7 | 协程被取消 | CancelledError 不被吞（返回 False），__aexit__ 的清理仍执行 |
| 8 | @contextmanager 对象复用两次 | 第二次抛 RuntimeError: generator didn't yield，必须每次新建 |
| 9 | 在 finally 里再抛异常 | 新异常会覆盖原异常并以 during handling of... 链接，清理代码要自己兜住 |
| 10 | 多资源同时管理 | 用 with a() as x, b() as y: 或 ExitStack，保证逆序释放且前者失败后者不泄漏 |
🚀 变种拓展题
- 用
contextlib.ExitStack/AsyncExitStack动态管理数量不定的资源（例如同时打开 N 个分片文件）。 - 让
Span同时可以当装饰器用（提示：继承contextlib.ContextDecorator），并说明异步版本要怎么做。 - 实现一个可重入（reentrant）的锁上下文管理器，并解释为什么
@contextmanager生成的对象天然不可重入。 - 实现
timed_lock(timeout)：拿不到锁就在超时后抛TimeoutError，并保证已拿到的锁一定释放。 - 基于
__aexit__实现一个「异步生成器安全关闭」工具，对比contextlib.aclosing的实现。
题 5：手写带 TTL 的 LRU 缓存装饰器（支持异步 + 线程安全 + 防缓存惊群）
场景：同一个 prompt 在 5 分钟内重复问，直接走缓存省钱省时延。要求：容量上限 LRU 淘汰、每条带 TTL 过期、同步异步双支持、线程安全、并且同一 key 的并发穿透只打一次下游（防惊群 / dog-pile）。
核心实现
python
import asyncio
import functools
import inspect
import threading
import time
from collections import OrderedDict
from typing import Any, Callable, Hashable
class _Entry:
__slots__ = ("value", "expire_at")
def __init__(self, value: Any, expire_at: float) -> None:
self.value = value
self.expire_at = expire_at
def make_key(args: tuple, kwargs: dict) -> Hashable:
"""稳定 key：kwargs 排序，保证 f(a=1,b=2) 与 f(b=2,a=1) 命中同一条。"""
return (args, tuple(sorted(kwargs.items())))
def ttl_lru_cache(maxsize: int = 128, ttl: float = 300.0):
def decorator(func: Callable):
store: OrderedDict[Hashable, _Entry] = OrderedDict()
lock = threading.Lock() # 同步版：保护 dict
inflight: dict[Hashable, asyncio.Future] = {} # 异步版：防惊群
stats = {"hit": 0, "miss": 0, "expired": 0}
def _get(key: Hashable) -> tuple[bool, Any]:
now = time.monotonic()
entry = store.get(key)
if entry is None:
stats["miss"] += 1
return False, None
if entry.expire_at <= now: # 先判过期再判命中
del store[key]
stats["expired"] += 1
stats["miss"] += 1
return False, None
store.move_to_end(key) # LRU：命中即刷新热度
stats["hit"] += 1
return True, entry.value
def _set(key: Hashable, value: Any) -> None:
store[key] = _Entry(value, time.monotonic() + ttl)
store.move_to_end(key)
while len(store) > maxsize:
store.popitem(last=False) # 淘汰最久未使用
if inspect.iscoroutinefunction(func):
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
key = make_key(args, kwargs)
hit, value = _get(key)
if hit:
return value
# 防惊群：同一 key 并发只放一个请求打下游，其余等同一个 Future
fut = inflight.get(key)
if fut is not None:
return await asyncio.shield(fut)
loop = asyncio.get_running_loop()
fut = loop.create_future()
inflight[key] = fut
try:
result = await func(*args, **kwargs)
except BaseException as exc:
if not fut.done():
fut.set_exception(exc)
raise
else:
_set(key, result)
if not fut.done():
fut.set_result(result)
return result
finally:
inflight.pop(key, None)
async_wrapper.cache_info = lambda: dict(stats, size=len(store)) # type: ignore[attr-defined]
async_wrapper.cache_clear = lambda: (store.clear(), stats.update(hit=0, miss=0, expired=0)) # type: ignore[attr-defined]
return async_wrapper
@functools.wraps(func)
def sync_wrapper(*args, **kwargs):
key = make_key(args, kwargs)
with lock:
hit, value = _get(key)
if hit:
return value
result = func(*args, **kwargs) # 锁外执行，避免长时间持锁
with lock:
_set(key, result)
return result
sync_wrapper.cache_info = lambda: dict(stats, size=len(store)) # type: ignore[attr-defined]
sync_wrapper.cache_clear = lambda: (store.clear(), stats.update(hit=0, miss=0, expired=0)) # type: ignore[attr-defined]
return sync_wrapper
return decoratorpython
# ❌ 错误写法：直接用 functools.lru_cache 缓存协程函数
@functools.lru_cache(maxsize=128)
async def call_llm(prompt: str) -> str:
return await llm.acreate(prompt)
# 缓存住的是「协程对象」，第二次命中拿到的是已经 await 过的协程
# -> RuntimeError: cannot reuse already awaited coroutine
# ✅ 正确写法：缓存 await 之后的「结果」，或缓存可复用的 Future/Task
@ttl_lru_cache(maxsize=1024, ttl=300)
async def call_llm(prompt: str) -> str:
return await llm.acreate(prompt)⚠️ 另一个经典坑：
functools.lru_cache装在实例方法上会把self一起作为 key 缓存，持有实例强引用导致对象永不释放（内存泄漏）。正确做法是用functools.cached_property、或把缓存建在实例属性上、或用weakref。
边界 Case 自测清单
| # | Case | 期望行为 |
|---|---|---|
| 1 | 同参数连续两次调用 | 第二次不执行函数体，cache_info() 的 hit 加 1 |
| 2 | TTL 到期后再调用 | 旧条目被删除，重新执行函数体，expired 计数加 1 |
| 3 | 写入超过 maxsize | 最久未使用的条目被淘汰，size 恒等于 maxsize |
| 4 | 命中后立刻再写新 key | 被命中的 key 因为 move_to_end 不会被优先淘汰 |
| 5 | f(1, b=2) 与 f(1, **{"b": 2}) | 命中同一条缓存（kwargs 已排序归一化） |
| 6 | 不可 hash 的参数（list / dict） | 抛 TypeError，不静默降级为不缓存（或显式序列化成 JSON 字符串） |
| 7 | 被装饰函数抛异常 | 不写入缓存，异常原样抛出，下一次仍会真实调用 |
| 8 | 异步版同一 key 并发 50 次 | 下游只被调用 1 次，其余 49 个复用同一 Future（防惊群） |
| 9 | 等待 Future 的协程被取消 | 用 shield 保护，发起方不受影响；inflight 在 finally 中一定被清理 |
| 10 | 多线程同时读写（同步版） | store 不出现 RuntimeError: dictionary changed size，统计值不错乱 |
🚀 变种拓展题
- 把时间源抽成可注入的
clock()，用假时钟在单测里零等待验证 TTL 过期逻辑。 - 支持「软过期 + 后台刷新」（stale-while-revalidate）：过期后先返回旧值，同时异步刷新。
- 把本地缓存换成 Redis 两级缓存，并处理「本地命中但 Redis 已失效」的一致性问题。
- 给缓存 key 加上模型名、温度、system prompt 版本号，解释为什么漏掉任意一个都会导致脏命中。
- 用
asyncio.Lock的 per-key 字典替代 Future 方案实现防惊群，并对比两者在异常传播上的差异。
⚠️ 当日踩坑实录
| ❌ 坑 | ✅ 解决方案 |
|---|---|
函数默认参数写成 def f(items=[])，多次调用之间数据互相串，调试了半天以为是并发问题 | 默认值在函数定义时只求值一次。可变默认参数一律写成 def f(items=None)，函数体内 items = items or []（注意空列表也会被 or 判假，严格场景用 if items is None）。上 ruff 的 B006 规则自动卡住 |
协程里写了 time.sleep(1) 做限速，结果 8 并发退化成串行，整个服务所有请求一起卡 1 秒 | 协程是单线程协作式调度，同步阻塞会卡死整个事件循环。改用 await asyncio.sleep(1)；不可避免的阻塞库用 await asyncio.to_thread(fn, ...) 丢到线程池；装 ruff 的 ASYNC 规则集自动扫出 time.sleep、requests 等阻塞调用 |
写装饰器忘了 functools.wraps，FastAPI 路由全部注册成同名 wrapper，OpenAPI 文档塌了，pytest 也收集不到用例 | 所有包装函数必须加 @functools.wraps(func)，它会复制 __name__、__doc__、__module__、__qualname__、__dict__ 和 __wrapped__。需要拿原函数时用 fn.__wrapped__ |
copy.copy 以为能隔离配置，改了副本里的嵌套 dict，原始默认配置也被改了，影响到其他请求 | 浅拷贝只复制最外层容器，内层还是同一个对象。嵌套结构用 copy.deepcopy，或者直接用不可变结构（frozen=True 的 dataclass、tuple）从源头规避 |
asyncio.create_task 后没保存返回值，任务跑到一半「凭空消失」，日志里只有一条 Task was destroyed but it is pending | 事件循环只持有 Task 的弱引用。必须把 task 存进集合保持强引用，并 add_done_callback 在完成后移除 + 读取异常；或者直接用 asyncio.TaskGroup 托管 |
gather 没传 return_exceptions=True，一个请求超时直接让整批 200 个结果全丢，用户看到的是整页空白 | 批量场景用 gather(..., return_exceptions=True) 做「部分成功」语义，再自己区分成功/失败；需要「一个失败全体取消」的事务语义时才用默认行为或 TaskGroup |
用 except Exception 包住整段协程做兜底，结果超时取消后函数还在重试，接口迟迟不返回 | asyncio.CancelledError 在 3.8+ 继承 BaseException，不会被 except Exception 捕获——但如果写了 except BaseException 或裸 except: 就会被吞。重试逻辑里必须显式 except asyncio.CancelledError: raise 放在最前面 |
functools.lru_cache 直接装在 async def 上，第二次命中报 cannot reuse already awaited coroutine | lru_cache 缓存的是协程对象而不是结果。要缓存异步函数必须自己实现（缓存 await 后的值或可复用的 Future），或用 async-lru 等库 |
lru_cache 装在实例方法上，缓存 key 里含 self，服务常驻内存只涨不降 | 这是典型的缓存导致的内存泄漏。改用 functools.cached_property、把缓存放到实例属性里，或用 weakref.WeakKeyDictionary 做 key |
循环里创建闭包 handlers.append(lambda: run(cfg))，所有 handler 最终都指向最后一个 cfg，还没有任何报错 | 闭包捕获的是变量本身而不是当时的值（延迟绑定）。用默认参数固定 lambda cfg=cfg: run(cfg)，或用 functools.partial(run, cfg) 显式传参 |
| 多进程跑 CPU 密集预处理，Windows 上无限递归启动子进程，最后把机器打满 | Windows / macOS 默认 spawn 启动方式会重新导入主模块。入口必须包在 if __name__ == "__main__": 里；Linux 的 fork 不会暴露这个问题，所以本地没测出来 |
自定义事务上下文管理器的 __exit__ 无脑 return True，把 CancelledError 一起吞了，请求取消后协程仍在跑 | __exit__ 返回真值代表「异常已处理」。只应在明确声明的异常类型上返回 True，默认返回 False/None。更稳的做法是直接用 contextlib.suppress(具体异常) |
@contextmanager 生成的上下文对象存成模块级变量复用，第二次 with 报 generator didn't yield | 基于生成器的上下文管理器是一次性的，每次都要重新调用工厂函数创建；需要可复用就写成类并实现 __enter__/__exit__ |
| 异步生成器（SSE 流式）在客户端断开后没有被关闭，连接和下游 LLM 请求一直挂着不释放 | 用 contextlib.aclosing 或 try/finally 中 await agen.aclose() 显式关闭；FastAPI 里结合 request.is_disconnected() 主动跳出循环 |
重试逻辑里写了裸 except:，把 CancelledError 和 KeyboardInterrupt 一起吞了，Ctrl+C 按半天停不下来 | 裸 except: 等价于 except BaseException。业务兜底一律用 except Exception，并在最前面补一条 except asyncio.CancelledError: raise 显式放行取消信号 |
| 重试装饰器对所有异常一视同仁，400 参数错误也被重试 3 次，白烧 3 倍 token 还把错误延迟放大了 4 倍 | 重试前必须先分类：参数错误、鉴权失败、内容审核拒绝属于 FatalError 立即抛；超时、429、5xx 属于 RetryableError 才退避重试。给装饰器加 exceptions= 白名单，别用 Exception 兜底 |
finally 块里写了 return，原本要往上抛的异常被静默吞掉，线上报错日志里什么都没有 | finally 里的 return/break/continue 会丢弃正在传播的异常。finally 只做清理，清理代码本身也要用 try/except 兜住，不要让它抛出新异常覆盖原始异常 |
用 raise CustomError("失败") 重新包装异常但没写 from e，traceback 丢了根因，排查时只看到一句「调用失败」 | 包装异常统一用 raise CustomError(...) from e 设置 __cause__；确实要隐藏内部细节时显式写 from None，表达「这是故意的」而不是「忘了写」 |
迁到 asyncio.TaskGroup 后，原来 except ValueError 的兜底全部失效，异常直接冒到最外层 500 | TaskGroup 抛的是 ExceptionGroup，不会被 except ValueError 接住。改用 except* 分流，或 except BaseExceptionGroup as eg 后用 eg.split() 拆分；测试侧的 pytest.raises 也要同步改 |
异步测试函数写成 async def 但忘配 asyncio_mode，pytest 把它当普通函数「跑通」了，断言其实一次都没执行 | 装 pytest-asyncio 并在 pyproject.toml 配 asyncio_mode = "auto"；同时把 -W error::RuntimeWarning 打开，让 coroutine was never awaited 直接变成测试失败而不是一行警告 |
mock 异步下游用了 MagicMock，await 的时候报 object MagicMock can't be used in await expression | 异步函数一律用 unittest.mock.AsyncMock；验证重试次数要断言 await_count 而不是 call_count——拿到协程就算一次 call，根本不代表执行完了 |
| 单测里真等超时和指数退避，一个用例跑 30 秒，CI 全流程从 2 分钟涨到 20 分钟 | 超时、退避、TTL 的时间参数全部做成可注入，测试传极小值；TTL 用可替换的 clock 假时钟推进，退避直接 patch 掉 asyncio.sleep 并断言入参序列是 [1, 2, 4] |
📌 当日复盘任务
一、GIL 与并发模型（必做）
- [ ] 用一段纯计算函数（如
fib(32)）分别跑单线程、4 线程、4 进程，记录三者耗时，用数据解释 GIL 为什么让多线程在 CPU 密集场景无效甚至更慢 - [ ] 把上面的计算换成
time.sleep(1)模拟 IO，重跑三种模型，总结「CPU 密集用多进程 / IO 密集用协程或线程」的判断依据 - [ ] 画一张「进程 / 线程 / 协程」对比表，至少覆盖：调度方、切换成本、内存开销、能否绕过 GIL、典型适用场景、调试难度
- [ ] 用
sys.setswitchinterval()调整切换间隔，观察对多线程吞吐的影响，理解 5ms 时间片和护航效应
二、可变对象、拷贝与内存（必做）
- [ ] 手写 3 个可变默认参数翻车的例子（
list/dict/ 带默认值的 dataclass），并各写出正确版本 - [ ] 用
id()+copy+deepcopy验证「赋值 / 浅拷贝 / 深拷贝」三层语义，包含嵌套 dict 和自引用列表两种情况 - [ ] 给一个含循环引用的对象做
deepcopy，确认memo机制避免了无限递归 - [ ] 给一个高频创建的小类加上
__slots__，用tracemalloc或sys.getsizeof对比创建 100 万个实例的内存差异
三、装饰器与闭包（必做）
- [ ] 独立手写出题 1 的三合一装饰器（不看答案），并用 pytest 覆盖自测清单里至少 6 个 Case
- [ ] 故意删掉
functools.wraps，观察__name__、__doc__、inspect.signature的变化，并在 FastAPI 里复现路由冲突 - [ ] 写一个「叠放三个装饰器」的例子，用打印顺序验证装饰器的应用顺序（自下而上）与执行顺序（自上而下）
- [ ] 复现一次闭包延迟绑定的 bug，并分别用默认参数和
functools.partial两种方式修复
四、生成器与迭代器（必做）
- [ ] 造一个 50 万行的 JSONL 测试文件，分别用
readlines()全量读和生成器流式读，用tracemalloc对比峰值内存 - [ ] 独立实现
batch_by_budget，并专门为「单条超预算」「恰好整除」「残批收尾」三个 Case 写断言 - [ ] 实现一个三级生成器管道（读 → 清洗 → 攒批），并在中途
break，验证文件句柄被正确关闭 - [ ] 对比
yield from和await的字节码（dis模块），写下两者在委托语义上的异同
五、asyncio 协程（重点）
- [ ] 独立手写出题 3 的
batch_call，用计数器断言在途并发数不超过信号量上限 - [ ] 分别用
gather/gather(return_exceptions=True)/as_completed/TaskGroup实现同一批调用，总结四者在「部分失败」语义上的差异 - [ ] 在协程里故意调用
time.sleep和requests.get，用loop.set_debug(True)观察 slow callback 告警，再换成asyncio.to_thread对比 - [ ] 实现一次「外部取消 → 子任务全部回收」的完整链路，确认控制台没有
Task was destroyed but it is pending - [ ] 用
contextvars把trace_id透传到线程池里的同步函数，验证日志可以按请求聚合 - [ ] 写一个
classify()函数把下游异常分成RetryableError/FatalError/DegradableError，接到题 1 的重试装饰器上，验证 400 参数错误不会被重试 - [ ] 用
raise ... from e、from None、裸raise三种写法各抛一次异常，对比 traceback 输出和__cause__/__context__的取值 - [ ] 用
asyncio.TaskGroup并发抛出ValueError和KeyError，分别用except*和eg.split()两种方式分流处理
六、上下文管理器与缓存（选做加分）
- [ ] 实现题 4 的
Span，同时支持with和async with，并补上嵌套场景的contextvarsreset 断言 - [ ] 用
ExitStack管理数量不定的资源，验证某个资源获取失败时前面已获取的资源都被正确释放 - [ ] 实现题 5 的 TTL+LRU 缓存，并用假时钟单测 TTL 过期，用 50 并发验证防惊群只打一次下游
- [ ] 把
functools.lru_cache装到async def上复现报错，写下为什么以及正确的替代方案
七、异步测试工程化（选做加分）
- [ ] 在项目里配好
pytest-asyncio的asyncio_mode = "auto"与-W error::RuntimeWarning，故意漏写一个await验证它会直接测试失败 - [ ] 用
AsyncMock的side_effect=[TimeoutError(), TimeoutError(), "ok"]验证重试装饰器恰好 await 了 3 次 - [ ] 用
patch("asyncio.sleep", ...)记录退避入参，断言指数退避序列为[1, 2, 4]且单测耗时在毫秒级 - [ ] 给题 3 的
batch_call写一个 inflight 计数器测试，断言并发峰值不超过信号量上限，并确认 fake 函数里有await asyncio.sleep(0)制造交错 - [ ] 为异步生成器写「客户端提前 break」的用例，断言
aclose后finally里的清理代码确实执行过
八、面试表达（必做）
- [ ] 挑出 12 道面试题里自己最不熟的 4 道，对着镜子或录音做 90 秒口述，确保结论先行、有数据、有项目落点
- [ ] 整理一份属于自己的「Python 并发决策话术」：遇到一个新需求，你按什么顺序判断用线程、进程还是协程
- [ ] 把今天的 5 道手撕题整理成可运行的代码仓库，附 README 说明每道题的核心考点和踩坑点