数据处理与数据集评估笔记

1. 为什么数据决定模型上限

对句子嵌入、检索召回和 RAG 场景来说，模型当然重要，但很多时候真正决定上限的是数据。

常见原因有：

• 正样本质量决定模型学到什么叫“相关”
• 负样本质量决定模型能否真正学会区分
• 样本分布决定模型上线后是否能泛化
• 标签噪声决定训练信号是否稳定
• 评估集质量决定你是否能看清真实问题

先记住一个判断：模型是在数据里学规律。如果数据本身混乱、偏斜或失真，训练出来的结果大概率也会带着同样的问题。

2. 数据处理到底包括什么

很多人一提数据处理，只想到“删乱码、去空值”，但在 embedding 项目里，数据处理通常至少包括下面这些部分：

• 数据采集
• 样本构造
• 数据清洗
• 去重和归一化
• 标签治理
• 样本分布分析
• 训练 / 验证 / 测试切分
• 评估集构建
• 数据版本管理

换个更完整的说法：

数据处理 = 数据工程 + 数据质量控制 + 数据评估设计。

3. 先定义样本单位，再谈处理策略

在做数据处理之前，第一件事不是清洗，而是先明确“一个样本到底是什么”。

不同任务里，样本单位并不一样：

3.1 相似度任务

通常是句子对：

• sentence_a
• sentence_b
• label

3.2 检索任务

通常是：

• query
• positive
• negative 或 hard_negative

3.3 文档向量库构建

通常是：

• doc_id
• chunk_id
• text
• metadata

如果样本单位没定义清楚，后面很多统计都会失真。
比如你以“句子对”为单位去算分布，和以“query”为单位去算分布，结论会完全不同。

4. 数据处理的主线流程

一个比较稳的流程通常是：

1. 明确任务目标和样本单位
2. 采集原始数据
3. 构造正样本、负样本、hard negative
4. 做文本清洗和标准化
5. 做去重、去模板、去噪
6. 分析样本分布和标签分布
7. 切分 train / valid / test
8. 构建标准评估集和 badcase 集
9. 训练第一版模型
10. 用 badcase 反推数据问题，再回流修正

这条链路里，最容易被忽视的不是清洗，而是：

• 分布分析
• 泄漏排查
• 评估集设计

5. 数据清洗要做什么

5.1 基础清洗

最基础的一层通常包括：

• 去空文本
• 去乱码
• 去无意义符号
• 去极短文本
• 去极长文本
• 统一空白符
• 统一全角半角
• 统一大小写

如果是中文场景，还经常要考虑：

• 简繁统一
• 标点统一
• URL、邮箱、手机号脱敏
• 时间、数字、金额是否归一化

5.2 噪声清洗

很多业务数据不是“脏”，而是“有大量无效结构”。

例如：

• 页眉页脚
• 导航栏
• 广告词
• 模板开场白
• 重复免责声明
• OCR 识别错误文本

这些内容如果不清掉，会让模型学到大量无关模式。

5.3 去重

去重非常重要，因为重复样本会让训练和评估都失真。

常见去重对象：

• 完全重复文本
• 近重复文本
• 同一 query 的重复点击对
• 同一文档切块后的重复片段

如果不做去重，常见后果有：

• 模型过度拟合高频表达
• 评估结果虚高
• 数据规模看起来很大，但有效信息量并不高

6. 样本分布为什么必须看

很多训练失败，不是模型没学会，而是数据分布本身有问题。

至少要看下面几类分布。

6.1 文本长度分布

要看：

• query 长度分布
• document / passage 长度分布
• chunk 长度分布

如果训练集里大部分文本很短，但线上是长 query 或长段落，模型上线后就容易掉点。

6.2 标签分布

例如：

• 正负样本比例
• 2 / 1 / 0 多档标签比例
• query 是否只有正样本没有负样本

如果标签极度失衡，模型往往会学到偏置策略，而不是学到真正的语义边界。

6.3 场景分布

要看数据是否覆盖不同业务场景，例如：

• FAQ
• 搜索
• 商品检索
• 知识库问答
• 代码检索

如果训练样本只覆盖热门场景，长尾场景通常会明显掉效果。

6.4 热门与长尾分布

要区分：

• 高频 query
• 长尾 query
• 高频文档
• 稀有文档

如果训练集被热门 query 主导，模型可能对高频表达很强，但对长尾表达泛化很差。

6.5 正样本来源分布

正样本如果同时来自多种来源，例如：

• 点击日志
• 人工标注
• FAQ 映射
• 改写生成

最好分别统计占比，因为不同来源噪声强度不一样。

6.6 负样本难度分布

不能只看负样本数量，还要看难度结构。

通常可以分成：

• 随机负样本
• 同域负样本
• hard negative

如果全是简单负样本，模型离线 loss 可能很好看，但上线区分能力依然不够。

7. 标签质量怎么检查

数据量大不代表标签质量高。

标签问题通常来自：

• 点击不等于相关
• 曝光位置偏置
• 人工标注标准不一致
• 自动构造规则太粗糙
• 历史系统的错误被继承进新数据

可以从下面几方面检查：

7.1 抽样人工复核

最直接也最有效。

重点抽样：

• 高频 query
• 长尾 query
• 高相似但错召回样本
• 模型和标签明显冲突的样本

7.2 看标签一致性

例如同一个 query 是否出现下面这种情况：

• 同一文档有时标正样本
• 有时又标成负样本

这种冲突标签会直接污染训练信号。

7.3 看模型反常样本

如果一个样本长期表现为：

• 标签是正样本，但模型始终给很低分
• 标签是负样本，但模型和人工都觉得很相关

那就要怀疑数据或标签本身有问题，而不是先怀疑模型。

8. 正样本和负样本要怎么治理

8.1 正样本治理

正样本要优先保证“真相关”。

对于日志数据，常见过滤规则包括：

• 去掉极短停留
• 去掉误点击
• 去掉曝光高但满意度低的内容
• 去掉跳出率特别高的点击对

如果是结构化映射数据，还要检查：

• 映射关系是否过期
• 标题和正文是否错配
• 类目标签是否长期维护

8.2 负样本治理

负样本的核心不是“越多越好”，而是“越像线上干扰项越好”。

常见做法：

• 先用随机负样本保证基础区分
• 再补同域负样本
• 最后重点挖 hard negative

8.3 Hard Negative 是最值得花时间的部分

hard negative 常见来源：

• BM25 高分误召回
• 旧 embedding 模型的错召回
• Cross-Encoder 高混淆样本
• 人工 badcase 回流

很多时候，模型的真实提升不来自更多数据，而来自更强的 hard negative。

9. 训练集、验证集、测试集怎么切

切分数据时，重点不是比例，而是避免数据泄漏。

9.1 常见切分比例

• 8:1:1
• 9:0.5:0.5

9.2 更重要的是避免泄漏

例如：

• 同一问题的改写版本，不要分到不同集合
• 同一文档的多个 chunk，不要乱落到 train 和 test
• 同一用户会话，不要一部分在训练、一部分在测试
• 同一商品不同标题变体，不要跨集合乱分

如果泄漏存在，评估结果通常会明显虚高。

9.3 切分方式最好按业务实体来做

例如按下面这些维度切，通常更稳：

• 按 query_id
• 按 doc_id
• 按会话
• 按用户

而不是简单随机切行。

10. 评估集应该怎么设计

评估集不是简单从训练数据里抽一点出来，而是要刻意设计。

一个好的评估集通常要满足：

• 覆盖核心业务场景
• 覆盖热门 query 和长尾 query
• 覆盖高混淆样本
• 尽量有人工校验
• 分布尽量接近线上真实请求

10.1 标准评估集

这是每次训练都要跑的稳定基线。

要求：

• 版本稳定
• 标注质量高
• 用来做横向对比

10.2 Badcase 集

这是最重要的补充集。

来源通常是：

• 线上错召回
• 用户投诉
• 人工复核发现的高价值错误
• 新业务新术语

badcase 集的作用不是看平均分，而是防止关键错误反复出现。

10.3 分层评估

如果场景复杂，建议把评估集再分层：

• 热门 query 集
• 长尾 query 集
• 专业术语集
• 否定表达集
• 数字、时间、版本号敏感集

这样你才能知道模型到底是“整体提升”，还是只在某一类场景提升。

11. 数据集评估到底评什么

除了看模型指标，还应该单独评估数据集本身。

至少可以看下面几类指标。

11.1 规模指标

• 样本总量
• 唯一 query 数
• 唯一 doc 数
• 平均每个 query 的正样本数
• 平均每个 query 的负样本数

11.2 分布指标

• 文本长度分布
• 标签分布
• 场景分布
• 来源分布
• 热门 / 长尾分布

11.3 质量指标

• 重复率
• 近重复率
• 空文本比例
• 模板文本比例
• 乱码比例
• 标签冲突率

11.4 难度指标

• hard negative 占比
• 高混淆样本占比
• 多跳或长文本样本占比

如果这些指标长期不看，训练往往会变成“看 loss 猜问题”。

12. 一个很实用的数据诊断思路

如果模型效果不好，可以先按下面顺序排查：

1. 看训练和测试是否泄漏
2. 看是否有大量重复样本
3. 看标签是否冲突
4. 看负样本是不是太简单
5. 看热门样本是否占比过高
6. 看长尾 query 是否覆盖不足
7. 看评估集是否真的接近线上

这条排查顺序很实用，因为很多问题根本不是模型结构导致的。

13. 数据版本管理为什么重要

如果你每次训练都在用“最新数据”，但没有快照和版本号，后面几乎一定会遇到这类问题：

• 效果变好了，不知道是哪批数据带来的
• 效果变差了，不知道是哪次清洗规则改坏了
• badcase 修好了，但下个版本又回来了

