DEV Community: Maximilien Soviche

chatbot videos

Maximilien Soviche — Tue, 17 Dec 2024 21:30:12 +0000

Un Chatbot RAG pour explorer du contenu vidéo : une architecture event-driven et serverless sur Google Cloud

Maximilien Soviche — Mon, 09 Dec 2024 16:02:09 +0000

Un chatbot capable de répondre à vos questions tout en vous fournissant un lien cliquable vers la bonne vidéo YouTube, et au bon endroit dans la vidéo. C'est ce que permet cette architecture conçue sur Google Cloud, mêlant IA générative et approche serverless.

Dans cet article, nous explorerons comment mettre en place un chatbot RAG (Retrieval-Augmented Generation) qui utilise un corpus de vidéos pour fournir des réponses enrichies et pertinentes. Ce système repose sur une architecture event-driven pilotée par EventArc et des microservices déployés avec Cloud Run.

Une telle solution peut être appliquée dans divers contextes :

Formation : Accès rapide à des segments vidéo éducatifs pertinents.
Entreprise : Recherche d'informations dans des vidéos internes comme des réunions ou des formations.
Médias et divertissement : Exploration d'archives vidéo ou de contenus créatifs selon des requêtes contextuelles.
Parlement : Exploration de contenus vidéos longs : séances, commissions, etc

L'application proposée ici est l'exploration d'une playlist Youtube de formation sur les services Google Cloud.

Qu'est ce que le RAG ?

Le RAG veut dire Retrieval-Augmented Generation, et est la plupart du temps appliqué sur des documents textuels. Plutôt que de poser une question à un LLM, on va effectuer les étapes suivantes :

Ingestion

On prend des documents pertinents pour l'application souhaitée.
On découpe ces documents en morçeaux, appelés chunks.
On construit une représentation de ces chunks appelée embedding, lisible par un algorithme de recherche par similarité.
On stocke ces embeddings dans une base de données.

Récupération

L'utilisateur pose une question qui est transformée en embedding.
L'embedding est comparé à ceux de la base de données afin de retrouver du contexte pertinent.
Un construit un prompt augmenté qui comporte la question de l'utilisateur ainsi que les éléments de contexte appropriés.
Le LLM peut alors fournir une réponse pertinente sur ce prompt grâce au contexte.

Le RAG sur du contenu vidéo

Pour effectuer du RAG sur du contenu vidéo, l'idée principale est d'effectuer une transcription du contenu, afin de se ramener à un système RAG classique avec des documents.

Vue d’ensemble de l’architecture

Le système se divise en deux parties :

Ingestion des vidéos : Un pipeline de traitement divise les vidéos en segments et génère des données prêtes pour la recherche.
Retrieval et Chatbot : Le chatbot trouve les segments pertinents, fournit une réponse augmentée et un lien vers la vidéo correspondante.

Partie 1 : Ingestion des Vidéos

L’ingestion repose sur cinq microservices Cloud Run. Ces services traitent les vidéos, segmentent leur contenu, et transforment les données en embeddings vectoriels stockés dans Firestore pour la recherche. Chaque microservice produit un résultat dans un bucket spécifique, ce qui déclenche le traitement suivant via un déclencheur EventArc.

Étape 1 : Téléchargement des vidéos avec le service Downloader

L'utilisateur fournit une playlist YouTube via une API, et déclenche son téléchargement via un appel API sur ce service.

Les métadonnées de la playlist (titres, descriptions, ID des vidéos) sont enregistrées dans une collection Firestore nommée metadata.

Étape 2 : Segmentation des vidéos avec le service Splitter

Chaque vidéo est découpée en segments de 2 minutes avec un recouvrement de 30 secondes. Ce recouvrement permet d'éviter la perte de contexte lorsqu'une phrase est coupée en deux. La piste audio est extraite pour chaque segment et stockée dans le bucket dédié.

La collection metadata dans Firestore est mise à jour. On associe à chaque vidéo ses segments avec leurs informations, notamment les timestamps de début et de fin.

Le choix de la durée des segments (2 minutes) et de la taille de recouvrement peuvent être modifiés selon les besoins et le cas traité.

Étape 3 : Transcription des segments audio avec le service Transcriber

