Celery 架构

TFRobotServer 使用 两个独立的 Celery 应用 来处理不同类型的异步任务。这种分离架构是由于 TFRobot 框架的技术限制和设计权衡导致的。

架构概览

┌─────────────────────────────────────────────────────────────┐
│                      RabbitMQ Broker                        │
│  ┌──────────────────┐        ┌──────────────────┐           │
│  │  robot_tasks     │        │  memory_tasks    │           │
│  └────────┬─────────┘        └────────┬─────────┘           │
└───────────┼───────────────────────────┼──────────────────────┘
            │                           │
            ▼                           ▼
┌───────────────────────┐    ┌───────────────────────┐
│   TFSRobotCelery      │    │  TFSMemoryCelery      │
│   (robot_celery.py)   │    │  (celery.py)          │
├───────────────────────┤    ├───────────────────────┤
│ • 响应用户消息          │    │ • 内存管理             │
│ • Robot 对话执行       │    │ • 学习任务              │
│ • 单进程 + 线程池       │    │ • 文档处理             │
│ • 绑定 Robot 实例      │    │ • 多进程并发            │
└───────────────────────┘    └───────────────────────┘

队列命名约定（TFRS-217）：同一 K8s namespace 下可能部署多个 Robot 实例（多个 TFRServer CR）共享同一 RabbitMQ broker。若都消费同名 robot_tasks / memory_tasks，competing-consumer 会让任务被错误实例的 worker 抢走 → 用户消息不回复。因此每个实例消费专属队列 robot_tasks.<robot_id> / memory_tasks.<robot_id>：

派发端由 tfrobotserver/dependencies/celery/queue_routing.py::per_robot_queue(base) 派生队列名（<robot_id> = RobotServerConfig().robot.robot_id，来自挂载 TOML [robot].robot_id）。

消费端由 tfrs-operator 渲染 worker 启动参数 -Q <queue>.<robot_id>（本地开发见 supervisord.conf）。派发端与消费端同源 robot_id，故队列名必然一致。

下文示例中的 <robot_id> 为占位符，请替换为实际 robot_id（debug profile 即 tfrobotserver-debug.toml 的 robot.robot_id）。仅当 robot_id 未配置（NOT_SET，极少数纯本地场景）时才回退为裸队列名。

1. TFSRobotCelery (robot_celery.py)

用途

专门用于响应用户消息和执行 Robot 对话任务。它是 TFRobotServer 的核心对话处理组件。

关键限制 ⚠️

只能启动一个 Worker 进程，原因如下：

TFRobot 配置依赖文件系统存储
配置存储路径：~/tfrobot/configs
向量库路径：~/tfrobot/configs/memos_vector 和 knowledge_vector
这些配置必须在整个进程中保持一致
多进程同时修改会导致状态不一致
Robot 使用单例模式 python robot = providers.ThreadSafeSingleton( init_tfrobot_for_celery, db=db, ... )
设计上只允许一个 Robot 实例存在
多 Worker 会创建多个 Robot 实例，导致冲突
Faiss 向量库的并发限制 python faiss.omp_set_num_threads(1) # 在 robot_celery.py:98
Faiss 的 OMP 并发设置为 1，避免进程竞争导致的运行错误
这是已知的问题，参考：飞书文档
Robot 运行时的全局状态影响
Robot 对话上下文、环境变量是全局的
多任务并发会互相干扰，导致上下文不全面

并发能力

虽然限制为单进程，但可以通过 线程池 提供有限的并发能力：

celery -A tfrobotserver.robot_celery:tfs_celery worker \
  --loglevel=info \
  --concurrency=4 \
  --pool=threads \     # 使用线程池而非进程池
  -Q robot_tasks.<robot_id>   # TFRS-217：按 robot_id 隔离，<robot_id> 替换为实际值

并发能力说明： - 线程池可以处理 I/O 密集型任务（网络请求、数据库查询） - 受 Python GIL 限制，CPU 密集型任务无法真正并行 - 适合处理用户消息的并发场景

代码位置

定义：tfrobotserver/robot_celery.py
任务协议：tfrobotserver/protocols/robot_run.py
队列：robot_tasks.<robot_id>（TFRS-217 按实例隔离；robot_id 未配置时回退 robot_tasks）

重要提示

不要将 TFSRobotCelery 转换为多进程部署

文档中明确标注：

"!!! 一定要确保此TFRobotCelery是以单进程模式启动，无并发，否则多个Robot实例会对环境造成冲突性的影响。"

