Generating Custom Blog Post Images with AI using a Serverless Azure Function

Denis Gontcharov — Fri, 03 Mar 2023 19:46:08 +0000

An Artificial Intelligence generated these pictures for my personal website...

Every article I publish on my website contains an image. In the past I handpicked these images myself (usually from Unsplash).

In 2023 I switched to daily publishing. Choosing a picture by hand every day was too cumbersome. This is why I had the same photo for every blog post which looked rather boring.

Two weeks ago I decided to solve this problem as a fun weekend project. I wanted to build an AI solution that would automatically generate an image whenever I published a new article on my website.

I wanted the image to complement the text so I chose a text-to-image model that creates an image based on a string of text, for example “man riding a horse on a beach at sunset”.

Of course, the entire article text would be too long for the image generation model. This is why I let a second AI model generate a three-sentence summary of my article to use as the prompt for the image generation model.

Here’s an overview of the automated workflow:

I make a new Git commit with today’s article on the main branch
This triggers a GitHub Action workflow that executes a Python script
The Python script makes an HTTP POST request with the article text to an Azure Function App in the cloud
The Azure Function App executes a serverless function that:
a) Runs the text summary model (Azure Cognitive Service for Language)
b) Runs the image generation model (stable-diffusion on Replicate AI)
c) Uploads the image to Azure Blob Storage
The Azure Static Webb App that hosts my website reads the image from Azure Blob Storage

The entire workflow runs without me - I just see the new image appear on my website. To me this is the beauty of automation.

Here's the workflow in detail with code:

1. Git commit

My articles are treated the same way as code. Pushing a commit to the main brain triggers the GitHub Action workflow below.

2. GitHub Action

This workflow file runs the Python file generate_article_image.py in the next step. Note that the Python file relies on an API key that is stored as a GitHub Secret.

name: Run Python script on push to function-app branch

on:
  push:
    branches: [ main ]

jobs:
  run-script:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v2

      - name: Run Python script
        env:
            FUNCTION_APP_HOST_KEY: ${{ secrets.FUNCTION_APP_HOST_KEY }}
        run: python python/generate_article_image.py

3. The HTTP request in the Python file

The details of this file aren't terribly important. The gist of it is that the script parses the latest Markdown file with the article text and returns the article's text and tags.

A second function makes a HTTP request to the Azure Function App (shown later).

import requests
import glob
import re
import os


def get_last_article(path):
    files = glob.glob(os.path.join(path, '[0-9][0-9][0-9][0-9]*.md'))
    files.sort(reverse=True)
    if not files:
        return None

    with open(files[0], 'r') as f:
        lines = f.readlines()
        if len(lines) < 4:
            return None
        tags_line = lines[3].strip()
        image_line = lines[4].strip()
        article_text = ''.join(lines[7:]).strip()

    image_uuid = re.search(r'\/(\w{8}-\w{4}-\w{4}-\w{4}-\w{12})\.', image_line).group(1)
    article_tags = re.search(r'\[(.*?)\]', tags_line).group(1)
    return image_uuid, article_tags, article_text


def post_request(image_uuid, article_tags, article_text):
    function_app_host_key = os.environ.get('FUNCTION_APP_HOST_KEY')
    url = f"https://functionAppbvmu3zl2dwsgc.azurewebsites.net/api/generate_image?code={function_app_host_key}"
    payload = {
        "image_uuid": f"{image_uuid}",
        "article_tags": f"{article_tags}",
        "article_text": f"{article_text}"
    }
    response = requests.post(url, json=payload)
    return response


if __name__ == '__main__':
    path = '_posts'
    image_uuid, article_tags, article_text = get_last_article(path)
    response = post_request(image_uuid, article_tags, article_text)
    print(response.content)

4. Azure Function App with serverless function

This is a Python file that contains a number of functions to:

Runs the text summary model
Runs the image generation model
Upload the image to Azure Blob Storage

Note that the Function App retrieves the necessary secrets for the Replicate and Azure Blob Storage API from Azure Key Vault.

import azure.functions as func
import logging
import os
import replicate
import requests

from azure.ai.textanalytics import TextAnalyticsClient, ExtractSummaryAction
from azure.core.credentials import AzureKeyCredential
from azure.identity import ManagedIdentityCredential
from azure.keyvault.secrets import SecretClient
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient
from azure.core.exceptions import ResourceExistsError