Les pistes audio des segments sont converties en texte grâce à l’API Speech-to-Text de Google Cloud (modèle Chirp). On obtient des transcriptions brutes dans un bucket.

Chirp est un modèle de transcription speech-to-text de pointe, conçu pour fournir des services de transcription extrêmement précis et rapides. Il prend en charge une large gamme de langues et de dialectes, ce qui en fait un choix polyvalent pour répondre à divers besoins de transcription. Il est disponible via l'API Speech-to-Text.

Étape 4 : Formatage des transcriptions avec le service Formatter

Les transcriptions brutes sont corrigées pour améliorer la grammaire et la clarté grâce à Gemini 1.5 Flash. On obtient ainsi des documents textuels dans un bucket, ce qui permet d'implémenter une logique RAG classique par la suite.

Étape 5 : Générer des embeddings avec le service Feeder

Les transcriptions finales sont transformées en embeddings vectoriels à l’aide de LangChain.

Le service Firestore sert de base de données pour les embeddings, avec une collection dédiée nommée embeddings.

Chaque embedding est stocké avec l'identifiant du segment audio auquel il appartient, ce qui permettra de retrouver les métadonnées dans la collection metadata.

Partie 2 : Retrieval et Chatbot

Une fois le corpus indexé, le chatbot répond aux questions des utilisateurs en recherchant les segments pertinents dans les embeddings et en fournissant un lien précis vers la vidéo source. Le timestamp correspondra au point de départ du segment audio le plus pertinent.

Étapes du chatbot :

Requête utilisateur : L'utilisateur pose une question via Google Chat.
Recherche par similarité :
- Les embeddings correspondant à la question sont récupérés dans Firestore.
- Les métadonnées associées (ID vidéo, segment, timestamps) sont extraites.
Reranking : La recherche par similarité permet d'obtenir les N chunks les plus pertinents (N=5 dans notre cas). L'étape de reranking consiste à utiliser un LLM pour réordonner ces chunks par ordre de pertinence, et n'en sélectionner qu'un.
Génération de réponse :
- Un prompt enrichi par le contexte du segment (le chunk retrouvé après reranking) est envoyé au LLM.
- Une URL est reconstruite avec le bon ID vidéo et le timestamp exact.
Validation finale avec Gemini 1.5 Flash :

On ne veut pas que l'URL de la vidéo soit fournie pour chaque réponse, sinon une question qui n'a rien à voir avec le contexte aboutirait quand même vers le contenu le moins éloigné en termes de similarité.

Le prompt augmenté est le suivant :

"""
You are a chatbot working for Stack Labs company. Your job is to retrieve information from videos.
You have to answer users' questions using the context given below, taken from video transcripts.
Don't invent anything, just use the context.
If you don't have a context, answer that you can't find one.

You must answer in english.

Context:
{context}

Question:
{query}
"""

Par conséquent la réponse du LLM peut comporter trois types de réponses :

Le chatbot dit qui il est et à quoi il sert
Le chatbot fournit un discours technique sur la base du contexte retrouvé dans Firestore
Le chatbot dit qu'il n'a pas de contexte

On ne veut fournir un lien cliquable à l'utilisateur que s'il s'agit d'un discours technique. On va donc faire un appel supplémentaire à Gemini 1.5 Flash, en le forçant à fournir l'un des trois flags suivants :
- who_am_i : Le chatbot se présente.
- technical_speech : Réponse technique.
- no_context : Aucune information trouvée.

Il est possible de configurer Gemini pour qu'il retourne une réponse au format JSON en respectant un schéma :


classification_schema = {
    "type": "object",
    "properties": {
        "response_type": {
            "type": "string",
            "enum": [
                "who_am_i",
                "technical_speech",
                "no_context"
            ],
            "description": "Message type based on context."
        }
    }
}

Le prompt donné à Gemini pour la classification est le suivant :

"""
Message : {message}

Classify this message by categorizing it into one of the following categories:
    - who_am_i: the message is someone presenting himself
    - technical_speech: Contains technical jargon, code-related keywords, or specific terminologies.
    - no_context: the message says that no context is found
"""

Si la réponse appartient à la catégorie technical_speech, l’URL est ajoutée à la réponse.

Voici un aperçu du résultat :

Avantages de cette architecture

1. Liens contextuels précis