所以至少要给下面这些东西做版本管理：

• 原始数据快照
• 清洗规则版本
• 样本构造规则版本
• hard negative 生成版本
• 评估集版本
• badcase 集版本

14. 一个适合落地的默认做法

如果你现在要从零做一套数据处理和数据集评估流程，可以先按这个 baseline 走：

1. 先定义样本单位和标签规范
2. 从业务日志和结构化关系里构造正样本
3. 补随机负样本、同域负样本和 hard negative
4. 做基础清洗、去重和归一化
5. 统计长度分布、标签分布、场景分布和来源分布
6. 按 query_id 或 doc_id 切 train / valid / test
7. 建一份小而稳定的标准评估集
8. 再维护一份持续追加的 badcase 集
9. 每次训练都固定跑离线评估和分层评估
10. 把线上 badcase 持续回流

这套流程不花哨，但很稳，也最容易持续迭代。

15. 一份可直接抄的检查清单

1. 是否明确了样本单位
2. 是否定义了标注规则
3. 是否做了基础清洗和噪声清洗
4. 是否做了完全去重和近重复检查
5. 是否统计了长度分布和标签分布
6. 是否区分了热门和长尾样本
7. 是否统计了正样本和负样本来源
8. 是否有足够强的 hard negative
9. 是否避免了 train / test 泄漏
10. 是否有稳定的评估集和 badcase 集
11. 是否给数据和规则做了版本管理

16. 一句话总结

数据处理和数据集评估的核心不是“把脏数据洗干净”，而是：

让训练数据更真实、分布更合理、标签更稳定、评估更可信。

对 embedding 项目来说，很多时候真正决定模型上限的，不是你换了多大的模型，而是你把数据治理做到了什么程度。

Drogon 学习笔记

1. Drogon 是什么

Drogon 是一个基于 C++ 的现代 Web 应用框架，适合用来开发：

• 高性能 HTTP API 服务
• C++ 网关服务
• 需要低延迟的业务接口
• WebSocket 服务
• 需要接 MySQL / PostgreSQL / Redis 的后端系统

它的几个核心特点：

• 性能高，底层是非阻塞 IO
• 支持异步编程
• 支持路由、控制器、过滤器、中间件
• 支持数据库、Redis、Session、文件上传
• 支持 HTTP Client，也能调用其他内部服务
• 支持 C++20 协程

如果先抓核心区别，可以这样记：FastAPI 更偏 Python 生态里的高效 API 框架，Drogon 则更像适合高性能 C++ 服务的全功能 Web 框架。

2. 为什么很多后端项目会用 Drogon

如果你的系统有这些特点：

• 接口并发高
• 对延迟敏感
• 业务层想用 C++ 实现
• 需要把网络、数据库、缓存、会话统一放在一个服务里

那 Drogon 会很合适。

尤其是在你现在这类 AI 项目架构里，Drogon 很适合承担：

• 对外 API 网关
• 会话和鉴权层
• 请求转发层
• MySQL 持久化层
• Redis 状态层

而 Python 生态仍然更适合承担：

• 模型推理
• 文档解析
• RAG 流程
• Celery 异步任务

因此，很多混合架构会拆成：

• Drogon：对外服务、状态治理、性能敏感层
• FastAPI：内部模型服务、任务接口
• Celery：后台异步执行

3. 安装和工程初始化

Drogon 的安装方式比 FastAPI 重一些，因为它是 C++ 框架，需要编译环境和依赖。

常见方式有：

• vcpkg
• 源码编译
• Docker

如果你想先快速上手，最常见的是源码安装：

1
2
3
4
5
6
7
8

git clone https://github.com/drogonframework/drogon.git
cd drogon
git submodule update --init
mkdir build
cd build
cmake ..
make -j4
sudo make install

安装完成后，通常还会有一个命令行工具：

1

drogon_ctl

它常用来：

• 创建项目骨架
• 创建 Controller
• 创建 Filter
• 生成 ORM Model

例如创建一个项目：

1

drogon_ctl create project demo

4. 最小可运行示例

和 FastAPI 的 main.py 一样，Drogon 也可以先从一个最小服务开始。

4.1 CMakeLists.txt

1
2
3
4
5
6
7
8
9
10

cmake_minimum_required(VERSION 3.16)
project(drogon_demo)

set(CMAKE_CXX_STANDARD 17)
set(CMAKE_CXX_STANDARD_REQUIRED ON)

find_package(Drogon CONFIG REQUIRED)

add_executable(${PROJECT_NAME} main.cc)
target_link_libraries(${PROJECT_NAME} PRIVATE Drogon::Drogon)

4.2 main.cc

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

#include <drogon/drogon.h>

using namespace drogon;

int main()
{
    app().registerHandler(
        "/",
        [](const HttpRequestPtr &req,
           std::function<void(const HttpResponsePtr &)> &&callback) {
            Json::Value json;
            json["message"] = "hello drogon";
            callback(HttpResponse::newHttpJsonResponse(json));
        },
        {Get});

    app().addListener("0.0.0.0", 8080);
    app().setThreadNum(1);
    app().run();
}

编译和运行：

1
2
3

cmake -S . -B build
cmake --build build
./build/drogon_demo

访问：

• http://127.0.0.1:8080/

如果对照 FastAPI，可以这样看：

• app = FastAPI() 对应 app()
• @app.get("/") 对应 registerHandler("/", ...)
• return {...} 对应 callback(HttpResponse::newHttpJsonResponse(...))

5. Drogon 的基本工作方式

可以把一次请求理解成下面这条链路：

1. 客户端发来 HTTP 请求
2. Drogon 根据路径和方法匹配路由
3. 先经过中间件和过滤器
4. 进入 Handler 或 Controller 方法
5. 业务代码处理请求
6. 构造 HttpResponse
7. 通过 callback 返回给客户端

和 FastAPI 很像，但有一个非常重要的区别：

Drogon 不会像 FastAPI 那样自动把 Python 函数返回值转成 JSON，你需要更显式地构造响应对象。

6. 直接注册 Handler

最简单的方式就是直接在 main() 里注册路由。

1
2
3
4
5
6
7
8
9

app().registerHandler(
    "/health",
    [](const HttpRequestPtr &req,
       std::function<void(const HttpResponsePtr &)> &&callback) {
        Json::Value json;
        json["status"] = "ok";
        callback(HttpResponse::newHttpJsonResponse(json));
    },
    {Get});

适合：

• 学习阶段
• 很小的 demo
• 临时测试接口

但项目一大，通常更推荐 Controller。

7. 三种常见 Controller

Drogon 常见的控制器主要有三类：

• HttpSimpleController
• HttpController
• WebSocketController

7.1 HttpSimpleController

适合单个简单接口。

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

#include <drogon/HttpSimpleController.h>

class Ping : public drogon::HttpSimpleController<Ping>
{
  public:
    PATH_LIST_BEGIN
    PATH_ADD("/ping", drogon::Get);
    PATH_LIST_END

    void asyncHandleHttpRequest(
        const drogon::HttpRequestPtr &req,
        std::function<void(const drogon::HttpResponsePtr &)> &&callback) override
    {
        Json::Value json;
        json["message"] = "pong";
        callback(drogon::HttpResponse::newHttpJsonResponse(json));
    }
};

7.2 HttpController

适合一组 REST 风格接口，功能更完整。

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

#include <drogon/HttpController.h>

namespace api::v1
{
class User : public drogon::HttpController<User>
{
  public:
    METHOD_LIST_BEGIN
    ADD_METHOD_TO(User::getUser, "/api/v1/users/{id}", drogon::Get);
    ADD_METHOD_TO(User::createUser, "/api/v1/users", drogon::Post);
    METHOD_LIST_END

    void getUser(
        const drogon::HttpRequestPtr &req,
        std::function<void(const drogon::HttpResponsePtr &)> &&callback,
        int id) const
    {
        Json::Value json;
        json["id"] = id;
        callback(drogon::HttpResponse::newHttpJsonResponse(json));
    }

    void createUser(
        const drogon::HttpRequestPtr &req,
        std::function<void(const drogon::HttpResponsePtr &)> &&callback) const
    {
        auto jsonPtr = req->getJsonObject();

        Json::Value json;
        json["name"] = jsonPtr ? (*jsonPtr)["name"].asString() : "";
        callback(drogon::HttpResponse::newHttpJsonResponse(json));
    }
};
}  // namespace api::v1

7.3 WebSocketController

适合：

• 实时聊天
• 实时协作
• 推送类业务

如果你后面要做双向实时通信，可以重点看这一类；如果只是 AI 输出流式结果，很多时候 SSE 就够了，不一定非要 WebSocket。

8. 路径参数、查询参数、JSON 请求体

这部分可以类比 FastAPI 的：

• 路径参数
• 查询参数
• 请求体

8.1 路径参数

在 Drogon 里，路径参数可以直接映射到函数参数。

1

ADD_METHOD_TO(BookController::getBook, "/api/v1/books/{book_id}", drogon::Get);

对应的处理函数：

1
2
3
4

void getBook(
    const drogon::HttpRequestPtr &req,
    std::function<void(const drogon::HttpResponsePtr &)> &&callback,
    int bookId) const;

放到 FastAPI 的语境里，大致对应：

1
2
3

@app.get("/books/{book_id}")
def get_book(book_id: int):
    ...

8.2 查询参数

查询参数通常从 req 里读取：

1
2

auto keyword = req->getParameter("keyword");
auto page = req->getParameter("page");

例如：

1

/api/v1/search?keyword=llm&page=1

8.3 JSON 请求体

如果客户端发的是 JSON，请求体常见写法是：

1
2
3
4
5
6
7
8
9
10

auto jsonPtr = req->getJsonObject();
if (!jsonPtr)
{
    auto resp = drogon::HttpResponse::newHttpResponse();
    resp->setStatusCode(drogon::k400BadRequest);
    callback(resp);
    return;
}