def authenticate_client(key):
    """Authenticate against Azure cognitiveservices. """
    ta_credential = AzureKeyCredential(key)
    text_analytics_client = TextAnalyticsClient(
            endpoint='https://summarize-daily-newsletter.cognitiveservices.azure.com/',
            credential=ta_credential)
    return text_analytics_client


def sample_extractive_summarization(client, text, num_sentence=3):
    """Summarize a large body of text into a specified number of sentences. """
    poller = client.begin_analyze_actions(
        text,
        actions=[
            ExtractSummaryAction(max_sentence_count=num_sentence)
        ],
    )

    document_results = poller.result()
    for result in document_results:
        extract_summary_result = result[0]  # first document, first result
        if extract_summary_result.is_error:
            logging.error("...Is an error with code '{}' and message '{}'".format(
                extract_summary_result.code, extract_summary_result.message
            ))
        else:
            return " ".join([sentence.text for sentence in extract_summary_result.sentences])


def generate_image(text):
    """Run an API call against stable diffusion to convert a string into an image. """
    model = replicate.models.get("stability-ai/stable-diffusion")
    version = model.versions.get("f178fa7a1ae43a9a9af01b833b9d2ecf97b1bcb0acfd2dc5dd04895e042863f1")

    inputs = {
        'prompt': text,
        'negative_prompt': "person arm hand body human",
        'width': 1024,
        'height': 768,
        'prompt_strength': 1,
        'num_outputs': 1,
        'num_inference_steps': 25,
        'guidance_scale': 7.5,
        'scheduler': "DPMSolverMultistep",
    }

    output = version.predict(**inputs)
    url = output[0]
    return url


def download_image_from_url(url, image_uuid):
    """Download the image from Replicate's URL to the local filesystem. """
    response = requests.get(url)
    local_filename = f'/tmp/{image_uuid}.png'
    with open(local_filename, 'wb') as f:
        f.write(response.content)
    return local_filename


def upload_image_to_azure_blob(image_path, connection_string, container_name, blob_name):
    """Upload the downloaded image to Azure Blob Storage """
    blob_service_client = BlobServiceClient.from_connection_string(connection_string)
    container_client = blob_service_client.get_container_client(container_name)
    blob_client = container_client.get_blob_client(blob_name)

    with open(image_path, "rb") as data:
        try:
            blob_client.upload_blob(data, overwrite=False)
            logging.info(f'Uploaded image {image_blob} to Azure Blob Storage')

        except ResourceExistsError:
            logging.info(f"Image {blob_name} already exists in container {container_name}, skipping upload")


def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    key_vault_name = os.environ.get('KEY_VAULT_NAME')

    credential = ManagedIdentityCredential()
    secret_client = SecretClient(
        vault_url=f"https://{key_vault_name}.vault.azure.net/",
        credential=credential
    )

    try:
        req_body = req.get_json()
    except ValueError:
        logging.exception('Error: failed to parse request body')
    else:
        image_uuid = req_body.get('image_uuid')
        article_tags = req_body.get('article_tags')
        article_text = req_body.get('article_text')

    azure_language_key = secret_client.get_secret('azureLanguageKey').value
    client = authenticate_client(azure_language_key)
    # model expects the article_text argument to must be a list
    article_summary = sample_extractive_summarization(client, [article_text])

    model_input = f'{article_tags} {article_summary}'
    logging.info(f'model_input: {model_input}')

    os.environ['REPLICATE_API_TOKEN'] = secret_client.get_secret('replicateAPIToken').value
    logging.info(os.environ.get('REPLICATE_API_TOKEN'))
    image_url = generate_image(model_input)

    image_path = download_image_from_url(image_url, image_uuid)
    image_blob = f'{image_uuid}.png'

    azure_storage_account_key = secret_client.get_secret('azureStorageAccountKey').value
    azure_storage_account_conn= ';'.join([
        'DefaultEndpointsProtocol=https',
        'AccountName=gontcharovdata',
        f'AccountKey={azure_storage_account_key}',
        'EndpointSuffix=core.windows.net'
    ])

    upload_image_to_azure_blob(
        image_path,
        azure_storage_account_conn,
        'images-generated',
        image_blob
    )

5. Azure Static Webb App

My personal website is also hosted on Azure as a Static Web App.

The pictures are uploaded to a public Azure Blob Storage container - that's where my website is reading the images from.