Dify 用 Celery 做任务队列，初始化在 extensions/ext_celery.py（extensions/ext_celery.py:99）：

# api/extensions/ext_celery.py:99
def init_app(app: DifyApp) -> Celery:
    class FlaskTask(Task):                                   # ★ 让每个后台任务也能用 Flask 上下文
        def __call__(self, *args, **kwargs):
            from core.logging.context import init_request_context
            with app.app_context():                          #    进入应用上下文（能访问 db 等）
                init_request_context()
                return self.run(*args, **kwargs)

    celery_app = Celery(
        app.name, task_cls=FlaskTask,
        broker=dify_config.CELERY_BROKER_URL,                # ★ broker：任务往哪丢（如 Redis）
        backend=dify_config.CELERY_BACKEND,                  #    backend：结果存哪
    )
    celery_app.conf.update(
        ...
        task_ignore_result=True,                             # 多数任务不关心返回值，省存储
        task_annotations=dify_config.CELERY_TASK_ANNOTATIONS,
    )

一个具体任务长这样——看 tasks/document_indexing_task.py（tasks/document_indexing_task.py:32）：

# api/tasks/document_indexing_task.py:32
@shared_task(queue="dataset")                    # ★ 声明这是个 Celery 任务，丢进 "dataset" 队列
def document_indexing_task(dataset_id: str, document_ids: list):
    """
    Async process document
    Usage: document_indexing_task.delay(dataset_id, document_ids)   # ★ 调用方用 .delay() 异步触发
    """
    ...                                          # 切片、建索引的慢活都在这里，后台跑

追踪的总管在 core/ops/ops_trace_manager.py。核心角色两个：TraceTask（core/ops/ops_trace_manager.py:651，"一条待上报的追踪任务"）和 TraceQueueManager（ops_trace_manager.py:1491，"攒批调度器"）。你在 Day16 见过的 TraceQueueManager、Day18 注入的追踪头，都汇到这里。

看 TraceQueueManager 怎么"攒"和"发"（core/ops/ops_trace_manager.py:1506）：

# api/core/ops/ops_trace_manager.py:1506
def add_trace_task(self, trace_task: TraceTask):
    global trace_manager_timer, trace_manager_queue
    try:
        if self._enterprise_telemetry_enabled or self.trace_instance:
            trace_task.app_id = self.app_id
            trace_manager_queue.put(trace_task)      # ★① 只是塞进内存队列，瞬间返回（不阻塞对话）
    except Exception:
        logger.exception("Error adding trace task, trace_type %s", trace_task.trace_type)
    finally:
        self.start_timer()                           # ② 确保后台定时器在跑

def collect_tasks(self):                             # ③ 定时器触发时：从队列捞一批
    tasks = []
    while len(tasks) < trace_manager_batch_size and not trace_manager_queue.empty():
        task = trace_manager_queue.get_nowait()
        tasks.append(task)
    return tasks

def run(self):
    tasks = self.collect_tasks()
    if tasks:
        self.send_to_celery(tasks)                   # ★④ 一批一起交给 Celery 异步上报

定时器每隔几秒跑一次（间隔来自 ops_trace_manager.py:1487 的 TRACE_QUEUE_MANAGER_INTERVAL，默认 5 秒；每批上限 BATCH_SIZE 默认 100），send_to_celery（ops_trace_manager.py:1542）把追踪数据存成文件、再丢 Celery 的 ops_trace 队列：

# api/tasks/ops_trace_task.py:41
@shared_task(queue="ops_trace")                      # ★ 追踪上报有专属队列，和业务任务隔离
def process_trace_tasks(self, file_info):
    ...                                              # worker 里真正把数据推给 Langfuse/LangSmith

Dify 不绑死某一家追踪平台，而是做成"可插拔"。看 OpsTraceProviderConfigMap（core/ops/ops_trace_manager.py:223 附近），按需懒加载各家的配置和追踪实现：

# api/core/ops/ops_trace_manager.py (Provider 映射节选)
match key:
    case TracingProviderEnum.LANGFUSE:                       # ★ Langfuse
        from dify_trace_langfuse.langfuse_trace import LangFuseDataTrace
        return {"config_class": LangfuseConfig,
                "secret_keys": ["public_key", "secret_key"],
                "other_keys": ["host", "project_key"],
                "trace_instance": LangFuseDataTrace}
    case TracingProviderEnum.LANGSMITH:                      # ★ LangSmith
        from dify_trace_langsmith.langsmith_trace import LangSmithDataTrace
        return {"config_class": LangSmithConfig,
                "secret_keys": ["api_key"],
                "other_keys": ["project", "endpoint"],
                "trace_instance": LangSmithDataTrace}
    case TracingProviderEnum.OPIK: ...                       # 还有 Opik 等