std::string name = (*jsonPtr)["name"].asString();

结论：

• 路径参数：函数参数映射
• 查询参数：req->getParameter()
• JSON 请求体：req->getJsonObject()

9. JSON 响应和状态码

Drogon 里最常见的响应方式是手动构造 JSON。

1
2
3
4
5
6
7

Json::Value json;
json["message"] = "created";
json["id"] = 1001;

auto resp = drogon::HttpResponse::newHttpJsonResponse(json);
resp->setStatusCode(drogon::k201Created);
callback(resp);

这和 FastAPI 的差异在于：

• FastAPI 更自动
• Drogon 更显式

显式的好处是：

• 更清楚知道返回了什么
• 更方便做细粒度控制
• 更符合 C++ 项目风格

10. 配置文件 config.json

很多 Drogon 项目不会把所有配置都写死在 main() 里，而是放到配置文件。

最常见的主程序写法是：

1
2
3
4
5
6
7

#include <drogon/drogon.h>

int main()
{
    drogon::app().loadConfigFile("config.json");
    drogon::app().run();
}

一个简化版配置示意：

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

{
  "listeners": [
    {
      "address": "0.0.0.0",
      "port": 8080
    }
  ],
  "db_clients": [
    {
      "name": "default",
      "rdbms": "mysql",
      "host": "127.0.0.1",
      "port": 3306,
      "dbname": "demo",
      "user": "root",
      "passwd": "123456",
      "is_fast": false,
      "number_of_connections": 4
    }
  ],
  "redis_clients": [
    {
      "name": "default",
      "host": "127.0.0.1",
      "port": 6379,
      "number_of_connections": 2
    }
  ],
  "app": {
    "number_of_threads": 4,
    "enable_session": true,
    "upload_path": "uploads",
    "client_max_body_size": "50M"
  }
}

这里最重要的点是：

• listeners：监听地址和端口
• db_clients：数据库连接
• redis_clients：Redis 连接
• app：线程数、Session、上传路径等全局设置

11. 过滤器和中间件

这部分很像 FastAPI 的依赖、鉴权逻辑、中间件，但 Drogon 拆得更明确。

简单理解：

• Filter：更像“接口前置校验”
• Middleware：更像“围绕请求处理链做前后处理”

例如做一个简单鉴权 Filter：

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

#include <drogon/HttpFilter.h>

class AuthFilter : public drogon::HttpFilter<AuthFilter>
{
  public:
    void doFilter(const drogon::HttpRequestPtr &req,
                  FilterCallback &&fcb,
                  FilterChainCallback &&fccb) override
    {
        auto token = req->getHeader("x-api-key");
        if (token != "secret")
        {
            Json::Value json;
            json["error"] = "unauthorized";

            auto resp = drogon::HttpResponse::newHttpJsonResponse(json);
            resp->setStatusCode(drogon::k401Unauthorized);
            fcb(resp);
            return;
        }

        fccb();
    }
};

挂到某个接口上：

1

ADD_METHOD_TO(User::getUser, "/api/v1/users/{id}", drogon::Get, "AuthFilter");

适合抽出来做的逻辑通常有：

• 登录校验
• API Key 校验
• 频率限制
• 内网访问限制
• 统一日志

12. Session

Drogon 内置了 Session 支持，但默认不是强制开启的。

它适合做：

• 登录态
• 短期会话数据
• 访问频率控制
• 一些轻量服务端状态

开启方式可以在配置文件里写：

1
2
3
4
5

{
  "app": {
    "enable_session": true
  }
}

或者代码里显式开启：

1

drogon::app().enableSession(1200);

使用示例：

1
2
3
4
5
6

req->session()->insert("user_id", 1001);

if (req->session()->find("user_id"))
{
    auto userId = req->session()->get<int>("user_id");
}

注意：

Session 依赖 Cookie。如果客户端不保存 Cookie，每次请求都可能被当成新会话。

13. 文件上传

在 AI 项目里，文件上传非常常见，比如：

• 上传 PDF
• 上传图片
• 上传音频
• 上传数据文件

Drogon 常用 MultiPartParser 来处理 multipart 上传。

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

#include <drogon/drogon.h>

void upload(const drogon::HttpRequestPtr &req,
            std::function<void(const drogon::HttpResponsePtr &)> &&callback)
{
    drogon::MultiPartParser parser;
    if (parser.parse(req) != 0 || parser.getFiles().empty())
    {
        Json::Value json;
        json["error"] = "no file";

        auto resp = drogon::HttpResponse::newHttpJsonResponse(json);
        resp->setStatusCode(drogon::k400BadRequest);
        callback(resp);
        return;
    }

    auto file = parser.getFiles()[0];
    file.save("./uploads");

    Json::Value json;
    json["filename"] = file.getFileName();
    auto resp = drogon::HttpResponse::newHttpJsonResponse(json);
    callback(resp);
}

实际项目里要特别注意：

• 不要直接信任原始文件名
• 要限制文件大小
• 要校验扩展名和 MIME 类型
• 最好自己生成保存文件名

对于 AI 系统，更推荐的流程通常是：

1. Drogon 接收上传
2. 保存到磁盘或对象存储
3. 写一条数据库记录
4. 调内部 FastAPI / Celery 提交处理任务

14. 数据库访问：DbClient 和 ORM

Drogon 自带数据库支持，常见关系型数据库包括：

• MySQL
• PostgreSQL
• SQLite3

最常见的是通过 DbClient 执行 SQL。

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

auto client = drogon::app().getDbClient();

client->execSqlAsync(
    "select id, name from users where id = ?",
    [callback](const drogon::orm::Result &result) {
        Json::Value json;

        if (result.empty())
        {
            json["error"] = "not found";
            auto resp = drogon::HttpResponse::newHttpJsonResponse(json);
            resp->setStatusCode(drogon::k404NotFound);
            callback(resp);
            return;
        }

        json["id"] = result[0]["id"].as<int>();
        json["name"] = result[0]["name"].as<std::string>();
        callback(drogon::HttpResponse::newHttpJsonResponse(json));
    },
    [callback](const drogon::orm::DrogonDbException &e) {
        Json::Value json;
        json["error"] = e.base().what();
        auto resp = drogon::HttpResponse::newHttpJsonResponse(json);
        resp->setStatusCode(drogon::k500InternalServerError);
        callback(resp);
    },
    1001);

要点：

• MySQL 占位符通常用 ?
• PostgreSQL 占位符通常用 $1、$2
• 异步接口更符合 Drogon 的整体风格
• 不要在回调里做很重的阻塞操作

如果你不想手写太多 SQL，还可以用 drogon_ctl 根据表结构生成 ORM Model。

适合：

• 表结构比较稳定
• 想减少手写样板代码
• 想把数据库访问写得更结构化

15. Redis 支持

Drogon 也支持 Redis，而且同样是异步风格。

如果已经在 config.json 里配置了 redis_clients，运行后就可以获取客户端：

1

auto redisClient = drogon::app().getRedisClient();

执行命令示例：

1
2
3
4
5
6
7
8
9

redisClient->execCommandAsync(
    [](const drogon::nosql::RedisResult &r) {
        LOG_INFO << "redis ok: " << r.asString();
    },
    [](const std::exception &err) {
        LOG_ERROR << "redis error: " << err.what();
    },
    "get session:%s",
    "1001");

它适合做：

• 缓存
• 任务状态存储
• 限流计数器
• 会话辅助状态
• 流式输出中间层

如果把你现在的整条链路连起来理解，就是：

• Drogon：对外接口
• Redis：缓存 / 状态 / 消息中间层
• FastAPI / Celery：后台处理

16. C++20 协程 Task<>

如果你用的是较新的编译器，Drogon 支持协程，这可以让异步代码更接近同步写法。

例如数据库查询可以写成：

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

#include <drogon/drogon.h>

drogon::Task<drogon::HttpResponsePtr> getUserCoro(int userId)
{
    auto db = drogon::app().getDbClient();

    try
    {
        auto result = co_await db->execSqlCoro(
            "select id, name from users where id = ?",
            userId);

        Json::Value json;
        json["id"] = result[0]["id"].as<int>();
        json["name"] = result[0]["name"].as<std::string>();
        co_return drogon::HttpResponse::newHttpJsonResponse(json);
    }
    catch (const drogon::orm::DrogonDbException &e)
    {
        Json::Value json;
        json["error"] = e.base().what();
        auto resp = drogon::HttpResponse::newHttpJsonResponse(json);
        resp->setStatusCode(drogon::k500InternalServerError);
        co_return resp;
    }
}

它的好处是：

• 少写回调嵌套
• 控制流更清晰
• 复杂异步逻辑更好维护

但你要注意：

• 需要合适的编译器和 C++20 支持
• 协程让代码更好写，不代表可以随便阻塞线程

17. Drogon 和 FastAPI 怎么选

这两个框架不是简单的“谁替代谁”，而是各自适合不同层。

如果只抓结论，可以这样选：

• 想更快把模型接口和内部服务跑起来，用 FastAPI
• 想把网关、状态治理和性能敏感层放在 C++ 里，用 Drogon

18. Drogon + FastAPI + Celery 的典型配合方式

这是你当前项目最值得掌握的一块。

一个典型流程可以这样拆：

1. 客户端请求先到 Drogon
2. Drogon 做鉴权、Session、参数检查、落库
3. Drogon 调用内部 FastAPI 接口提交任务
4. FastAPI 再把重任务交给 Celery
5. Celery Worker 真正处理文档、推理、索引构建
6. 结果写回 MySQL / Redis
7. Drogon 再把状态或结果返回给前端

你可以把它们理解成固定分工：

