# Celery Task Queue Integration with Flask for Async LLM Operations: Architecture and Implementation

> **The lmforge platform implements a Flask-aware Celery extension that wraps tasks in application context, enabling background workers to access Flask configurations, database sessions, and dependency-injected services while pro...

- Repository: [Haohao/lmforge-end-to-end-llmops-platform-for-multi-model-agents](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents)
- Tags: architecture
- Published: 2026-03-03

---

**The lmforge platform implements a Flask-aware Celery extension that wraps tasks in application context, enabling background workers to access Flask configurations, database sessions, and dependency-injected services while processing long-running LLM operations asynchronously.**

The lmforge-end-to-end-llmops-platform-for-multi-model-agents repository implements a robust Celery task queue integration with Flask to handle asynchronous LLM operations without blocking HTTP requests. This architecture decouples client-facing API endpoints from computationally intensive workloads like document indexing and agent application creation, ensuring responsive user experiences while leveraging Flask's ecosystem within background workers.

## Flask-Aware Celery Extension with Application Context

The core integration resides in [`api/internal/extension/celery_extension.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/extension/celery_extension.py), which defines a custom `FlaskTask` class that subclasses `celery.Task`. This class overrides the `__call__` method to wrap task execution within `app.app_context()`, ensuring that each worker process has access to Flask configurations, database connections, and other initialized extensions.

The extension creates a Celery instance using `task_cls=FlaskTask`, configures it from `app.config["CELERY"]`, and registers it in `app.extensions["celery"]` for global access throughout the application.

## Application Bootstrap and Initialization Order

During startup, the `Http` class in [`api/internal/server/http.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/server/http.py) orchestrates extension initialization. The Celery extension initializes after database, Weaviate, migrations, logging, and Redis extensions to ensure that background tasks can leverage these resources.

The initialization sequence ensures that when workers spawn, they inherit a fully configured Flask application context with active database sessions and vector store connections ready for LLM operations.

## Defining Asynchronous Tasks for LLM Operations

Background jobs are defined as standard Celery tasks using the `@shared_task` decorator, located in modules like [`api/internal/task/document_task.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/task/document_task.py) and [`api/internal/task/app_task.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/task/app_task.py). These tasks follow a lazy import pattern to avoid circular dependencies and minimize worker startup time.

For example, the `build_documents` task in [`document_task.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/document_task.py) imports the `IndexingService` from the Flask dependency injector only when the task executes:

```python
@shared_task
def build_documents(document_ids: list[UUID]) -> None:
    from app.http.module import injector
    from internal.service.indexing_service import IndexingService
    indexing_service = injector.get(IndexingService)
    indexing_service.build_documents(document_ids)

```

Similarly, `auto_create_app` in [`app_task.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/app_task.py) handles asynchronous agent application creation, allowing the HTTP layer to return immediately while the worker processes LLM configuration and vector store indexing.

## Triggering Tasks from Business Logic

The integration demonstrates practical usage in [`api/internal/service/assistant_agent_service.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/service/assistant_agent_service.py), where LangChain tools enqueue background work. The `create_app` tool accepts parameters from an LLM agent and immediately delegates to `auto_create_app.delay()`, returning a confirmation message to the agent without waiting for completion:

```python
@tool("create_app", args_schema=CreateAppInput)
def create_app(name: str, description: str) -> str:
    auto_create_app.delay(name, description, account_id)
    return f"已调用后端异步任务创建Agent应用。\n应用名称: {name}\n应用描述: {description}"

```

This pattern ensures that conversational AI interfaces remain responsive while complex LLM operations execute in dedicated worker processes.

## Worker Configuration and Broker Setup

Celery configuration resides in [`api/config/default_config.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/config/default_config.py), specifying Redis as both broker and result backend:

```python
"CELERY_BROKER_DB": 1,
"CELERY_RESULT_BACKEND_DB": 1,

```

The configuration dictionary maps to standard Celery settings, allowing seamless switching to RabbitMQ or Amazon SQS by updating configuration values without modifying application code. Workers consume from the Redis database specified, ensuring isolated task queues for the LLMOps platform.

## Summary

- **Flask-aware task context**: The custom `FlaskTask` class in [`celery_extension.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/celery_extension.py) ensures every worker executes within the Flask application context, providing access to configurations and database sessions.
- **Lazy-loaded dependencies**: Tasks in [`document_task.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/document_task.py) and [`app_task.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/app_task.py) use runtime imports and dependency injection to avoid circular imports and keep workers lightweight.
- **Async workflow integration**: Services like `AssistantAgentService` trigger tasks via `.delay()` to offload LLM operations, maintaining responsive HTTP endpoints while background workers handle document indexing and app creation.
- **Redis-backed configuration**: Default settings in [`default_config.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/default_config.py) use Redis for both broker and result backend, with configurable databases for environment isolation.

## Frequently Asked Questions

### How does the Flask application context get passed to Celery workers?

The `FlaskTask` class defined in [`api/internal/extension/celery_extension.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/extension/celery_extension.py) subclasses `celery.Task` and overrides the `__call__` method. When a worker executes a task, this wrapper method enters the Flask application context using `with app.app_context():` before running the actual task logic, ensuring database connections and configurations are available.

### Why do Celery tasks use lazy imports instead of top-level imports?

Tasks in files like [`api/internal/task/document_task.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/task/document_task.py) import services inside the function body rather than at the module level to prevent circular dependency issues during worker startup. Since the Flask application context initializes fully before tasks execute, lazy imports combined with the dependency injector (`injector.get`) ensure services are resolved only when needed within the active application context.

### What happens when an LLM agent triggers a long-running operation?

When the Assistant Agent service in [`api/internal/service/assistant_agent_service.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/service/assistant_agent_service.py) receives a tool call requiring heavy processing (like creating an application), it immediately calls `auto_create_app.delay()` to enqueue the task. The HTTP response returns to the LLM agent instantly with a confirmation message, while a separate Celery worker process picks up the task from the Redis broker and executes the lengthy LLM configuration and indexing operations in the background.

### Can the Celery broker be switched from Redis to RabbitMQ or SQS?

Yes. The configuration in [`api/config/default_config.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/config/default_config.py) uses standard Celery configuration keys for the broker and result backend. By updating the `CELERY` dictionary in the Flask config—specifically the broker URL and result backend settings—you can switch to RabbitMQ, Amazon SQS, or other supported brokers without modifying the task definitions or extension code.