FAISS 学习笔记

1. FAISS 是什么

FAISS，全称 Facebook AI Similarity Search，是一个用于高效向量相似度搜索和聚类的库。

它最常见的用途是：

• 语义检索
• 图片检索
• 推荐系统召回
• RAG 文档检索
• 大规模 embedding 向量查找

可以先这样理解：

Embedding 模型负责把文本变成向量，FAISS 负责在一堆向量里快速找到最相似的几个。

在 RAG 里，FAISS 通常位于这条链路中：

1
2
3
4
5
6
7

文档
  -> 切分 chunk
  -> embedding 模型生成向量
  -> FAISS 建索引
  -> 用户问题生成 query embedding
  -> FAISS Top-K 检索
  -> 把检索结果交给 LLM 生成回答

FAISS 本身不是完整数据库，它主要管理向量索引。文档标题、chunk 内容、文件路径、页码、用户 ID、权限等元数据，通常还要放在 MySQL、PostgreSQL、MongoDB 或其他业务数据库里。

2. 为什么需要 FAISS

如果有 10 个向量，直接遍历计算相似度就够了。

但如果有：

• 10 万个 chunk
• 100 万个商品向量
• 1000 万张图片特征

每次查询都全量遍历会很慢。

FAISS 解决的是：

• 怎么快速找到最近邻
• 怎么在速度、召回率、内存之间做取舍
• 怎么支持百万级、千万级甚至更大规模的向量检索

在小规模 RAG 项目里，IndexFlat 就能跑通完整链路；数据量上来后，再考虑 IVF、HNSW、PQ 等索引结构。

3. FAISS 的核心概念

3.1 向量

FAISS 处理的是固定维度的向量。

例如一个 embedding 模型输出 768 维向量：

1

[0.12, -0.03, 0.88, ..., 0.41]

那么 FAISS 里的所有向量都必须是 768 维。

如果数据库向量是 768 维，而查询向量是 1024 维，就不能放在同一个 index 里检索。

3.2 d

d 表示向量维度。

1
2

d = 768
index = faiss.IndexFlatL2(d)

这个 d 必须和 embedding 模型输出维度一致。

3.3 xb 和 xq

FAISS 官方示例里经常用两个名字：

• xb：database vectors，要被索引的向量集合
• xq：query vectors，查询向量

它们通常都是 NumPy 数组：

1
2

xb.shape == (num_vectors, dimension)
xq.shape == (num_queries, dimension)

例如：

1
2

xb: 10000 个文档 chunk 向量，每个 768 维
xq: 1 个用户问题向量，每个 768 维

3.4 Index

FAISS 的核心对象是 Index。

Index 可以理解成：

存放向量并支持相似度搜索的数据结构。

常用操作：

• add()：添加向量
• search()：查询最近的 Top-K 向量
• train()：训练索引，部分索引需要
• ntotal：当前索引中有多少条向量
• is_trained：索引是否已经训练完成

3.5 D 和 I

FAISS 搜索结果通常返回两个数组：

1

D, I = index.search(query_vectors, k)

含义：

• D：距离或相似度分数
• I：检索到的向量 ID 或内部下标

例如：

1
2

I = [[12, 88, 102]]
D = [[0.13, 0.20, 0.35]]

表示当前查询最相关的 3 个结果分别是第 12、88、102 个向量。

注意：

• L2 距离下，D 越小越相似。
• Inner Product 下，D 越大越相似。

4. 安装

CPU 版本：

1

pip install faiss-cpu

GPU 版本通常需要匹配 CUDA 环境，安装方式会和系统、CUDA、Python 版本有关。初学和普通 RAG 后端项目，先用 CPU 版本就够。

5. 最小可运行示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import faiss
import numpy as np

d = 4
xb = np.array(
    [
        [1.0, 0.0, 0.0, 0.0],
        [0.9, 0.1, 0.0, 0.0],
        [0.0, 1.0, 0.0, 0.0],
        [0.0, 0.0, 1.0, 0.0],
    ],
    dtype="float32",
)

xq = np.array([[1.0, 0.0, 0.0, 0.0]], dtype="float32")

index = faiss.IndexFlatL2(d)
index.add(xb)

k = 2
D, I = index.search(xq, k)

print(D)
print(I)

这段代码做了几件事：

1. 准备一批数据库向量 xb
2. 准备查询向量 xq
3. 创建 L2 距离索引 IndexFlatL2
4. 把数据库向量加入索引
5. 查询最相近的 2 个向量

6. 数据格式要求

FAISS 对输入数据有几个常见要求：

• NumPy 数组通常要是 float32
• shape 要是二维：(n, d)
• 数据最好是连续内存
• 所有向量维度必须一致

推荐这样处理：

1
2

vectors = np.asarray(vectors, dtype="float32")
vectors = np.ascontiguousarray(vectors)

如果只有一个查询向量，不要传一维数组：

1
2
3
4
5

# 不推荐
query.shape == (768,)

# 推荐
query.shape == (1, 768)

可以这样修正：

1

query = np.asarray(query, dtype="float32").reshape(1, -1)

7. 距离和相似度

FAISS 里最常见的两种检索方式：

• L2 距离
• Inner Product 内积

7.1 L2 距离

L2 距离就是欧氏距离。

FAISS 的 METRIC_L2 返回的是平方 L2 距离，不开根号。排序结果不受影响，因为平方距离和真实欧氏距离是单调一致的。

使用方式：

1

index = faiss.IndexFlatL2(d)

特点：

• D 越小越相似
• 适合直接按距离找最近邻

7.2 Inner Product

Inner Product 是内积，常用于最大内积搜索。

使用方式：

1

index = faiss.IndexFlatIP(d)

特点：

• D 越大越相似
• 常用于已经归一化的 embedding 检索

7.3 余弦相似度怎么做

很多文本 embedding 检索习惯使用余弦相似度。

FAISS 里常见做法是：

1. 使用 IndexFlatIP
2. 添加向量前先做 L2 归一化
3. 查询向量也做 L2 归一化

代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import faiss
import numpy as np

d = 768
xb = np.asarray(xb, dtype="float32")
xq = np.asarray(xq, dtype="float32")

faiss.normalize_L2(xb)
faiss.normalize_L2(xq)

index = faiss.IndexFlatIP(d)
index.add(xb)

D, I = index.search(xq, k=5)

注意：

如果只归一化文档向量，不归一化查询向量，或者反过来，只归一化查询向量不归一化文档向量，结果都会不一致。

8. 常见索引类型

FAISS 有很多索引。初学阶段先掌握这几类就够。

8.1 IndexFlatL2

精确 L2 检索。

1

index = faiss.IndexFlatL2(d)

特点：

• 不需要训练
• 暴力搜索，结果精确
• 数据量小到中等时很好用
• 内存占用约等于 向量数量 * 维度 * 4 字节

适合：

• 本地 demo
• 小规模 RAG
• 先验证业务链路
• 作为评测近似索引 recall 的 ground truth

8.2 IndexFlatIP

精确内积检索。

1

index = faiss.IndexFlatIP(d)

特点：

• 不需要训练
• 暴力搜索，结果精确
• 搭配向量归一化后，常用于余弦相似度检索

适合：

• 文本 embedding 语义检索
• RAG 原型
• 中小规模知识库

8.3 IndexHNSWFlat

HNSW 是一种基于图的近似最近邻搜索方法。

1

index = faiss.IndexHNSWFlat(d, 32)

第二个参数通常叫 M，可以理解成图里每个节点连接的邻居数量。

特点：

• 不需要像 IVF 那样单独训练
• 搜索速度快
• 召回率通常不错
• 内存比 Flat 更高，因为要额外存图结构

常见调参：

1
2

index.hnsw.efSearch = 64
index.hnsw.efConstruction = 200

简单理解：

• M 越大，图更密，召回更好，内存更高。
• efSearch 越大，搜索更认真，召回更好，速度更慢。
• efConstruction 越大，建图更慢，但图质量可能更好。

适合：

• 希望查询快
• 数据规模比 Flat 更大
• 可以接受近似结果

8.4 IndexIVFFlat

IVF，全称 Inverted File，可以理解成“先粗分类，再局部搜索”。

基本思路：

1. 先把向量空间分成很多个簇
2. 添加向量时，把每个向量放进对应簇里
3. 查询时，只搜索最相关的几个簇

代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

import faiss

d = 768
nlist = 100

quantizer = faiss.IndexFlatIP(d)
index = faiss.IndexIVFFlat(
    quantizer,
    d,
    nlist,
    faiss.METRIC_INNER_PRODUCT,
)

faiss.normalize_L2(train_vectors)
faiss.normalize_L2(database_vectors)
faiss.normalize_L2(query_vectors)

index.train(train_vectors)
index.add(database_vectors)

index.nprobe = 10
D, I = index.search(query_vectors, k=5)

关键参数：

• nlist：聚类中心数量，也就是分多少个桶
• nprobe：查询时搜索多少个桶

简单理解：

• nlist 越大，桶越多，每个桶里数据越少，但训练和管理更复杂。
• nprobe 越大，搜的桶越多，召回更高，速度更慢。

注意：

IVF 类索引通常需要先 train()，再 add()。

适合：

• 数据量较大
• 希望比 Flat 更快
• 能接受近似检索

8.5 IndexIVFPQ

PQ，全称 Product Quantization，核心目的是压缩向量。

IndexIVFPQ 可以理解成：

IVF 负责缩小搜索范围，PQ 负责压缩每个向量，降低内存占用。

特点：

• 内存占用明显降低
• 查询速度可以更快
• 结果是近似的
• 实现和调参比 Flat、HNSW、IVFFlat 更复杂

适合：

• 向量数量很大
• 内存压力明显
• 可以接受召回率下降

对于普通 RAG 项目，不建议一开始就用 IVFPQ。先用 IndexFlatIP 或 IndexHNSWFlat 跑通业务，再根据数据量和性能瓶颈升级。

9. index_factory

FAISS 提供 index_factory 用字符串快速创建索引。

例如：

1

index = faiss.index_factory(d, "Flat", faiss.METRIC_INNER_PRODUCT)

HNSW：

1

index = faiss.index_factory(d, "HNSW32,Flat", faiss.METRIC_INNER_PRODUCT)

IVF：

1

index = faiss.index_factory(d, "IVF100,Flat", faiss.METRIC_INNER_PRODUCT)

IVFPQ：

1

index = faiss.index_factory(d, "IVF100,PQ16", faiss.METRIC_INNER_PRODUCT)

index_factory 的好处是：

• 创建复合索引更方便
• 配置可以字符串化
• 适合写到配置文件里

缺点是：

• 初学时不如显式构造容易理解
• 写错字符串时排查成本更高

