Comment créer votre propre NotebookLM de Google

Avec la popularité croissante de la consommation de contenu audio, la possibilité de convertir vos documents ou votre contenu écrit dans des formats audio réalistes est devenue une tendance plus récente.

Bien que NotebookLM de Google ait retenu l'attention dans ce domaine, je souhaitais explorer la création d'un système similaire à l'aide de services cloud modernes. Dans cet article, je vais vous expliquer comment j'ai créé un système cloud natif évolutif qui convertit les documents en podcasts de haute qualité à l'aide de FastAPI, Firebase, Google Cloud Pub/Sub et du service Text-to-Speech d'Azure.

Voici une vitrine à laquelle vous pouvez vous référer pour les résultats de ce système : MyPodify Showcase

Le défi

Convertir des documents en podcasts n'est pas aussi simple que de faire passer du texte via un moteur de synthèse vocale. Cela nécessite un traitement minutieux, une compréhension du langage naturel et la capacité de gérer différents formats de documents tout en conservant une expérience utilisateur fluide. Le système doit :

• Traitez efficacement plusieurs formats de documents
• Générez un son naturel avec plusieurs voix
• Gérer le traitement de documents à grande échelle sans affecter l'expérience utilisateur
• Fournir des mises à jour de statut en temps réel aux utilisateurs
• Maintenir la haute disponibilité et l'évolutivité

Plongée en profondeur sur l'architecture

Décomposons les composants clés et comprenons comment ils fonctionnent ensemble :

1. Back-end FastAPI

FastAPI sert de framework backend, choisi pour plusieurs raisons impérieuses :

• Prise en charge asynchrone : construites sur Starlette, les capacités asynchrones de FastAPI permettent une gestion efficace des requêtes simultanées
• Documentation automatique OpenAPI : génère une documentation API interactive prête à l'emploi
• Sécurité des types : exploite les astuces de type de Python pour la validation de l'exécution
• Haute Performance : Comparable à Node.js et Go en termes de vitesse

Voici un aperçu détaillé de notre point de terminaison de téléchargement :

@app.post('/upload')
async def upload_files(
    token: Annotated[ParsedToken, Depends(verify_firebase_token)],
    project_name: str,
    description: str,
    website_link: str,
    host_count: int,
    files: Optional[List[UploadFile]] = File(None)
):
    # Validate token
    user_id = token['uid']

    # Generate unique identifiers
    project_id = str(uuid.uuid4())
    podcast_id = str(uuid.uuid4())

    # Process and store files
    file_urls = await process_uploads(files, user_id, project_id)

    # Create Firestore document
    await create_project_document(user_id, project_id, {
        'status': 'pending',
        'created_at': datetime.now(),
        'project_name': project_name,
        'description': description,
        'file_urls': file_urls
    })

    # Trigger async processing
    await publish_to_pubsub(user_id, project_id, podcast_id, file_urls)

    return {'project_id': project_id, 'status': 'processing'}

2. Intégration de Firebase

Firebase fournit deux services cruciaux pour notre application :

Stockage Firebase

• Gère les téléchargements de fichiers sécurisés avec mise à l'échelle automatique
• Fournit une distribution basée sur CDN pour les fichiers audio générés
• Prend en charge les téléchargements pouvant être repris pour les fichiers volumineux

Firestore

• Base de données en temps réel pour le suivi de l'état du projet
• Structure basée sur des documents parfaite pour les métadonnées du projet
• Mise à l'échelle automatique sans partitionnement manuel requis

Voici comment nous mettons en œuvre des mises à jour de statut en temps réel :

async def update_status(user_id: str, project_id: str, status: str, metadata: dict = None):
    doc_ref = db.collection('projects').document(f'{user_id}/{project_id}')

    update_data = {
        'status': status,
        'updated_at': datetime.now()
    }

    if metadata:
        update_data.update(metadata)

    await doc_ref.update(update_data)

3. Google Cloud Pub/Sub

Pub/Sub sert de base de messagerie, permettant :

• Architecture découplée pour une meilleure évolutivité
• Garantie de livraison au moins une fois
• Conservation et relecture automatiques des messages
• Files d'attente de lettres mortes pour les messages ayant échoué

