自分だけの Google NotebookLM を構築する方法

オーディオ コンテンツの利用の人気の高まりに伴い、ドキュメントや書かれたコンテンツをリアルなオーディオ形式に変換する機能が最近トレンドになっています。

この分野では Google の NotebookLM が注目を集めていますが、私は最新のクラウド サービスを使用して同様のシステムの構築を検討したいと考えました。この記事では、FastAPI、Firebase、Google Cloud Pub/Sub、Azure の Text-to-Speech サービスを使用してドキュメントを高品質のポッドキャストに変換する、スケーラブルなクラウドネイティブ システムを作成した方法を説明します。

このシステムの結果について参照できるショーケースは次のとおりです: MyPodify Showcase

挑戦

ドキュメントをポッドキャストに変換することは、テキスト読み上げエンジンを通じてテキストを実行するほど簡単ではありません。スムーズなユーザー エクスペリエンスを維持しながら、慎重な処理、自然言語の理解、さまざまなドキュメント形式を処理する能力が必要です。システムは次のことを行う必要があります:

• 複数のドキュメント形式を効率的に処理します
• 複数の音声で自然なサウンドのオーディオを生成
• ユーザー エクスペリエンスに影響を与えることなく、大規模なドキュメント処理を処理します
• リアルタイムのステータス更新をユーザーに提供します
• 高可用性とスケーラビリティを維持する

アーキテクチャの詳細

主要なコンポーネントを分解して、それらがどのように連携するかを理解しましょう:

1. FastAPI バックエンド

FastAPI はバックエンド フレームワークとして機能し、以下のようないくつかのやむを得ない理由から選択されています。

• 非同期サポート: Starlette 上に構築された FastAPI の非同期機能により、同時リクエストを効率的に処理できます
• 自動 OpenAPI ドキュメント: すぐに使えるインタラクティブな API ドキュメントを生成します
• タイプ セーフティ: 実行時検証に Python の型ヒントを利用します
• 高パフォーマンス: 速度の点で Node.js や Go に匹敵します

アップロード エンドポイントの詳細は次のとおりです。

@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. Firebaseの統合

Firebase は、アプリケーションに 2 つの重要なサービスを提供します。

Firebaseストレージ

• 自動スケーリングを使用して安全なファイルのアップロードを処理します
• 生成されたオーディオ ファイルに対して CDN を利用した配布を提供します
• 大きなファイルの再開可能なアップロードをサポート

ファイヤーストア

• プロジェクトのステータスを追跡するためのリアルタイム データベース
• プロジェクトのメタデータに最適なドキュメントベースの構造
• 手動シャーディング不要の自動スケーリング

リアルタイムのステータス更新を実装する方法は次のとおりです:

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 はメッセージング バックボーンとして機能し、次のことを可能にします。

• スケーラビリティを向上させる分離アーキテクチャ
• 少なくとも 1 回の配信保証
• メッセージの自動保存と再生
• 失敗したメッセージのデッドレターキュー

メッセージ構造の例:

@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. Azure Speech Service による音声生成

音声生成の中核には、Azure の Cognitive Services Speech SDK が使用されます。自然な音声合成を実装する方法を見てみましょう:

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)

私たちのシステムのユニークな機能の 1 つは、AI を使用してマルチ音声ポッドキャストを生成する機能です。さまざまなホストのスクリプト生成を処理する方法は次のとおりです:

{
    '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'
}

音声合成の場合、さまざまなスピーカーを特定の Azure 音声にマッピングします。

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. バックグラウンド処理ワーカー

ワーカー コンポーネントは面倒な作業を処理します。

1. 文書分析

   • さまざまなドキュメント形式からテキストを抽出します
   • 文書の構造と内容を分析する
   • 主要なトピックとセクションを特定する

2. コンテンツ処理

   • 自然な会話の流れを生成する
   • コンテンツを話者セグメントに分割する
   • トピック間のトランジションを作成する

3. オーディオ生成

   • Azure のニューラル音声を使用してテキストを音声に変換します
   • 複数の話者の音声を処理する
   • オーディオの後処理を適用する

ワーカー ロジックを簡略化して示します。

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

エラー処理と信頼性

システムは包括的なエラー処理を実装しています:

1. 再試行ロジック

   • 失敗した API 呼び出しに対する指数関数的バックオフ
   • 最大再試行回数の構成
   • 失敗したメッセージのデッドレターキュー

2. ステータス追跡

   • Firestore に保存される詳細なエラー メッセージ
   • ユーザーへのリアルタイムのステータス更新
   • モニタリングのためのエラー集計

3. リソースのクリーンアップ

   • 一時ファイルの自動削除
   • アップロードのクリーンアップに失敗しました
   • 孤立したリソースの検出

スケーリングとパフォーマンスの最適化

本番環境の負荷を処理するために、いくつかの最適化を実装しました。

1. ワーカーのスケーリング

   • キューの長さに基づく水平方向のスケーリング
   • リソースベースの自動スケーリング
   • 低遅延のためのリージョン展開

2. ストレージの最適化

   • コンテンツの重複排除
   • 圧縮オーディオストレージ
   • 配信のための CDN 統合

3. 処理の最適化

   • 類似ドキュメントのバッチ処理
   • 繰り返されるコンテンツのキャッシュ
   • 可能な場合は並列処理

監視と可観測性

システムには包括的な監視が含まれています:

@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'}

将来の機能強化

現在のシステムはうまく機能しますが、将来の改善にはいくつかの興味深い可能性があります。

1. 強化されたオーディオ処理

   • BGM の統合
   • 高度なオーディオ効果
   • カスタムボイストレーニング

2. コンテンツの強化

   • 自動チャプターマーカー
   • インタラクティブなトランスクリプト
   • 多言語サポート

3. プラットフォーム統合

   • ポッドキャスト プラットフォームへの直接公開
   • RSS フィードの生成
   • ソーシャルメディア共有

---

ドキュメントからポッドキャストへのコンバーターの構築は、最新のクラウド アーキテクチャへのエキサイティングな旅でした。 FastAPI、Firebase、Google Cloud Pub/Sub、Azure の Text-to-Speech サービスを組み合わせることで、大規模な複雑なドキュメント処理を処理するための堅牢な基盤が提供されます。

イベント駆動型のアーキテクチャにより、負荷がかかってもシステムの応答性が維持され、マネージド サービスの使用により運用上のオーバーヘッドが削減されます。同様のシステムを構築している場合でも、単にクラウド ネイティブ アーキテクチャを検討している場合でも、この詳細な説明が、スケーラブルな実稼働対応アプリケーションの構築に関する貴重な洞察を提供することを願っています。

---

クラウド アーキテクチャと最新のアプリケーション開発について詳しく知りたいですか?より技術的で実践的なチュートリアルについては、私に従ってください。

以上が自分だけの Google NotebookLM を構築する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。