• Drogon：网关和业务编排层
• FastAPI：Python 服务入口
• Celery：后台任务执行层
• Redis：状态、缓存、队列中间层
• MySQL：长期持久化

这套架构特别适合：

• 文档上传处理
• RAG 建库
• 长耗时模型推理
• 流式输出聊天系统

19. 一个比较常见的项目结构

如果用 Drogon 做 C++ 网关，一个比较实用的结构可以是：

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

cpp_gateway/
├── CMakeLists.txt
├── config.json
├── main.cc
├── controllers/
│   ├── HealthController.h
│   ├── HealthController.cc
│   ├── DocumentsController.h
│   └── DocumentsController.cc
├── filters/
│   ├── AuthFilter.h
│   └── AuthFilter.cc
├── services/
│   ├── TaskService.h
│   ├── TaskService.cc
│   ├── PythonClient.h
│   └── PythonClient.cc
├── repositories/
│   ├── DocumentRepo.h
│   └── DocumentRepo.cc
└── models/

建议职责划分：

• controllers/：只处理 HTTP 层
• filters/：鉴权、限流、前置校验
• services/：业务编排、调内部服务
• repositories/：数据库读写
• models/：ORM 生成模型或业务数据结构

也就是说：

不要把所有逻辑都堆在 Controller 里。

20. 常见坑

20.1 在 Handler 里做阻塞重活

例如：

• 大文件解析
• 模型推理
• 长时间等待第三方服务

这些都不应该直接堵在 Drogon 的请求处理线程里。

更合理的做法是：

• Drogon 负责接请求
• 重活交给 FastAPI / Celery

20.2 误以为性能高就应该把所有东西都写进 C++

不是。

对于 AI 项目：

• 模型生态、向量库、文档处理库，大部分都在 Python 侧更成熟
• Drogon 更适合做高性能 API 和状态治理

20.3 忽视文件上传安全

例如：

• 直接使用原始文件名保存
• 不限制 body 大小
• 不校验扩展名

这在上传接口里很危险。

20.4 Session 开了但客户端不带 Cookie

这样会导致服务端不断创建新 Session，白白浪费资源。

20.5 数据库回调里继续做阻塞操作

这会拖慢整个异步链路。

20.6 Controller 过胖

如果 Controller 里同时写：

• 参数解析
• SQL
• Redis
• 调内部 HTTP
• 业务编排

后面会非常难维护。

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

如果你是为了做 AI 项目里的 C++ 网关，推荐这样学：

1. 先跑通最小 Drogon 服务
2. 学会 registerHandler() 和 HttpController
3. 学会 JSON 请求和响应
4. 学会 config.json
5. 学会 Filter 和 Session
6. 学会文件上传
7. 学会 DbClient 和 Redis
8. 最后再接 FastAPI 和 Celery

22. 总结

Drogon 的核心价值在于：

• 用 C++ 写高性能 Web 服务
• 把 HTTP、数据库、Redis、Session、文件上传放到一套框架里
• 适合做网关层、业务层、低延迟接口层

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

• 对外 API 入口
• 会话治理层
• 状态协调层
• 数据持久化入口
• 调用 Python 服务的编排层

如果你把前面几份笔记连起来看，可以把它们理解成一条链：

• Drogon：对外网关 / C++ 服务层
• FastAPI：Python API 层
• Redis：缓存 / 状态 / 中间层
• Celery：后台异步任务层
• MySQL：持久化数据层

23. 参考

• Drogon GitHub: https://github.com/drogonframework/drogon
• Drogon Wiki: https://github.com/drogonframework/drogon/wiki
• Drogon Overview: https://github.com/drogonframework/drogon/wiki/ENG-01-Overview
• Drogon Installation: https://github.com/drogonframework/drogon/wiki/ENG-02-Installation
• Drogon HttpController: https://github.com/drogonframework/drogon/wiki/ENG-04-2-Controller-HttpController
• Drogon Middleware and Filter: https://github.com/drogonframework/drogon/wiki/ENG-05-Middleware-and-Filter
• Drogon Session: https://github.com/drogonframework/drogon/wiki/ENG-07-Session
• Drogon DbClient: https://github.com/drogonframework/drogon/wiki/ENG-08-1-Database-DbClient
• Drogon ORM: https://github.com/drogonframework/drogon/wiki/ENG-08-3-Database-ORM
• Drogon File Handler: https://github.com/drogonframework/drogon/wiki/ENG-09-1-File-Handler
• Drogon Configuration File: https://github.com/drogonframework/drogon/wiki/ENG-11-Configuration-File
• Drogon drogon_ctl: https://github.com/drogonframework/drogon/wiki/ENG-12-drogon_ctl-Command
• Drogon Coroutines: https://github.com/drogonframework/drogon/wiki/ENG-17-Coroutines
• Drogon Redis: https://github.com/drogonframework/drogon/wiki/ENG-18-Redis

Celery 学习笔记

1. Celery 是什么

Celery 是一个基于 Python 的分布式任务队列，用来把“耗时任务”从主业务流程里拆出去异步执行。

典型场景：

• 发送邮件、短信、消息通知
• 文件转码、图片处理、PDF 解析
• 大模型推理、批量数据处理
• 定时任务、周期性任务
• 需要失败重试的后台任务

先记住最实用的分工：FastAPI 负责接请求和返回结果，Celery 负责把耗时任务放到后台慢慢执行。

2. 为什么要用 Celery

如果一个接口内部要做很久的事情，比如：

• 调第三方接口要 10 秒
• 处理一批文件要 2 分钟
• 模型推理要占满 CPU / GPU

那就不适合一直阻塞 HTTP 请求。更合理的做法是：

1. FastAPI 收到请求
2. 把任务丢给 Celery
3. 立刻返回一个 task_id
4. 前端或调用方再通过 task_id 查询执行结果

这样接口响应会更快，系统也更容易扩展。

3. Celery 核心架构

Celery 主要由 3 个角色组成：

3.1 Broker

消息中间件，负责存放“待执行任务”。

常见选择：

• Redis
• RabbitMQ

它就是任务排队和等待消费的地方。

3.2 Worker

执行任务的工作进程。

它会不断从 Broker 中取任务，然后真正执行 Python 函数。

3.3 Result Backend

结果存储，用来保存任务状态和返回值。

常见选择：

• Redis
• Database
• RPC

如果你只关心“任务有没有执行”，不关心返回结果，也可以不配置结果后端；但实际项目里通常还是会配置，方便查状态。

4. 工作流程

Celery 的完整链路可以这样理解：

1. 客户端或 FastAPI 调用一个 Celery 任务
2. 任务消息被发送到 Broker
3. Worker 从 Broker 取出任务
4. Worker 执行任务函数
5. 执行结果和状态写入 Result Backend
6. 业务侧通过 task_id 查询任务状态和结果

对应的状态通常有：

• PENDING：任务还没开始，或者还查不到
• STARTED：任务已开始执行
• SUCCESS：执行成功
• FAILURE：执行失败
• RETRY：正在重试
• REVOKED：任务被撤销

说明：

• 默认不一定会看到 STARTED
• 如果想更明确地看到运行中状态，可以开启 task_track_started=True

5. 安装

1

pip install celery redis

如果使用 Redis 作为 Broker 和 Result Backend，需要先启动 Redis。

6. 最小可运行示例

6.1 编写 Celery 应用

文件：celery_app.py

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

from celery import Celery

app = Celery(
    "demo",
    broker="redis://127.0.0.1:6379/1",
    backend="redis://127.0.0.1:6379/2",
)

app.conf.update(
    task_track_started=True,
    timezone="Asia/Shanghai",
)

说明：

• /1 表示 Redis 的第 1 个逻辑库，用来当 Broker
• /2 表示 Redis 的第 2 个逻辑库，用来存任务结果
• Broker 和 Backend 分开写更清晰，排查问题也更方便

6.2 定义任务

文件：tasks.py

1
2
3
4
5
6
7
8

import time

from celery_app import app

@app.task
def add(a: int, b: int) -> int:
    time.sleep(3)
    return a + b

这里的 add 虽然是普通 Python 函数，但加了 @app.task 之后，它就成了 Celery 任务。

6.3 启动 Worker

1

celery -A tasks worker -l info

含义：

• -A tasks：表示 Celery 应用和任务定义在 tasks.py
• worker：启动 worker 进程
• -l info：日志级别为 info

6.4 提交任务

1
2
3
4

from tasks import add

result = add.delay(3, 5)
print(result.id)

注意：

• delay() 只是“提交任务”
• 它不会在当前进程里同步执行
• 返回的是 AsyncResult
• 真正执行任务的是独立的 Worker

6.5 查询任务结果

1
2
3
4
5
6
7
8
9
10
11

from celery.result import AsyncResult

from celery_app import app

task = AsyncResult("任务ID", app=app)

print(task.status)
print(task.successful())

if task.successful():
    print(task.get())

常用属性和方法：

• task.id：任务 ID
• task.status：任务状态
• task.successful()：是否执行成功
• task.failed()：是否执行失败
• task.ready()：任务是否结束
• task.get()：获取结果，可能会阻塞

7. delay() 和 apply_async() 的区别

7.1 delay()

最简单的调用方式，其实就是 apply_async() 的简写。

1

add.delay(3, 5)

适合：

• 普通异步提交
• 不需要额外参数时

7.2 apply_async()

更灵活，支持倒计时、定时执行、路由到指定队列等。

1

add.apply_async(args=[3, 5], countdown=10)

常见参数：

• args / kwargs：任务参数
• countdown=10：10 秒后执行
• eta=...：指定某个时间点执行
• expires=60：60 秒后任务过期
• queue="email"：指定进入哪个队列
• routing_key="task.email"：指定路由键

结论：

• 简单任务用 delay()
• 有调度需求时用 apply_async()

