如何建立您自己的 Google NotebookLM

隨著音訊內容消費的日益普及，將文件或書面內容轉換為真實音訊格式的能力最近已成為趨勢。

雖然 Google 的 NotebookLM 在這個領域引起了關注，但我想探索使用現代雲端服務建立類似的系統。在本文中，我將向您介紹如何建立一個可擴展的雲端原生系統，該系統使用 FastAPI、Firebase、Google Cloud Pub/Sub 和 Azure 的文字轉語音服務將文件轉換為高品質的播客。

這裡有一個展示，您可以參考該系統的結果：MyPodify Showcase

挑戰

將文件轉換為播客並不像透過文字轉語音引擎運作文字那麼簡單。它需要仔細的處理、自然語言的理解以及處理各種文件格式的能力，同時保持流暢的使用者體驗。系統需要：

• 高效處理多種文件格式
• 產生多種聲音的自然聲音音訊
• 應對大規模文件處理而不影響使用者體驗
• 提供使用者即時狀態更新
• 保持高可用性和可擴展性

架構深度探究

讓我們分解關鍵組件並了解它們如何協同工作：

1.FastAPI後端

FastAPI 作為我們的後端框架，選擇它有幾個令人信服的原因：

• 非同步支援：FastAPI 的非同步功能建構於 Starlette 之上，可有效處理並發請求
• 自動 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 為我們的應用程式提供了兩個關鍵服務：

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. 谷歌雲端發布/訂閱

Pub/Sub 作為我們的訊息傳遞主幹，支援：

• 解耦架構以實現更好的可擴展性
• 至少一次交貨保證
• 自動訊息保留與重播
• 失敗訊息的死信佇列

訊息結構範例：

@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 語音服務產生語音

我們音訊產生的核心使用 Azure 的認知服務語音 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)

我們系統的獨特功能之一是能夠使用人工智慧產生多語音播客。以下是我們如何處理不同主機的腳本產生：

{
    '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. 增強的音頻處理

   • 背景音樂整合
   • 進階音訊效果
   • 自訂語音訓練

2. 內容增強

   • 自動章節標記
   • 互動文字記錄
   • 多語言支援

3. 平台整合

   • 直接播客平台發布
   • RSS feed 產生
   • 社群媒體分享

---

建立文件到播客轉換器是進入現代雲端架構的令人興奮的旅程。 FastAPI、Firebase、Google Cloud Pub/Sub 和 Azure 的文字轉語音服務的組合為大規模處理複雜的文件處理提供了堅實的基礎。

事件驅動的架構確保系統在負載下保持回應，而託管服務的使用減少了營運開銷。無論您是建立類似的系統還是只是探索雲端原生架構，我希望這次深入研究能為建立可擴展、生產就緒的應用程式提供寶貴的見解。

---

想了解更多有關雲端架構和現代應用程式開發的資訊嗎？追蹤我，以取得更多技術實用教學。

以上是如何建立您自己的 Google NotebookLM的詳細內容。更多資訊請關注PHP中文網其他相關文章！