教程 | 如何使用Keras、Redis、Flask和Apache把深度学习模型部署到生产环境？

├── helpers.py
├── jemma.png
├── keras_rest_api_app.wsgi
├── run_model_server.py
├── run_web_server.py
├── settings.py
├── simple_request.py
└── stress_test.py

# initialize Redis connection settings
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0

# initialize constants used to control image spatial dimensions and
# data type
IMAGE_WIDTH = 224
IMAGE_HEIGHT = 224
IMAGE_CHANS = 3
IMAGE_DTYPE = "float32"

# initialize constants used for server queuing
IMAGE_QUEUE = "image_queue"
BATCH_SIZE = 32
SERVER_SLEEP = 0.25
CLIENT_SLEEP = 0.25

# import the necessary packages
import numpy as np
import base64
import sys

def base64_encode_image(a):
    # base64 encode the input NumPy array
    return base64.b64encode(a).decode("utf-8")

def base64_decode_image(a, dtype, shape):
    # if this is Python 3, we need the extra step of encoding the
    # serialized NumPy string as a byte object
    if sys.version_info.major == 3:
        a = bytes(a, encoding="utf-8")

    # convert the string to a NumPy array using the supplied data
    # type and target shape
    a = np.frombuffer(base64.decodestring(a), dtype=dtype)
    a = a.reshape(shape)

    # return the decoded image
    return a

# import the necessary packages
from keras.preprocessing.image import img_to_array
from keras.applications import imagenet_utils
from PIL import Image
import numpy as np
import settings
import helpers
import flask
import redis
import uuid
import time
import json
import io

# initialize our Flask application and Redis server
app = flask.Flask(__name__)
db = redis.StrictRedis(host=settings.REDIS_HOST,
    port=settings.REDIS_PORT, db=settings.REDIS_DB)

def prepare_image(image, target):
    # if the image mode is not RGB, convert it
    if image.mode != "RGB":
        image = image.convert("RGB")

    # resize the input image and preprocess it
    image = image.resize(target)
    image = img_to_array(image)
    image = np.expand_dims(image, axis=0)
    image = imagenet_utils.preprocess_input(image)

    # return the processed image
    return image

@app.route("/")
def homepage():
    return "Welcome to the PyImageSearch Keras REST API!"

@app.route("/predict", methods=["POST"])
def predict():
    # initialize the data dictionary that will be returned from the
    # view
    data = {"success": False}

    # ensure an image was properly uploaded to our endpoint
    if flask.request.method == "POST":
        if flask.request.files.get("image"):
            # read the image in PIL format and prepare it for
            # classification
            image = flask.request.files["image"].read()
            image = Image.open(io.BytesIO(image))
            image = prepare_image(image,
                (settings.IMAGE_WIDTH, settings.IMAGE_HEIGHT))

            # ensure our NumPy array is C-contiguous as well,
            # otherwise we won't be able to serialize it
            image = image.copy(order="C")

            # generate an ID for the classification then add the
            # classification ID + image to the queue
            k = str(uuid.uuid4())
            image = helpers.base64_encode_image(image)
            d = {"id": k, "image": image}
            db.rpush(settings.IMAGE_QUEUE, json.dumps(d))

            # keep looping until our model server returns the output
            # predictions
            while True:
                # attempt to grab the output predictions
                output = db.get(k)

                # check to see if our model has classified the input
                # image
                if output is not None:
                    # add the output predictions to our data
                    # dictionary so we can return it to the client
                    output = output.decode("utf-8")
                    data["predictions"] = json.loads(output)

                    # delete the result from the database and break
                    # from the polling loop
                    db.delete(k)
                    break

                # sleep for a small amount to give the model a chance
                # to classify the input image
                time.sleep(settings.CLIENT_SLEEP)

            # indicate that the request was a success
            data["success"] = True

    # return the data dictionary as a JSON response
    return flask.jsonify(data)

# for debugging purposes, it's helpful to start the Flask testing
# server (don't use this for production
if __name__ == "__main__":
    print("* Starting web service...")
    app.run()

# import the necessary packages
from keras.applications import ResNet50
from keras.applications import imagenet_utils
import numpy as np
import settings
import helpers
import redis
import time
import json

# connect to Redis server
db = redis.StrictRedis(host=settings.REDIS_HOST,
    port=settings.REDIS_PORT, db=settings.REDIS_DB)

def classify_process():
    # load the pre-trained Keras model (here we are using a model
    # pre-trained on ImageNet and provided by Keras, but you can
    # substitute in your own networks just as easily)
    print("* Loading model...")
    model = ResNet50(weights="imagenet")
    print("* Model loaded")

    # continually pool for new images to classify
    while True:
        # attempt to grab a batch of images from the database, then
        # initialize the image IDs and batch of images themselves
        queue = db.lrange(settings.IMAGE_QUEUE, 0,
            settings.BATCH_SIZE - 1)
        imageIDs = []
        batch = None

        # loop over the queue
        for q in queue:
            # deserialize the object and obtain the input image
            q = json.loads(q.decode("utf-8"))
            image = helpers.base64_decode_image(q["image"],
                settings.IMAGE_DTYPE,
                (1, settings.IMAGE_HEIGHT, settings.IMAGE_WIDTH,
                    settings.IMAGE_CHANS))

            # check to see if the batch list is None
            if batch is None:
                batch = image

            # otherwise, stack the data
            else:
                batch = np.vstack([batch, image])

            # update the list of image IDs
            imageIDs.append(q["id"])

        # check to see if we need to process the batch
        if len(imageIDs) > 0:
            # classify the batch
            print("* Batch size: {}".format(batch.shape))
            preds = model.predict(batch)
            results = imagenet_utils.decode_predictions(preds)

            # loop over the image IDs and their corresponding set of
            # results from our model
            for (imageID, resultSet) in zip(imageIDs, results):
                # initialize the list of output predictions
                output = []

                # loop over the results and add them to the list of
                # output predictions
                for (imagenetID, label, prob) in resultSet:
                    r = {"label": label, "probability": float(prob)}
                    output.append(r)

                # store the output predictions in the database, using
                # the image ID as the key so we can fetch the results
                db.set(imageID, json.dumps(output))

            # remove the set of images from our queue
            db.ltrim(settings.IMAGE_QUEUE, len(imageIDs), -1)

        # sleep for a small amount
        time.sleep(settings.SERVER_SLEEP)