8. 任务重试

很多后台任务依赖外部资源，比如：

• 第三方 API
• 数据库
• 对象存储
• 模型服务

这些服务不稳定时，直接失败往往不合理，更适合自动重试。

1
2
3
4
5
6
7
8
9

from celery_app import app

@app.task(bind=True, max_retries=3, default_retry_delay=5)
def fetch_remote_data(self, url: str):
    try:
        # 这里假设会调用外部接口
        raise RuntimeError("temporary error")
    except Exception as exc:
        raise self.retry(exc=exc)

说明：

• bind=True 后，任务第一个参数会变成 self
• self.retry(...) 可以重新入队
• max_retries=3 表示最多重试 3 次
• default_retry_delay=5 表示默认每次间隔 5 秒

适合有“短暂失败可能恢复”的任务，不适合参数本身就错误的任务。

9. 队列与路由

当任务类型越来越多时，通常不会把所有任务都丢到一个队列里。

常见拆法：

• default：普通任务
• email：发邮件
• file：文件处理
• gpu：模型推理

这样可以让不同 Worker 只处理自己擅长的任务。

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

from celery import Celery
from kombu import Exchange, Queue

app = Celery("demo")

app.conf.task_queues = (
    Queue("default", Exchange("default"), routing_key="default"),
    Queue("file", Exchange("file"), routing_key="file.process"),
    Queue("gpu", Exchange("gpu"), routing_key="gpu.infer"),
)

app.conf.task_routes = {
    "tasks.process_pdf": {
        "queue": "file",
        "routing_key": "file.process",
    },
    "tasks.run_inference": {
        "queue": "gpu",
        "routing_key": "gpu.infer",
    },
}

启动指定队列的 Worker：

1
2

celery -A tasks worker -Q file -l info
celery -A tasks worker -Q gpu -l info

这样做的好处：

• 避免耗时任务阻塞普通任务
• 可以按机器能力拆分 Worker
• GPU 任务和 CPU 任务可以隔离

10. 定时任务

Celery 可以配合 celery beat 做周期任务。

例如：

• 每天凌晨清理临时文件
• 每小时刷新缓存
• 每 10 分钟同步一次数据

1
2
3
4
5
6
7
8

from celery.schedules import crontab

app.conf.beat_schedule = {
    "cleanup-temp-files": {
        "task": "tasks.cleanup_temp_files",
        "schedule": crontab(hour=3, minute=0),
    },
}

启动：

1

celery -A tasks beat -l info

如果要实际执行任务，除了 beat 还要有 worker 在跑。

11. FastAPI 和 Celery 怎么配合

最常见的配合方式是：

1. FastAPI 暴露 HTTP 接口
2. 接口中调用 Celery 提交任务
3. Celery Worker 在后台执行
4. 再通过一个查询接口返回任务状态/结果

11.1 什么时候用 BackgroundTasks，什么时候用 Celery

FastAPI 自带 BackgroundTasks，但它和 Celery 不是一类东西。

BackgroundTasks 更适合：

• 很轻量的后台操作
• 跟当前服务进程强绑定
• 不要求失败重试
• 不要求多机器扩展

Celery 更适合：

• 任务耗时长
• 任务量大
• 需要重试
• 需要队列隔离
• 需要多个 Worker 扩展
• 需要独立于 Web 进程运行

一句话：

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

12. FastAPI + Celery 案例 1：提交任务并查询状态

这是最常见的模式。

12.1 目录结构

1
2
3
4

project/
├── main.py
├── celery_app.py
└── tasks.py

12.2 celery_app.py

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

from celery import Celery

celery_app = Celery(
    "fastapi_demo",
    broker="redis://127.0.0.1:6379/1",
    backend="redis://127.0.0.1:6379/2",
)

celery_app.conf.update(
    task_track_started=True,
    result_expires=3600,
)

12.3 tasks.py

1
2
3
4
5
6
7
8

import time

from celery_app import celery_app

@celery_app.task(name="tasks.long_time_add")
def long_time_add(a: int, b: int) -> int:
    time.sleep(10)
    return a + b

12.4 main.py

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

from fastapi import FastAPI
from celery.result import AsyncResult

from celery_app import celery_app
from tasks import long_time_add

app = FastAPI()

@app.post("/tasks/add")
def create_add_task(a: int, b: int):
    task = long_time_add.delay(a, b)
    return {
        "message": "task submitted",
        "task_id": task.id,
    }

@app.get("/tasks/{task_id}")
def get_task_result(task_id: str):
    task = AsyncResult(task_id, app=celery_app)

    response = {
        "task_id": task.id,
        "status": task.status,
    }

    if task.successful():
        response["result"] = task.result
    elif task.failed():
        response["error"] = str(task.result)

    return response

12.5 启动方式

启动 FastAPI：

1

uvicorn main:app --reload

启动 Worker：

1

celery -A tasks worker -l info

请求流程：

1. 调 POST /tasks/add?a=3&b=5
2. 接口立即返回 task_id
3. 再调 GET /tasks/{task_id}
4. 看任务是 PENDING、STARTED 还是 SUCCESS

这个模式非常适合前后端分离项目。

13. FastAPI + Celery 案例 2：文件处理任务

这个场景很常见，比如：

• 上传 PDF 后做文本提取
• 上传图片后做缩略图
• 上传音频后做转写

13.1 设计原则

不要把 UploadFile、数据库连接、请求对象直接传给 Celery。

原因：

• Celery 任务参数通常需要能序列化
• 这些对象往往不能直接被序列化
• 跨进程后对象本身也失效了

正确做法：

1. FastAPI 先把文件保存到磁盘或对象存储
2. 把 file_path 或 file_url 传给 Celery
3. Worker 根据路径再去处理文件

13.2 示例

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

from pathlib import Path
import shutil
import uuid

from fastapi import FastAPI, File, UploadFile

from tasks import process_pdf

app = FastAPI()

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

@app.post("/files/pdf")
def upload_pdf(file: UploadFile = File(...)):
    file_id = f"{uuid.uuid4()}_{file.filename}"
    save_path = UPLOAD_DIR / file_id

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

    task = process_pdf.delay(str(save_path))

    return {
        "file_path": str(save_path),
        "task_id": task.id,
    }

对应的 Celery 任务：

1
2
3
4
5
6
7
8
9

from celery_app import celery_app

@celery_app.task(name="tasks.process_pdf")
def process_pdf(file_path: str):
    # 这里做 PDF 文本抽取、切分、向量化等操作
    return {
        "file_path": file_path,
        "message": "processed",
    }

这个思路很适合 AI 项目里的文档解析链路。

14. FastAPI + Celery 案例 3：模型推理任务

对于 AI 模型开发，这个场景更有代表性。

例如：

• 文本摘要
• 图片分类
• 批量 embedding
• 长文本抽取

14.1 为什么适合 Celery

因为模型推理通常具备这些特点：

• 耗时不稳定
• 资源消耗高
• 可能需要排队
• 可能要分 CPU / GPU Worker

14.2 一个简化版例子

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

from fastapi import FastAPI

from tasks import run_inference

app = FastAPI()

@app.post("/infer")
def create_inference_task(text: str):
    task = run_inference.apply_async(
        kwargs={"text": text},
        queue="gpu",
        routing_key="gpu.infer",
    )
    return {"task_id": task.id}

对应任务：

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

import time

from celery_app import celery_app

@celery_app.task(name="tasks.run_inference")
def run_inference(text: str):
    time.sleep(5)
    return {
        "text": text,
        "label": "positive",
        "score": 0.98,
    }

这里的重点不在模型代码本身，而在架构思路：

• API 层只负责接收请求
• 推理层在 Worker 中执行
• GPU 任务路由到专门队列
• 可以单独扩容推理 Worker

如果将来一张 GPU 卡只跑一个 Worker，也很好管理。

15. 实战里很重要的细节

15.1 任务参数尽量简单

建议只传这些：

• 字符串
• 数字
• 布尔值
• 列表 / 字典
• 文件路径
• 数据库主键 ID

不建议直接传：

• 数据库 Session
• 请求对象 Request
• UploadFile
• 大模型实例
• 打开的文件句柄

原则：

任务参数最好是“可序列化、可重建、可追踪”的。

15.2 任务尽量幂等

幂等的意思是：同一个任务重复执行多次，结果最好一致，或者至少不会造成严重副作用。

因为这些情况都可能导致重复执行：

• Worker 崩溃
• 消息重复投递
• 任务重试
• 手动补偿

例如发通知、写数据库、扣费等场景，幂等性非常重要。

15.3 不要在 API 进程里执行重活

FastAPI 本身应该尽量保持：

• 响应快
• 逻辑薄
• 只做参数校验、鉴权、任务分发

重计算、长耗时 IO、批处理，尽量放 Worker。

15.4 结果不是必须永久保存

很多项目会犯一个问题：任务结果一直存 Redis，时间一久就堆积很多垃圾数据。

可以设置：

• result_expires
• 任务完成后主动清理
• 只保留必要结果，不保留大对象

如果结果很大，不要直接把大块内容塞进 Redis，建议落库或存文件，再返回引用地址。

15.5 错误处理不要只靠日志

建议至少做到：

• 任务失败后能看到 FAILURE
• 能根据 task_id 查到失败原因
• 关键任务开启自动重试
• 必要时记录业务日志或落库

15.6 CPU / GPU / IO 任务分开

这点在 AI 项目里尤其重要。

建议：

• 文件下载、上传、请求第三方 API 放一个队列
• CPU 预处理放一个队列
• GPU 推理放一个队列

这样不会出现“一个长推理任务把所有普通任务都堵住”的问题。

15.7 Worker 并发要按任务类型调

Celery 的并发不是越大越好，要看任务类型。

一般经验：