Le chatbot fournit des réponses enrichies avec des liens vidéo cliquables, garantissant une navigation rapide vers l’information.

2. Scalabilité et efficacité

L’utilisation de Cloud Run et EventArc permet une exécution à la demande et une scalabilité horizontale, tout en garantissant des coûts contenus.

3. Modularité

Chaque microservice est indépendant, ce qui facilite les mises à jour et l’ajout de nouvelles fonctionnalités. Chaque microservice Cloud Run peut être géré de façon indépendante : gestion de la capacité mémoire et CPU, montage d'un volume, etc.

4. Personnalisation

Cette solution peut être facilement personnalisée pour différents contextes, notamment le choix de la taille et du recouvrement des différents segments audios.

Conclusion

Cette architecture montre comment Google Cloud peut permettre une recherche d'informations efficace dans un corpus de vidéos. Le choix d’une approche serverless et event-driven garantit un système flexible, scalable, facile à maintenir et peu coûteux.

Merci d'avoir lu ! Nous sommes Victor et Maximilien, développeur et data engineer chez Stack Labs. Si vous souhaitez découvrir la Stack Labs Data Platform ou rejoindre une équipe tech motivée, n'hésitez pas à nous contacter.

Enhancing Video to Text Transcription with AI: An Asynchronous Solution on Google Cloud Platform

Maximilien Soviche — Wed, 03 Jul 2024 09:50:46 +0000

Asynchronous transcription can be applied in various contexts. For developers looking to implement a robust, scalable, and efficient transcription solution, the Google Cloud Platform (GCP) offers an ideal environment. In this article, we’ll explore an asynchronous video-to-text transcription solution built with GCP using an event-driven and serverless architecture.

Potential Applications

The provided solution is particularly well-suited for long video-to-text transcriptions, efficiently handling videos that are more than an hour long. This makes it ideal for a wide array of applications across various sectors. Here are some examples:

State Institutions or local authorities: Transcribing meetings, hearings, and other official recordings to ensure transparency and accessibility.
Company Meetings: Creating accurate records of internal meetings, conferences, and training sessions to enhance communication.
Educational Institutions: Transcribing lectures, seminars, and workshops to aid in learning and research.

For shorter videos, the Gemini Pro API can handle the entire video-to-text transcription process, offering a streamlined and efficient solution for quicker, smaller-scale transcription needs.

Solution Overview

Our solution comprises three event-driven Cloud Functions, each triggered by specific events in different Cloud Storage buckets using an Eventarc trigger. Event-driven Cloud Functions are deployed pieces of code on GCP, invoked in response to an event in the cloud environment. In our case, we want our functions to be invoked when a file is upload in a specific Cloud Storage bucket. Eventarc is a standardized solution to manage events on GCP. Eventarc triggers route these events between resources. In this particular case, each Eventarc trigger listen to new objects in a specific Cloud Storage bucket, and then triggers the associated Cloud Function. The event data is passed to the function.

More information about Cloud Functions

More information about Eventarc triggers.

The four buckets used in our architecture are:

Video Files Bucket: Where users upload their video files.
Audio Files Bucket: Stores the extracted audio files.
Raw Transcriptions Bucket: Contains the initial transcriptions generated by the Chirp speech-to-text model.
Curated Transcriptions Bucket: Stores the curated transcriptions, enhanced by Gemini.

The application architecture is designed to be modular and scalable. Here’s a step-by-step breakdown of the workflow:

Video Upload and Audio Extraction

When a user uploads a video file to the video-files bucket, the video-to-audio Cloud Function is triggered. This function uses ffmpeg to extract the audio from the video file and save it in the audio-files bucket.

import os
import subprocess
from google.cloud import storage
import functions_framework
import logging

logger = logging.getLogger(__name__)