建议：

• 学习阶段先显式构造
• 熟悉索引类型后再用 index_factory

10. 向量 ID 和业务 ID

10.1 默认 ID

如果直接这样添加：

1

index.add(vectors)

FAISS 会使用内部连续 ID：

1

0, 1, 2, 3, ...

这适合简单 demo，但真实项目里通常不够。

因为 RAG 里你真正关心的是：

• 这个向量属于哪个文档
• 是第几个 chunk
• 对应 MySQL 里的哪条记录
• 原文内容是什么
• 页码、标题、来源是什么

10.2 使用 IndexIDMap

可以给向量指定自己的 ID：

1
2
3
4
5
6
7
8
9
10
11
12
13

import faiss
import numpy as np

base_index = faiss.IndexFlatIP(d)
index = faiss.IndexIDMap(base_index)

vectors = np.asarray(vectors, dtype="float32")
ids = np.asarray([101, 102, 103], dtype="int64")

faiss.normalize_L2(vectors)
index.add_with_ids(vectors, ids)

D, I = index.search(query, k=3)

这里返回的 I 就是你传入的业务 ID。

10.3 RAG 项目里的 ID 设计

推荐做法：

1
2
3
4
5
6
7
8
9
10
11
12

MySQL chunk 表：
- id: chunk_id
- document_id
- chunk_index
- content
- page
- metadata
- embedding_model
- created_at

FAISS：
- 向量 ID 使用 chunk_id

搜索后：

1
2
3
4

FAISS 返回 chunk_id
  -> 用 chunk_id 查 MySQL
  -> 取 content、document_id、page、source
  -> 组装 prompt 和 citation

不要把大量元数据硬塞进 FAISS。FAISS 适合管向量，不适合替代业务数据库。

11. 索引保存和加载

FAISS index 可以保存到磁盘。

1

faiss.write_index(index, "docs.index")

加载：

1

index = faiss.read_index("docs.index")

RAG 项目里常见做法：

1
2
3
4
5

data/
└── indexes/
    ├── kb_1.index
    ├── kb_2.index
    └── kb_3.index

同时在数据库里记录：

1
2
3
4
5
6
7
8
9

knowledge_base_id
index_path
embedding_model
embedding_dim
metric_type
index_type
chunk_count
version
updated_at

这样做的好处：

• 服务重启后可以直接加载 index
• 可以知道 index 对应哪个 embedding 模型
• 模型切换后能判断旧索引是否需要重建

12. RAG 项目中的 FAISS 封装

在项目里不要把 FAISS 操作散落在 router 或 service 里，建议封装成 VectorStore。

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

from pathlib import Path

import faiss
import numpy as np

class FaissVectorStore:
    def __init__(self, dim: int, index_path: str | None = None):
        self.dim = dim
        self.index_path = Path(index_path) if index_path else None

        if self.index_path and self.index_path.exists():
            self.index = faiss.read_index(str(self.index_path))
            if self.index.d != dim:
                raise ValueError("index dimension mismatch")
        else:
            base_index = faiss.IndexFlatIP(dim)
            self.index = faiss.IndexIDMap(base_index)

    def add(self, vectors: np.ndarray, ids: np.ndarray) -> None:
        vectors = np.asarray(vectors, dtype="float32")
        vectors = np.ascontiguousarray(vectors)
        ids = np.asarray(ids, dtype="int64")

        if vectors.ndim != 2 or vectors.shape[1] != self.dim:
            raise ValueError("invalid vector shape")

        faiss.normalize_L2(vectors)
        self.index.add_with_ids(vectors, ids)

    def search(self, query: np.ndarray, top_k: int = 5):
        query = np.asarray(query, dtype="float32").reshape(1, -1)
        query = np.ascontiguousarray(query)

        if query.shape[1] != self.dim:
            raise ValueError("invalid query dimension")

        faiss.normalize_L2(query)
        scores, ids = self.index.search(query, top_k)
        return scores[0], ids[0]

    def save(self) -> None:
        if not self.index_path:
            raise ValueError("index_path is required")
        self.index_path.parent.mkdir(parents=True, exist_ok=True)
        faiss.write_index(self.index, str(self.index_path))

这层封装负责：

• 统一检查向量维度
• 统一转 float32
• 统一做归一化
• 统一保存和加载 index
• 对上层隐藏 FAISS 细节

13. 在 FastAPI + Celery + FAISS 中的位置

结合你的 RAG 项目，可以这样拆：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

FastAPI：
- 接收上传请求
- 创建 document 和 task 记录
- 返回 task_id

Celery：
- 解析文档
- chunk 切分
- 调 embedding 模型
- 调 VectorStore.add()
- 保存 FAISS index
- 更新任务状态

FastAPI chat 接口：
- 接收用户问题
- 调 embedding 模型生成 query vector
- 调 VectorStore.search()
- 根据 chunk_id 查 MySQL
- 拼接上下文
- 调 LLM
- 保存 message 和 citation

一个简化链路：

1
2
3
4
5
6
7
8
9
10
11
12

upload document
  -> Celery ingest task
  -> chunks
  -> embeddings
  -> FAISS index
  -> MySQL chunk metadata

chat question
  -> query embedding
  -> FAISS top-k chunk_id
  -> MySQL load chunk content
  -> LLM answer

14. 单文档索引和知识库索引

14.1 单文档索引

每个文档一个 FAISS index。

优点：

• 实现简单
• 删除文档方便
• 文件和索引一一对应

缺点：

• 多文档检索麻烦
• 查询多个文档时要搜多个 index
• 不适合知识库级检索

14.2 知识库索引

一个知识库一个 FAISS index，里面包含多个文档的 chunk。

优点：

• 支持多文档统一检索
• 更接近真实 RAG 系统
• 检索流程简单

缺点：

• 删除单个文档更麻烦
• 索引版本管理更重要
• 需要维护 chunk_id 到文档元数据的映射

推荐：

• demo 阶段可以单文档索引
• 简历项目和真实项目更建议做知识库索引

15. 更新和删除

FAISS 不是传统数据库，更新和删除要谨慎设计。

15.1 添加

添加新 chunk 比较简单：

1
2

index.add_with_ids(new_vectors, new_ids)
faiss.write_index(index, index_path)

15.2 删除

部分索引支持 remove_ids()，但工程上仍然要考虑：

• 删除后 MySQL 元数据是否同步
• index 文件是否需要重新保存
• 删除大量数据后索引质量是否下降
• 多进程读写是否冲突

更稳妥的做法：

1
2
3
4
5
6
7

少量删除：
- 数据库标记 chunk/document 为 deleted
- 检索后过滤 deleted 结果

大量删除或周期整理：
- 根据有效 chunk 重新构建 index
- 原子替换旧 index 文件

15.3 更新

更新文档一般等价于：

1. 删除旧文档对应 chunk
2. 重新解析文档
3. 重新生成 embedding
4. 重新加入 index

如果更新频率高，就要认真设计索引重建策略。

16. 检索参数怎么选

16.1 top_k

top_k 表示返回几个候选 chunk。

RAG 常见取值：

• top_k = 3
• top_k = 5
• top_k = 10

取太小：

• 可能漏掉关键信息

取太大：

• prompt 变长
• 噪声变多
• LLM 成本上升

可以先用 top_k=5，后续根据实际效果调。

16.2 nprobe

IVF 索引里最重要的搜索参数是 nprobe。

1

index.nprobe = 10

简单理解：

• nprobe 小：速度快，可能漏结果
• nprobe 大：召回高，速度慢

调参时要看：

• 查询耗时
• recall
• 最终回答质量

16.3 HNSW 参数

HNSW 常看：

• M
• efConstruction
• efSearch

经验理解：

• M 控制图连接密度
• efConstruction 控制建图质量
• efSearch 控制查询时探索范围

如果检索结果质量不够，先尝试调大 efSearch。

17. 如何评估 FAISS 检索效果

只看“能不能返回结果”是不够的。

RAG 项目里至少可以做几个简单评估：

17.1 Recall@K

如果你有标准答案 chunk，可以看：

1

Recall@K = 正确 chunk 是否出现在 Top-K 结果里

例如：

• Recall@1
• Recall@3
• Recall@5

17.2 MRR

MRR 关注正确结果排在第几位。

如果正确 chunk 总是排在第一名，效果就很好。

17.3 人工问题集

最实用的方式是为每份文档准备一批问题：

1
2

问题：论文里使用的数据集是什么？
期望命中的 chunk：chunk_id = 123

然后统计：

• 是否命中
• 排名第几
• 分数是多少
• LLM 最终回答是否引用了正确片段

17.4 检索耗时

RAG 链路里建议记录：

• embedding 耗时
• FAISS search 耗时
• MySQL 查 chunk 耗时
• LLM 生成耗时

这样才能知道瓶颈在哪里。

18. FAISS 和向量数据库的区别

FAISS 更像一个向量检索引擎或向量索引库。

向量数据库通常还提供：

• 数据持久化管理
• 元数据过滤
• 分布式存储
• 多租户
• 权限管理
• 在线增删改查
• 服务化 API

常见向量数据库：

• Milvus
• Qdrant
• Weaviate
• pgvector

对比：

1
2
3
4
5
6
7
8
9
10
11
12

FAISS：
- 轻量
- 本地库
- 性能强
- 灵活
- 需要自己做元数据和服务封装

向量数据库：
- 功能完整
- 更偏生产服务
- 自带元数据过滤和管理能力
- 部署和维护成本更高

你的 RAG 项目如果是学习和简历展示，用 FAISS 很合适。它能让你真正理解向量检索、索引、召回率、Top-K、元数据映射这些底层概念。

19. 常见坑

19.1 忘记转 float32

很多 embedding 模型或 Python 处理流程会给出 float64。

FAISS 通常期望 float32。

1

vectors = vectors.astype("float32")

19.2 向量维度不一致

比如：

1
2

索引维度：768
查询向量：1024

这通常是因为：

• embedding 模型换了
• 文档向量和查询向量用了不同模型
• reshape 错了

解决：

• 数据库记录 embedding_model 和 embedding_dim
• 加载 index 时检查配置
• 查询前检查 query shape

19.3 做余弦检索但忘记归一化

如果用 IndexFlatIP 模拟余弦相似度，一定要：

• 入库向量归一化
• 查询向量归一化

否则内积会受到向量长度影响。

19.4 IVF 忘记 train()

IVF、PQ 这类索引通常需要训练。

错误流程：

1
2

index = faiss.IndexIVFFlat(...)
index.add(vectors)

正确流程：

1
2

index.train(train_vectors)
index.add(vectors)

可以检查：

1

print(index.is_trained)

19.5 只保存 FAISS，不保存元数据