• IO 密集型任务可以适当提高并发
• CPU 密集型任务并发通常不要超过 CPU 核心数太多
• GPU 推理任务通常一个 Worker 进程绑定一张卡，甚至一个队列只开 --concurrency=1

例如 GPU Worker 常见写法：

1

celery -A tasks worker -Q gpu --concurrency=1 -l info

如果推理模型本身已经很吃显存，再盲目开高并发，反而容易把机器打爆。

16. 常见坑

16.1 任务提交成功，但一直不执行

通常排查：

• Redis 是否启动
• Worker 是否启动
• -A 指向的模块是否正确
• 任务有没有注册到当前 Worker
• 是否路由到了某个没人消费的队列

16.2 能查到任务 ID，但拿不到结果

通常是：

• 没配置 backend
• task_id 查错了
• 结果过期被清掉了

16.3 任务函数里引用了 Web 层对象

比如：

• 直接把 Request 传进任务
• 把 SQLAlchemy Session 传进任务
• 把上传文件对象传进任务

这类设计一般都会有问题。任务应该依赖“可序列化参数”，不要依赖当前请求上下文。

16.4 把 Celery 当成同步 RPC 用

有些人虽然用了 Celery，但提交完任务后立刻 .get() 等待结果，这样就又退回同步调用了。

如果接口本身必须立刻拿到结果，那就要重新评估：

• 这个任务是否真的适合放到 Celery
• 是否应该走同步推理接口
• 是否应该拆成“提交任务 + 轮询结果”

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

建议按这个顺序掌握：

1. 先跑通最小示例
2. 理解 delay()、AsyncResult
3. 学会查任务状态
4. 学会重试
5. 学会队列与路由
6. 再接入 FastAPI
7. 最后再做定时任务、监控、扩容

18. 总结

Celery 的本质就是：

• 把耗时任务从主线程 / 主服务里剥离出去
• 通过消息队列异步执行
• 通过 Worker 扩展执行能力
• 通过结果后端查询状态和结果

对于 FastAPI 项目，可以把它理解成一组固定分工：

• FastAPI：对外提供 API
• Redis / RabbitMQ：负责排队
• Celery Worker：真正执行任务
• Result Backend：保存状态和结果

如果项目里已经出现“接口慢、任务重、需要重试、需要排队、需要异步处理”的问题，那 Celery 基本就是很自然的选择。

19. 参考

• Celery First Steps: https://docs.celeryq.dev/en/main/getting-started/first-steps-with-celery.html
• Celery Calling Tasks: https://docs.celeryq.dev/en/main/userguide/calling.html
• Celery Tasks Guide: https://docs.celeryq.dev/en/main/userguide/tasks.html
• Celery Routing Tasks: https://docs.celeryq.dev/en/main/userguide/routing.html
• Celery Configuration: https://docs.celeryq.dev/en/main/userguide/configuration.html
• FastAPI Background Tasks: https://fastapi.tiangolo.com/tutorial/background-tasks/

Ray 学习笔记

截至 2026-03-30，这份笔记主要参考：

• Ray 官方文档 2.54.x
• Celery 官方稳定版文档 5.6.x

如果当前目标是做 AI 模型开发，可以先抓住这个区分：

Celery 更像“任务队列和后台调度器”，Ray 更像“分布式计算和 GPU 资源调度层”。

1. Ray 是什么

Ray 是一个 Python 分布式计算框架，用来把普通 Python 代码扩展到：

• 多进程
• 多机
• 多 GPU
• 分布式训练
• 批量推理
• 在线推理服务

它的核心不是“消息队列”，而是“把 Python 函数和类，变成可调度的分布式任务”。

Ray 真正提供的是一套调度能力：普通的 Python 函数和类，加上 @ray.remote 之后，就能被当成分布式任务或远程 actor 来运行。

2. Ray 适合什么场景

Ray 典型适合这些场景：

• 批量 embedding 生成
• 多卡模型推理
• 大规模数据预处理
• 分布式训练
• 超参搜索
• 需要把模型常驻在 GPU 上的服务
• 复杂 Python 工作流并行化

如果只是这些场景，通常不一定要用 Ray：

• 发送短信、邮件、通知
• 定时任务
• 普通后台异步任务
• 只需要重试、延迟执行、任务链

这些更偏 Celery 的强项。

3. Ray 的核心概念

Ray 入门先掌握 4 个东西：

3.1 Task

Task 就是“远程函数”。

特点：

• 无状态
• 适合并行执行独立任务
• 用 @ray.remote 标记

3.2 Actor

Actor 就是“远程对象”。

特点：

• 有状态
• 适合模型常驻内存 / GPU
• 适合连接池、模型池、缓存、参数服务

3.3 ObjectRef

Ray 不会马上把结果返回给你，而是先返回一个“对象引用”。

• xxx.remote() 返回的通常是 ObjectRef
• 用 ray.get() 才真正取回结果

这点很像 Future。

3.4 Object Store

Ray 有自己的对象存储。

• task / actor 的返回值会放到对象存储里
• 大对象可以先 ray.put()
• 后续多个任务共享这个引用，避免重复拷贝

4. 最小入门示例

4.1 安装

1

pip install -U "ray[default]"

如果你还要用训练或数据处理：

1

pip install -U "ray[train]" "ray[data]" torch

4.2 启动本地 Ray

最简单的方式：

1
2
3

import ray

ray.init()

说明：

• ray.init() 会连接已有本地 Ray，或者启动一个新的本地实例
• 如果你是 Celery worker 或生产环境进程，通常更建议显式连接已有集群：ray.init(address="auto")

4.3 第一个 Task

1
2
3
4
5
6
7
8
9
10
11

import ray

ray.init()

@ray.remote
def square(x: int) -> int:
    return x * x

refs = [square.remote(i) for i in range(8)]
results = ray.get(refs)
print(results)

你要关注的是：

• @ray.remote：把普通函数变成远程函数
• square.remote(i)：不是本地调用，而是提交任务
• ray.get(refs)：拿回结果

4.4 第一个 Actor

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

import ray

ray.init()

@ray.remote
class Counter:
    def __init__(self):
        self.value = 0

    def incr(self, n: int = 1):
        self.value += n
        return self.value

    def get(self):
        return self.value

counter = Counter.remote()

ray.get(counter.incr.remote())
ray.get(counter.incr.remote(10))
print(ray.get(counter.get.remote()))

什么时候该用 Actor：

• 模型加载很慢，不想每次任务都重新加载
• 需要在内存里保留状态
• 要做 GPU 常驻推理

4.5 共享大对象

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

import numpy as np
import ray

ray.init()

@ray.remote
def compute_sum(arr):
    return arr.sum()

big_array_ref = ray.put(np.ones((10000, 1000)))
result = ray.get(compute_sum.remote(big_array_ref))
print(result)

经验：

• 小参数直接传
• 大对象尽量 ray.put()

5. Ray 常用 API

5.1 ray.init()

用于启动或连接 Ray。

常见写法：

1
2
3

ray.init()
ray.init(address="auto")
ray.init(address="ray://<head-ip>:10001")

含义：

• ray.init()：本地开发最方便
• address="auto"：只连接已有集群，如果没集群会报错，适合生产环境
• ray://...：Ray Client 方式连接远端集群

5.2 ray.get()

取结果。

1
2

value = ray.get(ref)
values = ray.get(refs)

5.3 ray.wait()

适合任务很多时，先消费一部分结果，避免一次性堆太多任务或结果。

1
2

ready, pending = ray.wait(refs, num_returns=1)
result = ray.get(ready[0])

5.4 ray.put()

把大对象放进 Ray 对象存储。

1

obj_ref = ray.put(big_object)

6. Task 和 Actor 怎么选

最常见的 AI 场景经验是：

• CPU 并行清洗数据：Task
• 模型推理：Actor
• 分布式训练：优先用 Ray Train，而不是自己手搓 Ray Core

7. 从本地到集群

7.1 单机开发

直接：

1

ray.init()

7.2 手动启动单机 / 多机集群

头节点：

1

ray start --head --port=6379

工作节点：

1

ray start --address=<head-node-ip>:6379

Python 里连接：

1
2
3

import ray

ray.init(address="auto")

7.3 GPU 机器上限制可见显卡

如果你只想让 Ray 看到部分 GPU：

1

CUDA_VISIBLE_DEVICES=1,3 ray start --head --num-gpus=2

这在一台机器上做资源隔离时很有用。

8. 多张显卡怎么用

这部分是 Ray 很强的一块。

你先记住结论：

• 单个任务 / Actor 用几张卡，就声明几张
• 多个任务并行吃多卡，让 Ray 调度
• 训练任务优先用 Ray Train
• 批量推理优先用 Actor 池或 Ray Data

8.1 一张卡对应一个 Actor

这是最常见的“多卡推理”写法。

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

import os
import ray

ray.init(num_gpus=4)

@ray.remote(num_gpus=1)
class GPUWorker:
    def info(self):
        return {
            "gpu_ids": ray.get_runtime_context().get_accelerator_ids()["GPU"],
            "cuda_visible_devices": os.environ.get("CUDA_VISIBLE_DEVICES"),
        }

workers = [GPUWorker.remote() for _ in range(4)]
print(ray.get([worker.info.remote() for worker in workers]))

效果：

• 每个 Actor 占 1 张 GPU
• Ray 会把不同 Actor 调度到不同卡
• 大多数深度学习框架会自动遵守 CUDA_VISIBLE_DEVICES

8.2 单个任务使用多张卡

如果一个任务本身就要吃多卡：

1
2
3
4
5
6
7
8
9
10

import ray

ray.init(num_gpus=4)

@ray.remote(num_gpus=2)
def train_one_job():
    import os
    return os.environ.get("CUDA_VISIBLE_DEVICES")

print(ray.get(train_one_job.remote()))