@functions_framework.cloud_event
def convert_video_to_audio(cloud_event):
    """Video to audio event-triggered cloud function."""

    data = cloud_event.data
    bucket_name = data['bucket']
    video_file_name = data['name']

    destination_bucket_name = os.environ.get("AUDIO_FILES_BUCKET_NAME")

    if not video_file_name.endswith(('.mp4', '.mov', '.avi', '.mkv')):
        logger.info(f"File {video_file_name} is not a supported video format.")
        return

    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(video_file_name)

    tmp_video_file = f"/tmp/{video_file_name}"
    blob.download_to_filename(tmp_video_file)

    audio_file_name = os.path.splitext(video_file_name)[0] + '.mp3'
    tmp_audio_file = f"/tmp/{audio_file_name}"

    command = f"ffmpeg -i {tmp_video_file} -vn -acodec libmp3lame -q:a 2 {tmp_audio_file}"
    subprocess.call(command, shell=True)

    destination_bucket = storage_client.bucket(destination_bucket_name)
    destination_blob = destination_bucket.blob(audio_file_name)
    destination_blob.upload_from_filename(tmp_audio_file)

    os.remove(tmp_video_file)
    os.remove(tmp_audio_file)

    logger.info(f"Converted {video_file_name} to {audio_file_name} and uploaded to {destination_bucket_name}.")

Audio to Text Transcription

The upload of the audio file to the audio-files bucket triggers the audio-to-text Cloud Function. This function uses Chirp, a highly accurate speech-to-text model, and the Speech-to-Text API, to transcribe the audio and stores the raw transcription in the raw-transcriptions bucket.

from google.cloud.speech_v2 import SpeechClient
from google.cloud.speech_v2.types import cloud_speech
from google.api_core.client_options import ClientOptions
from google.cloud import storage
import functions_framework
from typing import Dict
import logging
import time
import os

from . import chirp_model_long


logger = logging.getLogger(__file__)


def transcribe_batch_gcs(
    project_id: str,
    gcs_uri: str,
    region: str = "us-central1"
) -> cloud_speech.BatchRecognizeResults:
    """Transcribes audio from a Google Cloud Storage URI.

    Parameters
    ----------
        project_id: The Google Cloud project ID.
        gcs_uri: The Google Cloud Storage URI.

    Returns
    -------
        The RecognizeResponse.
    """

    client = SpeechClient(
        client_options=ClientOptions(
            api_endpoint=f"{region}-speech.googleapis.com",
        )
    )

    config = cloud_speech.RecognitionConfig(
        auto_decoding_config=cloud_speech.AutoDetectDecodingConfig(),
        language_codes=["fr-FR"],
        model="chirp",
    )

    file_metadata = cloud_speech.BatchRecognizeFileMetadata(uri=gcs_uri)

    request = cloud_speech.BatchRecognizeRequest(
        recognizer=f"projects/{project_id}/locations/{region}/recognizers/_",
        config=config,
        files=[file_metadata],
        recognition_output_config=cloud_speech.RecognitionOutputConfig(
            inline_response_config=cloud_speech.InlineOutputConfig(),
        ),
        processing_strategy=cloud_speech.BatchRecognizeRequest.ProcessingStrategy.DYNAMIC_BATCHING
    )

    operation = client.batch_recognize(request=request)

    logger.info("Waiting for operation to complete...")
    response = operation.result(timeout=1000)

    transcript = ""

    for result in response.results[gcs_uri].transcript.results:
        if len(result.alternatives) > 0:
            logger.info(f"Transcript: {result.alternatives[0].transcript}")
            transcript += f" \n{result.alternatives[0].transcript}"

    logger.debug(f"Transcript: {transcript}")

    return transcript


@functions_framework.cloud_event
def speech_to_text_transcription(cloud_event):
    """Audio file transcription via Speech-to-text API call."""

    data: Dict = cloud_event.data

    event_id = cloud_event["id"]
    event_type = cloud_event["type"]

    input_bucket = data["bucket"]
    audio_file_name = data["name"]

    destination_bucket_name = os.environ.get("RAW_TRANSCRIPTIONS_BUCKET_NAME")

    logger.info(f"Event ID: {event_id}")
    logger.info(f"Event type: {event_type}")
    logger.info(f"Bucket: {input_bucket}")
    logger.info(f"File: {audio_file_name}")

    storage_client = storage.Client()

    start = time.time()

    transcript = transcribe_batch_gcs(
        project_id=os.environ.get("PROJECT_ID"),
        gcs_uri=f"gs://{input_bucket}/{audio_file_name}"
    )

    stop = time.time()

    raw_transcription_file_name = os.path.splitext(audio_file_name)[0] + '_raw.txt'

    destination_bucket = storage_client.bucket(destination_bucket_name)
    destination_blob = destination_bucket.blob(raw_transcription_file_name)
    destination_blob.upload_from_string(transcript, content_type='text/plain; charset=utf-8')

    logger.debug(transcript)
    logger.info(f"JOB DONE IN {round(stop - start, 2)} SECONDS.")

