AI 系统架构
你在笔记本上跑通了一个大模型 Demo，效果不错，但当你想把它变成一个真正的产品时，问题就来了：
模型参数量太大，单张 GPU 装不下怎么办？
-
训练一次要几周，中间出了故障怎么断点续训？
-
每天有几百万用户请求，怎么保证延迟在 1 秒以内？
-
怎么收集用户反馈，让模型持续进化？
这些问题，就是 AI 系统架构要解决的。
Demo 关注的是能不能跑通，生产系统关注的是能不能稳定、高效、低成本地跑。
生产级 AI 系统的特点：7x24 小时可用、支持百万级并发、可观测、可扩展、灾备可靠、成本可控。
AI 系统的特殊挑战
和传统 Web 服务相比，AI 系统有三个独特的挑战。
挑战一：不确定性输出
传统系统的输出是确定的——你输入 1+1，它永远返回 2。
AI 系统的输出是概率性的——同样的提示词，两次生成可能结果不一样。
这带来几个问题：怎么保证输出质量？怎么评估效果？怎么处理幻觉？
解决方案通常是：输出层加采样策略、结果后处理过滤、人工反馈闭环、多模型投票。
挑战二：延迟与成本的权衡
AI 推理需要大量计算，这意味着延迟和成本之间有个天然的矛盾。
要快？用更多 GPU，成本飙升。
要省钱？排队处理，用户体验差。
生产系统的核心是在 SLA（服务水平协议）和成本之间找平衡点。
| 优化方向 | 常用技术 | 效果 |
|---|---|---|
| 模型压缩 | 量化、剪枝、蒸馏 | 速度提升 2-4 倍，精度微降 |
| 推理优化 | vLLM、TensorRT、FlashAttention | 吞吐量提升 3-10 倍 |
| 架构设计 | 批量处理、多级缓存 | 单位请求成本降低 50%-80% |
挑战三：数据飞轮的构建
AI 系统不是"上线就完事了"，它需要持续迭代。
用户用得越多，反馈数据越多，模型就能训练得越好，用户就更愿意用——这就是数据飞轮。
但飞轮转起来不容易：怎么收集有效反馈？怎么标注数据？怎么持续训练？怎么评估新版本？
这些问题没有标准答案，但每个成功的 AI 产品都有自己的飞轮设计。
大规模训练基础设施
训练千亿甚至万亿参数的大模型，需要超级计算基础设施。
GPU 集群架构
现代 AI 训练集群通常由成百上千张 GPU 组成。
以 H100 GPU 为例，一张 H100 有 80GB 显存，算力是 FP8 精度下 1979 TFLOPS。
但单张 GPU 远远不够——GPT-3 训练用了约 355 张 V100，训了 3 个月。
典型的集群拓扑是：
| 层级 | 设备 | 连接方式 | 带宽 |
|---|---|---|---|
| 单机内 | GPU-GPU | NVLink | 900GB/s |
| 同机架 | 服务器-服务器 | InfiniBand | 400Gb/s |
| 跨机架 | 交换机-交换机 | InfiniBand Fabric | 400Gb/s |
网络是分布式训练的瓶颈。如果通信带宽不够，GPU 利用率可能从 90% 降到 30%，大部分时间在等数据。
InfiniBand 高速互联
普通以太网太慢，分布式训练用的是 InfiniBand。
InfiniBand 的特点是：极低延迟（微秒级）、极高带宽、支持 RDMA（远程直接内存访问）。
RDMA 让一张 GPU 可以直接读写另一台服务器的 GPU 显存，不需要经过操作系统内核，速度快很多。
存储系统设计
训练数据通常是 TB 甚至 PB 级的，存储系统也很讲究。
典型的分层存储设计：
- 热数据：SSD 或 NVMe，存放当前正在训练的批次
- 温数据：分布式存储（如 Ceph、Lustre），存放完整训练集
- 冷数据：对象存储（如 S3），存放历史数据和备份
容错与检查点
训一次几周，期间某张 GPU 挂了怎么办？从头再来太浪费。
解决方案是检查点（Checkpointing）——定期把模型状态存到磁盘，出错了从最近的检查点恢复。
但检查点也有成本：存一次可能要几分钟，占几十 GB 空间。
通常的策略是：每几百步存一次，保留最近几个检查点，旧的自动清理。
分布式训练策略
单张 GPU 装不下大模型，需要把训练任务拆分到多 GPU。
主要有三种并行策略：数据并行、张量并行、流水线并行，三者结合就是 3D 并行。
数据并行（Data Parallelism）
最简单也最常用的策略：每块 GPU 都放完整模型，但处理不同的数据。
比如有 8 张 GPU，批次大小是 1024，那每张 GPU 处理 128 条数据。
前向传播各自算各自的，反向传播后把梯度汇总取平均，然后更新模型。
数据并行的问题是：显存仍然是瓶颈——模型太大单 GPU 装不下的话，数据并行没用。
ZeRO 显存优化
ZeRO（Zero Redundancy Optimizer）是数据并行的增强版，能进一步节省显存。
普通数据并行中，每张 GPU 都存完整的模型参数、梯度、优化器状态——这很冗余。
ZeRO 的思路是：把这些状态拆分到不同 GPU，要用的时候再通信。
| ZeRO 阶段 | 拆分内容 | 显存节省 |
|---|---|---|
| ZeRO-1 | 优化器状态 | 约 4 倍 |
| ZeRO-2 | 优化器状态 + 梯度 | 约 8 倍 |
| ZeRO-3 | 优化器状态 + 梯度 + 参数 | 线性于 GPU 数量 |
用 DeepSpeed 配置 ZeRO 很简单：
"train_batch_size": 1024,
"train_micro_batch_size_per_gpu": 16,
"optimizer": {
"type": "Adam",
"params": {
"lr": 0.0001,
"betas": [0.9, 0.95],
"eps": 1e-8,
"weight_decay": 0.01
}
},
"zero_optimization": {
"stage": 3,
"allgather_partitions": true,
"allgather_bucket_size": 2e8,
"overlap_comm": true,
"reduce_scatter": true,
"reduce_bucket_size": 2e8,
"contiguous_gradients": true,
"stage3_prefetch_bucket_size": 1e8,
"stage3_param_persistence_threshold": 1e5,
"stage3_max_live_parameters": 1e9,
"stage3_max_reuse_distance": 1e9
},
"gradient_clipping": 1.0,
"fp16": {
"enabled": true,
"loss_scale": 0,
"loss_scale_window": 1000,
"initial_scale_power": 16,
"hysteresis": 2,
"min_loss_scale": 1
},
"checkpoint": {
"tag": "runoob-checkpoint",
"load_universal": true
}
}
这个配置用了 ZeRO-3，可以把模型状态分散到所有 GPU，显存占用随 GPU 数量线性降低。
张量并行（Tensor Parallelism）
如果 ZeRO 还不够，就得用张量并行——把一层的计算拆分到多 GPU。
Transformer 中的矩阵乘法可以按行或列拆分：
- 把 A 矩阵按行拆成 A₁ 和 A₂，分别在 GPU 0 和 GPU 1 计算 A₁×B 和 A₂×B
- 最后把结果拼起来
这样每层计算都需要通信，但显存占用也减半了。
Megatron-LM 是英伟达做的张量并行库，和 PyTorch 兼容很好。
流水线并行（Pipeline Parallelism）
张量并行是"层内拆分"，流水线并行是"层间拆分"。
比如 32 层的模型，GPU 0 放前 8 层，GPU 1 放中间 8 层，GPU 2 放下 8 层，GPU 3 放最后 8 层。
数据从 GPU 0 流到 GPU 3，就像工厂流水线一样。
但流水线有个问题：气泡（Bubble）——GPU 0 开始算的时候，GPU 1-3 没事干；等数据到 GPU 1 了，GPU 0 又没事干了。
解决方案是把数据分成"微批次"（Micro-batch），像流水线一样填进去，减少气泡时间。
3D 并行（DP+TP+PP）
三种策略可以组合使用：
- 流水线并行：跨节点拆分模型层
- 张量并行：节点内拆分层内计算
- 数据并行：在更大范围内复制整个流水线
比如有 64 张 GPU，可以这样规划：
- 8 个流水线阶段（PP=8）
- 每个阶段内用 2 张 GPU 做张量并行（TP=2）
- 然后复制 4 份做数据并行（DP=4）
- 总共：8 × 2 × 4 = 64 张 GPU
这就是 3D 并行——现代大模型训练的标准配置。
数据工程
好模型的前提是好数据——数据工程占 AI 系统工作量的 60% 以上。
数据收集与清洗流水线
训练数据通常来自多个来源：网页、书籍、代码、对话等。
典型的处理流程：
- 去重：删除重复或高度相似的文档
- 质量过滤：去掉低质量、有毒、有偏见的内容
- 格式统一：不同来源转成统一格式
- token 化：文本转成模型输入序列
数据去重：MinHash LSH
直接计算两两文档相似度太慢，常用方法是 MinHash + LSH（局部敏感哈希）。
思路是：把每个文档变成一个短的"指纹"，相似文档的指纹大概率相同或相近，然后按指纹分组。
实例
import re
from typing import List, Set, Dict, Tuple
def generate_shingles(text: str, k: int = 5) -> Set[str]:
"""生成 k-shingles：连续 k 个词组成的序列
比如 "我爱 runoob 教程"，k=2 → {"我爱", "爱 runoob", "runoob 教程"}
"""
# 简单分词（实际场景可用专业分词工具）
words = re.findall(r'\w+', text.lower())
shingles = set()
for i in range(len(words) - k + 1):
shingle = ' '.join(words[i:i+k])
shingles.add(shingle)
return shingles
def minhash_signature(shingles: Set[str], num_hashes: int = 100) -> List[int]:
"""生成 MinHash 签名
用多个哈希函数，每个取最小值
"""
signature = []
for i in range(num_hashes):
# 用 i 作为种子生成不同的哈希函数
min_hash = None
for shingle in shingles:
# 组合 shingle 和 i，生成哈希值
h = hashlib.sha256(f"{shingle}-{i}".encode()).hexdigest()
h_int = int(h, 16)
if min_hash is None or h_int < min_hash:
min_hash = h_int
signature.append(min_hash)
return signature
def lsh_banding(signature: List[int], bands: int = 20) -> List[str]:
"""用分桶法（Banding）生成 LSH 键
把签名分成多个 band，每个 band 单独哈希
"""
keys = []
rows_per_band = len(signature) // bands
for i in range(bands):
start = i * rows_per_band
end = start + rows_per_band
band = tuple(signature[start:end])
# 对这个 band 哈希生成一个键
band_hash = hashlib.sha256(str(band).encode()).hexdigest()[:16]
keys.append(f"band-{i}-{band_hash}")
return keys
def deduplicate_documents(documents: List[str],
threshold: float = 0.7) -> List[str]:
"""用 MinHash + LSH 去重文档
返回去重后的文档列表
"""
# 存储：LSH 键 → 文档索引列表
buckets: Dict[str, List[int]] = {}
# 存储：文档索引 → 签名
signatures: Dict[int, List[int]] = {}
# 标记：哪些文档是重复的
duplicates: Set[int] = set()
for idx, doc in enumerate(documents):
shingles = generate_shingles(doc)
sig = minhash_signature(shingles)
signatures[idx] = sig
keys = lsh_banding(sig)
# 检查是否已经有相似文档
is_duplicate = False
for key in keys:
if key in buckets:
# 这个桶里已有文档，逐一比较签名相似度
for other_idx in buckets[key]:
other_sig = signatures[other_idx]
# 计算签名相似度（Jaccard 近似）
matches = sum(1 for a, b in zip(sig, other_sig) if a == b)
similarity = matches / len(sig)
if similarity >= threshold:
# 超过阈值，认为是重复
is_duplicate = True
duplicates.add(idx)
break
if is_duplicate:
break
if not is_duplicate:
# 不是重复，把自己加到各个桶里
for key in keys:
if key not in buckets:
buckets[key] = []
buckets[key].append(idx)
# 返回非重复文档
return [doc for idx, doc in enumerate(documents) if idx not in duplicates]
# ============================================
# 测试 runoob 数据去重
# ============================================
if __name__ == "__main__":
documents = [
"欢迎来到 runoob 教程，这是一个学习编程的好地方。",
"欢迎来到 runoob 教程，这是学习编程的好地方。", # 高度相似
"Python 是一门简洁优雅的语言，适合初学者。",
"Python 是一门简洁优雅的编程语言，非常适合初学者。", # 高度相似
"机器学习让计算机从数据中学习规律。",
"这是完全不同的另一篇文章。",
]
print(f"去重前：{len(documents)} 篇文档")
deduplicated = deduplicate_documents(documents, threshold=0.6)
print(f"去重后：{len(deduplicated)} 篇文档\n")
print("保留的文档：")
for i, doc in enumerate(deduplicated):
print(f" [{i}] {doc}")
# 输出：
# 去重前：6 篇文档
# 去重后：4 篇文档
#
# 保留的文档：
# [0] 欢迎来到 runoob 教程，这是一个学习编程的好地方。
# [1] Python 是一门简洁优雅的语言，适合初学者。
# [2] 机器学习让计算机从数据中学习规律。
# [3] 这是完全不同的另一篇文章。
实际生产中，还会用更高效的实现（如 datasketch 库），但核心思路是一样的。
数据格式：WebDataset
小数据集随便存，但 TB 级的数据集要用专门的格式。
WebDataset 是常用的一种：它把文件打包成 tar 包，每个 tar 包含几千个样本，支持随机读取和顺序读取。
好处是：
- 减少文件系统压力（几百万个小文件很慢）
- 支持流式读取，不用把整个数据集读到内存
- 可以分布式加载，每个 worker 读不同的 tar
数据飞轮设计
数据飞轮是 AI 产品的护城河——用户越多，数据越多，模型越好，用户更多。
用户反馈数据收集
反馈有显式和隐式两种：
- 显式反馈：用户点赞、点踩、编辑、重新生成
- 隐式反馈：用户停留时间、复制、分享、会话长度
显式反馈质量高，但数量少；隐式反馈数量多，但噪音大。
好的反馈系统是两者结合——用显式反馈训练奖励模型，用隐式反馈做 A/B 测试。
自动数据标注
人工标注太贵太慢，现在的趋势是"用模型标注模型"。
常用策略：
- 强模型标注弱模型：用 GPT-4 标注的数据训小模型
- 自举（Bootstrap）：用现有模型生成候选，人工筛选
- 合成数据：用模型生成多样化的训练数据
持续训练策略
模型不是"训一次就完事了"，世界在变，模型也要跟着变。
典型的持续训练流程：
- 每天收集新的用户交互数据
- 每周做一次小更新（SFT，有监督微调）
- 每月做一次大更新（继续预训练 + SFT + RLHF）
- 每次更新都做 A/B 测试，确认效果再全量
持续训练要注意"灾难性遗忘"——新数据训多了，模型可能忘记以前的能力。解决方案是保留一个"回放 buffer"，每次训练混合新旧数据。
企业 AI 平台架构
一个完整的企业 AI 平台通常包含以下组件。
模型注册与版本管理
模型越多，管理越麻烦——需要一个中心化的模型注册表。
注册表要记录：
- 模型文件（权重、配置、tokenizer）
- 版本号和变更记录
- 训练数据来源和超参数
- 评估指标
- 部署状态
MLflow、Weights & Biases、Hugging Face Hub 都是常用工具。
Feature Store
很多 AI 应用需要特征工程——用户画像、历史行为、上下文信息等。
Feature Store 是特征管理系统，解决几个问题：
- 训练/推理 skew：训练和推理用的特征不一致
- 特征复用：不同模型可以共享特征
- 在线/离线一致：离线计算的特征，在线低延迟读取
典型架构是：离线用 Spark 或 Flink 批量计算，在线用 Redis 或 Cassandra 低延迟读取。
在线/离线推理服务
推理分两种场景，架构不一样：
| 场景 | 延迟要求 | 架构 | 例子 |
|---|---|---|---|
| 在线推理 | 毫秒级 | 实时 API + 批处理 | 聊天机器人、搜索 |
| 离线推理 | 小时/天级 | 批量任务队列 | 内容审核、报表生成 |
在线推理服务需要考虑：
- 批处理（Batching）：把多个请求凑到一起算，提升吞吐量
- 动态批：用 vLLM 或 Text Generation Inference 做动态批处理
- 缓存：K/V Cache 加速自回归生成，热门 prompt 直接缓存结果
- 负载均衡：多模型实例，智能路由
全链路监控
AI 系统出问题很难查——需要可观测性。
监控的几个维度：
- 系统指标：GPU 利用率、显存、延迟、吞吐量、错误率
- 模型指标：输出长度、重复率、停止词分布
- 业务指标：用户满意度、留存率、任务完成率
经典组合：Prometheus + Grafana 看指标，ELK 看日志，Jaeger 看链路追踪。
评估基准体系
怎么知道模型变好了还是变差了？需要一套评估体系。
通用能力基准
学术界常用的基准测试：
| 基准 | 测试内容 | 典型任务 |
|---|---|---|
| MMLU | 多任务语言理解 | 57 个学科的选择题 |
| HumanEval | 代码生成 | 164 个编程题 |
| MBPP | 代码生成 | 974 个 Python 题 |
| TruthfulQA | 事实性 | 817 个问题，测幻觉 |
| GSM8K | 数学推理 | 8000 道小学数学题 |
中文基准
英文基准好不等于中文好，需要中文专用基准：
- C-Eval：中文多任务语言理解，13948 道选择题
- CMMLU：中文多模态理解（现在主要是文本）
- AGIEval：中国高考题、公务员考试题
自定义业务评估集
公开基准是基础，但更重要的是自己的业务评估集。
怎么建：
- 收集真实用户请求（几百到几千条）
- 人工标注"好/中/差"或打分
- 分成测试集和验证集（测试集固定，不要动）
- 每次模型更新都跑一遍，看指标变化
业务评估集是你的"金标准"——它比任何公开基准都重要。
多租户 AI 服务
如果做的是 ToB 产品，多租户是绕不开的话题。
租户隔离
隔离有几个层级：
| 隔离层级 | 资源 | 优点 | 缺点 |
|---|---|---|---|
| 物理隔离 | 独立 GPU 机器 | 完全隔离，安全 | 成本高 |
| K8s 隔离 | 独立 Pod/Namespace | 平衡成本与安全 | 需要调度 |
| 逻辑隔离 | 共享资源，权限控制 | 成本最低 | 有泄露风险 |
通常的策略是：大客户物理隔离，中小客户逻辑隔离。
模型共享与微调管理
多租户常有的需求：用我的数据训个专有模型。
架构上需要支持：
- 基础模型共享存储
- 每个租户的 LoRA 适配器单独存放
- 推理时动态加载对应租户的 LoRA
这样可以节省大量显存——不用为每个租户都加载完整模型。
计费与配额
AI 服务成本很高，计费要精细化：
- 按 token 计费（输入和输出分开算）
- 按请求次数计费
- 按 GPU 时间计费（微调场景）
- 配额限制（每分钟多少请求、每天多少 token）
灾备与高可用
企业级系统不能挂——需要高可用设计。
多可用区部署
最基本的要求：跨可用区（AZ）部署。
比如：
- 主集群在 AZ A
- 热备集群在 AZ B
- 数据异步复制
- 负载均衡自动切流量
目标是：单 AZ 挂了，服务不中断，数据不丢失。
模型快照与回滚
新版本模型可能出问题——需要能快速回滚。
策略：
- 每次发布前存模型快照
- 保留最近 N 个版本
- 监控异常指标，触发自动回滚
- 灰度发布：先切 1% 流量，没问题再全量
降级与限流
流量突增怎么办？不能直接挂，要有预案：
- 限流：超过配额返回 429
- 降级：用更小的模型，或返回缓存结果
- 排队：返回 job ID，稍后查结果
核心原则：优雅降级，而不是彻底不可用。