Exemple de structure de message :

@app.post('/upload')
async def upload_files(
    token: Annotated[ParsedToken, Depends(verify_firebase_token)],
    project_name: str,
    description: str,
    website_link: str,
    host_count: int,
    files: Optional[List[UploadFile]] = File(None)
):
    # Validate token
    user_id = token['uid']

    # Generate unique identifiers
    project_id = str(uuid.uuid4())
    podcast_id = str(uuid.uuid4())

    # Process and store files
    file_urls = await process_uploads(files, user_id, project_id)

    # Create Firestore document
    await create_project_document(user_id, project_id, {
        'status': 'pending',
        'created_at': datetime.now(),
        'project_name': project_name,
        'description': description,
        'file_urls': file_urls
    })

    # Trigger async processing
    await publish_to_pubsub(user_id, project_id, podcast_id, file_urls)

    return {'project_id': project_id, 'status': 'processing'}

4. Génération vocale avec Azure Speech Service

Le cœur de notre génération audio utilise le SDK Cognitive Services Speech d'Azure. Voyons comment nous implémentons une synthèse vocale au son naturel :

async def update_status(user_id: str, project_id: str, status: str, metadata: dict = None):
    doc_ref = db.collection('projects').document(f'{user_id}/{project_id}')

    update_data = {
        'status': status,
        'updated_at': datetime.now()
    }

    if metadata:
        update_data.update(metadata)

    await doc_ref.update(update_data)

L'une des caractéristiques uniques de notre système est la possibilité de générer des podcasts multi-voix à l'aide de l'IA. Voici comment nous gérons la génération de scripts pour différents hôtes :

{
    'user_id': 'uid_123',
    'project_id': 'proj_456',
    'podcast_id': 'pod_789',
    'file_urls': ['gs://bucket/file1.pdf'],
    'description': 'Technical blog post about cloud architecture',
    'host_count': 2,
    'action': 'CREATE_PROJECT'
}

Pour la synthèse vocale, nous mappons différents locuteurs sur des voix Azure spécifiques :

import azure.cognitiveservices.speech as speechsdk
from pathlib import Path

class SpeechGenerator:
    def __init__(self):
        self.speech_config = speechsdk.SpeechConfig(
            subscription=os.getenv("AZURE_SPEECH_KEY"),
            region=os.getenv("AZURE_SPEECH_REGION")
        )

    async def create_speech_segment(self, text, voice, output_file):
        try:
            self.speech_config.speech_synthesis_voice_name = voice
            synthesizer = speechsdk.SpeechSynthesizer(
                speech_config=self.speech_config,
                audio_config=None
            )

            # Generate speech from text
            result = synthesizer.speak_text_async(text).get()

            if result.reason == speechsdk.ResultReason.SynthesizingAudioCompleted:
                with open(output_file, "wb") as audio_file:
                    audio_file.write(result.audio_data)
                return True

            return False

        except Exception as e:
            logger.error(f"Speech synthesis failed: {str(e)}")
            return False

5. Travailleur de traitement en arrière-plan

La composante travailleur s'occupe du gros travail :

1. Analyse de documents

   • Extraire du texte de différents formats de documents
   • Analyser la structure et le contenu du document
   • Identifier les sujets et sections clés

2. Traitement du contenu

   • Générer un flux de conversation naturel
   • Diviser le contenu en segments de haut-parleurs
   • Créer des transitions entre les sujets

3. Génération audio

   • Convertir la synthèse vocale à l'aide des voix neuronales d'Azure
   • Gérer les voix de plusieurs locuteurs
   • Appliquer le post-traitement audio

Voici une vue simplifiée de notre logique de travail :