About Chirp:

Chirp is a state-of-the-art speech-to-text model developed to provide highly accurate and fast transcription services. It supports a wide range of languages and dialects, making it a versatile choice for diverse transcription needs. It is available in the Speech-to-Text API.

More information about long audio to text transcription.

Transcription Curation

Finally, the curate-transcription Cloud Function is triggered by the new transcription file in the raw-transcriptions bucket. This function sends the raw transcription to the Gemini API that uses gemini-pro model for curation and stores the refined transcription in the curated-transcriptions bucket.

import vertexai
from vertexai.generative_models import GenerativeModel

import functions_framework
from google.cloud import storage

import time
import logging
import os
from typing import Dict

logger = logging.getLogger(__file__)


@functions_framework.cloud_event
def transcription_correction(cloud_event):
    """Gemini API call to correct and enhance speech-to-text transcription."""

    data: Dict = cloud_event.data

    event_id = cloud_event["id"]
    event_type = cloud_event["type"]

    input_bucket = data["bucket"]
    raw_transcription_filename = data["name"]

    destination_bucket_name = os.environ.get("CURATED_TRANSCRIPTIONS_BUCKET_NAME")

    logger.info(f"Event ID: {event_id}")
    logger.info(f"Event type: {event_type}")
    logger.info(f"Bucket: {input_bucket}")
    logger.info(f"File: {raw_transcription_filename}")

    storage_client = storage.Client()

    input_bucket = storage_client.get_bucket(input_bucket)
    input_blob = input_bucket.get_blob(raw_transcription_filename)
    transcript = input_blob.download_as_string()

    vertexai.init(project=os.environ.get("PROJECT_ID"), location="us-central1")
    model = GenerativeModel(model_name="gemini-1.0-pro-002")

    prompt = f"""
        YOUR CUSTOM PROMPT GOES HERE. 
        PROVIDING CONTEXT AND GIVING INFORMATION ABOUT THE RESULT YOU EXPECT IS NECESSARY.

        {transcript}
    """

    n_tokens = model.count_tokens(prompt)

    logger.info(f"JOB : SPEECH-TO-TEXT TRANSCRIPTION CORRECTION. \n{n_tokens.total_billable_characters} BILLABLE CHARACTERS")
    logger.info(f"RESPONSE WILL PROCESS {n_tokens.total_tokens} TOKENS.")

    start = time.time()

    response = model.generate_content(prompt)

    stop = time.time()

    curated_filename = raw_transcription_filename.replace("_raw", "_curated")

    destination_bucket = storage_client.bucket(destination_bucket_name)
    destination_blob = destination_bucket.blob(curated_filename)
    destination_blob.upload_from_string(response.text, content_type='text/plain; charset=utf-8')

    logger.debug(response.text)
    logger.info(f"JOB DONE IN {round(stop - start, 2)} SECONDS.")

The chosen architecture is modular and event-driven, which brings several advantages:

Scalability: This application can handle short or long videos, up to 8 hours.
Flexibility: The separation of concerns allows for easy maintenance and upgrades. If the user uploads a video in the video-files bucket, the three cloud function will be triggered. But if the user upload an audio in the audio-files bucket, then only the two last cloud functions will be triggered.
Cost-Efficiency: Cloud Functions are serverless. Using Cloud Functions ensures that resources are only used when necessary, reducing costs.

Deployment with Terraform

To ensure our solution is not only powerful but also easily manageable and deployable, we use Terraform for infrastructure as code (IaC). Terraform allows us to define our cloud resources in declarative configuration files, providing several key benefits:

Infrastructure configurations can be version-controlled using Git, following GitOps principles. This means changes to the infrastructure are tracked, reviewed, and can be rolled back if necessary.
As our application grows, Terraform makes it easy to manage our infrastructure by simply updating the configuration files.
Terraform makes the deployment of our application reliable and repeatable.

In this particular case, four cloud storage buckets and three cloud functions are needed. We use one terraform resource for the cloud functions and another for the buckets. This provides a flexible code, and makes it easier to integrate and manage new buckets or cloud functions. More information about terraform : Terraform documentation.

