异步任务编排：Celery + Redis

学习目标

• 理解 Dify 如何通过 Celery + Redis 实现后台任务编排
• 掌握任务幂等性、失败重试与健康检查策略
• 学会分析 Celery Worker 与 Flask 应用的进程模型

前置知识

• Celery 基本概念（Task、Broker、Worker）
• Redis 作为消息中间人的角色

---

项目实践

Celery + Redis 架构

任务发布

通过 async_workflow_service 发布异步任务：

# 伪代码
class AsyncWorkflowService:
    def publish_task(self, task_type: str, **kwargs):
        # 通过 Celery 的 delay/apply_async 发布任务到 Redis
        task.delay(**kwargs)

任务实现

任务定义在 api/tasks/ 目录下：

api/tasks/
├── annotation_batch_import_task.py    # 批量导入标注
├── clean_document_task.py             # 清理文档
├── create_segment_to_index_task.py    # 创建文档段并索引
├── deal_dataset_vector_index_task.py  # 数据集向量索引
├── delete_account_task.py             # 删除账户
├── disable_segments_from_index_task.py # 禁用索引
├── document_indexing_task.py          # 文档索引
├── enable_segments_to_index_task.py   # 启用索引
├── mail_email_webhook_task.py         # 发送邮件
├── ops_trace_task.py                  # 追踪任务
├── recover_document_task.py           # 恢复文档
└── ...

典型任务模式：

# 伪代码
@celery.task(
    bind=True,
    max_retries=3,                    # 最大重试次数
    default_retry_delay=60,           # 重试间隔
    rate_limit="10/m",               # 速率限制
)
def document_indexing_task(self, document_id: str):
    try:
        # 1. 获取文档
        # 2. 文本提取
        # 3. 分片
        # 4. 向量化
        # 5. 写入向量数据库
    except Exception as exc:
        self.retry(exc=exc, countdown=60)  # 重试

入口点

api/celery_entrypoint.py    # Celery 应用入口
api/celery_healthcheck.py   # 健康检查

Gunicorn + gevent 异步 Worker 模型

Flask 应用使用 Gunicorn + gevent 运行：

# api/gunicorn.conf.py 伪代码
worker_class = "gevent"           # 使用 gevent 异步 Worker
workers = multiprocessing.cpu_count() * 2 + 1  # Worker 数量
worker_connections = 1000         # 每个 Worker 的并发连接数

进程模型：

┌─────────────────────────────────────────┐
│ Gunicorn Master                         │
├───────────┬───────────┬─────────────────┤
│ Worker 1  │ Worker 2  │ Worker N        │
│ (gevent)  │ (gevent)  │ (gevent)        │
│ Greenlets │ Greenlets │ Greenlets       │
└───────────┴───────────┴─────────────────┘

┌─────────────────────────────────────────┐
│ Celery Workers (separate processes)     │
├───────────┬───────────┬─────────────────┤
│ Worker 1  │ Worker 2  │ Beat Scheduler  │
│ (task)    │ (task)    │ (periodic)      │
└───────────┴───────────┴─────────────────┘

健康检查

celery_healthcheck.py 提供 Celery 健康检查端点：

# 伪代码
def celery_healthcheck():
    # 检查 Celery Worker 是否在线
    # 检查任务队列是否正常
    # 返回健康状态

问题与规避

陷阱 1：任务幂等性

网络问题或重试可能导致同一任务被多次执行。

Dify 的规避策略：

• 任务实现应保证幂等性（多次执行结果相同）
• 文档索引任务通过 doc_id 去重
• 向量存储的 text_exists() 方法检查重复

陷阱 2：任务失败后的处理

长时间任务失败后需要恢复，而非从头开始。

Dify 的规避策略：

• 设置 max_retries 和 default_retry_delay
• 失败任务记录在数据库中，支持手动重试
• 关键任务（如文档索引）支持断点续传

陷阱 3：Celery 与 Redis 的连接问题

Redis 故障会导致整个任务系统不可用。

Dify 的规避策略：

• Celery 配置中设置 Redis 连接超时和重试
• 健康检查端点监控 Celery Worker 状态
• 任务发布时捕获连接异常，返回友好的错误信息

设计取舍

Celery vs 其他异步框架

优势：

• 成熟的 Python 异步任务框架
• Redis 作为 Broker 易于部署
• 支持任务重试、速率限制、定时任务
• 丰富的监控和管理工具

代价：

• 增加部署复杂度（需要单独的 Celery Worker 进程）
• 任务状态持久化依赖 Redis/数据库
• 调试相对困难（跨进程）

参考来源

• Dify Tasks 目录：api/tasks/
• Celery 入口：api/celery_entrypoint.py
• Celery 健康检查：api/celery_healthcheck.py