async def generate_podcast_script(outline: str, analysis: str, host_count: int):
    # System instructions for different podcast formats
    system_instructions = TWO_HOST_SYSTEM_PROMPT if host_count > 1 else ONE_HOST_SYSTEM_PROMPT

    # Example of how we structure the AI conversation
    if host_count > 1:
        script_format = """
        **Alex**: "Hello and welcome to MyPodify! I'm your host Alex, joined by..."
        **Jane**: "Hi everyone! I'm Jane, and today we're diving into {topic}..."
        """
    else:
        script_format = """
        **Alex**: "Welcome to MyPodify! Today we're exploring {topic}..."
        """

    # Generate the complete script using AI
    script = await generate_content_from_openai(
        content=f"{outline}\n\nContent Details:{analysis}",
        system_instructions=system_instructions,
        purpose="Podcast Script"
    )

    return script

Gestion des erreurs et fiabilité

Le système implémente une gestion complète des erreurs :

1. Réessayer la logique

   • Intervalle exponentiel en cas d'échec des appels d'API
   • Configuration du nombre maximum de nouvelles tentatives
   • File d'attente des lettres mortes pour les messages ayant échoué

2. Suivi du statut

   • Messages d'erreur détaillés stockés dans Firestore
   • Mises à jour du statut en temps réel des utilisateurs
   • Agrégation d'erreurs pour la surveillance

3. Nettoyage des ressources

   • Suppression automatique des fichiers temporaires
   • Échec du nettoyage du téléchargement
   • Détection des ressources orphelines

Mise à l'échelle et optimisations des performances

Pour gérer les charges de production, nous avons mis en œuvre plusieurs optimisations :

1. Mise à l'échelle des travailleurs

   • Mise à l'échelle horizontale basée sur la longueur de la file d'attente
   • Autoscaling basé sur les ressources
   • Déploiement régional pour une latence plus faible

2. Optimisation du stockage

   • Déduplication de contenu
   • Stockage audio compressé
   • Intégration CDN pour la livraison

3. Optimisation du traitement

   • Traitement par lots de documents similaires
   • Mise en cache du contenu répété
   • Traitement parallèle si possible

Surveillance et observabilité

Le système comprend une surveillance complète :

@app.post('/upload')
async def upload_files(
    token: Annotated[ParsedToken, Depends(verify_firebase_token)],
    project_name: str,
    description: str,
    website_link: str,
    host_count: int,
    files: Optional[List[UploadFile]] = File(None)
):
    # Validate token
    user_id = token['uid']

    # Generate unique identifiers
    project_id = str(uuid.uuid4())
    podcast_id = str(uuid.uuid4())

    # Process and store files
    file_urls = await process_uploads(files, user_id, project_id)

    # Create Firestore document
    await create_project_document(user_id, project_id, {
        'status': 'pending',
        'created_at': datetime.now(),
        'project_name': project_name,
        'description': description,
        'file_urls': file_urls
    })

    # Trigger async processing
    await publish_to_pubsub(user_id, project_id, podcast_id, file_urls)

    return {'project_id': project_id, 'status': 'processing'}

Améliorations futures

Bien que le système actuel fonctionne bien, il existe plusieurs possibilités intéressantes d'améliorations futures :

1. Traitement audio amélioré

   • Intégration de la musique de fond
   • Effets audio avancés
   • Formation vocale personnalisée

2. Amélioration du contenu

   • Marqueurs de chapitre automatiques
   • Transcriptions interactives
   • Support multilingue

3. Intégration de plateforme

   • Publication directe sur plateforme de podcasts
   • Génération de flux RSS
   • Partage sur les réseaux sociaux

---

La création d'un convertisseur de document en podcast a été un voyage passionnant vers l'architecture cloud moderne. La combinaison de FastAPI, Firebase, Google Cloud Pub/Sub et des services de synthèse vocale d'Azure constitue une base solide pour gérer le traitement de documents complexes à grande échelle.

L'architecture basée sur les événements garantit que le système reste réactif sous charge, tandis que l'utilisation de services gérés réduit les frais opérationnels. Que vous construisiez un système similaire ou que vous exploriez simplement des architectures cloud natives, j'espère que cette étude approfondie vous a fourni des informations précieuses sur la création d'applications évolutives et prêtes pour la production.

---

Vous souhaitez en savoir plus sur l'architecture cloud et le développement d'applications modernes ? Suivez-moi pour des tutoriels plus techniques et pratiques.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!