# locals.tf

locals {
    function_definitions = [
    {
        name = "convert_video_to_audio",
        source_dir = "../services/video_to_audio_cloud_function"
        input_bucket = var.video_files_bucket_name
    },
    {
        name = "speech_to_text_transcription",
        source_dir = "../services/transcript_cloud_function"
        input_bucket = var.audio_files_bucket_name
    },
    {
        name = "transcription_correction",
        source_dir = "../services/gemini_cloud_function"
        input_bucket = var.raw_transcriptions_bucket_name
    }
]
}

From this locals.tf file, the user can add, configure or remove cloud functions very easily. The cloud_functions.tf file uses one terraform resource for all cloud functions, and loops over these function definitions.

# cloud_functions.tf

resource "random_id" "bucket_prefix" {
  byte_length = 8
}

resource "google_storage_bucket" "source_code_bucket" {
  name                        = "${random_id.bucket_prefix.hex}-source-code-bucket"
  location                    = var.location
  force_destroy = true

  uniform_bucket_level_access = true
}

data "archive_file" "function_sources" {
  for_each    = { for def in local.function_definitions : def.name => def }
  type        = "zip"
  output_path = "/tmp/${each.value.name}-source.zip"
  source_dir  = each.value.source_dir
}

resource "google_storage_bucket_object" "function_sources" {
  for_each = data.archive_file.function_sources
  name     = "${basename(each.value.output_path)}-${each.value.output_md5}.zip"
  bucket   = google_storage_bucket.source_code_bucket.name
  source   = each.value.output_path
}

data "google_storage_project_service_account" "default" {}

resource "google_project_iam_member" "gcs_pubsub_publishing" {
  project = var.deploy_project
  role    = "roles/pubsub.publisher"
  member  = "serviceAccount:${data.google_storage_project_service_account.default.email_address}"
}

resource "google_service_account" "account" {
  account_id   = "gcf-sa"
  display_name = "Test Service Account - used for both the cloud function and eventarc trigger in the test"
}

resource "google_project_iam_member" "roles" {
  for_each = {
    "invoking"                   = "roles/run.invoker"
    "event_receiving"            = "roles/eventarc.eventReceiver"
    "artifactregistry_reader"    = "roles/artifactregistry.reader"
    "storage_object_admin"       = "roles/storage.objectUser"
    "speech_client"              = "roles/speech.client"
    "insights_collector_service" = "roles/storage.insightsCollectorService"
    "aiplatform_user"            = "roles/aiplatform.user"
  }
  project    = var.deploy_project
  role       = each.value
  member     = "serviceAccount:${google_service_account.account.email}"
  depends_on = [google_project_iam_member.gcs_pubsub_publishing]
}

resource "google_cloudfunctions2_function" "functions" {
  for_each = { for def in local.function_definitions : def.name => def }
  depends_on = [
    google_project_iam_member.roles["event_receiving"],
    google_project_iam_member.roles["artifactregistry_reader"],
  ]
  name        = each.value.name
  location    = var.location
  description = "Function to process ${each.value.name}"

  build_config {
    runtime     = "python39"
    entry_point = each.value.name
    environment_variables = {
      BUILD_CONFIG_TEST = "build_test"
    }
    source {
      storage_source {
        bucket = google_storage_bucket.source_code_bucket.name
        object = google_storage_bucket_object.function_sources[each.key].name
      }
    }
  }

  service_config {
    min_instance_count = 1
    max_instance_count = 3
    available_memory = "256M"
    timeout_seconds    = 60
    available_cpu = 4
    environment_variables = {
      PROJECT_ID                         = var.deploy_project
      AUDIO_FILES_BUCKET_NAME            = var.audio_files_bucket_name
      RAW_TRANSCRIPTIONS_BUCKET_NAME     = var.raw_transcriptions_bucket_name
      CURATED_TRANSCRIPTIONS_BUCKET_NAME = var.curated_transcriptions_bucket_name
    }
    ingress_settings               = "ALLOW_INTERNAL_ONLY"
    all_traffic_on_latest_revision = true
    service_account_email          = google_service_account.account.email
  }

  event_trigger {
    trigger_region        = var.location
    event_type            = "google.cloud.storage.object.v1.finalized"
    retry_policy          = "RETRY_POLICY_RETRY"
    service_account_email = google_service_account.account.email
    event_filters {
      attribute = "bucket"
      value     = google_storage_bucket.video_transcription_bucket_set[each.value.input_bucket].name
    }
  }
}