如果只保存 index 文件，不保存 chunk 元数据，搜索结果只会给你 ID，无法还原原文。

正确做法：

• FAISS 存向量
• MySQL 存 chunk 内容和元数据
• FAISS ID 对应 MySQL chunk_id

19.6 多进程同时写 index

如果 FastAPI 和多个 Celery worker 同时写同一个 index 文件，容易出问题。

建议：

• 写操作集中到单个任务队列
• 使用文件锁或数据库状态锁
• 写新文件后原子替换旧文件
• 查询服务读取稳定版本

19.7 把 FAISS 当数据库用

FAISS 适合相似度搜索，不适合做复杂业务查询。

这些事情应该交给业务数据库：

• 用户权限
• 文档状态
• 删除标记
• 文件来源
• 页码
• 文本内容
• citation 信息

20. RAG 项目里的推荐方案

如果是你的当前 RAG 文档检索项目，我建议这样选：

20.1 初版

1
2
3
4
5

索引：IndexIDMap(IndexFlatIP)
相似度：归一化向量 + Inner Product
元数据：MySQL chunk 表
持久化：faiss.write_index()
任务：Celery 构建索引

优点：

• 简单
• 准确
• 方便调试
• 很适合简历项目说明

20.2 数据量变大后

如果 Flat 查询变慢，可以升级：

1

IndexHNSWFlat

或者：

1

IndexIVFFlat

建议顺序：

1. 先用 Flat 建基准
2. 再上 HNSW 或 IVF
3. 用 Recall@K 对比召回损失
4. 记录查询耗时变化

20.3 简历里可以怎么讲

可以写成：

1

使用 FAISS 构建文档 chunk 向量索引，采用 chunk_id 作为向量 ID，并在 MySQL 中维护文档、chunk、索引版本和 citation 元数据；查询阶段对用户问题生成 embedding 后执行 Top-K 召回，再根据 chunk_id 回表获取原文内容并拼接上下文供 LLM 生成回答。

如果你做了评测，还可以加：

1

基于人工问题集评估 Recall@K 和检索耗时，对比 IndexFlatIP、HNSW、IVFFlat 在召回率、延迟和内存占用上的差异。

21. 一个完整的 RAG 检索示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

import faiss
import numpy as np

class SimpleRagIndex:
    def __init__(self, dim: int):
        self.dim = dim
        self.index = faiss.IndexIDMap(faiss.IndexFlatIP(dim))

    def add_chunks(self, embeddings: list[list[float]], chunk_ids: list[int]) -> None:
        vectors = np.asarray(embeddings, dtype="float32")
        vectors = np.ascontiguousarray(vectors)

        ids = np.asarray(chunk_ids, dtype="int64")

        if vectors.ndim != 2:
            raise ValueError("embeddings must be 2D")
        if vectors.shape[1] != self.dim:
            raise ValueError("embedding dimension mismatch")
        if len(ids) != len(vectors):
            raise ValueError("ids length mismatch")

        faiss.normalize_L2(vectors)
        self.index.add_with_ids(vectors, ids)

    def search(self, query_embedding: list[float], top_k: int = 5):
        query = np.asarray(query_embedding, dtype="float32").reshape(1, -1)
        query = np.ascontiguousarray(query)

        if query.shape[1] != self.dim:
            raise ValueError("query dimension mismatch")

        faiss.normalize_L2(query)
        scores, chunk_ids = self.index.search(query, top_k)

        results = []
        for score, chunk_id in zip(scores[0], chunk_ids[0]):
            if chunk_id == -1:
                continue
            results.append(
                {
                    "chunk_id": int(chunk_id),
                    "score": float(score),
                }
            )
        return results

rag_index = SimpleRagIndex(dim=4)

rag_index.add_chunks(
    embeddings=[
        [1.0, 0.0, 0.0, 0.0],
        [0.9, 0.1, 0.0, 0.0],
        [0.0, 1.0, 0.0, 0.0],
    ],
    chunk_ids=[1001, 1002, 1003],
)

results = rag_index.search([1.0, 0.0, 0.0, 0.0], top_k=2)
print(results)

真实项目里，拿到 chunk_id 后再去 MySQL 查：

1

chunk_id -> chunk content -> document metadata -> citation

22. 学习顺序

建议按这个顺序学：

1. 理解 embedding 和向量相似度
2. 跑通 IndexFlatL2
3. 跑通 IndexFlatIP + normalize_L2
4. 学会 add()、search()、D/I
5. 学会 IndexIDMap 绑定业务 ID
6. 学会 write_index() 和 read_index()
7. 接入 RAG 项目：chunk_id 回表查 MySQL
8. 数据量变大后再学 HNSW、IVF、PQ
9. 用 Recall@K 和耗时评估索引效果

23. 总结

FAISS 的核心价值是：

• 快速做向量相似度搜索
• 支持精确检索和近似检索
• 支持不同索引结构，在速度、内存、召回率之间取舍
• 非常适合 RAG 项目里的 Top-K chunk 召回

初学时先记住：

1
2
3
4

Embedding 负责把文本变成向量
FAISS 负责从向量里找相似内容
MySQL 负责保存 chunk 内容和元数据
LLM 负责基于检索结果生成回答

对于你的 RAG 项目，最实用的起点是：

1

IndexIDMap(IndexFlatIP) + normalize_L2 + chunk_id 回表查 MySQL

等数据量、延迟和内存真的成为问题，再升级 HNSW、IVF 或 PQ。

24. 参考

• FAISS GitHub: https://github.com/facebookresearch/faiss
• FAISS Wiki: https://github.com/facebookresearch/faiss/wiki
• FAISS Getting Started: https://github.com/facebookresearch/faiss/wiki/Getting-started
• FAISS Indexes: https://github.com/facebookresearch/faiss/wiki/Faiss-indexes
• FAISS MetricType and distances: https://github.com/facebookresearch/faiss/wiki/MetricType-and-distances
• FAISS Index Factory: https://github.com/facebookresearch/faiss/wiki/The-index-factory

FastAPI 学习笔记

1. FastAPI 是什么

FastAPI 是一个基于 Python 类型注解构建的 Web API 框架，适合用来开发：

• 后端接口服务
• AI 模型推理接口
• 文件上传和处理接口
• 内部微服务
• 需要自动生成接口文档的项目

它的几个核心特点：

• 开发效率高，写法接近普通 Python 函数
• 自动参数校验
• 自动生成 OpenAPI 文档
• 支持异步
• 和 Pydantic 配合很好，适合结构化数据处理

FastAPI 最实用的地方在于：你写的是普通 Python 函数，但它会顺手帮你把 HTTP 请求、参数校验和接口文档这些事情一起处理掉。

2. 为什么很多 AI 项目喜欢用 FastAPI

在 AI 项目里，FastAPI 很常见，因为它很适合做这些事情：

• 暴露模型推理接口
• 上传文档、图片、音频
• 提供任务提交和状态查询接口
• 给前端或其他服务提供统一 API
• 和 Celery、Redis、数据库组合成一套服务

尤其是下面这种场景：

1. 前端上传文件
2. FastAPI 接收请求
3. 把耗时任务交给 Celery
4. 再通过 FastAPI 提供任务状态查询接口

这类架构在 RAG、推理平台、数据处理平台里非常常见。

3. 安装

最基础的安装：

1

pip install fastapi uvicorn

如果要做文件上传，还需要：

1

pip install python-multipart

说明：

• fastapi：框架本身
• uvicorn：ASGI Server，用来启动服务
• python-multipart：处理表单和文件上传

4. 最小可运行示例

文件：main.py

1
2
3
4
5
6
7

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "hello fastapi"}

启动命令：

1

uvicorn main:app --reload

含义：

• main：Python 文件名 main.py
• app：文件中的 FastAPI 实例
• --reload：代码变更后自动重启，适合开发环境

启动后可以访问：

• http://127.0.0.1:8000/
• http://127.0.0.1:8000/docs
• http://127.0.0.1:8000/redoc

其中：

• /docs：Swagger UI
• /redoc：ReDoc 文档页面

5. FastAPI 的基本工作方式

可以把一个接口理解成：

1. 定义一个路径
2. 指定请求方法，比如 GET、POST
3. 声明这个接口需要哪些参数
4. FastAPI 自动帮你解析参数并做校验
5. 函数返回值自动转换为 JSON 响应

示例：

1
2
3
4
5
6
7

from fastapi import FastAPI

app = FastAPI()

@app.get("/hello")
def hello(name: str):
    return {"message": f"hello {name}"}

访问：

1

/hello?name=tom

返回：

1

{"message": "hello tom"}

这里的 name: str 会被 FastAPI 识别为查询参数。

6. 路由基础

FastAPI 支持常见的 HTTP 方法：

• @app.get()
• @app.post()
• @app.put()
• @app.delete()
• @app.patch()

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

from fastapi import FastAPI

app = FastAPI()

@app.get("/users")
def get_users():
    return {"action": "list users"}

@app.post("/users")
def create_user():
    return {"action": "create user"}

@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

说明：

• /users 通常表示资源集合
• /users/{user_id} 表示某个具体资源
• user_id: int 会自动做类型校验

如果路径参数类型不对，FastAPI 会直接返回校验错误。

7. 路径参数、查询参数、请求体

这是 FastAPI 最重要的基础之一。

7.1 路径参数

路径里写在 {} 中的内容就是路径参数。

1
2
3

@app.get("/items/{item_id}")
def get_item(item_id: int):
    return {"item_id": item_id}

例如：

1

/items/100

这里的 item_id 就是路径参数。

7.2 查询参数

函数里那些不在路径中的普通基础类型参数，通常会被当成查询参数。

1
2
3

@app.get("/items")
def list_items(page: int = 1, size: int = 10):
    return {"page": page, "size": size}

例如：

1

/items?page=2&size=20

7.3 请求体

如果参数是 Pydantic 模型，FastAPI 会把它当成请求体。

1
2
3
4
5
6
7
8
9

from pydantic import BaseModel

class UserCreate(BaseModel):
    name: str
    age: int

@app.post("/users")
def create_user(user: UserCreate):
    return user

请求：

1
2
3
4

{
  "name": "alice",
  "age": 18
}

结论：

• 路径里的参数是路径参数
• 简单类型参数通常是查询参数
• Pydantic 模型通常是请求体

8. Pydantic 模型

FastAPI 的数据校验非常依赖 Pydantic。

最常见的用途：

• 定义请求体结构
• 定义返回体结构
• 做字段校验
• 约束字段类型

8.1 基本示例

1
2
3
4
5
6

from pydantic import BaseModel, Field