来源：robot_celery.py:17

2. TFSMemoryCelery (celery.py)

用途

处理与 TFRobot 运行时无关的后台任务，包括： - 内存管理（MemStore 清理、索引优化） - 学习任务（文档向量化、知识库更新） - 文档处理（partition_doc、格式转换） - 定时任务（刷新 AccessToken）

并发能力

支持多进程并发部署，无特殊限制：

celery -A tfrobotserver.celery:memory_manage_celery worker \
  --loglevel=info \
  --concurrency=4 \
  -Q memory_tasks.<robot_id>   # TFRS-217：按 robot_id 隔离，<robot_id> 替换为实际值

特性

不依赖 TFRobot 实例
无状态设计，可水平扩展
支持多种池类型（prefork、gevent）
可使用 AutoScaling 动态调整进程数

代码位置

定义：tfrobotserver/celery.py
任务协议：
tfrobotserver/protocols/sync_task_manage_celery.py
tfrobotserver/protocols/learn_task_celery.py
队列：memory_tasks.<robot_id>（TFRS-217 按实例隔离；robot_id 未配置时回退 memory_tasks）

AutoScaling

适用范围：仅适用于 TFSMemoryCelery

自动缩放组件用于根据负载动态调整池的大小。自动缩放器在有工作要做时增加更多的池进程，并在工作负载低时开始减少进程。

支持的池类型

prefork（默认，多进程）
gevent（协程）

使用方法

通过 --autoscale 选项启用自动缩放，需要提供两个数字：最大和最小进程数。

celery -A tfrobotserver.celery:memory_manage_celery worker \
  --loglevel=info \
  --autoscale=10,3 \    # 最大 10 个进程，最小保持 3 个
  -Q memory_tasks.<robot_id>   # TFRS-217：按 robot_id 隔离，<robot_id> 替换为实际值

自定义自动缩放规则

可以继承 Autoscaler 类定义自己的自动缩放规则。常见的度量标准包括： - 负载平均值 - 可用内存量 - 队列长度

通过 worker_autoscaler 设置指定自定义的自动缩放器。

为什么 TFSRobotCelery 不支持 AutoScaling？

由于 单进程限制，TFSRobotCelery 无法使用 AutoScaling。即使使用线程池，线程数量也应在启动时固定，因为： 1. 线程过多会导致 Robot 实例的资源竞争 2. Faiss OMP 已限制为单线程 3. Robot 的状态管理不适合动态调整并发度

Kubernetes 部署示例

参考 k8s/tfrs-all-in-one.yaml：

# TFSRobotCelery - 单进程 + 线程池
- name: robot-worker
  command: ["celery"]
  args: [
    "-A", "tfrobotserver.robot_celery:tfs_celery",
    "worker",
    "--loglevel=info",
    "--concurrency=4",
    "--pool=threads",      # 线程池
    "-Q", "robot_tasks"
  ]

# TFSMemoryCelery - 多进程
- name: memory-worker
  command: ["celery"]
  args: [
    "-A", "tfrobotserver.celery:memory_manage_celery",
    "worker",
    "--loglevel=info",
    "--concurrency=4",
    "-Q", "memory_tasks"
  ]

设计权衡与最佳实践

何时使用 TFSRobotCelery？

✅ 适合： - 需要与 Robot 对话交互的任务 - 需要访问 Robot 运行时状态的任务 - 用户消息处理和响应

❌ 不适合： - 纯计算任务（受 GIL 限制） - 需要高并发的任务 - 与 Robot 无关的后台任务

何时使用 TFSMemoryCelery？

✅ 适合： - 文档处理和向量化 - 定时任务和清理任务 - 需要高并发的后台任务 - 与 Robot 运行时无关的任务

❌ 不适合： - 需要与 Robot 对话的任务 - 需要访问 Robot 运行时状态的任务

性能优化建议

TFSRobotCelery
使用线程池处理 I/O 密集型任务
合理设置 --concurrency（建议 2-8）
监控线程池的使用率，避免过载
TFSMemoryCelery
根据任务类型选择合适的池类型
CPU 密集型任务使用 prefork
I/O 密集型任务使用 gevent
使用 AutoScaling 动态调整资源

参考资料

Celery 官方文档 - Workers Guide
Celery 官方文档 - Autoscaling
TFRobotServer 源码：
tfrobotserver/robot_celery.py
tfrobotserver/celery.py
tfrobotserver/dependencies/celery/celery_container.py
K8s 部署配置：k8s/tfrs-all-in-one.yaml