Similarly the buckets.tf file uses only one terraform resource for all Cloud Storage buckets.

# buckets.tf

resource "google_storage_bucket" "video_transcription_bucket_set" {
  for_each = toset([
    var.video_files_bucket_name,
    var.audio_files_bucket_name,
    var.raw_transcriptions_bucket_name,
    var.curated_transcriptions_bucket_name
  ])
  name = each.value
  location = var.location
  storage_class = "STANDARD"
  force_destroy = true

  uniform_bucket_level_access = true
}

Costs

Storage:

$0.026 per gigabyte per month

Speech-to-Text API v2:

Depends on the amount of audio you plan to process:

$0.016 per minute processed per month for 0 to 500,000 minutes of audio
$0.01 per minute processed per month for 500,000 to 1,000,000 minutes of audio
$0.008 per minute processed per month for 1,000,000 to 2,000,000 minutes of audio
$0.004 per minute processed per month for over 2,000,000 minutes of audio

Pricing details

Gemini API:

Under the following limits, the service is free of charge:

15 requests per minute
1 million tokens per minute
1,500 requests per day If you want to exceed these limits, a pay-as-you-go policy is applied. Pricing details

Cloud Functions:

The pricing depends on how long the function runs, how many times it is triggered, and the resources that are provisioned. The following link explains the pricing policy for event-driven cloud functions.

Estimate the costs of your solution with Google Cloud’s pricing calculator.

Example:
A state institution wants to automate transcript generation for meetings. The average duration of these meetings is 4 hours. The records are uploaded to GCP using this solution. Let’s simulate the costs of one transcription for this specific use case using the simulator:

The final cost per month, with one transcription per month, is estimated to be $5.14. More than half the costs are due to Speech-to-Text API use.

Service Display Name	Name	Quantity	Region	Total Price (USD)
Speech-to-Text V2	Cloud Speech-to-Text Recognition	240.0	global	3.84
Cloud Functions 1	CPU Allocation Time (2nd Gen)	40080000	us-central1	0.96192
Cloud Functions 1	Memory Allocation Time (2nd Gen)	25600000000	us-central1	0.0625
Cloud Functions 1	Invocations (2nd Gen)	1000.0	global	0
Cloud Functions 2	CPU Allocation Time (2nd Gen)	4008000.0	europe-west1	0.09619
Cloud Functions 2	Memory Allocation Time (2nd Gen)	2560000000	europe-west1	0.00625
Cloud Functions 2	Invocations (2nd Gen)	1000.0	global	0
Cloud Functions 3	CPU Allocation Time (2nd Gen)	4008000.0	europe-west1	0.09619
Cloud Functions 3	Memory Allocation Time (2nd Gen)	2560000000	europe-west1	0.00625
Cloud Functions 3	Invocations (2nd Gen)	1000.0	global	0
Cloud Storage 1	Standard Storage Belgium	3.0	europe-west1	0.06
Cloud Storage 2	Standard Storage Belgium	0.5	europe-west1	0.01
Cloud Storage 3	Standard Storage Belgium	0.01	europe-west1	0.0002
Cloud Storage 4	Standard Storage Belgium	0.01	europe-west1	0.0002

	Total Price:			5.1397

Prices are in US dollars, effective date is 2024-07-01T08:32:56.935Z

The estimated fees provided by Google Cloud Pricing Calculator are for discussion purposes only and are not binding on either you or Google. Your actual fees may be higher or lower than the estimate.

Url to the estimate:	Link to estimate

Conclusion

Leveraging AI to enhance video-to-text transcription on Google Cloud Platform offers significant benefits in scalability, flexibility, and efficiency. By integrating Chirp for speech-to-text conversion and Gemini Pro for transcription curation, and managing the deployment with Terraform, this solution provides a robust, easily deployable framework for high-quality transcriptions across various applications.

Thanks for reading! I’m Maximilien, data engineer at Stack Labs.
If you want to discover the Stack Labs Data Platform or join an enthousiast Data Engineering team, please contact us.