class ArticleCreate(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    content: str
    author: str

这里的好处是：

• 字段缺失会报错
• 类型不对会报错
• 长度不满足约束会报错

8.2 作为返回模型

1
2
3
4
5
6
7
8
9
10
11
12

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserOut(BaseModel):
    id: int
    name: str

@app.get("/users/{user_id}", response_model=UserOut)
def get_user(user_id: int):
    return {"id": user_id, "name": "tom", "password": "secret"}

返回给客户端时，password 这类不在 response_model 中的字段会被过滤掉。

这非常适合做：

• 隐藏敏感字段
• 固定接口返回结构
• 保证前后端契约一致

9. def 和 async def 怎么选

这是 FastAPI 初学时很容易混淆的点。

9.1 用 async def 的情况

适合：

• 调用异步数据库客户端
• 调用异步 HTTP 客户端
• WebSocket
• 其他真正支持异步的 IO 场景

9.2 用普通 def 的情况

适合：

• 普通同步逻辑
• CPU 密集型计算
• 调用同步库
• 本来就不是异步的代码

注意：

async def 不是“更快”的意思，它只是更适合异步 IO。

如果你在 async def 里直接跑很重的同步任务，照样会阻塞。

对于重任务，通常应该交给：

• Celery
• 任务队列
• 独立 Worker

10. 请求参数的更明确写法

FastAPI 虽然能自动推断参数来源，但实际项目里更推荐写明确一些。

可以使用：

• Path
• Query
• Body

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

from fastapi import Body, FastAPI, Path, Query

app = FastAPI()

@app.post("/items/{item_id}")
def update_item(
    item_id: int = Path(..., ge=1),
    keyword: str | None = Query(default=None, min_length=1),
    data: dict = Body(...),
):
    return {
        "item_id": item_id,
        "keyword": keyword,
        "data": data,
    }

这里：

• Path(..., ge=1) 表示路径参数必须大于等于 1
• Query(...) 用来描述查询参数校验
• Body(...) 明确说明这个参数来自请求体

这种写法更清晰，也更适合团队协作。

11. APIRouter 路由拆分

项目一大，通常不会把所有接口都写在一个 main.py 里。

FastAPI 推荐使用 APIRouter 拆模块。

11.1 基本示例

文件：routers/health.py

1
2
3
4
5
6
7

from fastapi import APIRouter

router = APIRouter(prefix="/health", tags=["health"])

@router.get("")
def health_check():
    return {"status": "ok"}

主文件：main.py

1
2
3
4
5
6

from fastapi import FastAPI

from routers.health import router as health_router

app = FastAPI()
app.include_router(health_router)

访问路径：

1

/health

11.2 为什么要拆 Router

好处：

• 按业务模块组织代码
• 主入口文件更简洁
• 适合多人协作
• 后续做版本管理更方便

常见拆分方式：

• routers/user.py
• routers/auth.py
• routers/files.py
• routers/tasks.py

12. 路由前缀和 API 版本管理

你原来的笔记重点写的是这部分，这里整理成更清晰的版本。

12.1 统一加前缀

如果整个项目都希望走 /api/v1 前缀，可以在 include_router() 时统一加上。

1
2
3
4
5
6
7
8

from fastapi import FastAPI

from routers.health import router as health_router

app = FastAPI()

root_prefix = "/api/v1"
app.include_router(health_router, prefix=root_prefix)

如果 health_router 本身是：

1

router = APIRouter(prefix="/health", tags=["health"])

那么最终路径就是：

1

/api/v1/health

12.2 在 Router 上直接带业务前缀

例如：

1

router = APIRouter(prefix="/users", tags=["users"])

再在主应用里挂上版本前缀：

1

app.include_router(user_router, prefix="/api/v1")

最后效果就是：

1

/api/v1/users

12.3 多版本应用挂载

如果你想同时保留多个版本，也可以挂多个 FastAPI 子应用。

1
2
3
4
5
6
7
8
9

from fastapi import FastAPI

app = FastAPI()

app_v1 = FastAPI(title="API V1")
app_v2 = FastAPI(title="API V2")

app.mount("/api/v1", app_v1)
app.mount("/api/v2", app_v2)

这种方式适合：

• 两个版本差异非常大
• 想彻底隔离文档和路由
• 历史接口需要长期兼容

但如果只是小规模版本演进，通常直接用：

• include_router(..., prefix="/api/v1")
• include_router(..., prefix="/api/v2")

就够了。

12.4 版本管理建议

实际项目里一般这样做：

• 先按模块拆 Router
• 再统一加版本前缀
• 不要一开始就把版本设计得过于复杂

多数项目初期使用：

1

/api/v1/xxx

已经足够。

13. 依赖注入 Depends

Depends 是 FastAPI 很重要的能力。

它的核心作用是：

• 把公共逻辑抽出来复用
• 给接口注入数据库连接、用户信息、分页参数等
• 让路由函数更干净

13.1 基本示例

1
2
3
4
5
6
7
8
9
10

from fastapi import Depends, FastAPI

app = FastAPI()

def common_pagination(page: int = 1, size: int = 10):
    return {"page": page, "size": size}

@app.get("/articles")
def list_articles(pagination=Depends(common_pagination)):
    return pagination

说明：

• page 和 size 可以在多个接口中复用
• 路由函数里只拿最终整理好的结果

13.2 常见依赖场景

• 获取当前登录用户
• 获取数据库 Session
• 提取分页参数
• 权限校验
• 校验请求头

例如认证场景：

1
2
3
4
5
6
7
8
9
10
11
12

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()

def verify_token(x_token: str = Header(...)):
    if x_token != "demo-token":
        raise HTTPException(status_code=401, detail="invalid token")
    return x_token

@app.get("/secure")
def secure_api(token: str = Depends(verify_token)):
    return {"token": token}

14. 异常处理

FastAPI 常用 HTTPException 抛业务错误。

1
2
3
4
5
6
7
8
9

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/users/{user_id}")
def get_user(user_id: int):
    if user_id != 1:
        raise HTTPException(status_code=404, detail="user not found")
    return {"id": 1, "name": "tom"}

适合用在：

• 资源不存在
• 权限不足
• 参数不合法
• 业务状态不满足要求

补充理解：

• 参数校验错误通常由 FastAPI 自动处理
• 业务逻辑错误通常由你主动抛 HTTPException

15. 文件上传

这在 AI 项目里很常见，比如：

• 上传 PDF
• 上传图片
• 上传音频
• 上传知识库文档

15.1 单文件上传

1
2
3
4
5
6
7
8
9
10

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/upload")
def upload_file(file: UploadFile = File(...)):
    return {
        "filename": file.filename,
        "content_type": file.content_type,
    }

15.2 保存到本地

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

from pathlib import Path
import shutil

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/upload")
def upload_file(file: UploadFile = File(...)):
    save_path = UPLOAD_DIR / file.filename

    with save_path.open("wb") as buffer:
        shutil.copyfileobj(file.file, buffer)

    return {"path": str(save_path)}

15.3 实战建议

• 不要直接相信上传文件名
• 最好自己生成唯一文件名
• 大文件处理不要阻塞接口太久
• 上传成功后可以把处理任务交给 Celery

16. 表单与文件混合提交

有时接口既要文件，也要普通字段，比如：

• 文件 + 文档类型
• 图片 + 用户 ID
• 音频 + 语言参数

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13

from fastapi import FastAPI, File, Form, UploadFile

app = FastAPI()

@app.post("/documents")
def create_document(
    file: UploadFile = File(...),
    doc_type: str = Form(...),
):
    return {
        "filename": file.filename,
        "doc_type": doc_type,
    }

17. 中间件和 CORS

17.1 中间件是什么

中间件是在请求进入路由前、响应返回客户端前统一插入的一层处理逻辑。

常见用途：

• 记录请求日志
• 统计耗时
• 增加追踪 ID
• 统一鉴权

17.2 CORS

如果前后端分离，浏览器经常会遇到跨域问题。

FastAPI 通常这样加：

1
2
3
4
5
6
7
8
9
10
11
12

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

开发环境常见，生产环境一般要更严格地限制来源。

18. 生命周期事件

很多项目需要在服务启动时做初始化，比如：

• 建立连接池
• 加载配置
• 初始化日志
• 加载模型

FastAPI 支持应用生命周期管理。

1
2
3
4
5
6
7
8
9
10
11

from contextlib import asynccontextmanager

from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("service starting")
    yield
    print("service stopping")

app = FastAPI(lifespan=lifespan)

如果项目里有大模型、向量库客户端、数据库连接池，这一层会比较重要。

19. BackgroundTasks 和 Celery 的区别

这点和你刚整理的 Celery.md 是连着的。

19.1 BackgroundTasks

适合：

• 很轻量的后台操作
• 和当前 Web 进程生命周期绑定
• 不需要重试
• 不需要独立扩容

示例：

1
2
3
4
5
6
7
8
9
10
11
12

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_log(message: str):
    with open("app.log", "a", encoding="utf-8") as f:
        f.write(message + "\n")

@app.post("/notify")
def notify(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "new request")
    return {"message": "accepted"}

19.2 Celery

更适合：

• 长耗时任务
• 文件处理
• 模型推理
• 批量任务
• 需要失败重试
• 需要多机扩展

一句话：

轻任务用 BackgroundTasks，重任务用 Celery。

20. FastAPI 和 Celery 的典型配合方式

在 AI 项目里，比较常见的结构是：

1. FastAPI 负责接收请求
2. 接口做参数校验和鉴权
3. 把耗时任务提交给 Celery
4. 返回 task_id
5. 前端轮询任务状态

例如：

1
2
3
4
5
6
7
8
9
10

from fastapi import FastAPI

from tasks import process_document

app = FastAPI()

@app.post("/tasks/documents")
def create_document_task(file_path: str):
    task = process_document.delay(file_path)
    return {"task_id": task.id}

这个思路尤其适合：

• 文档解析
• embedding 计算
• 批量摘要
• 图片处理
• 音频转写

21. 响应模型和状态码

建议在正式项目里尽量把返回结构固定下来。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

from fastapi import FastAPI, status
from pydantic import BaseModel

app = FastAPI()

class UserCreate(BaseModel):
    name: str

class UserOut(BaseModel):
    id: int
    name: str

@app.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def create_user(user: UserCreate):
    return {"id": 1, "name": user.name}

这样做的好处：

• 文档更清晰
• 前端更好联调
• 返回结构更稳定

22. 测试

FastAPI 支持用 TestClient 做接口测试。

1
2
3
4
5
6
7
8
9
10

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "hello fastapi"}

这个对接口开发很有用，特别是：

• 改接口时防止回归
• 验证参数校验
• 验证权限逻辑

23. FastAPI 项目的工程结构

