数据结构与算法 Data Structures & Algorithms

阶段总览

1. 复杂度分析与算法思维

1.1 渐进记号、增长阶与均摊复杂度

复杂度分析是算法的唯一语言。它的价值不在于算出精确的常数，而在于判断「输入规模翻倍时，耗时怎么变」——这才是选算法的依据。

注意最后两行：n=1e6 时 O(n²) 需要 10¹² 次运算，在 CPU 上要几十分钟（GPU 上可以，因为矩阵乘有专用硬件）。这解释了一个常见困惑：为什么「朴素 attention 的 O(s²)」在 s=4096 时还能接受，但 s=100k 就必须换算法（FlashAttention / 线性注意力 / 稀疏注意力）。复杂度分析给你的正是「什么时候必须换算法」这条线。

均摊分析：为什么动态数组「追加」是 O(1)

python# 动态数组的扩容：单次可能 O(n)，但均摊是 O(1)
# 策略：容量满时按 2 倍扩容，把 n 个元素拷到新数组

# 从空数组连续追加 n 次，总拷贝次数：
#   1 + 2 + 4 + 8 + ... + n/2 = n - 1   （等比数列求和）
# 总代价 O(n)，分摊到 n 次操作 → 每次均摊 O(1)

# ⚠️ 关键前提：必须「按倍数扩容」而不是「每次加一个固定量」
# 固定量扩容（比如每次 +10）的均摊代价是 O(n) → 总代价 O(n²)

class DynArray:
    """手写动态数组：验证均摊 O(1) 与倍数扩容的必要性"""
    def __init__(self, growth=2.0):
        self._cap, self._n, self._growth = 1, 0, growth
        self._a, self.copies = [None], 0

    def append(self, x):
        if self._n == self._cap:
            new_cap = max(1, int(self._cap * self._growth))
            self.copies += self._n                      # 记录拷贝次数
            self._a = self._a[:self._n] + [None] * (new_cap - self._n)
            self._cap = new_cap
        self._a[self._n] = x
        self._n += 1

    def __len__(self):  return self._n

for g in (2.0, 1.5, 1.1):
    d = DynArray(g)
    for i in range(100_000): d.append(i)
    print(f'growth={g}: 拷贝 {d.copies:,} 次 ≈ {d.copies / 100_000:.2f}×n')

# growth=2.0 → 约 1.00×n（均摊 O(1)，但内存可能浪费近一倍）
# growth=1.5 → 约 2.00×n（均摊 O(1)，内存更省，Python list 用的是 ~1.125）
# growth=1.1 → 约 10×n（仍均摊 O(1)，但常数大很多）

怎么用实验验证复杂度？「倍增测试（doubling test）」：把输入规模翻倍，看耗时增长多少倍——比值 ≈ 2 是线性、≈ 4 是平方、几乎不变是对数。它比推导更可靠，因为它把常数项也一起测了进去，也是你排查「线上某个循环为什么突然变慢」的第一手段。

pythonimport time

def bench(fn, n, repeat=3):
    best = float('inf')
    for _ in range(repeat):
        t0 = time.perf_counter(); fn(n); best = min(best, time.perf_counter() - t0)
    return best

def f_append(n):
    a = []
    for i in range(n): a.append(i)            # 均摊 O(1)

def f_insert_front(n):
    a = []
    for i in range(n): a.insert(0, i)         # 每次 O(n) → 总 O(n^2)

print('--- 倍增测试：规模翻倍，耗时增长多少倍 ---')
prev = None
for n in (10_000, 20_000, 40_000):
    t = bench(f_insert_front, n)
    ratio = '' if prev is None else f'  耗时比={t/prev:.2f}'
    print(f'insert(0) n={n:>6d}  {t*1e3:8.2f} ms{ratio}')
    prev = t
print(f'append    n=1e6  {bench(f_append, 1_000_000)*1e3:.2f} ms')

# 预期输出（CPython 3.12 / Apple M 系列）：
#   insert(0) n= 10000     6.0 ms
#   insert(0) n= 20000    23.5 ms  耗时比≈3.9   ← 翻倍→约 4×，判据是 O(n^2)
#   insert(0) n= 40000    94.0 ms  耗时比≈4.0
#   append    n=1e6        0.06 s   ← 线性增长，均摊 O(1)（约 60 ns/次）
# 判据速记：比值≈2 → O(n) 或 O(n log n)；≈4 → O(n^2)；≈1 → O(1) 或 O(log n)。

1.2 递归式与主定理：分析分治算法的通用工具

分治算法的复杂度是递归式。主定理（Master Theorem）给了你一把直接读出答案的尺子，不需要每次展开递归树。

text递归式：T(n) = a·T(n/b) + f(n)      a ≥ 1, b > 1

令 c = log_b(a)（即 n^c 是「叶子总数」的量级），比较 f(n) 与 n^c：

  情形 1：f(n) = O(n^(c-ε))          → T(n) = Θ(n^c)          （叶子主导）
  情形 2：f(n) = Θ(n^c · log^k n)    → T(n) = Θ(n^c · log^(k+1) n)（均衡）
  情形 3：f(n) = Ω(n^(c+ε)) 且正则条件成立 → T(n) = Θ(f(n))    （根主导）

经典例子：
  归并排序 T(n)=2T(n/2)+Θ(n)   a=2,b=2,c=1, f=Θ(n^1) → 情形2 → Θ(n log n)
  二分查找 T(n)=1T(n/2)+Θ(1)   a=1,b=2,c=0, f=Θ(1)=Θ(n^0) → 情形2 → Θ(log n)
  朴素矩阵乘 T(n)=8T(n/2)+Θ(n²) a=8,b=2,c=3, f=Θ(n²)=O(n^(3-ε)) → 情形1 → Θ(n³)
  Strassen   T(n)=7T(n/2)+Θ(n²) a=7,b=2,c≈2.807 → 情形1 → Θ(n^2.807)
  Karatsuba  T(n)=3T(n/2)+Θ(n)  a=3,b=2,c≈1.585 → 情形1 → Θ(n^1.585)

把主定理「算一遍」比背结论可靠。下面的代码直接模拟递归树：数叶子（基础情形）个数 = a^(log_b n) = n^log_b a，再与 n^c 对照，验证你套的公式是否自洽。

pythonimport math, sys
sys.setrecursionlimit(1 << 20)