这适合：

• 单个训练任务需要 2 卡 / 4 卡
• 单个推理进程要占多卡

8.3 分数 GPU

如果模型很小，也可以共享 GPU：

1
2
3
4
5
6
7
8
9
10
11

import ray
import time

ray.init(num_gpus=1)

@ray.remote(num_gpus=0.25)
def small_infer(x):
    time.sleep(1)
    return x

print(ray.get([small_infer.remote(i) for i in range(4)]))

注意：

• 分数 GPU 只是“调度上的份额”
• 显存控制还是你自己负责
• 小模型、embedding、小批量推理更适合这样做

8.4 多卡训练优先用 Ray Train

如果你在做训练，不建议直接自己拼很多 @ray.remote。

更推荐：

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

from ray.train import ScalingConfig
from ray.train.torch import TorchTrainer, get_device

def train_loop(config):
    import torch

    device = get_device()
    model = torch.nn.Linear(10, 1).to(device)
    x = torch.randn(32, 10, device=device)
    y = model(x)
    loss = y.mean()
    loss.backward()

trainer = TorchTrainer(
    train_loop,
    scaling_config=ScalingConfig(
        num_workers=4,
        use_gpu=True,
    ),
)

trainer.fit()

这表示：

• 4 个 worker
• 每个 worker 1 张 GPU
• 总共 4 张 GPU

8.5 一个 worker 吃多张卡

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

from ray.train import ScalingConfig
from ray.train.torch import TorchTrainer, get_devices

def train_loop(config):
    import torch

    devices = get_devices()
    print(devices)

trainer = TorchTrainer(
    train_loop,
    scaling_config=ScalingConfig(
        num_workers=2,
        use_gpu=True,
        resources_per_worker={"GPU": 2},
    ),
)

trainer.fit()

这表示：

• 2 个 worker
• 每个 worker 2 张 GPU
• 总共 4 张 GPU

适合：

• 每个 worker 本身要做张量并行
• 每个训练进程要 2 卡或更多

8.6 需要成组预留资源时，用 Placement Group

有些任务需要“成组占住资源”，比如：

• 2 个 actor 必须同时启动
• 一个训练 job 要一次性拿到 2 CPU + 2 GPU
• 希望资源尽量在同一台机器

这时用 Placement Group。

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

import ray
from ray.util.placement_group import placement_group
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy

ray.init(num_gpus=2, num_cpus=8)

pg = placement_group(
    [{"CPU": 2, "GPU": 1}, {"CPU": 2, "GPU": 1}],
    strategy="STRICT_PACK",
)
ray.get(pg.ready())

@ray.remote(num_cpus=2, num_gpus=1)
class TrainerWorker:
    def ready(self):
        return "ok"

workers = [
    TrainerWorker.options(
        scheduling_strategy=PlacementGroupSchedulingStrategy(
            placement_group=pg,
            placement_group_bundle_index=i,
        )
    ).remote()
    for i in range(2)
]

print(ray.get([w.ready.remote() for w in workers]))

STRICT_PACK 的意思是：

• 所有 bundle 必须放在同一台节点上
• 如果放不下，就直接失败

这对多卡训练很常见。

9. AI 场景下更实用的写法

9.1 模型常驻 GPU 的 Actor

如果你做 embedding、rerank、分类、OCR、ASR，常见需求是：

• 模型初始化很慢
• 不想每个请求都重新加载模型
• 想让不同 worker 常驻不同 GPU

这时最实用的方案往往是：

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

import ray

ray.init(num_gpus=2)

@ray.remote(num_gpus=1)
class EmbeddingWorker:
    def __init__(self, model_name: str):
        from sentence_transformers import SentenceTransformer

        self.model = SentenceTransformer(model_name, device="cuda")

    def encode(self, texts: list[str]):
        vectors = self.model.encode(texts, normalize_embeddings=True)
        return vectors.tolist()

workers = [
    EmbeddingWorker.remote("BAAI/bge-small-zh-v1.5")
    for _ in range(2)
]

优势：

• 模型只加载一次
• Ray 自动把 worker 分到不同 GPU
• 适合批量任务和服务化场景

9.2 批量推理更推荐 Ray Data

如果是大批量离线推理，比如：

• 给 100 万条文本做 embedding
• 大量图片跑分类 / OCR
• 多卡批量推理

Ray Data 一般比你自己手写 for + remote() 更省心。

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

import pandas as pd
import ray

ray.init()

class EmbeddingPredictor:
    def __init__(self, model_name: str):
        from sentence_transformers import SentenceTransformer

        self.model = SentenceTransformer(model_name, device="cuda")

    def __call__(self, batch: pd.DataFrame) -> pd.DataFrame:
        batch = batch.copy()
        batch["embedding"] = self.model.encode(batch["text"].tolist()).tolist()
        return batch

ds = ray.data.from_items(
    [{"text": f"sample {i}"} for i in range(1000)]
)

result_ds = ds.map_batches(
    EmbeddingPredictor,
    fn_constructor_args=("BAAI/bge-small-zh-v1.5",),
    batch_size=64,
    batch_format="pandas",
    compute=ray.data.ActorPoolStrategy(size=4),
    num_gpus=1,
)

print(result_ds.take(2))

什么时候优先考虑 Ray Data：

• 数据量大
• 希望自动做并行读取和批处理
• 想用 actor pool 跑多卡离线推理

10. Ray 和 Celery 的区别

很多人第一次会问：

我已经有 Celery 了，还要不要 Ray？

答案通常是：

不是替代关系，而是职责不同。

10.1 一张表看区别

10.2 什么时候只用 Celery

• 发通知
• 发邮件
• 定时任务
• 任务重试
• 后台轻量业务任务

10.3 什么时候只用 Ray

• 分布式训练
• 多 GPU 推理
• 批量 embedding
• 大规模离线数据处理
• 需要模型常驻内存 / GPU

10.4 什么时候 Ray + Celery 联合

这是很常见、也很推荐的工程做法：

• Celery 负责“接住业务异步任务”
• Ray 负责“真正吃 CPU / GPU 算力”

最典型架构：

1
2
3
4
5
6

FastAPI / Django
    -> 提交 Celery 任务
    -> Celery worker 收到任务
    -> Celery worker 把重计算任务提交给 Ray
    -> Ray task / actor 在 CPU / GPU 上执行
    -> Celery 记录状态、重试、结果

这时职责很清晰：

• Celery：任务排队、重试、路由、延迟调度、业务集成
• Ray：并行计算、多机、多卡、模型常驻

11. Ray + Celery 推荐落地方式

这一节不是某个“官方一键集成组件”的说明，而是基于 Ray 官方能力和 Celery 官方能力整理出来的工程实践。

11.1 推荐原则

推荐这样设计：

• 不要让 Celery 自己直接“管理 GPU”
• 不要每个 Celery 任务里都临时启动一个 Ray 集群
• Celery worker 只负责连接已有 Ray 集群并下发任务
• 模型尽量放到 Ray Actor 里常驻

一句话：

Celery 做任务层，Ray 做算力层。

11.2 最实用的方案：Celery 任务内部调用 Ray Actor

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

from celery import Celery
import ray

celery_app = Celery(
    "demo",
    broker="redis://127.0.0.1:6379/1",
    backend="redis://127.0.0.1:6379/2",
)

def ensure_ray():
    if not ray.is_initialized():
        ray.init(address="auto", namespace="ml")

@ray.remote(num_gpus=1)
class EmbeddingWorker:
    def __init__(self, model_name: str):
        from sentence_transformers import SentenceTransformer

        self.model = SentenceTransformer(model_name, device="cuda")

    def encode(self, texts: list[str]):
        return self.model.encode(texts).tolist()

def get_embedding_worker():
    ensure_ray()
    try:
        return ray.get_actor("embedding_worker")
    except ValueError:
        return EmbeddingWorker.options(
            name="embedding_worker",
            lifetime="detached",
            get_if_exists=True,
        ).remote("BAAI/bge-small-zh-v1.5")

@celery_app.task(
    bind=True,
    autoretry_for=(Exception,),
    retry_backoff=True,
    max_retries=3,
)
def build_embeddings(self, texts: list[str]):
    worker = get_embedding_worker()
    ref = worker.encode.remote(texts)
    vectors = ray.get(ref)
    return {
        "count": len(vectors),
        "vectors": vectors,
    }

这套写法的好处：

• Celery 有重试和任务状态
• Ray Actor 让模型只加载一次
• GPU 真正由 Ray 调度
• Celery 不直接碰 CUDA_VISIBLE_DEVICES

11.3 为什么这里要用 Named Actor

Named Actor 很适合下面这种需求：

• 多个 Celery 任务都要访问同一个模型 worker
• 模型很大，不想重复加载
• 希望 worker 独立于单次任务生命周期存在

这里用到的 3 个点都很重要：

• name="embedding_worker"：给 actor 命名
• get_if_exists=True：如果已有就复用
• lifetime="detached"：Celery 任务结束后 actor 依然活着

补充：

• Detached Actor 不会自动回收
• 不再需要时要手动 ray.kill(actor_handle)，否则会一直占资源

11.4 如果要做多卡模型池

把一个 Named Actor 扩展成多个即可：

1
2
3
4
5
6
7
8

workers = [
    EmbeddingWorker.options(
        name=f"embedding_worker_{i}",
        lifetime="detached",
        get_if_exists=True,
    ).remote("BAAI/bge-small-zh-v1.5")
    for i in range(4)
]

然后 Celery 里：

• 轮询选一个
• 按队列分流
• 或者再包一个调度 Actor

11.5 Celery 队列怎么配合

重任务建议单独队列，比如：

1

build_embeddings.apply_async(args=[texts], queue="gpu")

对应 worker：

1

celery -A app worker -l info -Q gpu --concurrency=4