当项目从 demo 变成一个后端服务时，最重要的不是目录多，而是每一层职责清楚。

FastAPI 官方文档里会用 app/main.py、app/dependencies.py、app/routers/ 这种结构说明大型应用拆分。实际项目里可以在这个基础上继续细分成：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

project/
├── app/
│   ├── main.py
│   ├── api/
│   │   └── v1/
│   │       ├── router.py
│   │       └── endpoints/
│   │           ├── health.py
│   │           ├── documents.py
│   │           ├── chat.py
│   │           └── tasks.py
│   ├── core/
│   │   ├── config.py
│   │   ├── logging.py
│   │   ├── security.py
│   │   └── exceptions.py
│   ├── schemas/
│   │   ├── document.py
│   │   ├── chat.py
│   │   └── task.py
│   ├── models/
│   │   ├── document.py
│   │   ├── chunk.py
│   │   └── message.py
│   ├── crud/
│   │   ├── document.py
│   │   └── task.py
│   ├── services/
│   │   ├── document_service.py
│   │   ├── rag_service.py
│   │   ├── embedding_service.py
│   │   └── llm_service.py
│   ├── clients/
│   │   ├── redis_client.py
│   │   ├── vector_store.py
│   │   └── llm_client.py
│   ├── db/
│   │   ├── session.py
│   │   └── base.py
│   └── workers/
│       ├── celery_app.py
│       └── tasks.py
├── tests/
│   ├── test_documents.py
│   └── test_chat.py
└── requirements.txt

不一定每个项目都要这么全。目录结构应该跟项目复杂度匹配，小项目可以先从 main.py + routers + schemas + services 开始。

23.1 各目录负责什么

• main.py：应用入口，只负责创建 FastAPI 实例、注册中间件、注册路由、声明生命周期逻辑。
• api/v1/router.py：汇总某个 API 版本下的所有 router，例如 /documents、/chat、/tasks。
• api/v1/endpoints/：HTTP 接口层，只处理请求参数、依赖注入、状态码和响应模型，不直接堆复杂业务。
• schemas/：Pydantic 模型，定义请求体、返回体和接口契约。
• models/：数据库 ORM 模型，对应 MySQL 表结构。
• crud/：数据库读写封装，只做增删改查，不写复杂业务流程。
• services/：业务逻辑层，负责组织多个 crud、client、算法模块完成一个完整业务动作。
• clients/：外部服务或基础设施客户端，例如 Redis、向量库、LLM 服务、对象存储。
• core/：全局配置、日志、安全、异常处理、公共常量。
• workers/：Celery 应用和后台任务，处理文档解析、embedding、索引构建等耗时流程。
• tests/：接口测试、业务测试、E2E 测试。

最关键的原则：

Router 不写重业务，Service 不关心 HTTP，CRUD 不关心业务流程。

23.2 一次请求在结构里的流动

以创建文档任务为例：

1
2
3
4
5
6
7
8
9

Client
  -> FastAPI middleware
  -> api/v1/endpoints/documents.py
  -> Depends 获取配置、数据库 Session、用户信息
  -> schemas/document.py 校验请求体
  -> services/document_service.py 组织业务
  -> crud/document.py 写入文档记录
  -> workers/tasks.py 提交 Celery 任务
  -> 返回 task_id

接口函数应该尽量薄，例如：

1
2
3
4
5
6
7
8
9
10
11
12

from fastapi import APIRouter, Depends, UploadFile

from app.db.session import get_db
from app.schemas.document import DocumentUploadResponse
from app.services.document_service import DocumentService

router = APIRouter(prefix="/documents", tags=["documents"])

@router.post("", response_model=DocumentUploadResponse)
def upload_document(file: UploadFile, db=Depends(get_db)):
    service = DocumentService(db)
    return service.create_ingest_task(file)

这个接口只做三件事：

• 接收 HTTP 请求
• 通过 Depends 拿到依赖
• 调用 service 返回结果

文件保存、数据库写入、任务提交、错误处理都应该下沉到 service 或更底层。

23.3 main.py 应该保持很薄

main.py 更像“装配入口”，不要把业务逻辑都塞进去。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.api.v1.router import api_router
from app.core.config import settings

def create_app() -> FastAPI:
    app = FastAPI(title=settings.app_name)

    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    app.include_router(api_router, prefix="/api/v1")

    return app

app = create_app()

好处：

• 应用创建逻辑清楚
• 测试时可以复用 create_app()
• 业务模块不会和框架入口互相缠住

23.4 API Router 汇总方式

可以用一个 api/v1/router.py 统一挂载各业务模块：

1
2
3
4
5
6
7
8
9

from fastapi import APIRouter

from app.api.v1.endpoints import chat, documents, health, tasks

api_router = APIRouter()
api_router.include_router(health.router)
api_router.include_router(documents.router)
api_router.include_router(tasks.router)
api_router.include_router(chat.router)

这样 main.py 只需要引入一个 api_router，不会随着接口数量变多而越来越乱。

23.5 配置层 core/config.py

正式项目不要把数据库地址、Redis 地址、模型地址直接写死在业务代码里。更好的做法是用配置对象统一管理。

1
2
3
4
5
6
7
8
9
10
11
12
13

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env")

    app_name: str = "rag-api"
    mysql_url: str
    redis_url: str
    llm_base_url: str
    embedding_model: str = "KaLM-Embedding-0.5B"
    cors_origins: list[str] = ["http://localhost:3000"]

settings = Settings()

这样可以做到：

• 本地、测试、生产环境分开配置
• 敏感信息不写进代码
• 配置类型可以被 Pydantic 校验

23.6 结合 RAG 项目的推荐结构

如果是你的 RAG 文档检索项目，可以这样对应：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

接口层：
- documents.py：上传文档、查询文档列表、删除文档
- tasks.py：查询 Celery 任务状态和进度
- chat.py：创建会话、提交问题、返回回答和 citations
- health.py：健康检查和依赖状态检查

业务层：
- document_service.py：保存文件、创建文档记录、提交 ingest 任务
- rag_service.py：Top-K 检索、上下文拼接、调用 LLM、保存 message/citation
- embedding_service.py：文本向量化、批量 embedding
- index_service.py：FAISS 索引构建、加载、保存

基础设施层：
- db/session.py：MySQL Session
- clients/redis_client.py：Redis 连接
- clients/llm_client.py：OpenAI-compatible LLM 调用
- clients/vector_store.py：FAISS 检索封装
- workers/tasks.py：文档解析、chunk、embedding、建索引

RAG 链路可以这样记：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

上传接口
  -> document_service 创建文档和任务
  -> Celery worker 解析文档
  -> chunk 切片
  -> embedding_service 生成向量
  -> index_service 写入 FAISS
  -> task 状态更新

问答接口
  -> rag_service 接收问题
  -> vector_store Top-K 检索
  -> 拼接上下文和 Prompt
  -> llm_client 调用模型
  -> 保存 answer 和 citation
  -> 返回给前端

23.7 什么时候需要拆得更细

如果项目只有几个接口：

1
2
3
4

main.py
routers/
schemas/
services/

就已经够用。

当出现下面情况时，再继续拆：

• 接口超过十几个
• 多个接口复用同一批业务逻辑
• 数据库表开始变多
• 出现 Celery、Redis、LLM、向量库等外部依赖
• 测试变得难写
• main.py 或某个 router 文件超过几百行

不要为了“看起来专业”一开始就堆很多目录。目录结构的目的不是复杂，而是让业务边界清楚。

24. 常见坑

24.1 把所有代码都写在一个文件里

初学时能跑，项目一大就很难维护。

建议尽快学会：

• APIRouter
• 模块拆分
• 业务逻辑下沉到 service 层

24.2 在 async def 里跑重同步任务

这会阻塞事件循环。

如果是：

• 长时间计算
• 文件处理
• 模型推理

通常不应该直接堆在接口函数里。

24.3 请求体验证和返回结构不统一

如果不定义 Pydantic 模型，接口会越来越乱。

建议：

• 请求体尽量用模型
• 返回值尽量用 response_model

24.4 忽略文件上传安全问题

例如：

• 直接使用用户传上来的文件名
• 不限制文件类型
• 不限制文件大小

这些在正式项目里都容易出问题。

24.5 把 FastAPI 当成“万能后台线程框架”

FastAPI 主要是 Web API 框架，不是专门的任务调度系统。

如果任务已经明显是：

• 重任务
• 批处理
• 需要排队
• 需要重试

就应该交给 Celery 之类的任务系统。

25. 一套比较实用的学习顺序

建议按这个顺序掌握：

1. 先跑通最小示例
2. 理解路由、路径参数、查询参数
3. 学会用 Pydantic 定义请求体和返回体
4. 学会拆 APIRouter
5. 学会 Depends
6. 学会文件上传
7. 学会 BackgroundTasks
8. 最后再接入 Celery、数据库、鉴权

26. 总结

FastAPI 的核心价值在于：

• 用很少的代码快速暴露 API
• 自动完成参数解析和校验
• 自动生成接口文档
• 很适合和 Pydantic、Celery、Redis、数据库组合

对于 AI 项目，可以把它理解成：

• 对外接口入口
• 请求校验层
• 文件上传入口
• 任务分发入口
• 状态查询入口

如果你后面要继续整理 AI 模型开发相关知识点，FastAPI、Redis、Celery 这三份笔记其实可以看成一条链：

• FastAPI：接请求
• Redis：做缓存 / 消息中间件
• Celery：处理后台任务

27. 参考

• FastAPI Official Docs: https://fastapi.tiangolo.com/
• FastAPI Path Params: https://fastapi.tiangolo.com/tutorial/path-params/
• FastAPI Query Params: https://fastapi.tiangolo.com/tutorial/query-params/
• FastAPI Request Body: https://fastapi.tiangolo.com/tutorial/body/
• FastAPI APIRouter: https://fastapi.tiangolo.com/tutorial/bigger-applications/
• FastAPI Dependencies: https://fastapi.tiangolo.com/tutorial/dependencies/
• FastAPI File Upload: https://fastapi.tiangolo.com/tutorial/request-files/
• FastAPI Background Tasks: https://fastapi.tiangolo.com/tutorial/background-tasks/
• FastAPI Testing: https://fastapi.tiangolo.com/tutorial/testing/
• FastAPI Middleware: https://fastapi.tiangolo.com/tutorial/middleware/
• FastAPI Lifespan Events: https://fastapi.tiangolo.com/advanced/events/
• FastAPI Settings: https://fastapi.tiangolo.com/advanced/settings/

Fighting Netcode 项目知识笔记

1. 这个项目是什么

Fighting 是一个用 C++ 写的联机动作游戏同步 Demo。

项目地址：

/Users/chutian/Desktop/Fighting