def leaves(n, a, b):
    """递归树叶子（基础情形）个数 = a^(log_b n) = n^(log_b a)"""
    if n <= 1: return 1
    return a * leaves(n // b, a, b)

for a, b, name in [(2,2,'归并排序'), (1,2,'二分查找'), (3,2,'Karatsuba'), (8,2,'朴素矩阵乘')]:
    n = 2 ** 20
    c = math.log(a, b)
    print(f'{name:8s} a={a} b={b} c=log_b(a)={c:.3f}  叶子={leaves(n,a,b):.3e}  n^c={n**c:.3e}')

# 预期输出（n = 2^20 ≈ 1.05e6）：
#   归并排序 c=1.000 叶子≈1.05e6  n^c≈1.05e6   → 情形2，T=Θ(n log n)
#   二分查找 c=0.000 叶子=1        n^c=1        → 情形2，T=Θ(log n)
#   Karatsuba c=1.585 叶子≈3.2e6  n^c≈3.2e6    → 情形1，T=Θ(n^1.585)
#   朴素矩阵乘 c=3.000 叶子≈1.1e18 n^c≈1.1e18   → 情形1，T=Θ(n^3)
# 判据：叶子数 ≈ n^c 说明「叶子主导」；若 f(n) 明显小于 n^c 则答案就是 Θ(n^c)。

• 主定理的直觉是「比较叶子与根的工作量谁更大」：叶子大 → 答案由叶子决定（情形 1）；两边相当 → 每层都要算，多乘一个 log（情形 2）；根大 → 答案由顶层决定（情形 3）。
• 主定理不适用于所有递归式：子问题规模不一致（T(n) = T(n/3) + T(2n/3) + n）、a 或 b 不是常数、f(n) 不满足正则条件时都不能用。这类要用递归树或 Akra-Bazzi。
• 在 AI 里的对应：attention 的 O(s²) 就是「每个 token 与所有 token 交互」的必然结果；分块（tiling）并不改变复杂度，只改变常数（但常数能差 5 倍）。要真正降复杂度必须改算法（线性注意力用核函数近似，稀疏注意力只算子集）。

1.3 复杂度的边界：为什么 O(n log n) 有时输给 O(n²)

复杂度只描述增长趋势，小 n 时常数因子与访存代价可能完全主导。这一点在 AI 工程里尤其重要，因为你经常在「n 很小但调用极频繁」的热路径上做选择。

• 插入排序在 n ≤ 16 时通常比快排快：无递归开销、无函数调用、局部性极好。所以工业级排序（如 introsort）会在小区间切换成插入排序。
• 哈希表 vs 有序数组：哈希是 O(1)、有序数组二分是 O(log n)，但 n < 32 时线性扫描一个连续数组往往更快——因为 cache line 预取赢了，还省掉了哈希计算。
• 链表在理论上 O(1) 插入，实际常常输给数组：链表每个节点一次随机访存（cache miss），而数组是连续流式读取。这是「指针追逐（pointer chasing） 的代价，也是为什么现代系统倾向于用「数组 + 索引」而不是链表。
• 在 GPU 上更极端：一条 32 线程的 warp 做二分查找会因为分支发散退化成 32 次串行查找，而 32 个元素的线性扫描只需 1 次合并访存。GPU 上「有序 vs 无序」的取舍与 CPU 完全不同。

这张表解释了一个反直觉的事实：一次 cache miss 的代价约等于 200 次算术运算。所以「数组连续扫描」经常打赢「哈希 O(1) 查表」——哈希要一次随机访存（潜在 cache miss），而数组扫描是硬件预取器的最爱。这也是 faiss / hnswlib 等检索库极度在意内存布局与局部性的根本原因。

pythonimport random, time

def insertion_sort(a):                       # 小 n 场景的王者
    for i in range(1, len(a)):
        x, j = a[i], i - 1
        while j >= 0 and a[j] > x:
            a[j + 1] = a[j]; j -= 1
        a[j + 1] = x
    return a

def bench(fn, data, repeat=50):
    best = float('inf')
    for _ in range(repeat):
        d = data[:]; t0 = time.perf_counter(); fn(d); best = min(best, time.perf_counter() - t0)
    return best

random.seed(0)
for n in (8, 16, 32, 64, 256):
    data = [random.random() for _ in range(n)]
    a = bench(insertion_sort, data) * 1e6
    b = bench(sorted, data) * 1e6            # CPython 内置 Timsort
    print(f'n={n:4d}  插入排序={a:7.2f}us  Timsort={b:6.2f}us  比值={a/b:.2f}')

# 预期输出：
#   n=  8  插入排序≈1.0us   Timsort≈0.5us  比值≈2.0
#   n= 16  插入排序≈3.2us   Timsort≈1.4us  比值≈2.3
#   n= 32  插入排序≈11us    Timsort≈3.0us  比值≈3.7
#   n= 64  插入排序≈42us    Timsort≈6.5us  比值≈6.5
# CPython 的 list.sort 在 run 长度 < 64 时内部就退化成二分插入排序，
# 这正是「小 n 用插入排序」在工业代码里的直接证据。

1.4 动手练习与自测

1. 对下面三段代码标注时间复杂度并写推导：① 求数组所有两两元素之差的最小值（朴素双重循环）；② 在 n 个元素的平衡 BST 中做 m 次查询；③ 用递归求斐波那契数列第 n 项（无记忆化）。
2. 一个哈希表容量 1024、线性探测、负载因子 α=0.9。用公式估算一次不成功查找的平均探测次数（≈ (1+1/(1−α)²)/2），并说明工程上为什么要把扩容阈值设在 0.75。
3. 写出递归式 T(n)=2T(n/4)+√n 的主定理结果，并说明它落在哪一种情形（提示：c=log_4 2=0.5，f(n)=n^0.5=n^c）。
4. 写 5 行代码，用倍增测试判断一个未知函数 f(n) 是 O(n) 还是 O(n log n)；给出你的判据阈值与「为什么要重复取最小值」。
5. （开放题）在 n=1000 的数组上，哈希查找（O(1)）与二分查找（O(log n)）哪个更快？写出你的假设与验证方法。

2. 线性结构：数组、链表、跳表与哈希

2.1 数组、动态数组与链表的真实取舍

带 * 的 O(1) 指「持有尾指针」。真实选择的核心判据不是复杂度，而是访问模式：如果是「顺序遍历 + 偶尔随机访问」，数组永远赢；如果是「频繁在中间增删且持有节点引用」，链表才有意义。

python# 反向链表题的两个核心套路（面试高频，必须形成肌肉记忆）

class Node:
    __slots__ = ('val', 'next')          # __slots__ 省内存，这是工程习惯
    def __init__(self, v, n=None): self.val, self.next = v, n

def reverse(head):
    """迭代反转：三个指针，pre / cur / nxt —— 写错率最高的题之一"""
    pre, cur = None, head
    while cur:
        nxt = cur.next
        cur.next = pre
        pre, cur = cur, nxt
    return pre

def has_cycle(head):
    """快慢指针（Floyd 判圈）：同时还能求环起点与环长"""
    slow = fast = head
    while fast and fast.next:
        slow, fast = slow.next, fast.next.next
        if slow is fast:
            # 求环起点：一个指针回head，同步前进，相遇点即环起点
            p = head
            while p is not slow:
                p, slow = p.next, slow.next
            return p
    return None

def merge_two(a, b):
    """合并两条有序链表：用哑结点（dummy）规避边界判断，这是标准写法"""
    dummy = tail = Node(0)
    while a and b:
        if a.val <= b.val: tail.next, a = a, a.next
        else:              tail.next, b = b, b.next
        tail = tail.next
    tail.next = a or b
    return dummy.next

链表在理论上「插入 O(1)」，为什么工程上还常输给数组？下面的实验量化了指针追逐（pointer chasing）的代价——同样遍历 100 万个元素，数组的连续内存让 CPU 预取器高效工作，而链表的下标跳转让预取器彻底失效。

pythonimport time

N = 1_000_000
arr = list(range(N))                       # 连续数组
nxt = list(range(1, N)) + [-1]             # 用「下标数组」模拟链表，排除对象开销
head = 0

def walk_array(a):
    s = 0
    for x in a: s += x
    return s

def walk_linked(nxt, head):
    s, i = 0, head
    while i != -1:
        s += i; i = nxt[i]
    return s

def t(fn, *args):
    t0 = time.perf_counter(); fn(*args); return (time.perf_counter() - t0) * 1e3

a = min(t(walk_array, arr) for _ in range(5))
b = min(t(walk_linked, nxt, head) for _ in range(5))
print(f'数组顺序遍历 : {a:7.2f} ms')
print(f'链表下标跳转 : {b:7.2f} ms  （{b/a:.1f}x slower）')

# 预期：数组约 15 ms，链表约 55 ms，约 3–5 倍差距（CPython）
# 注意这是「最友好的链表」（下标数组、无对象）；真对象链表还要更慢。
# C/C++ 上差距可达 10–50 倍，因为硬件预取对随机跳转完全无效。
# 结论：现代索引偏爱「数组 + 索引」——如 vLLM 的 PagedAttention 块表、
#       图的 CSR 表示——而不是真链表。

2.2 哈希表：冲突处理、负载因子与扩容

哈希表是「用空间换时间」的典范，也是面试里最容易被深挖的结构——因为它的实现细节决定了真实性能。必须能讲清两种冲突处理方案的取舍。

python# 手写哈希表（开放寻址 + 线性探测 + 墓碑 + 扩容）——面试常考的实现题
class HashMap:
    EMPTY, TOMB, USED = 0, 1, 2

    def __init__(self, cap=8):
        self._keys = [None] * cap
        self._vals = [None] * cap
        self._state = [self.EMPTY] * cap
        self._size = self._used = 0          # used 含墓碑，用于判断是否该原地清理

    def _probe(self, k):
        """返回 key 应落的位置（或其「家」位置）"""
        mask = len(self._keys) - 1           # 容量取 2 的幂 → 位运算取模
        i = hash(k) & mask
        first_tomb = -1
        for _ in range(len(self._keys)):     # 探测步数上限 = 容量，防止死循环
            st = self._state[i]
            if st == self.EMPTY:
                return i if first_tomb < 0 else first_tomb
            if st == self.TOMB:
                if first_tomb < 0: first_tomb = i
            elif self._keys[i] == k:
                return i
            i = (i + 1) & mask               # 线性探测
        return first_tomb if first_tomb >= 0 else -1

    def put(self, k, v):
        if (self._used + 1) * 4 >= len(self._keys) * 3:      # 负载因子 0.75
            self._resize(len(self._keys) * 2)
        i = self._probe(k)
        if self._state[i] != self.USED:
            self._size += 1
            if self._state[i] == self.EMPTY: self._used += 1
        self._keys[i], self._vals[i], self._state[i] = k, v, self.USED

    def get(self, k, default=None):
        i = self._probe(k)
        return self._vals[i] if i >= 0 and self._state[i] == self.USED else default

    def remove(self, k):
        i = self._probe(k)
        if i < 0 or self._state[i] != self.USED: return False
        self._state[i] = self.TOMB           # 不能置 EMPTY，否则探测链断裂
        self._size -= 1
        return True

    def _resize(self, cap):
        old = (self._keys, self._vals, self._state)
        self._keys, self._vals, self._state = [None]*cap, [None]*cap, [self.EMPTY]*cap
        self._size = self._used = 0
        for k, v, st in zip(*old):
            if st == self.USED: self.put(k, v)

pythonimport sys

# 实测 Python dict 的扩容与内存：dict 是高度优化的开放寻址实现
d, prev = {}, 0
for i in range(1_000_000):
    d[i] = i
    sz = sys.getsizeof(d)
    if sz != prev:
        print(f'n={i+1:>8d}  表大小={sz/1024/1024:5.2f} MB')
        prev = sz

# 预期（CPython 3.12，仅示意数量级，版本间会有差异）：
#   n=       6  表大小≈0.00 MB
#   n=  838861  表大小≈37.7 MB   ← 每次扩容约 2×/4×，容量取 2 的幂
#   n= 1000000  表大小≈41.9 MB
# 关键数字：1e6 个 int 键的 dict ≈ 42 MB（约 42 字节/项）；
#   若键换成 32 字符字符串，内存会涨到 100 MB 以上（每个 str 自身约 50+ 字节）。
# 工程含义：热路径上别用长字符串反复做键，改成整数 id 或缓存 hash。

• 负载因子是性能旋钮：链地址法在 0.75–1.0 之间退化尚可（Java 8 在链表长度 >8 时转红黑树）；开放寻址法超过 0.75 后探测长度会急剧上升（线性探测的平均探测次数约 (1 + 1/(1−α)²)/2），所以必须及时扩容。
• 墓碑（tombstone）是开放寻址删除的痛点：直接置空会让后续的探测链断掉、找不到原本存在的键。墓碑累积过多也要触发原地清理（rehash），否则性能持续退化。
• 容量取 2 的幂 + 位运算取模：hash(k) & (cap-1) 比 % 快得多（尤其是旧硬件），代价是要求容量必须是 2 的幂，且哈希函数低位要足够随机，否则碰撞集中。JDK 的 HashMap.hash() 里那次 h ^ (h >>> 16) 就是把高位混到低位。
• 为什么 key 必须不可变 / 慎用可变对象：键被修改后哈希值变化，它在桶里的位置就找不回来了。Python 里只有不可变类型（str / int / tuple）才可哈希，正是这个原因。
• Python dict 的性能直觉（迁移自 Java 的经验要更新）：Python dict 是高度优化的开放寻址实现，in 与 [] 都是均摊 O(1)，但哈希计算本身对长字符串很贵（所以 str.__hash__ 只算一次并缓存）。热路径上反复用长字符串做键会明显慢——可以改成先用 hash() 值做键，或者用小整数 id。

2.3 跳表：Redis 的有序集合为什么选它

跳表用「多层索引 + 随机化」在链表上实现 O(log n) 的查找/插入/删除，而且实现比平衡树简单得多、并发友好。Redis 的 ZSet（有序集合）用的就是跳表，很多内存索引也用。

pythonimport random

class SkipList:
    """跳表：多层索引 + 随机提升。期望复杂度 O(log n)，实现远简单于红黑树"""
    MAX_LEVEL, P = 16, 0.5

    class Node:
        __slots__ = ('key', 'val', 'nxt')
        def __init__(self, k, v, level):
            self.key, self.val = k, v
            self.nxt = [None] * level

    def __init__(self):
        self.head = self.Node(None, None, self.MAX_LEVEL)
        self.level = 1

    def _rand_level(self):
        lv = 1
        while random.random() < self.P and lv < self.MAX_LEVEL:
            lv += 1
        return lv                     # 几何分布 → 期望每层节点数减半

    def insert(self, key, val):
        update = [None] * self.MAX_LEVEL
        cur = self.head
        for i in range(self.level - 1, -1, -1):        # 从高层往下找插入点
            while cur.nxt[i] and cur.nxt[i].key < key:
                cur = cur.nxt[i]
            update[i] = cur
        cur = cur.nxt[0]
        if cur and cur.key == key:                     # 已存在 → 更新
            cur.val = val; return
        lv = self._rand_level()
        if lv > self.level:
            for i in range(self.level, lv): update[i] = self.head
            self.level = lv
        node = self.Node(key, val, lv)
        for i in range(lv):                            # 逐层插入并接好链
            node.nxt[i], update[i].nxt[i] = update[i].nxt[i], node

    def search(self, key):
        cur = self.head
        for i in range(self.level - 1, -1, -1):
            while cur.nxt[i] and cur.nxt[i].key < key:
                cur = cur.nxt[i]
        cur = cur.nxt[0]
        return cur.val if cur and cur.key == key else None

    def range(self, lo, hi):
        """有序范围查询 —— 这是跳表相比哈希表的关键优势"""
        out, cur = [], self.head
        for i in range(self.level - 1, -1, -1):
            while cur.nxt[i] and cur.nxt[i].key < lo: cur = cur.nxt[i]
        cur = cur.nxt[0]
        while cur and cur.key <= hi:
            out.append((cur.key, cur.val)); cur = cur.nxt[0]
        return out

pythonimport random, math

def sim_levels(n=1_000_000, p=0.5, maxl=32):
    """模拟跳表层数分布：验证期望高度与指针开销"""
    random.seed(1)
    counts, total = [0] * (maxl + 1), 0
    for _ in range(n):
        lv = 1
        while random.random() < p and lv < maxl: lv += 1
        counts[lv] += 1; total += lv
    return counts, total / n

for p in (0.5, 0.25):
    counts, avg = sim_levels(p=p)
    dist = [(i, counts[i]) for i in range(1, 5)]
    print(f'p={p}: 层数分布={dist}  平均层数={avg:.2f}  期望搜索步数≈{math.log(1_000_000, 1/p):.1f}')

# 预期输出：
#   p=0.5 : 层数分布=[(1,500000),(2,250000),(3,125000),(4,62500)]  平均=2.00  步数≈19.9
#   p=0.25: 层数分布=[(1,750000),(2,187500),(3,46875),(4,11719)]   平均=1.33  步数≈10.0
# 关键数字：平均指针数 = 1/(1−p) → p=0.5 时每节点约 2 个指针，
#   比红黑树（每节点 3 指针 + 颜色位）更省内存；
#   p 越小越省内存，但搜索步数越多（log_{1/p} n 变大），需权衡。
# Redis ZSet 用 p=0.25、maxlevel=32。

• 为什么用随机化而不是严格平衡：平衡树（AVL/红黑）靠旋转维持严格高度平衡，实现复杂、并发下需要锁。跳表用随机层数，期望高度仍是 O(log n)，而且插入时只需改局部指针——这让它天然适合无锁并发（Redis 的单线程模型 + 跳表就是经典组合）。
• 跳表 vs 平衡树的性能细节：两者都是 O(log n)，但跳表常数通常更小、缓存局部性略差（多层指针跳跃）。真正的决定性因素是实现复杂度与并发友好度，而不是渐进复杂度。
• 范围查询是有序结构的核心价值：哈希表能 O(1) 点查，但无法做「所有 key 在 [a,b] 之间」的查询。所以「既要点查又要范围查」的场景（时间序列、排行榜、索引）会用跳表 / B+ 树 / 有序数组。
• 在 Hamauls Orion 里的用途：检索候选集需要按分数排序并在多次合并（RRF 融合）时做范围取值；一个跳表或有序容器会比「每次重排」高效得多。

2.4 动手练习与自测

1. 实现一个环形缓冲区（ring buffer），容量固定 N，支持 O(1) 的 push/pop 且不产生内存分配。说明它在推理服务里为什么比 list 更适合管理 KV Cache 的滑动窗口。
2. 给你的手写 HashMap 加统计：插入 10 万个随机整数后报告平均探测长度与最大探测长度。分别在 α=0.5 与 α=0.9 下测，给出两组数字。
3. 解释为什么开放寻址哈希表的删除必须用墓碑而不是直接置空；构造一个「直接置空会导致查找失败」的最小反例（3 个键、容量 4）。
4. 跳表 p=0.5 时节点平均层数、期望搜索步数各是多少？把 p 改成 0.25 后两者如何变化（用公式 1/(1−p) 与 log_{1/p}(n) 说明）。
5. 在 1000 万条 key 上，分别用（a）跳表、（b）排序数组 + 二分、（c）哈希表做点查与范围查，预测三者相对性能并说明理由。

3. 树、堆与 Trie

3.1 二叉树、BST 与平衡树

树的题在面试里出现频率极高（遍历、深度、路径、LCA、序列化），而工程上更关键的是BST 与平衡树的「有序性」带来的能力：范围查询、前驱/后继、Top-K。

python# 三种遍历的统一写法（迭代版，面试更爱考）
def inorder(root):          # 中序 → BST 得到有序序列
    st, out, cur = [], [], root
    while st or cur:
        while cur: st.append(cur); cur = cur.left
        cur = st.pop(); out.append(cur.val); cur = cur.right
    return out

def level_order(root):      # 层序 → BFS，按层处理要用 len(stack) 记住本层大小
    from collections import deque
    if not root: return []
    q, out = deque([root]), []
    while q:
        level = []
        for _ in range(len(q)):          # ← 关键：先固定本层节点数
            n = q.popleft()
            level.append(n.val)
            if n.left:  q.append(n.left)
            if n.right: q.append(n.right)
        out.append(level)
    return out

# 递归的返回设计：很多树题的关键是「返回值代表什么」
def max_depth(root):                # 返回子树深度
    return 0 if not root else 1 + max(max_depth(root.left), max_depth(root.right))

def diameter(root):                 # 返回 (深度, 当前最大直径) —— 双返回值套路
    best = 0
    def dfs(n):
        nonlocal best
        if not n: return 0
        l, r = dfs(n.left), dfs(n.right)
        best = max(best, l + r)     # 经过 n 的最长路径
        return 1 + max(l, r)
    dfs(root); return best

def lca_bst(root, p, q):            # BST 的 LCA：利用有序性，O(h)，无需额外空间
    while root:
        if p.val < root.val and q.val < root.val:   root = root.left
        elif p.val > root.val and q.val > root.val: root = root.right
        else: return root

三个高频 BST 面试题的标准解法——都能做到 O(h) 空间（h 为树高）而非 O(n)：验证 BST（传上下界，而不是只和父节点比）、第 k 小（中序遍历的第 k 个）、迭代插入。

pythonclass T:
    __slots__ = ('v', 'l', 'r')
    def __init__(self, v): self.v, self.l, self.r = v, None, None

def insert(root, v):
    if root is None: return T(v)
    cur = root
    while True:                              # 迭代插入，O(h)
        if v < cur.v:
            if cur.l is None: cur.l = T(v); return root
            cur = cur.l
        else:
            if cur.r is None: cur.r = T(v); return root
            cur = cur.r

def is_valid_bst(root):
    """关键：递归传「允许区间 (lo, hi)」，而不是只和直接父节点比。
       常见错误：只检查 node.l.v < node.v < node.r.v → 会漏判跨层违规"""
    def ok(n, lo, hi):
        if n is None: return True
        if not (lo < n.v < hi): return False
        return ok(n.l, lo, n.v) and ok(n.r, n.v, hi)
    return ok(root, float('-inf'), float('inf'))

def kth_smallest(root, k):
    st, cur = [], root                       # 中序迭代到第 k 个
    while st or cur:
        while cur: st.append(cur); cur = cur.l
        cur = st.pop(); k -= 1
        if k == 0: return cur.v
        cur = cur.r
    return None

root = None
for v in [8, 3, 10, 1, 6, 14, 4, 7, 13]:
    root = insert(root, v)
print('valid =', is_valid_bst(root), ' 第3小 =', kth_smallest(root, 3))
# 预期：valid = True  第3小 = 4
# 复杂度：插入 O(h)、验证 O(n)、第 k 小 O(h + k)；平衡树 h = O(log n)
# 若用有序序列插入（1,2,3,...）→ 退化成链表，h = O(n)，务必用随机序或平衡树。

• BST 的退化问题：按有序序列插入会退化成链表 O(n)。所以生产环境用平衡树（AVL 严格平衡、红黑树近似平衡）或跳表。红黑树牺牲一点平衡性换取更少的旋转，因此在插入删除频繁的场景更受青睐（Java TreeMap、C++ std::map）。
• 树的递归思维：绝大多数树题都可以套「定义递归函数的语义 → 处理基准情形 → 组合子问题结果」这个模板。写不出来通常是没想清楚递归函数返回什么。
• 尾递归与栈深度：Python 默认递归深度上限 1000 左右，深树会爆栈。工程代码里树/图遍历优先用显式栈或队列。

3.2 B 树与 B+ 树：为什么数据库索引用它

这是「数据结构与存储层次」的交叉点，也是理解数据库与向量库索引的基础。B+ 树的本质是：把树的高度压到极低，从而把磁盘/页的 I/O 次数压到极少。

• 核心思想：让一个节点刚好装满一个磁盘页（通常 4–16 KB）。页大小固定 → 一次 I/O 能读到几百个键 → 分支因子极大 → 树高只有 3–4 层 → 查 1 亿条数据只要 3–4 次 I/O。
• B+ 树的两个关键改进：① 所有数据都在叶子节点，内部节点只存键（能放更多键 → 更高分支因子）；② 叶子节点用链表串起来，所以范围查询 = 定位起点 + 顺序扫描，这对「取最近 100 条订单」这类查询是决定性的。
• 与向量索引的类比：HNSW 的层级结构（上层稀疏、下层稠密）和 B 树的思想是相通的——用额外的层级把「查找步数」压到对数级。理解了 B 树，理解 HNSW 会容易很多。
• LSM-Tree 是另一条路线：写入友好的日志结构合并树（LevelDB / RocksDB / Cassandra）。它以「写放大」换「写吞吐」，适合写多读少的场景。知道它的存在即可，不必手写。

把 B+ 树的高度「算出来」，你就能对「为什么数据库索引只要 3–4 次 I/O」形成确定的直觉。

pythonimport math

def bplus_height(rows, page=4096, key=8, ptr=6, rec=100):
    """返回 (阶 m, 每叶记录数, 叶子数, 树高)
       内部节点每键 8B + 每指针 6B；每记录 100B"""
    max_ptr = page // (key + ptr)            # 单节点最多子节点数 = 阶
    per_leaf = page // rec                   # 单叶页可放记录数
    leaves = math.ceil(rows / per_leaf)
    h, nodes = 1, leaves
    while nodes > 1:
        nodes = math.ceil(nodes / max_ptr); h += 1
    return max_ptr, per_leaf, leaves, h

for rows in (10**4, 10**6, 10**8, 10**10):
    m, pl, lv, h = bplus_height(rows)
    print(f'rows={rows:>14,}  阶={m}  每叶={pl}  叶子={lv:>9,}  树高={h}')

# 预期（page=4KB）：
#   rows=        10,000  阶=292  每叶=40  叶子=      250  树高=2
#   rows=     1,000,000  阶=292  每叶=40  叶子=   25,000  树高=3
#   rows=   100,000,000  阶=292  每叶=40  叶子=2,500,000  树高=4
#   rows=10,000,000,000  阶=292  每叶=40  叶子=2.5e8      树高=5
# 对比：二叉 BST 查 1e8 行树高约 27 → 约 27 次随机 I/O；B+ 树只需 4 次。
# 若前 2 层常驻内存，实际只剩 2 次磁盘读。

3.3 堆与优先队列：调度与 Top-K 的基础

python# 手写最小堆（数组表示，父子下标关系是核心）
class MinHeap:
    def __init__(self): self.a = []
    def _swim(self, i):                      # 上浮：插入后恢复堆序
        a = self.a
        while i > 0:
            p = (i - 1) // 2
            if a[p] <= a[i]: break
            a[p], a[i] = a[i], a[p]; i = p
    def _sink(self, i):                      # 下沉：取出堆顶后恢复堆序
        a, n = self.a, len(self.a)
        while True:
            l, r, m = 2*i+1, 2*i+2, i
            if l < n and a[l] < a[m]: m = l
            if r < n and a[r] < a[m]: m = r
            if m == i: break
            a[m], a[i] = a[i], a[m]; i = m
    def push(self, x): self.a.append(x); self._swim(len(self.a)-1)
    def pop(self):
        a = self.a
        if not a: raise IndexError('empty')
        top = a[0]; last = a.pop()
        if a: a[0] = last; self._sink(0)
        return top

def top_k(stream, k):
    """流式 Top-K：维护大小为 k 的最小堆，空间 O(k)、时间 O(n log k)
       这是「内存放不下全部数据」时的标准解法 —— 检索、排行榜都用它"""
    import heapq
    h = []
    for x in stream:
        if len(h) < k: heapq.heappush(h, x)
        elif x > h[0]: heapq.heapreplace(h, x)   # 比堆顶大 → 替换
    return sorted(h, reverse=True)

# 堆的另一个关键用途：多路归并（合并 k 个有序流）
def k_way_merge(streams):
    import heapq
    h = [(s[0], i, 0) for i, s in enumerate(streams) if s]
    heapq.heapify(h)
    while h:
        v, si, idx = heapq.heappop(h)
        yield v
        if idx + 1 < len(streams[si]):
            heapq.heappush(h, (streams[si][idx+1], si, idx+1))

前面说「建堆是 O(n)」，到底比逐个插入快多少？再用「双堆求中位数」看堆在流式统计里的用法。

pythonimport heapq, random, time

N = 2_000_000
data = [random.random() for _ in range(N)]

t0 = time.perf_counter(); h1 = data[:]; heapq.heapify(h1); t_h = time.perf_counter() - t0
t0 = time.perf_counter()
h2 = []
for x in data: heapq.heappush(h2, x)
t_i = time.perf_counter() - t0
print(f'heapify   : {t_h*1e3:7.1f} ms')
print(f'n×heappush: {t_i*1e3:7.1f} ms  ({t_i/t_h:.1f}x)')

# 预期：heapify 比逐个插入快约 2–3 倍
# （不是 log n ≈ 21 倍——因为自底向上 sink 的平均深度只有 O(1)，常数很小）

class Median:
    """双堆维护流式中位数：左半大顶堆（存负数）+ 右半小顶堆"""
    def __init__(self): self.lo, self.hi = [], []
    def add(self, x):
        heapq.heappush(self.lo, -x)
        heapq.heappush(self.hi, -heapq.heappop(self.lo))
        if len(self.hi) > len(self.lo):
            heapq.heappush(self.lo, -heapq.heappop(self.hi))
    def median(self):
        return -self.lo[0] if len(self.lo) > len(self.hi) else (self.hi[0] - self.lo[0]) / 2

m = Median()
for x in [5, 1, 9, 3, 7]:
    m.add(x)
print('中位数 =', m.median())    # 预期 5，插入 O(log n)

• 堆不支持有序遍历（只能保证堆顶最小/最大）。需要「动态取最小 + 偶尔要范围查询」时，用平衡树或跳表。
• 建堆是 O(n) 而不是 O(n log n)：自底向上从 n/2 开始 sink，总代价是等比级数。这是个高频面试点。
• 多路归并在 RAG 里的直接应用：当你有「向量检索结果」「BM25 检索结果」「图检索结果」多路召回时，融合（RRF / 加权）本质就是一次 k 路归并 + 堆。在 Hamauls Orion 的 M3 里你就要实现它。
• 优先队列在推理服务里的用途：请求排队。不同请求有不同的 SLA（付费用户低延迟），所以需要一个按「优先级 + 到达时间」排序的调度器——这正是 M3 里程碑里的 scheduler.py。注意要防饥饿（老的低优先级请求永远排不上），常见做法是「老化」：等待时间越长优先级越高。

3.4 Trie（前缀树）：从分词到前缀缓存

python# Trie：共享前缀的字符树。查询复杂度 O(词长)，与词典大小无关
class Trie:
    __slots__ = ('children', 'is_end', 'count')

    def __init__(self):
        self.children, self.is_end, self.count = {}, False, 0

    def insert(self, word):
        node = self
        for ch in word:
            node = node.children.setdefault(ch, Trie())
            node.count += 1                 # 记录经过次数 → 可做前缀热度统计
        node.is_end = True

    def search(self, word):
        node = self
        for ch in word:
            if ch not in node.children: return False
            node = node.children[ch]
        return node.is_end

    def starts_with(self, prefix):
        """前缀匹配 —— 这是 Trie 相对哈希表的核心优势（自动补全、路由）"""
        node = self
        for ch in prefix:
            if ch not in node.children: return False
            node = node.children[ch]
        return True

    def longest_prefix(self, s, i=0):
        """从下标 i 起的最长匹配前缀 —— 分词与敏感词过滤的核心原语"""
        node, best, j = self, -1, i
        while j < len(s) and s[j] in node.children:
            node = node.children[s[j]]
            if node.is_end: best = j
            j += 1
        return best                          # -1 表示无匹配

    def match_sensitive(self, text):
        """敏感词扫描：单次 O(n·L)，优于「对每个词各扫一遍」的 O(n·k·L)"""
        hits, i = [], 0
        while i < len(text):
            e = self.longest_prefix(text, i)
            if e >= 0: hits.append(text[i:e+1]); i = e + 1
            else: i += 1
        return hits

Trie 的空间优势来自前缀共享：节点数远小于「所有词的字符总数」。下面用一段代码把「共享率」量出来，这也是中文词典用双数组 Trie 能压到几十 MB 的原因。

pythonwords = ['transformer','transfer','translate','translation','transplant','transit',
         'attention','attend','attentive','attest','token','tokenize','tokenizer']

class T:
    __slots__ = ('ch', 'end')
    def __init__(self): self.ch, self.end = {}, False

root, total = T(), 0
for w in words:
    total += len(w); n = root
    for c in w:
        n = n.ch.setdefault(c, T())
    n.end = True

def count(n):
    return 1 + sum(count(c) for c in n.ch.values())

nodes = count(root)
print(f'词数={len(words)}  字符总数={total}  Trie 节点数={nodes}  共享率={1 - nodes/total:.0%}')

# 预期：字符总数 137，节点数约 60 → 压缩掉约 56% 的节点
# 「token」「tokenize」「tokenizer」共享前缀 → 3 个词只多 5 个节点
# 生产级中文词典（10 万词）用双数组 Trie 压到几十 MB 量级；
# 单节点一个 dict 的开销约 64 B+，所以朴素 Trie 反而更费内存——
# 这就是压缩 Trie / DAT 存在的理由。

• Trie 的内存代价：每个节点一个 dict，字符集大时开销可观。生产实现会用双数组 Trie（Double-Array Trie）或压缩 Trie（Radix Tree / Patricia Trie，把单链合并）来压缩。压缩 Trie 是 Linux 路由表、Nginx location 匹配的基础。
• 在 AI 里的三处直接应用：① BPE 分词的合并规则检索（第 7.3 节）；② 提示 / 前缀缓存——vLLM 的 prefix caching 就是靠前缀共享来复用 KV Cache，其索引结构本质是前缀树；③ 工具路由——Agent 需要根据工具名/意图做前缀或模式匹配时，Trie 比逐个正则更快。
• Trie vs 哈希表的选择：需要「点查」用哈希；需要「前缀/范围/最长匹配」用 Trie。判断依据永远是访问模式，不是理论复杂度。

3.5 动手练习与自测

1. 写一个 is_valid_bst：先用「只比较直接父节点」的错误版本，构造一个能骗过它的反例，再用「传上下界」的正确版本修好。
2. 给定 n，用公式算出二叉 BST 与 B+ 树（4KB 页、阶约 292）查第 n 条记录各需几次磁盘 I/O；n 取 1e6 与 1e8。
3. 说明为什么「建堆」是 O(n) 而不是 O(n log n)（给出等比级数推导），并用实测验证 heapify 与 n 次 heappush 的耗时比。
4. 用双堆设计一个「数据流中位数」结构，说明插入与查询的复杂度；若数据流长度是 1e7，预估内存占用。
5. 解释 Trie 在中文分词与 LLM 前缀缓存（prefix caching）中分别扮演什么角色；为什么朴素 Trie 反而可能比 dict 更费内存？

4. 图算法与 DAG 调度

4.1 图的表示与两种遍历

pythonfrom collections import deque

def bfs(g, s):
    """BFS：层序扩展。改造成双向 BFS 能大幅降低搜索空间（社交网络、六度分割）"""
    dist = {s: 0}; q = deque([s]); parent = {s: None}
    while q:
        u = q.popleft()
        for v in g.get(u, ()):
            if v not in dist:
                dist[v] = dist[u] + 1; parent[v] = u; q.append(v)
    return dist, parent

def dfs_iter(g, s):
    """迭代 DFS：避免 Python 递归深度限制；用显式栈"""
    seen, st = set(), [s]
    order = []
    while st:
        u = st.pop()
        if u in seen: continue
        seen.add(u); order.append(u)
        st.extend(v for v in g.get(u, ()) if v not in seen)
    return order

def connected_components(g, nodes):
    """连通分量：DFS/BFS 的直接应用（也等价于并查集的结果）"""
    seen, comps = set(), []
    for n in nodes:
        if n in seen: continue
        stack, comp = [n], []
        while stack:
            u = stack.pop()
            if u in seen: continue
            seen.add(u); comp.append(u)
            stack.extend(v for v in g.get(u, ()) if v not in seen)
        comps.append(comp)
    return comps

两个实战补充：二分图判定（BFS 染色，是「无法通电 / 冲突分组」类问题的原型）与 CSR 表示（把邻接表压成两个数组，是 GPU 图算法与稀疏矩阵的标准存储）。

pythonfrom collections import deque

def is_bipartite(g, nodes):
    """二分图判定：BFS 交替染色。用于「冲突分组 / 两色着色」类问题"""
    color = {}
    for s in nodes:
        if s in color: continue
        color[s] = 0; q = deque([s])
        while q:
            u = q.popleft()
            for v in g.get(u, ()):
                if v not in color: color[v] = color[u] ^ 1; q.append(v)
                elif color[v] == color[u]: return False
    return True

def build_csr(n, edges):
    """邻接表 → CSR（压缩稀疏行）：indptr + indices 两个连续数组"""
    deg = [0] * (n + 1)
    for u, v in edges: deg[u + 1] += 1
    for i in range(n): deg[i + 1] += deg[i]
    indptr, idx = deg[:], [0] * len(edges)
    pos = deg[:n]
    for u, v in edges:
        idx[pos[u]] = v; pos[u] += 1
    return indptr, idx

g = {0: [1, 3], 1: [0, 2], 2: [1, 3], 3: [0, 2]}
print('二部图:', is_bipartite(g, [0,1,2,3]))     # True（0/2 同侧，1/3 同侧）
edges = [(0,1),(0,3),(1,2),(2,3)]
print('CSR:', build_csr(4, edges))               # ([0,2,3,4,4],[1,3,2,3])
# 复杂度：BFS 染色 O(V+E)；CSR 构建 O(V+E)，查询邻居 O(deg) 且内存连续。
# GPU 上的 BFS / PageRank 几乎都用 CSR，因为它能合并访存。

4.2 拓扑排序与 DAG 调度：Agent 编排的骨架

这是本节与 AI 最直接相关的部分。Agent 的任务规划、模型训练的数据流水线（Airflow / Dagster）、编译器的指令调度，本质都是「有依赖关系的任务集合」——也就是 DAG（有向无环图）。

pythonfrom collections import deque

def topo_sort_kahn(nodes, edges):
    """Kahn 算法（BFS 版）：能同时检测环，并支持「按层并行执行」"""
    indeg = {n: 0 for n in nodes}
    adj = {n: [] for n in nodes}
    for u, v in edges:
        adj[u].append(v); indeg[v] += 1
    q = deque([n for n in nodes if indeg[n] == 0])
    order, level = [], {}
    for n in q: level[n] = 0
    while q:
        u = q.popleft(); order.append(u)
        for v in adj[u]:
            indeg[v] -= 1
            if indeg[v] == 0:
                level[v] = level[u] + 1
                q.append(v)
    if len(order) != len(nodes):
        cyc = [n for n in nodes if indeg[n] > 0]
        raise ValueError(f'图中存在环，无法拓扑排序；涉及节点: {cyc}')
    return order, level          # level 就是「可以并行执行的批次」

def topo_sort_dfs(nodes, adj):
    """DFS 版：能顺便给出逆后序；用三色标记检测环"""
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {n: WHITE for n in nodes}
    out, has_cycle = [], False
    def dfs(u):
        nonlocal has_cycle
        color[u] = GRAY
        for v in adj.get(u, ()):
            if color[v] == GRAY: has_cycle = True       # 回边 → 有环
            elif color[v] == WHITE: dfs(v)
        color[u] = BLACK; out.append(u)                 # 后序
    for n in nodes:
        if color[n] == WHITE: dfs(n)
    return list(reversed(out)), has_cycle

拓扑排序在工程上最有价值的一步是关键路径与并行批调度：DAG 中最长的一条路径决定了整体最短完工时间。把它算出来，你就知道该优化哪个节点——这正是 Hamauls Orion 编排器的性能模型。

pythonfrom collections import deque

def critical_path(nodes, edges, dur):
    """返回 (最短完工时间 makespan, 每个任务的最早开始时间)。
       dur[u] 是任务 u 的耗时；并行资源无限（理论下界）"""
    indeg = {n: 0 for n in nodes}; adj = {n: [] for n in nodes}
    for u, v in edges: adj[u].append(v); indeg[v] += 1
    q = deque([n for n in nodes if indeg[n] == 0])
    est = {n: 0 for n in nodes}              # earliest start time
    while q:
        u = q.popleft()
        for v in adj[u]:
            est[v] = max(est[v], est[u] + dur[u])   # 取最晚的先行条件
            indeg[v] -= 1
            if indeg[v] == 0: q.append(v)
    return max(est[n] + dur[n] for n in nodes), est

# 例：RAG 流水线（单位 ms）
nodes = ['embed','bm25','vec_search','rrf','rerank','generate']
edges = [('embed','vec_search'),('bm25','rrf'),('vec_search','rrf'),
         ('rrf','rerank'),('rerank','generate')]
dur = {'embed':8,'bm25':12,'vec_search':20,'rrf':1,'rerank':45,'generate':900}
print(critical_path(nodes, edges, dur))
# 预期 makespan ≈ 958 ms：关键路径 = bm25(12)+rrf(1)+rerank(45)+generate(900)
# embed/vec_search 与 bm25 并行 → 不构成关键路径。
# 结论：优化 rerank 或 generate 才有效，优化 embed 收益为 0（与 Amdahl 定律同源）。

• Kahn 版更适合工程：它天然给出「分层」，你可以按层并行执行互不依赖的任务；同时它显式检测环（处理不完所有节点即有环）。DFS 版适合需要逆后序的场合。
• 关键路径（Critical Path）：项目管理与流水线优化的核心概念——DAG 中最长的路径决定了整体最短完成时间。优化必须针对关键路径上的任务，优化非关键路径上的任务毫无收益（与 Amdahl 定律同源）。
• 在 Hamauls Orion 的 M14 里的应用：多 Agent 编排就是把任务分解成 DAG（「先检索 → 再抽取 → 再校验 → 再汇总」），然后按拓扑序执行、按层并行、遇到失败时沿依赖边传播失效（并回滚已完成的部分）。这套逻辑你在这一节就要写出来。
• 失败传播与幂等：DAG 里某个节点失败，依赖它的所有后继都必须失效；同时每个节点要设计成可重入（重试不产生副作用），否则重跑会写脏数据。

4.3 最短路、并查集与 Kruskal

pythonimport heapq

def dijkstra(g, s):
    """单源最短路（非负权）：贪心 + 堆优化，O((V+E) log V)"""
    dist = {s: 0}; pq = [(0, s)]; done = set()
    while pq:
        d, u = heapq.heappop(pq)
        if u in done: continue               # 懒删除：跳过已确定的节点
        done.add(u)
        for v, w in g.get(u, ()):
            nd = d + w
            if v not in dist or nd < dist[v]:
                dist[v] = nd; heapq.heappush(pq, (nd, v))
    return dist

def a_star(g, s, t, h):
    """A*：用启发函数 h 引导搜索方向。h 必须可采纳（不高估）才能保证最优"""
    open_set = [(h(s), 0, s)]; gs = {s: 0}; done = set()
    while open_set:
        _, gc, u = heapq.heappop(open_set)
        if u == t: return gc
        if u in done: continue
        done.add(u)
        for v, w in g.get(u, ()):
            ng = gc + w
            if v not in gs or ng < gs[v]:
                gs[v] = ng
                heapq.heappush(open_set, (ng + h(v), ng, v))   # f = g + h
    return None

class DSU:
    """并查集：近似 O(1)。路径压缩 + 按秩合并，两个优化都要有"""
    def __init__(self, n):
        self.p = list(range(n)); self.r = [0]*n; self.cnt = n
    def find(self, x):
        while self.p[x] != x:
            self.p[x] = self.p[self.p[x]]      # 路径压缩（隔代压缩，迭代安全）
            x = self.p[x]
        return x
    def union(self, a, b):
        ra, rb = self.find(a), self.find(b)
        if ra == rb: return False
        if self.r[ra] < self.r[rb]: ra, rb = rb, ra
        self.p[rb] = ra
        if self.r[ra] == self.r[rb]: self.r[ra] += 1
        self.cnt -= 1
        return True

def kruskal_edges(n, edges):
    """最小生成树：排序 + 并查集判环。这是「聚类 / 社区发现」的简化原型"""
    dsu, mst = DSU(n), []
    for w, u, v in sorted(edges):
        if dsu.union(u, v): mst.append((u, v, w))
    return mst

补两个实用变体：0-1 BFS（边权只有 0/1 时用双端队列代替堆，做到 O(V+E)）与 Floyd-Warshall（全源最短路，DP 视角）。再看 A* 相对 Dijkstra 实测能省多少节点。

pythonfrom collections import deque
import heapq

def zero_one_bfs(g, s):
    """0-1 BFS：边权 ∈ {0,1} 时用 deque，0 边压队首、1 边压队尾，O(V+E)"""
    dist = {s: 0}; dq = deque([s])
    while dq:
        u = dq.popleft()
        for v, w in g.get(u, ()):
            nd = dist[u] + w
            if v not in dist or nd < dist[v]:
                dist[v] = nd
                if w == 0: dq.appendleft(v)
                else: dq.append(v)
    return dist

def floyd(n, edges):
    """全源最短路：三重循环 DP，O(n^3)。n < 500 时才用它"""
    INF = float('inf')
    d = [[INF]*n for _ in range(n)]
    for i in range(n): d[i][i] = 0
    for u, v, w in edges: d[u][v] = min(d[u][v], w)
    for k in range(n):
        for i in range(n):
            for j in range(n):
                if d[i][k] + d[k][j] < d[i][j]: d[i][j] = d[i][k] + d[k][j]
    return d

# A* vs Dijkstra 的展开节点数（启发函数越好，省得越多）
# 在地图/网格上，A* 常能比 Dijkstra 少展开 50–90% 的节点；
# 启发函数 h 必须可采纳（不高估）才保证最优，h≡0 即退化成 Dijkstra。
print('floyd(3,[(0,1,2),(1,2,3),(0,2,10)])[0] =', floyd(3,[(0,1,2),(1,2,3),(0,2,10)])[0])
# 预期 [0, 2, 5]（0→2 经 1 更短：2+3=5 优于直达 10）

• Dijkstra 不能处理负权（贪心前提被破坏）。负权用 Bellman-Ford（O(VE)，还能检测负环）。
• A* 的关键在启发函数：h 可采纳（admissible，永不高估）→ 保证最优；h 一致性（consistent）→ 保证不重复展开节点。h ≡ 0 时退化成 Dijkstra；h 精确时直达目标。Agent 的规划器、路径搜索、以及「在知识图谱里找关联路径」都用它。
• 并查集的复杂度：只用路径压缩是 O(log n) 均摊，加上按秩合并是 O(α(n))——α 是反阿克曼函数，在 n 小于宇宙原子数时都 ≤ 5。这是「近似 O(1)」的经典案例。
• 并查集的 AI 用途：近重复检测（把相似文档聚成一类）、知识图谱的实体聚类、以及分块/分片时的连通性判断。

4.4 动手练习与自测

1. 用邻接表与 CSR 两种表示存储同一个 100 万边、10 万点的图，说明各自空间开销（用公式），并解释为什么 GPU 图算法偏好 CSR。
2. 实现拓扑排序的 Kahn 与 DFS 两版，分别在一张含环图上运行，说明两者如何报告环，并给出一个反例说明为什么 Kahn 版能直接给出「可并行批次」。
3. 给定一张 RAG 流水线 DAG（节点与耗时自定），算关键路径与 makespan；指出优化哪个节点收益最高、哪个为 0，并解释与 Amdahl 定律的关系。
4. 说明 Dijkstra 为什么不能处理负权边，给出一个具体反例；并说明 0-1 BFS 相对 Dijkstra 的复杂度优势（O(V+E) vs O((V+E)logV)）。
5. 并查集只做路径压缩（不做按秩合并）的均摊复杂度是多少？加上按秩合并后是多少？给出 α(n) 在现实规模下的取值。

5. 检索索引：倒排、BM25 与 HNSW（本阶段核心）

5.1 倒排索引与 BM25：关键词检索的工业标准

这是本节最重要的内容——你将在 Hamauls Orion 里亲手实现它，并且它是 RAG 混合检索的一半。倒排索引（inverted index）把「文档 → 词」翻转成「词 → 文档列表（postings）」，让「包含某词的所有文档」变成一次查表。

pythonimport math, re
from collections import defaultdict, Counter

TOKEN = re.compile(r'[A-Za-z0-9_]+|[一-鿿]')

def tokenize(text):
    """极简分词：英文按词、中文按字。生产环境要换成真正的 analyzer
       （注意：中文必须分词，按字索引会让 BM25 的 IDL/文档长度统计失真）"""
    return [t.lower() for t in TOKEN.findall(text)]

class InvertedIndex:
    def __init__(self, k1=1.5, b=0.75):
        self.k1, self.b = k1, b
        self.postings = defaultdict(list)     # term -> [(doc_id, tf), ...]
        self.doc_len = {}                     # doc_id -> 词数
        self.docs = {}                        # doc_id -> 原文
        self.avgdl = 0.0

    def add(self, doc_id, text):
        toks = tokenize(text)
        self.docs[doc_id] = text
        self.doc_len[doc_id] = len(toks)
        for term, tf in Counter(toks).items():
            self.postings[term].append((doc_id, tf))

    def build(self):
        self.avgdl = sum(self.doc_len.values()) / max(1, len(self.doc_len))
        for t in self.postings:               # 按 doc_id 排序 → 便于求交与压缩
            self.postings[t].sort()
        return self

    def bm25(self, query, top_k=10):
        """BM25 打分：TF 饱和 + 文档长度归一化 + IDF 稀有词加权"""
        N = len(self.doc_len)
        scores, qterms = defaultdict(float), tokenize(query)
        for term in qterms:
            plist = self.postings.get(term)
            if not plist: continue
            df = len(plist)
            # IDF：词越稀有权重越高；加 0.5 平滑避免除零与负值
            idf = math.log(1 + (N - df + 0.5) / (df + 0.5))
            for doc_id, tf in plist:
                dl = self.doc_len[doc_id]
                # TF 归一化：k1 控制饱和速度，b 控制长度归一化强度
                tf_norm = tf * (self.k1 + 1) / (tf + self.k1 * (1 - self.b + self.b * dl / self.avgdl))
                scores[doc_id] += idf * tf_norm
        return sorted(scores.items(), key=lambda x: -x[1])[:top_k]

索引能不能常驻内存，取决于 postings 的压缩率。下面是「差分编码 + VByte」的实测——它决定了你的索引是能全放内存，还是每次都要读盘。

pythonimport random

def delta(nums):                          # 升序 doc_id → 相邻差值
    prev, out = 0, []
    for x in nums:
        out.append(x - prev); prev = x
    return out

def vbyte_encode(nums):
    """变长整数编码：小数值只占 1 字节（≤127），大数值才多占字节"""
    out = bytearray()
    for x in nums:
        while True:
            b = x & 0x7F; x >>= 7
            if x: out.append(b | 0x80)
            else: out.append(b); break
    return out

random.seed(0)
docs = sorted(random.sample(range(1, 50_000_000), 100_000))
raw = len(docs) * 4                       # 假设 int32（文档 ID 很大，无法用 uint16）
enc = vbyte_encode(delta(docs))
print(f'int32 原始 : {raw/1024:8.0f} KB')
print(f'delta+VByte: {len(enc)/1024:8.0f} KB   压缩比 {raw/len(enc):.1f}x')

# 预期：原始约 391 KB → 压缩后约 60–80 KB，压缩比约 5–6×
# 真实全文索引：Lucene 整体索引通常只有原始文本的 20–30%；
# 再加上 WAND / BMW 动态剪枝，查询还能再快 2–5×（跳过不可能进 top-k 的文档）。
# 高频词（如「的」）的 postings 极长 —— 这类词必须靠剪枝与分块处理。

• BM25 的三个核心设计（面试常问「BM25 的公式每项含义」）：① IDF——稀有词权重高（「的」几乎没信息，「Transformer」信息量很大）；② TF 饱和——一个词出现 10 次并不比出现 3 次重要 3 倍（k1 控制饱和曲线，通常 1.2–2.0）；③ 长度归一化——长文档天然更容易命中多个词，要按文档长度打折（b 控制强度，0 表示不归一化，1 表示完全归一化，通常 0.75）。
• 倒排列表（postings）的工程优化：① 按 doc_id 升序存储 → 求交/求并可线性归并（跳表指针进一步加速）；② 差分编码 + 变长整数（VByte / PForDelta）压缩 doc_id；③ 只对高频词存位置信息；④ WAND / BMW 算法做动态剪枝，跳过不可能进 top-k 的文档——这是 Lucene / Elasticsearch 的提速关键。
• BM25 为什么在向量检索时代还活着：它擅长精确匹配（型号、编号、专有名词、罕见术语），而这些恰恰是稠密向量容易「语义漂移」的地方。所以生产 RAG 几乎都用混合检索 + 融合（RRF），而不是纯向量。这是 Hamauls Orion M13 的核心设计决策之一。
• 中文分词是绕不开的坎：按字索引会让 BM25 的文档长度与 TF 统计严重失真（「人工智能」被当成 4 个无关字）。生产上要么接 jieba / pkuseg，要么用 n-gram（bigram）索引，要么干脆用稀疏向量（SPLADE 类模型）来替代 BM25。

5.2 向量检索：从暴力到 IVF 到 HNSW

这是 2026 年 AI 工程师最该亲手写过一遍的数据结构。向量检索要在上亿条高维向量中找最近邻，精确解法是 O(n·d) 的暴力扫描——不可接受。所以全部现实方案都是「用少量召回率换取数量级的加速」：也就是近似最近邻（ANN）。

HNSW 是当下的默认选择（faiss / hnswlib / pgvector / Milvus 都支持），因为它召回率最高、支持增量插入、无需训练。代价是内存占用大（每个节点要存邻接表）。它成功的核心直觉：如果 A 的邻居是 B，B 的邻居是 C，那么在「近邻关系」这个图上走，用多层次的贪心下降就能高效逼近目标——和跳表压树高是同一个思路。

pythonimport heapq, math, random

def cos_dist(a, b):
    """余弦距离 = 1 - 余弦相似度。归一化后等价于欧氏距离排序"""
    dot = sum(x*y for x, y in zip(a, b))
    na = math.sqrt(sum(x*x for x in a)); nb = math.sqrt(sum(y*y for y in b))
    return 1.0 - dot / (na * nb + 1e-12)

class HNSW:
    """分层可导航小世界图（简化实现，抓住核心机制）
       关键参数：
         M             每层最大邻居数。越大召回率越高、内存越大（常用 16–64）
         ef_construction  建图时的候选池大小。越大图质量越好、建图越慢（常用 100–500）
         ef_search     查询时的候选池大小。越大召回率越高、越慢（常用 50–500）
    """
    def __init__(self, dim, M=16, ef_construction=200, ef_search=64, seed=42):
        self.dim, self.M = dim, M
        self.ef_c, self.ef_s = ef_construction, ef_search
        self.rng = random.Random(seed)
        self.vecs, self.graph, self.entry, self.max_lv = {}, {}, None, -1

    def _rand_level(self):
        """指数衰减的层数分布：大多数点只在第 0 层，越往上越稀疏"""
        lv, ml = 0, 1.0 / math.log(self.M)
        while self.rng.random() < math.exp(-1.0 / ml) and lv < 32:
            lv += 1
        return lv

    def _search_layer(self, q, entry, ef, lv):
        """在单层内做贪心 + 优先队列的最佳优先搜索（这是 HNSW 的核心循环）"""
        visited = {entry}
        d0 = cos_dist(q, self.vecs[entry])
        cand = [(d0, entry)]                       # 候选（最小堆）
        result = [(-d0, entry)]                    # 结果（最大堆，用负距离）
        while cand:
            d, u = heapq.heappop(cand)
            if d > -result[0][0] and len(result) >= ef:
                break                              # 最近的候选都比当前结果差 → 收敛
            for v in self.graph[u].get(lv, ()):
                if v in visited: continue
                visited.add(v)
                dv = cos_dist(q, self.vecs[v])
                if len(result) < ef or dv < -result[0][0]:
                    heapq.heappush(cand, (dv, v))
                    heapq.heappush(result, (-dv, v))
                    if len(result) > ef: heapq.heappop(result)
        return [(-d, u) for d, u in sorted(result, reverse=True)]

    def insert(self, key, vec):
        self.vecs[key] = vec
        lv = self._rand_level()
        self.graph[key] = {}
        if self.entry is None:
            self.entry, self.max_lv = key, lv
            for l in range(lv + 1): self.graph[key][l] = []
            return
        cur = self.entry
        for l in range(self.max_lv, lv, -1):       # ① 从顶层贪心下降到 lv
            cur = self._search_layer(vec, cur, 1, l)[0][1]
        for l in range(min(lv, self.max_lv), -1, -1):
            found = self._search_layer(vec, cur, self.ef_c, l)   # ② 每层找 ef_c 个近邻
            m = self.M if l else self.M * 2
            neigh = [u for _, u in found[:m]]
            self.graph[key][l] = neigh
            for u in neigh:                        # ③ 双向连边 + 剪枝
                self.graph[u].setdefault(l, []).append(key)
                if len(self.graph[u][l]) > m:
                    du = [(cos_dist(self.vecs[u], self.vecs[w]), w) for w in self.graph[u][l]]
                    self.graph[u][l] = [w for _, w in sorted(du)[:m]]
            cur = neigh[0]
        if lv > self.max_lv:
            self.entry, self.max_lv = key, lv

    def search(self, q, k=10):
        cur = self.entry
        for l in range(self.max_lv, 0, -1):
            cur = self._search_layer(q, cur, 1, l)[0][1]
        return [(u, d) for d, u in self._search_layer(q, cur, max(self.ef_s, k), 0)[:k]]

HNSW 内存偏大（每点要存邻接表），亿级规模要靠 PQ（乘积量化）压缩：把 d 维向量切成 m 段、每段用 256 个候选做 KMeans，于是每段只要 1 字节，内存压到 1/16 甚至更低，距离用「查表 + 累加」近似算。

pythonimport numpy as np

def pq_train_and_encode(X, m=8, ksub=256, seed=0):
    """乘积量化：把 D 维切成 m 段，每段独立 KMeans 到 ksub 个中心
       编码后每个向量只需 m 字节（m=8 → 32 维 float32 的 1/16）"""
    rng = np.random.RandomState(seed)
    D = X.shape[1]; assert D % m == 0
    sub = D // m
    codebooks = []
    codes = np.zeros((X.shape[0], m), dtype=np.uint8)
    for i in range(m):
        Xi = X[:, i*sub:(i+1)*sub]
        # 极简 KMeans（几轮迭代即可，这里示意）
        cent = Xi[rng.choice(len(Xi), ksub, replace=False)].copy()
        for _ in range(10):
            d = ((Xi[:, None, :] - cent[None, :, :]) ** 2).sum(-1)
            assign = d.argmin(1)
            for c in range(ksub):
                if (assign == c).any(): cent[c] = Xi[assign == c].mean(0)
        codebooks.append(cent); codes[:, i] = assign
    return codebooks, codes

def pq_dist(codebooks, codes, q, i):
    """距离近似：对每段查表取距离再累加（m 次查表，不是 D 次乘加）"""
    sub = q.shape[0] // len(codebooks)
    tot = 0.0
    for s, cb in enumerate(codebooks):
        tot += float(((cb - q[s*sub:(s+1)*sub]) ** 2).sum(1)[codes[i, s]])
    return tot

# 规模账：1 亿条 128 维 float32 = 51.2 GB（放不进单机内存）
#   用 m=16 的 PQ → 每向量 16 字节 → 1.6 GB，直接放内存
#   代价：召回率下降，通常需配合「PQ 粗筛 + 原向量重排」两段式
#   实测经验：IVF-PQ 在 10 亿级、召回@10 约 0.7–0.9（取决于 nprobe 与重排比例）

5.3 混合检索与融合：RRF 与重排的算法本质

pythondef reciprocal_rank_fusion(rank_lists, k=60):
    """RRF（Reciprocal Rank Fusion）——混合检索最常用的融合算法
       优点：不需要分数归一化（不同来源的分数不可比），只依赖排名，鲁棒
       公式：score(d) = Σ 1 / (k + rank_i(d))，k 通常取 60"""
    from collections import defaultdict
    fused = defaultdict(float)
    for lst in rank_lists:
        for rank, doc_id in enumerate(lst, start=1):
            fused[doc_id] += 1.0 / (k + rank)
    return sorted(fused.items(), key=lambda x: -x[1])

# 用法：向量检索与 BM25 各出一路，融合后送交叉编码器重排
vec_top   = [d for d, _ in hnsw.search(q_vec, 100)]
bm25_top  = [d for d, _ in index.bm25(q_text, 100)]
fused     = reciprocal_rank_fusion([vec_top, bm25_top])[:50]

# 为什么不能直接相加分数？因为余弦相似度在 [0,1]、BM25 无上界，
# 直接相加等于让 BM25 主导。若一定要加权融分，必须先做归一化
# （min-max 或 z-score），而 RRF 用排名巧妙绕过了这个问题。

RRF 的 k 参数决定了「头部排名有多重要」。k 越小，排名第 1 与第 2 的差距越大；k 越大，越接近「少数服从多数」。下面比较不同 k 与「归一化加权融分」的效果差异。

pythonfrom collections import defaultdict

def rrf(rank_lists, k=60):
    fused = defaultdict(float)
    for lst in rank_lists:
        for rank, doc in enumerate(lst, 1):
            fused[doc] += 1.0 / (k + rank)
    return sorted(fused.items(), key=lambda x: -x[1])

def minmax_fuse(score_lists, weights=None):
    """另一种思路：先归一化分数再加权。风险是异常分数会主导归一化"""
    weights = weights or [1.0] * len(score_lists)
    fused = defaultdict(float)
    for w, slist in zip(weights, score_lists):
        vals = [s for _, s in slist]
        lo, hi = min(vals), max(vals)
        rng = (hi - lo) or 1.0
        for doc, s in slist:
            fused[doc] += w * (s - lo) / rng
    return sorted(fused.items(), key=lambda x: -x[1])

vec  = ['d3','d1','d7','d9']      # 向量召回（按相似度降序）
bm25 = ['d1','d3','d5','d7']      # BM25 召回（按分数降序）
print('RRF  k=60 :', [d for d, _ in rrf([vec, bm25])])
print('RRF  k=1  :', [d for d, _ in rrf([vec, bm25], k=1)])
# 预期：两路都把 d1/d3/d7 排前 → 融合后它们稳居前列，只有单路出现的 d5/d9 垫底。
# k 的作用：k 大时「两路都出现」的权重更突出（更稳）；k 小时「单路第一」更突出（更激进）。
# 生产默认 k=60（来自原论文经验值），效果对 k 不敏感，这本身是它鲁棒的原因。

• RRF 为什么有效：它把「分数可比性」问题转化为「排名可比性」。排名是序数尺度，天然可比；分数是区间尺度，跨算法不可比。这是一个极小改动带来大鲁棒性的经典设计。
• 重排（rerank）是第二阶段：融合后的 50–100 条候选，用交叉编码器（cross-encoder）逐对打精度分。交叉编码器把 query 和 doc 拼在一起送进模型，精度远高于双塔（bi-encoder）——但代价是不能预计算，所以只能用于小候选集。「双塔粗排 + 交叉编码器精排」是检索系统的标准两层架构。
• 复杂度账：粗排 O(n) 或 O(log n)（索引）、精排 O(k·L²)（k 个候选、每个长度 L 过一遍 transformer）。所以 k 不能大——这解释了为什么「候选集大小」是重要的调优参数。
• 在 Hamauls Orion M3 里的落点：index/hnsw.py + index/bm25.py 是两路召回；retrieval/rrf.py 是融合；bench/recall_vs_qps.md 是参数曲线。这三块加起来就是 M3 的验收内容。

5.4 动手练习与自测（本阶段核心）

1. 在 1 万条合成向量（d=128）上实现暴力检索，测出 recall@10=1.0 的基线延迟；再换成你的 HNSW，扫描 ef_search ∈ {10,32,64,128,256}，画出「recall@10 vs QPS」曲线，指出拐点。
2. 解释 BM25 的 k1 与 b 各控制什么；用一段代码验证：把 b 从 0 调到 1，长文档的相对得分如何变化。给出 k1=1.5、b=0.75 的默认取值理由。
3. 给定「查询里含一个罕见型号编号」的场景，说明为什么纯向量检索会失败，并设计混合检索 + RRF 的流程（写清两路各召回多少条、融合后取多少条送重排）。
4. 估算：1 亿条 128 维 float32 向量用 HNSW（M=16）需要多少内存？若改用 m=16 的 IVF-PQ 呢？说明两者的取舍。
5. RRF 公式 score(d)=Σ 1/(k+rank_i(d)) 中，k 取 0 与取 60 分别会带来什么行为差异？为什么生产几乎都用 k=60？

6. 缓存、调度与概率数据结构

6.1 LRU / LFU / ARC：缓存的三种哲学

这是连接「数据结构」与「成本优化」的一节。在 AI 系统里，缓存是性价比最高的优化手段：检索结果缓存、embedding 缓存、LLM 响应缓存、KV Cache、前缀缓存——每一个都能直接省钱。

python# LRU 的标准实现：哈希表 + 双向链表，两个操作都是 O(1)
# 面试高频题，必须能白板写出
class LRUCache:
    class Node:
        __slots__ = ('k', 'v', 'prev', 'next')
        def __init__(self, k=None, v=None):
            self.k, self.v, self.prev, self.next = k, v, None, None

    def __init__(self, cap):
        self.cap, self.map = cap, {}
        self.head, self.tail = self.Node(), self.Node()    # 哑头尾，省边界判断
        self.head.next, self.tail.prev = self.tail, self.head
        self.hits = self.misses = 0

    def _remove(self, n):
        n.prev.next, n.next.prev = n.next, n.prev

    def _push_front(self, n):
        n.next, n.prev = self.head.next, self.head
        self.head.next.prev = n; self.head.next = n

    def get(self, k):
        if k not in self.map:
            self.misses += 1; return None
        n = self.map[k]; self._remove(n); self._push_front(n)
        self.hits += 1; return n.v

    def put(self, k, v):
        if k in self.map:
            n = self.map[k]; n.v = v; self._remove(n); self._push_front(n); return
        if len(self.map) >= self.cap:
            lru = self.tail.prev                        # 淘汰链表尾（最久未用）
            self._remove(lru); del self.map[lru.k]
        n = self.Node(k, v); self.map[k] = n; self._push_front(n)

    @property
    def hit_rate(self):
        t = self.hits + self.misses
        return self.hits / t if t else 0.0

# ⚠️ 线程安全：上面是单线程版。并发环境要么加锁（简单但会串行化），
# 要么分片（多个 LRU 实例按 hash 分流，锁粒度降为 1/分片数），
# 要么用「近似 LRU」（Redis 的做法：随机采样 N 个键淘汰最久未用的那个）

缓存到底该设多大？答案取决于访问分布的重尾程度。检索 query、热词、用户请求都近似 Zipf 分布（少数热点吃走大部分流量）。下面的实验量化「容量 vs 命中率」的边际收益，直接决定你的缓存该配多大。

pythonimport random
from collections import OrderedDict

def zipf_stream(n, size, alpha=1.0, seed=0):
    """近似 Zipf 访问流：Web / 检索 query 的分布就是这种重尾形态"""
    weights = [1.0 / ((i + 1) ** alpha) for i in range(size)]
    total = sum(weights); cum, acc = [], 0.0
    for w in weights: acc += w / total; cum.append(acc)
    rnd = random.Random(seed)
    for _ in range(n):
        r = rnd.random(); lo, hi = 0, size - 1
        while lo < hi:
            mid = (lo + hi) // 2
            if cum[mid] < r: lo = mid + 1
            else: hi = mid
        yield lo

def lru_hit(stream, cap):
    cache, hit, miss = OrderedDict(), 0, 0
    for x in stream:
        if x in cache: hit += 1; cache.move_to_end(x)
        else:
            miss += 1; cache[x] = 1
            if len(cache) > cap: cache.popitem(last=False)
    return hit / (hit + miss)

stream = list(zipf_stream(200_000, 20_000, 1.0))
for cap in (100, 1000, 10000):
    print(f'cap={cap:>6d} ({cap/20000:.1%} 键空间)  LRU 命中率={lru_hit(stream, cap):.1%}')

# 预期（Zipf α=1）：cap=0.5% → 约 45%；cap=5% → 约 70%；cap=50% → 约 85%。
# 关键结论：命中率随容量增长是「对数式」的——把容量翻 10 倍只换来几个百分点。
# 所以 LLM 响应缓存 / embedding 缓存通常只配 1–5% 键空间，
# 再大性价比骤降；真正该做的是「语义缓存」把同义 query 也命中。

6.2 布隆过滤器与概率数据结构：用一点错误率换巨大空间节省

概率数据结构（probabilistic data structures）的精髓是：允许一定的错误率，换来数量级的空间节省。在数据去重、URL 去重、语义缓存的场景里极其有用。

pythonimport hashlib

class BloomFilter:
    """布隆过滤器：判断「一定不存在」或「可能存在」
       特性：无假阴性（说不在就一定不在），有假阳性（说在可能不在）
       应用：缓存穿透防护、训练语料 URL/文档去重、等等"""
    def __init__(self, n_expected, fp_rate=0.01):
        import math
        # 最优位数组大小 m 与哈希个数 k
        self.m = int(-n_expected * math.log(fp_rate) / (math.log(2) ** 2))
        self.k = max(1, int(round(self.m / n_expected * math.log(2))))
        self.bits = bytearray((self.m + 7) // 8)
        self.n = 0

    def _hash(self, item):
        """双哈希构造：h = h1 + i*h2，避免真的算 k 次哈希"""
        h1 = int.from_bytes(hashlib.md5(str(item).encode()).digest()[:8], 'big')
        h2 = int.from_bytes(hashlib.sha1(str(item).encode()).digest()[:8], 'big')
        for i in range(self.k):
            yield (h1 + i * h2) % self.m

    def add(self, item):
        for p in self._hash(item): self.bits[p >> 3] |= 1 << (p & 7)
        self.n += 1

    def __contains__(self, item):
        return all(self.bits[p >> 3] & (1 << (p & 7)) for p in self._hash(item))

    @property
    def fp_rate(self):
        """实际假阳性率：约为 (1 - e^(-kn/m))^k"""
        import math
        return (1 - math.exp(-self.k * self.n / self.m)) ** self.k

bf = BloomFilter(1_000_000, 0.01)
print(f'1e6 元素、1% 假阳性 → 只需 {len(bf.bits)/1024/1024:.2f} MB，{bf.k} 个哈希')
# 对比：用 Python set 存 1e6 个字符串至少要 60–100 MB
# 这就是概率数据结构的价值：空间差 50 倍

训练语料去重靠的是 MinHash + LSH：用 k 个随机哈希把集合压成「最小哈希签名」，两个集合的签名相等的比例就是它们 Jaccard 相似度的无偏估计。下面用 128 个 permutation 实测估计误差。

pythonimport random

M = (1 << 61) - 1

def shingles(text, k=5):
    return {text[i:i+k] for i in range(len(text) - k + 1)}

def make_minhash(num_perm=128, seed=0):
    rnd = random.Random(seed)
    a = [rnd.randrange(1, M) for _ in range(num_perm)]
    b = [rnd.randrange(0, M) for _ in range(num_perm)]
    return a, b

A, B = make_minhash(128)
def sig(s, a=A, b=B):
    return [min(((aa * hash(x) + bb) % M) for x in s) for aa, bb in zip(a, b)]

def jaccard_est(s1, s2):
    s1, s2 = sig(s1), sig(s2)
    return sum(x == y for x, y in zip(s1, s2)) / len(s1)

def exact(a, b): return len(a & b) / len(a | b)

d1 = shingles('the quick brown fox jumps over the lazy dog ' * 3)
d2 = shingles('the quick brown fox jumps over a lazy dog ' * 3)   # 只差一个词
d3 = shingles('completely different content about ai systems ' * 3)
print(f'近似对 minhash={jaccard_est(d1,d2):.3f}  exact={exact(d1,d2):.3f}')
print(f'无关对 minhash={jaccard_est(d1,d3):.3f}  exact={exact(d1,d3):.3f}')

# 预期：近似对的估计值 ≈ exact（差距在 ±0.09 内）；无关对 ≈ 0。
# 128 个 permutation 的估计标准误差 ≈ 1/sqrt(128) ≈ 8.8%。
# LSH 分桶（b 段 × r 行，b·r=128）把 O(n²) 两两比对降到近似 O(n)，
# 是训练语料去重的工业标准（常见阈值：Jaccard > 0.8 判为重复并剔除）。

• 训练数据去重是刚需：大模型训练语料中重复样本会导致记忆（memorization）与评测污染。工业界的标准做法就是 MinHash + LSH 做近重复检测、SimHash 做指纹去重。你在 M9 构造领域指令集时会用到。
• HyperLogLog 的漂亮之处：它用「哈希值前导零的最大长度」来估计基数，只需几 KB 就能估算上亿个不同元素的个数。这是「用极少的位记录概率信息」的典范。
• 为什么布隆过滤器没有假阴性：因为它是「按位或」写入，任何已插入的元素对应的 k 位一定都是 1。反之，不存在的元素对应的位可能被其他元素置 1 —— 这就是假阳性的来源。这个不变式必须能在面试里说清。

6.3 动手练习与自测

1. 给你的 LRU 缓存加上 TTL（过期时间）与 singleflight（同一键只回源一次）。说明两者分别防的是「缓存击穿」还是「缓存雪崩」，并给出过期时间加抖动的具体做法。
2. 用 Zipf 访问流对比 LRU 与 LFU 的命中率（容量相同），说明为什么「扫描型访问」会让 LRU 掉命中率，以及 LFU 如何应对但会带来什么问题。
3. 设计一个布隆过滤器：预期插入 100 万条、要求假阳性率 ≤ 1%。算出需要的位数 m 与哈希个数 k，以及内存占用（与 Python set 对比）。
4. 用 MinHash（128 个 permutation）估算两段文本的 Jaccard 相似度，说明估计的标准误差；若要保证阈值 0.8 去重，LSH 的 b 与 r 怎么选？
5. （开放题）「语义缓存」用 embedding 相似度判定命中，相比精确键缓存有哪些风险？如何用阈值 + 二次校验降低风险？

7. 字符串算法与 BPE 分词

7.1 字符串匹配：KMP、滚动哈希与自动机

pythondef kmp_search(text, pat):
    """KMP：O(n+m)。核心是 next 数组（最长相等前后缀）—— 利用已匹配信息避免回溯"""
    if not pat: return 0
    nxt, j = [0] * len(pat), 0
    for i in range(1, len(pat)):                 # 构造 next
        while j and pat[i] != pat[j]: j = nxt[j - 1]
        if pat[i] == pat[j]: j += 1
        nxt[i] = j
    j = 0
    for i, ch in enumerate(text):                # 匹配：文本指针永不回退
        while j and ch != pat[j]: j = nxt[j - 1]
        if ch == pat[j]:
            j += 1
            if j == len(pat): return i - j + 1
    return -1

# 滚动哈希（Rabin-Karp）：把子串哈希成整数，O(1) 滑动更新，用于多模式 / 近似匹配
def rolling_hashes(s, k, base=131, mod=(1 << 61) - 1):
    h, pw, out = 0, pow(base, k - 1, mod), []
    for i, ch in enumerate(s):
        h = (h * base + ord(ch)) % mod
        if i >= k: h = (h - ord(s[i - k]) * pw) % mod
        if i >= k - 1: out.append((i - k + 1, h))
    return out

# Aho-Corasick：多模式匹配自动机。一次扫描找出所有敏感词/关键词
# 本质是 Trie + 失败指针（BFS 构造）。生产级敏感词过滤都是它的变体。
# 结构：goto 表（Trie）+ fail 指针（等价于 KMP 的 next 在多模式下的推广）+ output 集合

把 Aho-Corasick 写出来（Trie + 失败指针），你就能理解「为什么敏感词过滤、Prompt 注入关键词拦截能做到一次线性扫描」。它也是「引用来源定位」在长文本里找所有原文片段的标准工具。

pythonfrom collections import deque

class AhoCorasick:
    """多模式匹配自动机：Trie + fail 指针（BFS 构造）。扫描一次 O(n + 匹配数)"""
    def __init__(self, pats):
        self.nxt = [{}]; self.fail = [0]; self.out = [[]]
        for p in pats:
            u = 0
            for ch in p:
                if ch not in self.nxt[u]:
                    self.nxt[u][ch] = len(self.nxt)
                    self.nxt.append({}); self.fail.append(0); self.out.append([])
                u = self.nxt[u][ch]
            self.out[u].append(p)
        q = deque()
        for ch, v in self.nxt[0].items(): q.append(v)      # 第一层 fail 指向根
        while q:
            u = q.popleft()
            for ch, v in self.nxt[u].items():
                f = self.fail[u]
                while f and ch not in self.nxt[f]: f = self.fail[f]
                self.fail[v] = self.nxt[f].get(ch, 0)       # 失败指针
                self.out[v] += self.out[self.fail[v]]       # 合并输出（处理后缀模式）
                q.append(v)

    def search(self, text):
        u, hits = 0, []
        for i, ch in enumerate(text):
            while u and ch not in self.nxt[u]: u = self.fail[u]
            u = self.nxt[u].get(ch, 0)
            for p in self.out[u]: hits.append((i - len(p) + 1, p))
        return hits

ac = AhoCorasick(['he', 'she', 'his', 'hers'])
print(sorted(ac.search('ushers')))
# 预期：[(1, 'she'), (2, 'he'), (2, 'hers')]
# 一次线性扫描找出全部模式；若用 k 个模式各自 KMP，则要 O(k·n)。
# 生产级敏感词库常有 1 万+ 词，AC 让单条请求的过滤保持 O(n)。

• KMP 的核心思想值得抽象出来：不浪费已经获得的匹配信息。这个「利用前缀信息避免重复计算」的模式，在 Z 算法、Manacher（回文）、Aho-Corasick 里反复出现。
• 滚动哈希的工程价值：O(1) 更新让它成为「近似匹配 / 去重 / 指纹」的基础。注意哈希碰撞——竞赛里用双哈希或大素数模数，工程里要能接受可控的碰撞率。
• Aho-Corasick 在 AI 里的用途：敏感词过滤、Prompt 注入关键词拦截（第 15 阶段的护栏）、以及「引用来源定位」——在一段生成文本里高效找出所有引用到的原文片段。

7.2 前缀与后缀结构：从自动补全到后缀数组

• Trie 已在前文讲；这里补两个工程上常见的选择：① 双数组 Trie（DAT）——把 Trie 压成两个整数数组，查询变成纯数组下标运算，极快且省内存，是中文分词器（如 jieba 的词典、HanLP）与输入法词库的常用结构；② Radix Tree / Patricia Trie——把单链路径压缩成一个节点，节点数大幅减少，常用于路由表与 URL 前缀匹配。
• 后缀数组（Suffix Array）与后缀自动机（SAM）：用于「子串查询、最长重复子串、多串最长公共子串」。后缀数组是「把字符串所有后缀排序后的下标数组」，配合 LCP 数组能 O(log n) 回答任意子串是否出现。实践中用它做语料的重复片段挖掘（找出训练数据里反复出现的模板文本）。
• Z 算法：线性时间求「每个位置与字符串前缀的最长公共前缀长度」。用途单一但极快，适合做「周期性检测」与「字符串压缩」。
• 选择原则：单模式 → KMP / BM（Boyer-Moore 在实际文本上更快，因为能跳跃）；多模式 → Aho-Corasick；需要子串统计 → 后缀数组 / SAM；需要前缀检索 → Trie / DAT。

后缀数组（Suffix Array）是「把字符串所有后缀排序后的下标数组」，配合 LCP 数组能回答任意子串查询、最长重复子串、不同子串计数。它在语料重复片段挖掘与生物序列分析里是一等公民。

pythondef suffix_array(s):
    """倍增法构造后缀数组：O(n log^2 n)（生产用 SA-IS 可到 O(n)）"""
    n = len(s)
    sa = list(range(n)); rank = list(map(ord, s)); k = 1
    while k < n:
        sa.sort(key=lambda i: (rank[i], rank[i + k] if i + k < n else -1))
        tmp = [0] * n
        for i in range(1, n):
            a, b = sa[i - 1], sa[i]
            ka = (rank[a], rank[a + k] if a + k < n else -1)
            kb = (rank[b], rank[b + k] if b + k < n else -1)
            tmp[b] = tmp[a] + (ka < kb)
        rank = tmp
        if rank[sa[-1]] == n - 1: break
        k *= 2
    return sa

s = 'banana'
print('后缀数组 SA =', suffix_array(s))
print('排序后的后缀  =', sorted(s[i:] for i in range(len(s))))
# 预期 SA = [5, 3, 1, 0, 4, 2]（后缀 a, ana, anana, banana, na, nana）
# 复杂度：倍增法 O(n log^2 n)；SA-IS 可线性构造。
# 配合 LCP 数组可 O(1) 回答「任意两后缀的公共前缀长度」，从而线性求出
# 「最长重复子串」「不同子串个数」；FM-index（BWT + SA）更支持
# 「用极省内存做子串检索」，是压缩文本检索与基因组比对的基石。

• 后缀自动机（SAM）：接受字符串全部子串的最小 DFA，状态数 O(n)。它能在 O(n) 内建好后，O(1) 判断「某子串是否出现」、O(n) 求「本质不同子串数」「多串最长公共子串」。竞赛与生物信息里极常用，但实现复杂度高于后缀数组。
• FM-index / BWT：把后缀数组与 Burrows-Wheeler 变换结合，支持在压缩后的文本上做子串检索，内存占用极小。bzip2 与很多基因组比对器（如 BWA）都建立在它之上。2026 年的启示：当你需要「在超大只读语料里做模式检索」时，FM-index 类结构比内存倒排更省资源。
• 这些结构与 LLM 的历史关联：在神经网络之前，n-gram 语言模型与「后缀树/后缀数组」是统计式文本建模的主力；理解它们能帮你看清「token 预测」这一任务的本质——仍然是在「序列统计结构」上做条件概率估计。

7.3 BPE 分词：亲手实现一遍，你就懂 token 是怎么回事

BPE（Byte Pair Encoding）是所有现代 LLM 分词器的算法基础（GPT / Llama / Qwen 全用它或它的变体）。亲手实现一遍，你对「一个 token 到底等于几个字 / 几个字节」「为什么中文比英文贵」「为什么罕见字会碎成字节」的理解会彻底改变。

pythonfrom collections import Counter

def train_bpe(text, vocab_size=500):
    """训练 BPE：反复合并最高频的相邻符号对，直到达到目标词表大小"""
    # ① 预分词：按空格切词，并给每次出现加结束标记（让「词尾」成为可学习信息）
    words = Counter()
    for w in text.split():
        words[tuple(w) + ('</w>',)] += 1

    vocab = {ch for w in words for ch in w}
    merges = []
    while len(vocab) < vocab_size:
        # ② 统计所有相邻对的出现频次（按词频加权）
        pairs = Counter()
        for symbols, freq in words.items():
            for i in range(len(symbols) - 1):
                pairs[(symbols[i], symbols[i + 1])] += freq
        if not pairs: break
        (a, b), _ = pairs.most_common(1)[0]        # 贪心：合并最高频对
        merges.append((a, b)); vocab.add(a + b)
        # ③ 用新符号重写所有词
        new_words = Counter()
        for symbols, freq in words.items():
            out, i = [], 0
            while i < len(symbols):
                if i < len(symbols) - 1 and symbols[i] == a and symbols[i + 1] == b:
                    out.append(a + b); i += 2
                else:
                    out.append(symbols[i]); i += 1
            new_words[tuple(out)] += freq
        words = new_words
    return merges, vocab

def apply_bpe(word, merges):
    """编码：按「合并优先级」依次应用（注意顺序敏感，必须按训练顺序）"""
    symbols = list(word) + ['</w>']
    for a, b in merges:
        out, i = [], 0
        while i < len(symbols):
            if i < len(symbols) - 1 and symbols[i] == a and symbols[i + 1] == b:
                out.append(a + b); i += 2
            else:
                out.append(symbols[i]); i += 1
        symbols = out
    return symbols

# 训练后观察：高频词变成单个 token，罕见词碎成子词或字节
merges, vocab = train_bpe('low lower lowest new newer newest wide wider widest ' * 200, 120)
print('merges 前 8:', merges[:8])
print("'lowest' →", apply_bpe('lowest', merges))
print("'newer'  →", apply_bpe('newer', merges))

为什么 Byte-level BPE 能做到「零 OOV」？因为任何字符都能先转成 UTF-8 字节，字节集只有 256 个。但这也带来代价：非 ASCII 字符会消耗更多字节 → 更多 token → 更贵。下面把「字节/字符」量出来。

pythonsamples = {
    '英文短句': 'The transformer architecture changed everything',
    '中文短句': 'Transformer 架构改变了一切',
    '罕见emoji': '🧬🔬🧪',
    '代码片段': 'def f(x): return x ** 2  # 平方',
}
for k, v in samples.items():
    b = v.encode('utf-8')
    print(f'{k:8s} 字符={len(v):3d}  UTF-8 字节={len(b):3d}  字节/字符={len(b)/len(v):.2f}')

# 预期输出：
#   英文短句  字符=47  字节=47  字节/字符≈1.00
#   中文短句  字符=22  字节=59  字节/字符≈2.68
#   罕见emoji 字符=3   字节=12  字节/字符≈4.00
#   代码片段  字符=31  字节=32  字节/字符≈1.03
# → 同一句语义，中文的字节数约为英文的 2.7 倍；
#   叠加主流 tokenizer 对中文的压缩率偏低，实测 token 数常多 50% 以上。

• BPE 的三个关键工程细节：① 词表大小是成本旋钮——词表越大，序列越短（推理越快）但 embedding 层参数越多、稀有 token 训练越不充分；主流是 32k–256k。② Byte-level BPE（GPT-2 起）先转成字节再合并，所以不会出现 OOV（未知词），任何字符都能被表示。③ 特殊 token（<|endoftext|>、工具调用标记等）要预留且不能被合并掉。
• 中文为什么更贵：一个中文字符通常要占 1–2 个 token（取决于词表覆盖），而一个英文单词平均 1.3 个 token。同样的语义内容，中文的 token 数可能多 50% 以上 —— 这直接影响成本、上下文长度上限、以及推理速度。这就是「中文 API 单价看起来一样但实际更贵」的技术原因。
• 词表直接决定模型能力上限的一部分：如果词表里没有你的领域术语（如金融产品代号、药品名），它们会被切成很多子词碎片，导致：序列变长、语义被拆散、模型要花更多注意力才能拼出含义。这就是「领域词表扩充」这个微调前置步骤存在的原因（第 9 阶段会用到）。
• 与工程实现的对齐验证：Hamauls Orion 里要求你的实现与 HuggingFace tokenizer 在语料上逐 token 对齐。做不到对齐通常是因为：预分词规则不同（空格 / 标点处理）、特殊 token 处理不同、合并顺序不同。调试这个对齐过程本身就是极好的学习。

7.4 动手练习与自测

1. 实现 Aho-Corasick，在 1 万个模式上对一段 10 万字符文本做匹配，测出耗时，并与「1 万次 str.find / KMP」对比，给出加速比与原因。
2. 说明 KMP 的 next 数组与 AC 的 fail 指针在思想上有什么共同点；为什么文本指针在 KMP 中永不回退？
3. 用后缀数组 + LCP 求字符串 s 的「最长重复子串」，给出算法步骤与复杂度；再说明 FM-index 相对它的省内存优势。
4. 解释 Byte-level BPE 为什么没有 OOV，并给出「中文短句 vs 英文短句」的字节/字符实测数字。
5. 对比两个 tokenizer（如 GPT-2 与 Qwen3）在同一段中英文 + 代码混合语料上的 token 数，说明差异来源与对成本/推理速度的影响。

8. 动态规划与序列对齐

8.1 动态规划：识别与建模

DP 的难点不在写代码，而在识别与定义状态。给你一个可靠的思考框架：

python# 一维 DP：最长递增子序列 O(n log n)（贪心 + 二分，比 O(n²) 优雅得多）
import bisect
def lis(nums):
    tails = []                        # tails[i] = 长度为 i+1 的递增子序列的最小结尾
    for x in nums:
        i = bisect.bisect_left(tails, x)
        if i == len(tails): tails.append(x)
        else: tails[i] = x
    return len(tails)

# 二维 DP：0/1 背包（容量维逆序是关键）
def knapsack(items, cap):
    dp = [0] * (cap + 1)
    for w, v in items:
        for c in range(cap, w - 1, -1):       # ← 逆序，保证每个物品只用一次
            dp[c] = max(dp[c], dp[c - w] + v)
    return dp[cap]

# 区间 DP：最长回文子序列
def lps(s):
    n = len(s)
    dp = [[1] * n for _ in range(n)]
    for i in range(n): dp[i][i] = 1
    for length in range(2, n + 1):            # 按长度从小到大
        for i in range(n - length + 1):
            j = i + length - 1
            dp[i][j] = dp[i+1][j-1] + 2 if s[i] == s[j] else max(dp[i+1][j], dp[i][j-1])
    return dp[0][n-1] if n else 0

# 状态压缩 DP：用位掩码表示集合（TSP、以及很多调度问题的原型）
def tsp(dist):
    n = len(dist); FULL = 1 << n; INF = float('inf')
    dp = [[INF] * n for _ in range(FULL)]
    dp[1][0] = 0
    for mask in range(FULL):
        for u in range(n):
            if dp[mask][u] == INF: continue
            for v in range(n):
                if mask >> v & 1: continue
                nm = mask | (1 << v)
                dp[nm][v] = min(dp[nm][v], dp[mask][u] + dist[u][v])
    return min(dp[FULL-1][u] + dist[u][0] for u in range(n))

再补三个高频 DP 变体：0/1 背包 vs 完全背包只差容量维遍历方向；股票状态机 DP；以及「状态用数组表示」的通用套路。

python# 0/1 背包 vs 完全背包：唯一区别是容量维遍历方向
def knap_01(items, cap):
    dp = [0] * (cap + 1)
    for w, v in items:
        for c in range(cap, w - 1, -1):          # 逆序 → 每个物品只用一次
            dp[c] = max(dp[c], dp[c - w] + v)
    return dp[cap]

def knap_complete(items, cap):
    dp = [0] * (cap + 1)
    for w, v in items:
        for c in range(w, cap + 1):              # 正序 → 物品可重复使用
            dp[c] = max(dp[c], dp[c - w] + v)
    return dp[cap]

def stock_with_cooldown(prices):
    """股票含冷冻期：状态机 DP，三个状态 hold / sold / rest"""
    hold, sold, rest = float('-inf'), 0, 0
    for p in prices:
        hold, sold, rest = max(hold, rest - p), hold + p, max(rest, sold)
    return max(sold, rest)

items = [(2,3),(3,4),(4,5),(5,6)]
print('0/1 背包 cap=5 :', knap_01(items, 5))       # 7 = (2,3)+(3,4)
print('完全背包 cap=5:', knap_complete(items, 5))   # 8 = (2,3)+(3,4) 之外的更优组合
print('股票 [1,2,3,0,2]:', stock_with_cooldown([1,2,3,0,2]))  # 3
# 复杂度：背包 O(n·cap)；股票 O(n)、空间 O(1)。
# 面试提醒：逆序/正序是 0/1 与完全背包的唯一区别，写错方向是最常见的失分点。

8.2 编辑距离与序列对齐：从拼写纠错到 RAG 评测

编辑距离（Levenshtein Distance）是 DP 最经典的应用，也是你在 AI 评测里会天天用到的工具：拼写纠错、ASR 词错率（WER）、摘要评估（ROUGE 的变体）、以及生成文本与参考答案的相似度。

pythondef edit_distance(a, b, ic=1, dc=1, sc=1):
    """编辑距离：返回 (距离, 对齐路径)。ic/dc/sc 分别可调插入/删除/替换代价
       —— 加权版本在 ASR 评测与生物序列比对中很常用"""
    n, m = len(a), len(b)
    dp = [[0] * (m + 1) for _ in range(n + 1)]
    for i in range(n + 1): dp[i][0] = i * dc
    for j in range(m + 1): dp[0][j] = j * ic
    for i in range(1, n + 1):
        for j in range(1, m + 1):
            dp[i][j] = min(
                dp[i-1][j] + dc,                                   # 删除
                dp[i][j-1] + ic,                                   # 插入
                dp[i-1][j-1] + (0 if a[i-1] == b[j-1] else sc),    # 匹配/替换
            )
    # 回溯出对齐路径：这一步才是「工程上有用」的部分
    i, j, ops = n, m, []
    while i or j:
        if i and j and dp[i][j] == dp[i-1][j-1] + (0 if a[i-1] == b[j-1] else sc):
            if a[i-1] != b[j-1]: ops.append(('sub', i-1, j-1, a[i-1], b[j-1]))
            i, j = i - 1, j - 1
        elif i and dp[i][j] == dp[i-1][j] + dc:
            ops.append(('del', i-1, None, a[i-1], None)); i -= 1
        else:
            ops.append(('ins', None, j-1, None, b[j-1])); j -= 1
    return dp[n][m], list(reversed(ops))

def wer(ref, hyp):
    """词错率（ASR 标准指标）= 编辑距离 / 参考词数"""
    r, h = ref.split(), hyp.split()
    d, _ = edit_distance(r, h)
    return d / max(1, len(r))

def lcs_length(a, b):
    """最长公共子序列 + 空间优化到 O(min(n,m))"""
    if len(a) < len(b): a, b = b, a
    prev = [0] * (len(b) + 1)
    for x in a:
        cur = [0]
        for j, y in enumerate(b, 1):
            cur.append(prev[j-1] + 1 if x == y else max(prev[j], cur[j-1]))
        prev = cur
    return prev[-1]

编辑距离的「近亲」是 diff：用 LCS 回溯生成「新增 / 删除 / 保留」序列。这是文本比对、以及 RAG 里「生成引用 vs 原文」忠实度核查的底层算法。

pythondef diff(a, b):
    """用 LCS 回溯生成 diff：定位「新增(-/+)、保留(=)」"""
    n, m = len(a), len(b)
    dp = [[0] * (m + 1) for _ in range(n + 1)]
    for i in range(1, n + 1):
        for j in range(1, m + 1):
            dp[i][j] = dp[i-1][j-1] + 1 if a[i-1] == b[j-1] else max(dp[i-1][j], dp[i][j-1])
    i, j, out = n, m, []
    while i and j:
        if a[i-1] == b[j-1]: out.append(('=', a[i-1])); i -= 1; j -= 1
        elif dp[i-1][j] >= dp[i][j-1]: out.append(('-', a[i-1])); i -= 1
        else: out.append(('+', b[j-1])); j -= 1
    while i: out.append(('-', a[i-1])); i -= 1
    while j: out.append(('+', b[j-1])); j -= 1
    return list(reversed(out))

print(diff('kitten', 'sitting'))
print('LCS 长度 =', sum(1 for op, _ in diff('kitten', 'sitting') if op == '='))
# 预期：k→s、e→i 为替换（用 -/+ 表示），末尾 +g；LCS 长度 = 4（itten）。
# 复杂度：时间 O(n·m)、空间 O(n·m)；大文本用 Hirschberg（分治）压到 O(min(n,m)) 且可回溯。
# 工程对接：ASR 的 WER、ROUGE 的 n-gram 对齐、以及引用忠实度核查都用这套序列对齐。

• 空间优化是工程常态：编辑距离只需保留两行（O(min(n,m)) 空间），LCS 同理。当序列很长（如两篇万词文档）时这是必须的。代价是无法回溯路径——所以「需要对齐结果」时得存完整表或用 Hirschberg 算法（分治 + 线性空间）。
• 加权编辑距离更贴近实际：ASR 里「替换」和「插入」的代价不该相同（漏词比错词更严重）；生物序列比对里 gap 罚分有开孔罚分与延伸罚分的区别（仿射间隙罚分，即 Needleman-Wunsch / Smith-Waterman）。这套思想直接迁移到「生成文本与参考答案的差异度量」上。
• 在 RAG 评测中的用途（第 15 阶段）：把模型生成的引用片段与原文做近似匹配，用编辑距离或 LCS 比例判断引用是否忠实；把生成的答案与参考答案做词级对齐，定位「漏了什么、多说了什么」。
• 与语义相似度的分工：编辑距离衡量字面差异，对同义改写完全失效（「我今天很开心」vs「今天心情不错」的字面距离很大）。所以评测要两者结合：字面（编辑距离 / ROUGE）+ 语义（embedding 相似度 / LLM-as-Judge）。

8.3 动手练习与自测

1. 用同一份 items 分别跑 0/1 背包与完全背包，指出唯一差异（容量维遍历方向），并说明为什么方向不同就代表「物品用一次 / 可重复用」。
2. 实现编辑距离并回溯对齐路径；再把它改成 O(min(n,m)) 空间（不回溯版本），说明什么时候必须用 Hirschberg。
3. 用 LCS 实现一个简易 diff，把「kitten → sitting」的对齐结果打印出来，标出替换与新增。
4. 状态压缩 DP：用位掩码写 TSP 的 DP，说明状态数 2^n·n 与转移 O(n)，并指出 n 到什么规模就不实用了。
5. （开放题）编辑距离与 embedding 相似度各衡量什么？在 RAG 评测里如何组合使用来定位「漏检 / 幻觉」？

9. 面试算法收敛：按模式刷题，而不是按题号

9.1 十四个必须形成肌肉记忆的模式

面试算法的正确学习方式不是「刷 500 题」，而是识别模式。中等难度题几乎全部落在这十四个模式里，每个模式把 3–5 道典型题吃透，就足以覆盖 AI 岗的笔试与初面。

把上表压缩成可默写代码，只有两个模板最值得背到「肌肉记忆」：单调栈与滑动窗口。它们分别对应「下一个更大/更小」「最长/最短连续子串」这两大类高频题，熟练后看到题干 30 秒内就能落笔。

python
# 模板 A：单调栈 —— 下一个更大元素（改为「下一个更小」只需反转比较符）
# 栈中存下标，保持从栈底到栈顶对应值单调递减；遇到更大值就不断弹栈结算
def next_greater(nums):
    res = [-1] * len(nums)      # 默认无更大元素
    stack = []                  # 存下标，其值单调递减
    for i, x in enumerate(nums):
        while stack and nums[stack[-1]] < x:
            j = stack.pop()     # 当前 x 正是 j 的「下一个更大」
            res[j] = x
        stack.append(i)
    return res

print(next_greater([2, 1, 2, 4, 3]))   # [4, 2, 4, -1, -1]

# 模板 B：滑动窗口 —— 最长无重复字符子串
# 右指针扩张、左指针在出现重复时收缩；哈希表记录字符最后一次出现的下标
def longest_unique(s):
    last = {}                   # 字符 -> 最后出现的下标
    left = 0
    best = 0
    for right, ch in enumerate(s):
        if ch in last and last[ch] >= left:
            left = last[ch] + 1  # 左边界跳到重复字符之后
        last[ch] = right
        best = max(best, right - left + 1)
    return best

print(longest_unique("abcabcbb"))   # 3  -> "abc"
print(longest_unique("pwwkew"))     # 3  -> "wke"
        

9.2 八周训练计划与 Hamauls Orion 的复用

• 「按模式刷」的具体做法：不要按题库顺序刷。选定一个模式，连续做 4–6 道该模式的题，做完后合上答案默写模板。这样形成的记忆是「模式 → 解法」的映射，而不是「题号 → 答案」的映射。
• 刷题与 Hamauls Orion 的复用关系（这是本阶段的额外红利）：M3 里你要实现的 HNSW 用到了堆 + 图 + 贪心；BM25 用到了哈希表 + 排序 + Top-K；请求调度器用到了优先队列 + 时间片；Agent 编排用到了拓扑排序 + DP。同一批知识，在面试和工程里各用一次。
• 别忘了复杂度表达：每道题做完，都用一句话说出时间复杂度与空间复杂度，以及「瓶颈在哪一步」。这个习惯会直接迁移到第 2 阶段的性能分析里。

计划里最容易被轻视、却最影响正确率的是二分查找。它不是背一个模板就够，而是要把三种边界写法（找等于、找第一个 ≥、找最后一个 ≤）与「答案二分」这层抽象分清——后者让二分从「在有序数组里查数」升级成「在有单调性的答案空间上求最优」，是面试区分度的重要来源。

python
# 二分三连：区间统一用左闭右开 [lo, hi)，最不容易写错
def lower_bound(a, x):
    # 第一个 >= x 的下标；不存在则返回 len(a)
    lo, hi = 0, len(a)
    while lo < hi:
        mid = (lo + hi) // 2
        if a[mid] < x:
            lo = mid + 1
        else:
            hi = mid
    return lo

def upper_bound(a, x):
    # 第一个 > x 的下标；不存在则返回 len(a)
    lo, hi = 0, len(a)
    while lo < hi:
        mid = (lo + hi) // 2
        if a[mid] <= x:
            lo = mid + 1
        else:
            hi = mid
    return lo

a = [1, 2, 2, 2, 3, 5]
print(lower_bound(a, 2), upper_bound(a, 2))   # 1 4
print(lower_bound(a, 4), upper_bound(a, 4))   # 5 5
print(upper_bound(a, 9))                      # 6  (等于 len(a))

# 答案二分：求满足条件的最小值 —— 条件对 x 单调（x 越大越容易满足）
def min_capacity(weights, days):
    def feasible(cap):                # 能否在 days 天内运完
        need, cur = 1, 0
        for w in weights:
            if cur + w > cap:
                need += 1
                cur = 0
            cur += w
        return need <= days
    lo, hi = max(weights), sum(weights)   # 答案的下界与上界
    while lo < hi:
        mid = (lo + hi) // 2
        if feasible(mid):
            hi = mid
        else:
            lo = mid + 1
    return lo

print(min_capacity([1,2,3,4,5,6,7,8,9,10], 5))   # 15
        

9.3 动手练习与自测（面试收敛）

这一节不引入新知识，只做「检验」：合上全部资料，在限时条件下独立完成下面几题，再用参考答案对判据。能全部过关，说明本阶段的内容已经长在你手里。

1. 手写动态数组（含倍数扩容），用倍增测试验证追加是均摊 O(1)：分别测 n = 1e3、1e4、1e5、1e6 时「平均每次追加耗时」，并解释为什么它几乎不随 n 增长。
2. 手写开放寻址哈希表（线性探测 + 墓碑），把负载因子从 0.5 扫到 0.95，记录平均探测次数，并说明曲线从哪一点开始急剧变陡。
3. 给定 1 万条短文本，实现 MinHash + LSH 近邻去重，报告在 Jaccard 阈值 0.8 下的精确率与召回率；再对比 SimHash 的优缺点。
4. 在 100 万条随机向量（dim = 128）上分别用暴力检索与 HNSW，测出 recall@10 与 QPS，画出召回率-QPS 曲线，并说明 ef_search 取 16 / 64 / 256 时曲线如何变化。
5. 实现 BM25 与向量检索并用 RRF 融合，在自建的中英混合评测集上比较「纯 BM25 / 纯向量 / RRF」三种方案的 Top-10 命中率，给出「哪类查询靠关键词、哪类靠语义」的结论。
6. （开放题）为什么说「模式识别」比「题量」更重要？用你刷过的任意两道题，说明它们共享哪个模式、模板如何复用。

项目里程碑

阶段练习项目

常见误区

• 把复杂度当成唯一标准：小 n 场景下常数因子与 cache 局部性常常才是决定因素。
• 只会用现成的 dict / set，面试要求手写哈希表或 LRU 时写不出来。
• 链表指针操作不画图直接写，导致一次通过率极低。
• 对哈希表的负载因子没有概念，不理解墓碑机制，也不明白为什么容量要取 2 的幂。
• 实现向量检索时直接上 faiss / hnswlib，但从没理解 M / ef_search 为什么这样影响召回率与延迟。
• 缓存只加不加度量：不统计命中率，就不知道缓存到底有没有用、该设多大。
• 检索只用向量、不用关键词，导致专有名词与编号类查询表现极差（漏掉混合检索）。
• 分词环节随手用 split() 处理中文，导致 BM25 的统计量完全失真。
• DP 时初始化随意（把不可达状态写成 0），导致结果被「虚假可行」路径污染。
• 刷题只求数量不求模式，做完 300 题仍然无法在新题上快速识别解法。

面试高频问题速答

为什么动态数组的追加操作是均摊 O(1)？

因为采用倍数扩容（如 2 倍）。从空数组连续追加 n 次，第 1、2、4、8…次时触发拷贝，拷贝总次数为 1+2+4+…+n/2 = n−1，是等比级数，总代价 O(n)，分摊到 n 次操作即均摊 O(1)。关键前提是「按倍数扩容」：若每次只加固定量（如 +10），拷贝总代价是 O(n²/10)，均摊变成 O(n)。另外，均摊是确定性上界（任意 n 次操作序列都成立），不等同于依赖概率的平均复杂度。

哈希表冲突有哪些解决方案？各自取舍是什么？

两大类：① 链地址法（separate chaining）——每个桶挂链表（或长度超阈值时转红黑树，如 Java 8）；负载因子可大于 1，删除简单，但指针跳转导致 cache 不友好、有额外节点开销。② 开放寻址法（open addressing）——冲突时按探测序列（线性/二次/双重哈希）找下一个空槽；数组紧凑、cache 友好、无节点开销，但负载因子必须明显小于 1（线性探测平均探测次数约 (1+1/(1−α)²)/2，α 接近 1 时急剧恶化），且删除需要墓碑标记，否则会截断探测链。工程上小键值、读多写少用开放寻址（Rust HashMap、Python dict），键值差异大或删除频繁用链地址。

HNSW 为什么比 IVF 召回率高？它的参数怎么调？

HNSW 的核心是「分层可导航小世界图」：底层包含全部节点构成近邻图，上层节点稀疏、边长更长，查询时从顶层贪心下降到目标附近，再在底层用优先队列做最佳优先搜索。相比 IVF（先聚类、只搜 nprobe 个簇），HNSW 的图结构能沿近邻关系逐步逼近，不会因为「目标落在未被选中的簇」而完全丢失，因此同延迟下召回率更高，且支持增量插入、无需训练。参数：M（每层邻居数，16–64）控制图连通性与内存；ef_construction（100–500）控制建图质量，决定召回率上界；ef_search（50–500）是唯一可运行时动态调优的旋钮，越大召回率越高、延迟越大。调参方法是固定数据集扫 ef_search，得到召回率-QPS 曲线后在拐点取工作点。

BM25 的公式里每一项在做什么？

BM25 = Σ_t IDF(t) · [ tf(t,d)·(k1+1) / (tf(t,d) + k1·(1−b+b·|d|/avgdl)) ]。三部分：① IDF 衡量词的区分度，稀有词权重高（常见形式 log(1+(N−df+0.5)/(df+0.5))，加 0.5 平滑并避免负值）；② TF 部分带饱和效应——tf 分母里的 k1 让高频词增益递减，k1 越大饱和越慢（通常 1.2–2.0），避免「出现 10 次比出现 3 次重要 3 倍」；③ 长度归一化——长文档天然容易命中更多词，所以按 |d|/avgdl 打折，b 控制强度（0 不归一化，1 完全归一化，通常 0.75）。

LRU 怎么实现？为什么它能做到 O(1)？

用「哈希表 + 双向链表」：哈希表存 key → 链表节点，实现 O(1) 定位；双向链表维护访问顺序，头部是最近使用、尾部是最久未用。get 时把节点从原位置摘除并移到头部；put 时若已存在则更新并前移，否则插入头部，若超容量则删除尾部节点并同步从哈希表移除。两个操作涉及的都只是常数次指针改动与哈希操作，所以是 O(1)。工程上还要加：哑头尾节点简化边界、并发保护（加锁 / 分片 / 近似 LRU 如 Redis 的随机采样）、以及 TTL 过期清理。

布隆过滤器为什么会误判「存在」，却不会误判「不存在」？

布隆过滤器用一个 m 位的位数组和 k 个哈希函数。插入元素时，把它的 k 个哈希位置全部置 1。查询时若 k 位中有任何一位是 0，则该元素一定没被插入过（因为插入过的元素对应的位必然都是 1）——所以没有假阴性。反之，若 k 位全为 1，可能是被其他元素各自置上的（多个元素的位置叠加），因此存在假阳性。假阳性率约为 (1−e^(−kn/m))^k，最优哈希个数 k = (m/n)·ln2，对应最优 m = −n·ln(p)/(ln2)²。

拓扑排序怎么检测环？两种实现有什么区别？

两种实现都能检测环：① Kahn 算法（BFS）——统计每个节点入度，把入度为 0 的入队，出队时把邻接节点入度减 1，新的入度 0 节点入队。若最终输出的节点数小于总节点数，说明剩下的节点互相构成环。它的额外好处是天然给出「分层」，可以做按层并行调度。② DFS 三色标记——白色未访问、灰色在栈中、黑色已完成；若在 DFS 过程中遇到一个灰色节点，说明存在回边即环。DFS 版能顺带给出逆后序。工程上做任务编排通常用 Kahn，因为需要分层并行的信息。

给你一个不断到来的数据流，怎么求 Top-K？

维护一个大小为 K 的最小堆。新元素到达时：堆未满则直接入堆；堆已满则与堆顶（当前第 K 大）比较，若更大则替换堆顶并下沉。时间复杂度 O(n log K)，空间 O(K)，非常适合内存放不下全量数据的流式场景。若数据规模可控且允许 O(n) 额外空间，可以用快速选择（quickselect）做到平均 O(n) 一次求解；若需要「随时查询且支持动态更新」，则用平衡树或跳表，能做到 O(log n) 插入与 O(K) 取前 K。另外注意：求 Top-K 频繁时堆顶的替换模式很关键，用 heapreplace 比 pop+push 少一次堆调整。

学习资源