这样做的意义：

• 把 GPU 相关任务跟普通业务任务隔离
• Celery 侧也能做业务级限流

注意：

• Celery 的 --concurrency 不等于 GPU 数
• 真正的 GPU 分配还是 Ray 决定

12. 一个更合理的职责分层建议

如果你的项目同时有 Web、异步任务、模型推理，我更建议这样分：

这意味着：

• 后台异步批处理：Celery + Ray
• 在线低延迟模型服务：优先 Ray Serve

如果你把“在线推理 API”也硬塞进 Celery，一般不是最优解。

13. 常见坑

13.1 不要在循环里一边提交一边 ray.get()

错误思路：

1
2

for x in items:
    result = ray.get(task.remote(x))

这样会把并行写成串行。

更合理：

1
2

refs = [task.remote(x) for x in items]
results = ray.get(refs)

13.2 任务别切得太细

如果单个任务非常轻，Ray 的调度开销会吃掉收益。

经验：

• 小任务尽量 batch 化
• 推理尽量做批处理

13.3 大对象别重复按值传

错误思路：

• 每个 task 都传一份很大的模型参数

更合理：

• ray.put()
• 或者放进 Actor

13.4 GPU task 默认可能不复用 worker

Ray 文档提到：

• 某些 GPU 任务可能会残留显存占用
• 为避免资源泄漏，Ray 默认不会复用执行过 GPU task 的 worker 进程

如果你遇到 GPU 调度开销偏大，需要再回头看 worker 复用策略。

如果你确认自己的任务不会泄漏 GPU 资源，可以显式开启复用：

1
2
3

@ray.remote(num_gpus=1, max_calls=0)
def gpu_task():
    ...

13.5 Celery worker 不要偷偷起本地 Ray

在生产环境里，Celery 进程里更推荐：

1

ray.init(address="auto")

而不是：

1

ray.init()

原因：

• ray.init() 可能在没有集群时直接起一个本地 Ray
• 这样每台 Celery 机器都可能误起自己的小集群
• 排查起来很麻烦

14. 一个现实中的选型建议

如果你的任务是下面这些，我建议这样选：

14.1 批量 embedding / rerank / OCR / ASR

优先：

• Celery + Ray Actor
• 数据量特别大时用 Ray Data

14.2 分布式训练

优先：

• Ray Train

14.3 在线模型 API

优先：

• Ray Serve

14.4 只是后台异步业务

优先：

• Celery

15. 我建议你的学习顺序

如果你现在刚入门 Ray，可以按这个顺序学：

1. ray.init()、@ray.remote、ray.get()
2. Task 和 Actor 的区别
3. ray.put()、ray.wait()
4. GPU 调度：num_gpus
5. Named Actor / Detached Actor
6. Placement Group
7. Ray Data 批量推理
8. Ray Train 多卡训练
9. 再考虑和 Celery 联合

16. 官方文档和官方博客

下面这些我认为最值得先看。

16.1 Ray 官方文档

• Ray Core Walkthrough: https://docs.ray.io/en/latest/ray-core/walkthrough.html
• Starting Ray: https://docs.ray.io/en/latest/ray-core/starting-ray.html
• ray.init API: https://docs.ray.io/en/latest/ray-core/api/doc/ray.init.html
• Accelerator Support: https://docs.ray.io/en/latest/ray-core/scheduling/accelerators.html
• Placement Groups: https://docs.ray.io/en/latest/ray-core/scheduling/placement-group.html
• Named Actors: https://docs.ray.io/en/latest/ray-core/actors/named-actors.html
• Ray Train GPU Guide: https://docs.ray.io/en/latest/train/user-guides/using-gpus.html
• Ray Data map_batches: https://docs.ray.io/en/latest/data/api/doc/ray.data.Dataset.map_batches.html
• Offline Batch Inference: https://docs.ray.io/en/latest/data/batch_inference.html

16.2 Celery 官方文档

• Calling Tasks: https://docs.celeryq.dev/en/stable/userguide/calling.html
• Tasks: https://docs.celeryq.dev/en/stable/userguide/tasks.html
• Workers Guide: https://docs.celeryq.dev/en/stable/userguide/workers.html

16.3 官方博客 / 官方活动

• Scaling Model Batch Inference in Ray: https://www.anyscale.com/blog/scaling-model-batch-inference-in-ray-using-actors-actorpool-and-ray-data
• Ray Data: Scalable Data Processing for AI Workloads: https://www.anyscale.com/blog/ray-data-scalable-data-processing-for-ai-workloads
• Introduction to Ray Core and its Ecosystem: https://www.anyscale.com/events/2022/02/03/introduction-to-ray-core-and-its-ecosystem
• Ray Core Tutorial: https://www.anyscale.com/events/2021/06/24/ray-core-tutorial

17. 最后总结

你可以先把 Ray 理解成：

一个能把 Python 函数、对象、GPU、集群资源统一调度起来的分布式计算框架。

如果放到工程里：

• Celery 解决“任务怎么排队、重试、延迟、编排”
• Ray 解决“任务怎么真正并行地吃 CPU / GPU”

如果你后面是做 AI 系统，尤其是：

• 多卡推理
• 批量 embedding
• 大模型训练 / 推理

那 Ray 很值得认真学。

18. FastAPI + Celery + Ray + 4 张 GPU 最小可运行项目

下面给你一份“单机 4 张 GPU”的最小结构。

目标：

• FastAPI 负责接请求
• Celery 负责异步任务和状态查询
• Ray 负责 4 张 GPU 的模型调度
• 每张 GPU 常驻 1 个 embedding worker

18.1 适用场景

这个结构很适合：

• 批量生成 embedding
• OCR / ASR / rerank 这类 GPU 推理任务
• Web 请求不想阻塞
• 想把“业务异步”和“GPU 算力”分层

不太适合：

• 极低延迟在线推理

如果你追求在线低延迟，优先考虑 Ray Serve。

18.2 目录结构

1
2
3
4
5
6
7
8
9

ray_embedding_demo/
├── app/
│   ├── __init__.py
│   ├── celery_app.py
│   ├── main.py
│   ├── ray_actors.py
│   ├── ray_runtime.py
│   └── tasks.py
└── requirements.txt

18.3 requirements.txt

1
2
3
4
5
6
7
8

fastapi
uvicorn[standard]
celery
redis
ray[default]
sentence-transformers
torch
numpy

18.4 app/celery_app.py

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

from celery import Celery

celery_app = Celery(
    "ray_embedding_demo",
    broker="redis://127.0.0.1:6379/1",
    backend="redis://127.0.0.1:6379/2",
)

celery_app.conf.update(
    imports=("app.tasks",),
    task_track_started=True,
    task_default_queue="gpu",
    timezone="Asia/Shanghai",
)

18.5 app/ray_runtime.py

1
2
3
4
5
6
7

import ray

RAY_NAMESPACE = "embedding-demo"

def ensure_ray():
    if not ray.is_initialized():
        ray.init(address="auto", namespace=RAY_NAMESPACE)

18.6 app/ray_actors.py

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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

from __future__ import annotations

import os

import ray

from app.ray_runtime import ensure_ray

MODEL_NAME = "BAAI/bge-small-zh-v1.5"
WORKER_COUNT = 4
WORKER_PREFIX = "embedding_worker"
ROUTER_NAME = "embedding_router"

def worker_name(worker_id: int) -> str:
    return f"{WORKER_PREFIX}_{worker_id}"

@ray.remote(num_gpus=1)
class EmbeddingWorker:
    def __init__(self, model_name: str, worker_id: int):
        from sentence_transformers import SentenceTransformer

        self.worker_id = worker_id
        self.model = SentenceTransformer(model_name, device="cuda")

    def encode(self, texts: list[str]) -> list[list[float]]:
        vectors = self.model.encode(
            texts,
            batch_size=32,
            normalize_embeddings=True,
        )
        return vectors.tolist()

    def info(self) -> dict:
        return {
            "worker_id": self.worker_id,
            "gpu_ids": ray.get_runtime_context().get_accelerator_ids()["GPU"],
            "cuda_visible_devices": os.environ.get("CUDA_VISIBLE_DEVICES"),
        }

@ray.remote
class EmbeddingRouter:
    def __init__(self, worker_names: list[str]):
        self.worker_names = worker_names
        self.index = 0

    def encode(self, texts: list[str]) -> list[list[float]]:
        current_name = self.worker_names[self.index]
        self.index = (self.index + 1) % len(self.worker_names)
        worker = ray.get_actor(current_name)
        return ray.get(worker.encode.remote(texts))

    def state(self) -> dict:
        return {
            "worker_names": self.worker_names,
            "next_index": self.index,
        }

def ensure_embedding_cluster():
    ensure_ray()

    worker_names = []
    for worker_id in range(WORKER_COUNT):
        name = worker_name(worker_id)
        EmbeddingWorker.options(
            name=name,
            lifetime="detached",
            get_if_exists=True,
        ).remote(MODEL_NAME, worker_id)
        worker_names.append(name)

    EmbeddingRouter.options(
        name=ROUTER_NAME,
        lifetime="detached",
        get_if_exists=True,
    ).remote(worker_names)

def get_router():
    ensure_embedding_cluster()
    return ray.get_actor(ROUTER_NAME)

def list_workers() -> list[dict]:
    ensure_embedding_cluster()
    handles = [ray.get_actor(worker_name(i)) for i in range(WORKER_COUNT)]
    return ray.get([handle.info.remote() for handle in handles])

说明：

• EmbeddingWorker：每个 actor 绑定 1 张 GPU
• EmbeddingRouter：做一个最简单的轮询分发
• lifetime="detached"：worker 不跟着单个任务退出
• get_if_exists=True：多进程启动时避免重复创建

18.7 app/tasks.py

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