它不是 AI 模型训练项目，而是一个很适合学习“实时系统工程”的项目。它实现的是：

• 固定帧推进
• UDP 通信
• 服务端权威状态
• 客户端本地预测
• 输入冗余
• 回滚重放
• 状态快照
• 状态哈希对账
• 压力测试

可以把它理解成一个“小型实时分布式系统”。

如果放到 AI 模型开发学习体系里，它最值得参考的不是模型算法，而是工程能力：

• 如何让多个端的状态保持一致
• 如何处理延迟、丢包和乱序
• 如何用快照、日志和 hash 定位状态分叉
• 如何用固定 tick 管理实时流程
• 如何把核心逻辑拆成可测试的模块

一句话概括：

这个项目是一个 60Hz 回滚同步网络游戏 Demo，但它背后的工程思想可以迁移到 AI 推理服务、Agent 执行系统、实时任务调度和分布式状态一致性设计里。

2. 项目整体目标

这个项目不是为了做一个完整游戏，而是为了跑通联机动作游戏最难的几件事。

核心目标是：

• 玩家输入能快速反馈
• 服务端保持最终权威
• 网络延迟不会让本地操作卡顿
• 客户端预测错了以后可以回滚修正
• 服务端和客户端可以通过 hash 判断状态是否分叉

联机动作游戏里，一个常见问题是：

如果每次按键都等服务端确认，游戏会非常卡；如果完全相信客户端，又容易作弊和状态不一致。

这个项目采用的方案是：

• 客户端先预测，保证操作手感
• 服务端做权威推进，保证最终正确
• 客户端收到权威状态后回滚重放，修正预测误差

这就是 Rollback Netcode 的核心思想。

3. 项目目录结构

项目主要目录如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

apps/
  client_main.cpp
  server_main.cpp

include/lab/
  app/
  core/
  io/
  net/
  sim/
  time/
  util/

src/
  app/
  core/
  io/
  net/
  sim/
  time/

tests/
  core_tests.cpp
  stress_tests.cpp

docs/
  ARCHITECTURE.md

README.md
日志.md
CMakeLists.txt

可以按三层理解：

这个分层比较清晰：

• 模拟层不关心网络和窗口
• 网络层不关心游戏规则
• 应用层负责把模拟、网络、渲染拼起来

这也是很多 AI 工程项目应该学习的拆法：

• 模型推理核心逻辑
• API / 网络协议层
• 应用编排层
• 测试和压测层

4. 快速运行方式

项目依赖：

• CMake 3.20+
• C++20
• libevent
• SDL2
• SDL2_ttf
• nlohmann_json

构建：

1
2

cmake -S . -B build
cmake --build build

启动服务端：

1

./build/lab_server

启动两个客户端：

1
2

./build/lab_client
./build/lab_client

运行测试：

1

ctest --test-dir build --output-on-failure

运行压力测试：

1

./build/lab_stress

更长时间压力测试：

1

./build/lab_stress --ticks 200000 --history 8192 --state-delay 12

5. 核心概念总览

这个项目最重要的概念有 8 个。

5.1 Tick

Tick 是逻辑帧编号。

代码里定义：

1

using Tick = uint32_t;

游戏不是直接按真实时间推进，而是按离散帧推进。

例如 60Hz 表示：

1
2

1 秒 = 60 个 tick
1 tick = 1 / 60 秒

这样做的好处是：

• 逻辑顺序明确
• 输入可以绑定到某个 tick
• 状态可以按 tick 保存
• 回滚时可以从某个 tick 重新模拟
• 网络包可以明确说明自己属于哪一帧

AI 工程迁移理解：

Tick 类似任务系统里的 step id、事件序号、offset、version。只要系统存在异步、重试、回放，就需要这种明确的逻辑编号。

5.2 InputCmd

InputCmd 是每一帧的最小输入单位。

1
2
3
4
5
6

struct InputCmd {
    Tick tick = 0;
    uint16_t buttons = 0;
    int8_t moveX = 0;
    int8_t moveY = 0;
};

它包含：

• 属于哪个 tick
• 按键位图
• 水平方向
• 垂直方向

这个结构很小，适合频繁发送。

设计重点：

• 输入要带 tick
• 输入结构要尽量小
• 输入可以重复发送，服务端按 tick 去重覆盖

5.3 WorldSnapshot

WorldSnapshot 是世界状态快照。

它包含：

• 当前 tick
• 玩家状态
• 子弹状态
• 迷宫 seed
• 迷宫网格

它的作用是：

• 网络状态同步
• 回滚恢复
• hash 对账
• 测试验证

这说明项目把“状态数据”和“推进逻辑”分开了。

这种设计很重要：

只要系统支持回滚、回放、审计、调试，就应该有明确的 Snapshot 结构。

5.4 World::Step

World::Step(cmds, dt) 是世界推进入口。

它接收所有玩家在同一个 tick 的输入，然后推进一帧。

关键要求：

• cmds.size() 必须等于玩家数量
• 所有 cmd.tick 必须一致
• 同样输入和同样初始状态，应该得到同样输出

这就是确定性模拟的基础。

5.5 InputBuffer

InputBuffer 是按 tick 存输入的环形缓冲。

核心逻辑：

1

idx = tick % capacity

读取时不只看槽位，还要校验 tick：

1

if (ring_[idx].cmd.tick != tick) return std::nullopt;

这个校验非常关键。

因为环形缓冲会覆盖旧数据，如果只看下标，不看 tick，就可能把旧帧误当成当前帧。

5.6 StateHistory

StateHistory 是按 tick 存世界快照的环形缓冲。

它和 InputBuffer 思路一样：

• 写入时按 tick % cap
• 读取时校验真实 tick

它用于：

• 客户端保存预测状态
• 收到权威状态后比较误差
• 从某个权威 tick 恢复并重放
• hash 对账

5.7 State Hash

Hasher 对 WorldSnapshot 做 hash。

它不是直接 hash float 的内存，而是先做毫米级量化：

1
2
3

int32_t QuantizeMm(float v) {
  return std::lround(v * 1000.0f);
}

这样做是为了避免浮点误差造成假 mismatch。

hash 覆盖内容包括：

• tick
• 玩家位置和速度
• 玩家 HP
• action
• cooldown
• aim 方向
• 子弹状态
• 迷宫 seed 和网格

如果新增一个会影响模拟结果的字段，但忘了加入快照、网络包或 hash，就容易出现服务端和客户端状态分叉。

5.8 Rollback Replay

回滚重放是客户端收到权威状态后的修正流程。

基本公式：

1
2
3
4
5
6

Restore(authoritativeSnapshot)
for t in authoritativeTick + 1 .. localTick - 1:
  取本地输入
  取远端预测输入
  World::Step(cmds, dt)
  保存快照

它解决的问题是：

• 客户端先预测，保证响应快
• 权威状态来了以后，以权威状态为准
• 重新模拟未来帧，让当前画面追上本地时间

6. 服务端流程

服务端是权威端。

核心文件：

apps/server_main.cpp

服务端主要做 6 件事：

1. 启动 UDP socket
2. 接收客户端 hello / input
3. 按地址分配玩家 slot
4. 收齐玩家后发送 Start
5. 每个 tick 收集输入并推进权威世界
6. 下发 Ack 和 State

6.1 玩家分配

服务端用客户端地址作为 key。

每个客户端连接后分配：

• player1
• player2

如果人数满了，后续客户端会被拒绝。

6.2 开局同步

服务端收齐 kRequiredPlayers = 2 后，不是立刻从当前 tick 开始，而是设置：

1

startTick = tick + kStartDelayTicks;

这个设计用于给客户端一点时间接收开局包，减少起步不同步。

6.3 缺输入处理

服务端每个 tick 都要给每个玩家拿到输入。

如果该 tick 有输入：

• 直接使用

如果没有输入，但上次输入还在 hold 窗口内：

• 复用上一帧输入
• 把 tick 改成当前 tick

如果超出 hold 窗口：

• 使用默认空输入

这段逻辑本质上是在处理 UDP 丢包和延迟。

6.4 权威推进

服务端每帧执行：

1
2
3
4

cmds = 每个玩家当前 tick 的输入
world.Step(cmds, dt)
snapshot = world.Snapshot()
hash = Hasher::Hash(snapshot)

服务端的状态是最终可信状态。

客户端的预测状态只是为了体验。

6.5 状态下发

服务端每帧发 Ack，每隔若干帧发完整 State。

当前配置：

1

kStateEvery = 2

也就是每 2 tick 下发一次完整状态。

这种设计平衡了：

• 状态同步频率
• 带宽消耗
• 回滚修正速度

7. 客户端流程

客户端负责“先动起来，然后等权威校正”。

核心文件：

apps/client_main.cpp

客户端主要做 8 件事：

1. 未开局时定期发送 hello
2. 收到 Start 后对齐 startTick
3. 每 tick 采样键盘输入
4. 本地保存输入历史
5. 预测远端玩家输入
6. 本地推进预测世界
7. 发送带冗余的输入包
8. 收到权威 State 后恢复并重放

7.1 本地预测

本地玩家的输入来自键盘。

远端玩家的输入无法立即知道，只能预测。

当前预测逻辑比较简单：

• 根据远端玩家速度推测移动方向
• 如果远端处于 Hitstun，则预测不动
• 如果速度接近 0，则预测停止

相关文件：

src/app/InputPrediction.cpp

这不是高级 AI，而是一个经验规则。

7.2 输入冗余发送

客户端每个 tick 不只发送当前输入，而是发送最近 K 帧输入。

配置：

1

kInputRedundancy = 4

这样即使 UDP 丢了某个包，后续包也可能补上之前几帧输入。

输入包里包含：

• playerId
• seq
• newestTick
• clientAckServerTick
• 最近 K 个 InputCmd

7.3 权威状态应用

客户端收到 State 后，会把网络包还原成 WorldSnapshot。

然后做几件事：

• 校验 stateHash
• 判断本地玩家是否偏离权威状态
• 把权威快照写入 StateHistory
• 从权威快照恢复
• 重放后续输入到当前 tick

注意：

项目只用本地玩家差异来判断是否计入 rollback。

原因是：

远端玩家本来就是预测的，如果把远端误差也算进回滚触发条件，客户端会频繁回滚。

7.4 为什么即使没有计入 rollback 也要 rebase

代码里有一个重要设计：

即使本地玩家误差没有超过阈值，客户端仍然会从权威快照 rebase，然后 replay。

这样做的原因是：

• 远端玩家状态应该尽快采用权威值
• 子弹状态也应该采用权威值
• 不然画面可能长期偏离真实世界

也就是说：