# if this is the main thread of execution start the model server
# process
if __name__ == "__main__":
    classify_process()

# add our app to the system path
import sys
sys.path.insert(0, "/var/www/html/keras-complete-rest-api")

# import the application and away we go...
from run_web_server import app as application

# USAGE
# python stress_test.py

# import the necessary packages
from threading import Thread
import requests
import time

# initialize the Keras REST API endpoint URL along with the input
# image path
KERAS_REST_API_URL = "http://localhost/predict"
IMAGE_PATH = "jemma.png"

# initialize the number of requests for the stress test along with
# the sleep amount between requests
NUM_REQUESTS = 500
SLEEP_COUNT = 0.05

def call_predict_endpoint(n):
    # load the input image and construct the payload for the request
    image = open(IMAGE_PATH, "rb").read()
    payload = {"image": image}

    # submit the request
    r = requests.post(KERAS_REST_API_URL, files=payload).json()

    # ensure the request was sucessful
    if r["success"]:
        print("[INFO] thread {} OK".format(n))

    # otherwise, the request failed
    else:
        print("[INFO] thread {} FAILED".format(n))

# loop over the number of threads
for i in range(0, NUM_REQUESTS):
    # start a new thread to call the API
    t = Thread(target=call_predict_endpoint, args=(i,))
    t.daemon = True
    t.start()
    time.sleep(SLEEP_COUNT)

# insert a long sleep so we can wait until the server is finished
# processing the images
time.sleep(300)

$ wget http://download.redis.io/redis-stable.tar.gz
$ tar xvzf redis-stable.tar.gz
$ cd redis-stable
$ make
$ sudo make install

$ mkvirtualenv keras_flask -p python3

$ pip install numpy
$ pip install scipy h5py
$ pip install tensorflow==1.4.1 # tensorflow-gpu==1.4.1 for GPU machines
$ pip install keras
$ pip install flask gevent
$ pip install imutils requests
$ pip install redis
$ pip install Pillow

$ sudo apt-get install apache2

$ sudo apt-get install libapache2-mod-wsgi-py3
$ sudo a2enmod wsgi

$ sudo apt-get install libapache2-mod-wsgi
$ sudo a2enmod wsgi

$ ls ~
keras-complete-rest-api

$ cd /var/www/html/
$ sudo ln -s ~/keras-complete-rest-api keras-complete-rest-api

$ sudo vi /etc/apache2/sites-available/000-default.conf

WSGIPythonHome /home/ubuntu/.virtualenvs/keras_flask/bin
WSGIPythonPath /home/ubuntu/.virtualenvs/keras_flask/lib/python3.5/site-packages

<VirtualHost *:80>
    ...
</VirtualHost>

<VirtualHost *:80>
    ...

    WSGIDaemonProcess keras_rest_api_app threads=10
    WSGIScriptAlias / /var/www/html/keras-complete-rest-api/keras_rest_api_app.wsgi

    <Directory /var/www/html/keras-complete-rest-api>
        WSGIProcessGroup keras_rest_api_app
        WSGIApplicationGroup %{GLOBAL}
        Order deny,allow
        Allow from all
    </Directory>

    ...
</VirtualHost>

$ cd /usr/lib
$ sudo ln -s /usr/local/cuda/lib64/* ./

$ sudo service apache2 restart

$ ls /var/log/apache2
access.log error.log other_vhosts_access.log

$ tail -f /var/log/apache2/error.log

$ sudo service apache2 start

$ redis-server

$ python run_model_server.py
* Loading model...
...
* Model loaded

$ curl -X POST -F image=@jemma.png 'http://localhost/predict'
{
  "predictions": [
    {
      "label": "beagle", 
      "probability": 0.9461532831192017
    }, 
    {
      "label": "bluetick", 
      "probability": 0.031958963721990585
    }, 
    {
      "label": "redbone", 
      "probability": 0.0066171870566904545
    }, 
    {
      "label": "Walker_hound", 
      "probability": 0.003387963864952326
    }, 
    {
      "label": "Greater_Swiss_Mountain_dog", 
      "probability": 0.0025766845792531967
    }
  ], 
  "success": true
}

$ python stress_test.py 
[INFO] thread 3 OK
[INFO] thread 0 OK
[INFO] thread 1 OK
...
[INFO] thread 497 OK
[INFO] thread 499 OK
[INFO] thread 498 OK

* Batch size: (4, 224, 224, 3)
* Batch size: (9, 224, 224, 3)
* Batch size: (9, 224, 224, 3)
* Batch size: (8, 224, 224, 3)
...
* Batch size: (2, 224, 224, 3)
* Batch size: (10, 224, 224, 3)
* Batch size: (7, 224, 224, 3)

$ redis-cli
> FLUSHALL