• rollbackCount 只是统计“本地玩家明显错误”的次数
• rebase + replay 是状态校正流程本身

8. 网络协议设计

核心文件：

• include/lab/net/Packets.h
• src/net/NetCode.cpp

当前协议版本：

1

kVersion = 3

协议包类型：

8.1 二进制编解码

项目手写了二进制协议。

特点：

• 使用固定 magic
• 使用协议 version
• 使用 packet type
• 整数统一转网络字节序
• 64 位整数拆成两个 32 位写入
• float 不直接传，而是转毫米整数

例如状态里的位置：

1

ps.x_mm = std::lround(wp.x * 1000.0f);

接收端还原：

1

float x = x_mm / 1000.0f;

这比直接传 float 更稳定，也更容易做 hash 对账。

8.2 为什么要有 magic 和 version

magic 用来识别是不是本协议的数据包。

version 用来防止不同协议版本之间误解码。

这在 AI 服务协议里也很重要。

例如模型推理请求如果格式变了，最好也有：

• api version
• schema version
• feature version
• model version

否则新旧客户端混用时很难排查问题。

8.3 Input 包为什么带冗余

UDP 不保证可靠。

如果每个输入只发一次：

• 丢包就会丢输入
• 服务端只能默认空输入
• 玩家表现会突然停顿

所以 Input 包会带最近 K 帧输入。

这是一种简单可靠的抗丢包策略。

AI 工程里类似的思想是：

• 请求重试带幂等 id
• 消息队列消费保留 offset
• 事件流消费允许重复事件
• 服务端按版本号或序号去重

9. 世界模拟逻辑

核心文件：

src/sim/World.cpp

当前世界是一个顶视角小型“迷宫坦克”。

包含：

• 迷宫地图
• 玩家移动
• 墙体碰撞
• 玩家之间 pushbox 分离
• 子弹发射
• 子弹碰撞
• HP 扣减
• hitstun
• cooldown

9.1 迷宫生成

迷宫由固定 seed 生成。

服务端通过 mazeSeed 下发给客户端。

这样客户端不需要每次接收完整地图，只要用同样 seed 就可以生成同样迷宫。

但项目的 WorldSnapshot 里也保留了迷宫网格。

这样更稳：

• seed 用于重建
• maze 数据用于快照、hash 和回滚

9.2 玩家移动

玩家速度不是瞬间切换，而是用摩擦插值靠近目标速度：

1

v += (targetV - v) * alpha

这样移动更平滑。

但它依然是确定性的，因为每一帧都只由：

• 当前状态
• 当前输入
• 固定 dt

共同决定。

9.3 子弹状态

子弹包含：

• 位置
• 速度
• owner
• life

子弹撞墙或命中玩家后消失。

命中后：

• 被击中玩家 HP 减少
• 进入 Hitstun
• 产生轻微击退

子弹也进入快照、网络包和 hash。

这是必须的。

因为子弹会影响未来游戏状态。

9.4 aimX / aimY 的意义

玩家状态里有：

1
2

int8_t aimX;
int8_t aimY;

它保存最近的瞄准方向。

这个字段很关键。

如果玩家停下来以后再开火，当前 moveX / moveY 可能是 0。

这时开火方向就要依赖上一次方向。

如果回滚恢复时没有保存 aim 方向，重放结果就可能和服务端不同。

这个例子说明：

只要某个字段会影响未来模拟结果，它就必须进入 Snapshot、网络包和 hash。

10. 固定帧推进

服务端和客户端都使用 accumulator 控制固定时间步。

基本结构：

1
2
3
4
5
6

frame = now - prev
acc += min(frame, maxFrame)

while acc >= dt:
  step one tick
  acc -= dt

项目里：

1
2

dt = 1.0 / 60.0
maxFrame = 0.25

为什么不用真实 frameTime 直接推进？

因为真实时间不稳定：

• 系统调度会抖动
• 渲染耗时会变化
• 网络回调时间不可控

固定 dt 的好处是：

• 模拟稳定
• 回滚可重放
• hash 更容易一致
• 测试更容易复现

AI 工程迁移理解：

在 Agent、多轮任务执行、流式处理、训练调度里，也应该尽量区分：

• 真实时间
• 逻辑步数
• 状态版本

系统内部最好基于逻辑 step 推进，而不是让 wall clock 到处影响业务逻辑。

11. 回滚同步完整链路

完整链路可以这样理解。

11.1 正常预测阶段

每个客户端每 tick 做：

1
2
3
4
5
6
7

采样本地输入
保存 localHist[tick]
预测远端输入
World::Step(allCmds, dt)
保存 stateHist[tick]
发送最近 K 帧输入给服务端
tick++

11.2 服务端权威阶段

服务端每 tick 做：

1
2
3
4
5
6
7
8

读取每个玩家输入
缺输入则 hold/default
World::Step(allCmds, dt)
生成权威 snapshot
计算 hash
发送 Ack
按频率发送 State
tick++

11.3 客户端校正阶段

客户端收到权威 State：

1
2
3
4
5
6
7

解码 State
还原 auth snapshot
校验 stateHash
比较本地玩家误差
写入权威快照
Restore(auth)
Replay auth.tick + 1 到当前 tick

这个流程的本质是：

客户端永远允许自己先猜，但最终必须回到服务端认可的历史上。

12. 测试设计

项目测试分两类。

12.1 core_tests

文件：

tests/core_tests.cpp

主要验证：

• InputBuffer 零容量时会被修正为 1
• StateHistory 零容量时会被修正为 1
• InputPacket 编解码正确
• StatePacket 编解码正确
• Hasher 使用毫米精度
• Hasher 覆盖 shotCooldown
• World(0) 会至少创建 1 个玩家

这些是小而关键的回归测试。

12.2 stress_tests

文件：

tests/stress_tests.cpp

压力测试更重要。

它在单进程里模拟：

• 权威世界
• 客户端预测世界
• 输入历史
• 网络包编解码
• State 延迟
• State 抖动
• 回滚重放
• hash 校验

测试重点不是窗口和真实 socket，而是核心逻辑链路。

它验证：

• 从原始权威快照恢复并重放，能追上权威历史
• 从网络量化后的 State 恢复并重放，hash 仍然一致
• 输入冗余包可以正常编解码
• 长时间 tick 推进不会破坏状态一致性

这类测试对 AI 工程也有参考意义：

真正可靠的系统，不只测单个函数，还要测“数据经过网络、延迟、恢复、重放之后是否仍然正确”。

13. 这个项目值得学习的重点知识点

13.1 分层设计

项目没有把所有逻辑写在一个 main 里，而是拆成：

• sim
• net
• app
• tests

这种拆分让核心模拟可以脱离网络和渲染测试。

AI 工程对应：

• 模型调用逻辑不要和 HTTP handler 混在一起
• 数据处理逻辑不要和 UI 混在一起
• 推理服务核心最好能被单元测试和压测直接调用

13.2 状态快照

WorldSnapshot 是这个项目的核心中间结构。

它让系统可以：

• 保存状态
• 传输状态
• 恢复状态
• 比较状态
• hash 状态

AI 工程对应：

• Agent 运行状态
• RAG 查询状态
• 会话上下文快照
• 任务执行 checkpoint
• 分布式训练 checkpoint

13.3 输入日志

项目不是只保存当前输入，而是保存输入历史。

这让系统可以从任意历史点重新计算。

AI 工程对应：

• prompt history
• tool call history
• message queue event log
• workflow step log
• 训练样本处理日志

13.4 确定性推进

同样状态 + 同样输入 + 同样 dt，应该得到同样结果。

这对回滚极其重要。

AI 工程里虽然模型推理本身可能有随机性，但工程层也可以尽量确定：

• 固定 prompt 模板版本
• 固定模型版本
• 固定采样参数
• 固定工具调用输入
• 固定数据预处理逻辑

否则线上问题很难复现。

13.5 网络协议版本化

项目协议有：

• magic
• version
• packet type

这能避免错误解码。

AI 服务接口也应该重视：

• API version
• request schema version
• model version
• embedding version
• prompt version

13.6 冗余和幂等

Input 包重复携带最近 K 帧输入。

这是一种用冗余换可靠性的设计。

AI 工程对应：

• 任务重试要有 task id
• 消息重复消费要能去重
• 请求超时后重发不能重复扣费或重复写库
• 流式输出恢复要能从 offset 继续

13.7 Hash 对账

项目用 hash 判断状态是否一致。

AI 工程也可以用类似思想：

• 对输入文档算 hash
• 对 chunk 结果算 hash
• 对 embedding 输入算 hash
• 对模型配置算 hash
• 对 prompt 模板算 hash
• 对 workflow 状态算 hash

这能帮助定位：

• 是输入变了
• 是模型变了
• 是参数变了
• 是代码逻辑变了
• 是缓存污染了

13.8 压力测试

lab_stress 很值得参考。

它没有依赖窗口和真实 socket，而是直接压核心逻辑。

AI 工程里也可以这样做：

• 不一定先压真实 API
• 可以先压核心 pipeline
• 模拟延迟、乱序、失败和重试
• 验证恢复后结果是否一致

14. 和 AI 模型开发的关联

这个项目虽然不是 AI 项目，但可以帮助理解 AI 工程里的几个难点。

14.1 和推理服务的关系

推理服务也有类似问题：

• 请求并发
• 网络延迟
• 状态追踪
• 版本一致性
• 超时重试
• 结果校验

可以借鉴：

• 每个请求带 request id
• 每个模型有 model version
• 每次调用记录输入 hash
• 缓存 key 包含模型和参数版本
• 异步任务保存 checkpoint

14.2 和 Agent 系统的关系

Agent 执行多步任务时，很像 tick 推进。

每一步可以看成：

1
2
3
4
5
6

step_id
observation
decision
tool_call
tool_result
state_update

如果中途失败，就需要：

• 从某个 step 恢复
• 重放历史
• 判断状态是否一致
• 避免重复执行副作用工具

这和回滚同步的思路很接近。

14.3 和 RAG 系统的关系

RAG 里也需要状态和版本管理。

例如：

• 文档版本
• chunk 版本
• embedding 模型版本
• 向量库索引版本
• 查询改写版本
• reranker 版本

如果没有这些版本，线上效果变化时很难排查。

可以参考这个项目的 hash 思想，对关键中间结果做记录。

14.4 和分布式训练的关系

分布式训练也重视：

• step
• checkpoint
• deterministic replay
• 状态恢复
• 参数同步
• 异常恢复

WorldSnapshot 类似训练 checkpoint。

InputCmd 类似每一步的 batch / event。

StateHistory 类似最近 checkpoint 历史。

Hasher 类似一致性校验。

15. 如果继续完善这个项目

这个项目已经把核心链路跑通了，但如果要继续做，可以考虑这些方向。

15.1 网络层增强

可以增加：

• 延迟统计
• 丢包率统计
• RTT 估计
• 抖动缓冲
• 包乱序处理
• 客户端 ping / pong
• 输入确认窗口

15.2 协议层增强

可以增加：

• 协议 schema 文档
• 包大小统计
• 包字段版本兼容
• State 增量同步
• 压缩
• 加密或签名

15.3 回滚体验增强

可以增加：

• 插值平滑
• 远端玩家视觉修正
• 回滚次数可视化
• 回滚帧数统计
• 输入延迟动态调整

15.4 测试增强

可以增加：

• 真实 UDP bot 压测
• 随机丢包模拟
• 随机乱序模拟
• 多平台 hash 对比
• fuzz 解码测试
• 长时间 soak test

15.5 工程结构增强

可以增加：

• 更明确的协议文档
• 更统一的配置来源
• CI 自动测试
• clang-format
• sanitizers
• 性能 profiling

16. 学习路线建议

如果你要用这个项目作为学习材料，可以按这个顺序看。

16.1 第一遍：先看整体

先读：

• README.md
• docs/ARCHITECTURE.md
• CMakeLists.txt

目标是知道：

• 项目怎么构建
• 有哪些模块
• 服务端和客户端怎么交互

16.2 第二遍：看数据结构

重点看：

• InputCmd.h
• StateSnapshot.h
• InputBuffer.h
• StateHistory.h
• Packets.h

目标是理解：

• 输入怎么表示
• 状态怎么表示
• 网络包怎么表示
• 历史怎么保存

16.3 第三遍：看核心模拟

重点看：

• World.h
• World.cpp
• Rules.h
• Hasher.cpp

目标是理解：

• 一帧怎么推进
• 哪些字段影响确定性
• 为什么 hash 要覆盖这些字段

16.4 第四遍：看服务端

重点看：

• server_main.cpp

目标是理解：

• 玩家怎么分配
• 输入怎么接收
• 缺输入怎么处理
• 权威状态怎么广播

16.5 第五遍：看客户端

重点看：

• client_main.cpp
• InputPrediction.cpp

目标是理解：

• 本地预测怎么做
• 远端输入怎么猜
• 收到权威 State 后怎么回滚
• hash mismatch 怎么统计

16.6 第六遍：看测试

重点看：

• core_tests.cpp
• stress_tests.cpp

目标是理解：

• 这个项目如何验证状态一致性
• 压测为什么不依赖真实窗口和 socket
• 如何模拟延迟和重放

17. 重点代码阅读清单

建议重点阅读这些文件：

1
2
3
4
5
6
7
8
9
10
11

include/lab/sim/InputCmd.h
include/lab/sim/StateSnapshot.h
include/lab/sim/InputBuffer.h
include/lab/sim/StateHistory.h
include/lab/net/Packets.h
src/sim/World.cpp
src/sim/Hasher.cpp
src/net/NetCode.cpp
apps/server_main.cpp
apps/client_main.cpp
tests/stress_tests.cpp

阅读时重点问自己几个问题：

• 每个 tick 的输入从哪里来？
• 服务端和客户端的 tick 如何对齐？
• 客户端什么时候预测？
• 权威状态什么时候覆盖预测状态？
• 什么字段进入 Snapshot？
• 什么字段进入网络包？
• 什么字段进入 hash？
• 如果新增一个状态字段，需要改哪些地方？
• 如果 UDP 丢包，会发生什么？
• 如果 State 延迟到达，客户端如何追上当前 tick？

18. 常见易错点

18.1 新增字段只改了 World，没有改 Snapshot

如果新增字段会影响模拟，但没有进入 Snapshot，回滚恢复后就会丢状态。

18.2 新增字段只改了 Snapshot，没有改网络包

服务端有这个状态，但客户端收不到，预测和权威会分叉。

18.3 新增字段没有进入 hash

状态已经不同，但 hash 检查不出来。

18.4 直接 hash float 内存

不同平台或编解码后可能有微小浮点差异，导致假 mismatch。

项目用毫米量化规避了这个问题。

18.5 环形缓冲读取时不校验 tick

只用 tick % cap 会读到被覆盖的旧数据。

项目里读取时会校验真实 tick，这是正确做法。

18.6 把远端预测误差也算作 rollback 条件

远端玩家本来就是猜的。

如果把远端差异也作为回滚计数依据，会导致 rollback 统计爆炸。

项目只比较本地玩家，比较合理。

19. 可以写成博客的版本

标题

从一个 C++ 格斗游戏 Demo 学实时系统：固定帧、预测、回滚与状态一致性

开头

很多人以为游戏网络同步只是“把坐标发给别人”。

但动作游戏真正困难的地方在于：

• 玩家按键必须立刻有反馈
• 网络包可能延迟或丢失
• 服务端必须保持权威
• 客户端预测错了还要能修回来

这个项目用一个小型 60Hz 顶视角对战 Demo，把这些问题完整串了起来。

第一部分：为什么需要固定帧

实时系统最怕逻辑跟真实时间强绑定。

如果每一帧都直接用真实耗时推进，状态会受到系统调度、渲染耗时和网络回调影响。

所以项目采用固定 tick：

1
2
3

1 秒 60 帧
每帧 dt = 1 / 60
所有输入和状态都绑定 tick

这让回放、回滚和调试都变得可控。

第二部分：客户端为什么要预测

如果玩家按键后必须等服务端确认，游戏会非常卡。

所以客户端先假设自己的输入有效，立即推进本地世界。

这样玩家看到的是即时反馈。

但客户端不是最终权威。

服务端稍后会下发真实状态，客户端再根据权威状态修正。

第三部分：回滚的本质

回滚不是简单地把当前位置拉回服务端位置。

真正的流程是：

1
2
3

恢复到服务端给的历史快照
重新执行这之后的输入
追上当前本地 tick

这要求系统必须保存：

• 历史输入
• 历史状态
• 可恢复的快照
• 确定性的 Step 函数

第四部分：为什么 hash 很重要

服务端和客户端状态一旦分叉，肉眼很难定位原因。

项目用 Hasher 对关键状态做 hash。

只要 hash 不一致，就说明某个状态字段不同。

更重要的是，hash 不是直接混合 float，而是和网络包一样做毫米量化。

这样可以避免浮点误差造成误报。

第五部分：这个项目对 AI 工程的启发

虽然这是游戏项目，但它的工程思想非常适合迁移到 AI 系统。

AI Agent 也有 step。

RAG 也有中间状态。

推理服务也有请求版本、模型版本和缓存一致性。

分布式训练也有 checkpoint 和恢复。

如果一个 AI 系统需要稳定运行，就不能只关心模型输出，还要关心：

• 输入是否可追踪
• 中间状态是否可恢复
• 版本是否明确
• 错误是否可复现
• 结果是否可校验

结尾

这个 Fighting 项目真正值得学习的，不只是 Rollback Netcode，而是它展示了一套实时系统的基本工程方法：

• 用 tick 管理时间
• 用输入日志支撑回放
• 用快照支撑恢复
• 用 hash 支撑一致性检查
• 用压力测试验证长链路正确性

这些能力放到 AI 模型开发中，同样非常重要。

20. 可扩展博客选题

后续可以基于这份笔记拆成几篇博客。

20.1 选题一：Rollback Netcode 入门

重点讲：

• 为什么动作游戏不能只靠服务端确认
• 本地预测是什么
• 回滚重放是什么
• 输入历史和状态快照怎么设计

20.2 选题二：用 C++ 实现一个确定性模拟核心

重点讲：

• World::Step
• 固定 dt
• 状态字段管理
• 碰撞和子弹逻辑
• 如何避免不确定性

20.3 选题三：UDP 协议如何对抗丢包

重点讲：

• 为什么用 UDP
• Input 包为什么带冗余
• 服务端如何 hold last input
• Ack 和 State 分别解决什么问题

20.4 选题四：状态 hash 如何定位分布式系统分叉

重点讲：

• 为什么需要 hash
• 为什么不能直接 hash float
• 哪些字段必须进入 hash
• hash mismatch 如何排查

20.5 选题五：从游戏同步看 AI Agent 状态恢复

重点讲：

• tick 和 Agent step 的类比
• InputCmd 和 tool call 的类比
• Snapshot 和 checkpoint 的类比
• Replay 和失败恢复的类比

21. 总结

这个项目可以用一句话总结：

用 C++ 实现了一个小型实时联机同步系统，通过固定帧、输入冗余、本地预测、服务端权威、回滚重放和 hash 对账来解决网络延迟下的状态一致性问题。

最应该记住的重点：

• Tick 是逻辑时间基础
• InputCmd 是最小输入事件
• WorldSnapshot 是恢复和同步的核心
• World::Step 必须尽量确定性
• InputBuffer 和 StateHistory 用环形缓冲保存历史
• 客户端预测是为了体验，服务端权威是为了正确
• 回滚不是瞬移，而是恢复历史状态后重放输入
• hash 对账要和网络量化精度一致
• 压力测试应该覆盖完整链路，而不是只测单个函数

放到 AI 模型开发学习里，这个项目最有价值的地方是：

它训练的是工程化思维：状态、版本、恢复、重放、对账、压测。

Qwen 模型演进笔记

1. 先说明“最新模型”指的是谁

截至 2026-04-20，如果说“Qwen 最新模型”，其实要分两类看：

• 最新通用 API 旗舰模型：Qwen3.6-Plus
  发布时间：2026-04-02
• 最新开源主力模型：Qwen3.6-35B-A3B
  发布时间：2026-04-17

如果你更关心：

• 云端调用、Agent、编程能力：重点看 Qwen3.6-Plus
• 本地部署、开源权重、性价比：重点看 Qwen3.6-35B-A3B

2. Qwen2.5 是什么

Qwen2.5 是 Qwen 在 2024-09-19 发布的一代重要开源模型系列。

这一代的定位可以理解成：

把“通用聊天模型”真正做到了更稳、更全、更适合工程落地。

Qwen2.5 这一代包含：

• 通用语言模型：Qwen2.5
• 代码模型：Qwen2.5-Coder
• 数学模型：Qwen2.5-Math

开源尺寸覆盖比较广：

• 0.5B
• 1.5B
• 3B
• 7B
• 14B
• 32B
• 72B

3. Qwen2.5 的主要改进点

根据官方介绍，Qwen2.5 相比 Qwen2 的核心提升主要在下面几块：

3.1 知识、代码、数学能力更强

官方提到 Qwen2.5 基于最多 18T tokens 的大规模数据训练，相比 Qwen2：

• 知识能力更强
• 代码能力提升明显
• 数学能力提升明显