如果你的 Telegram 機器人可以監聽呢？

不僅僅是閱讀文字——而是真正理解語音訊息，進行推理，並用自然流暢的聲音回應。這就是我們今天正在建立的：一個基於Google Gemini API 的 Telegram 機器人，它能夠處理文字和語音，並具備多輪記憶功能和文字轉語音回覆功能。

以下是它的實際運作效果：

1. 你可以用任何語言發送語音留言。

2. Gemini 能理解音訊並產生文字回應。

3. 機器人發送文本，並以語音訊息的形式回應。

全部程式碼大約400行Python即可完成。讓我們開始建造吧。

我們正在使用什麼

• python-telegram-bot — 非同步 Telegram Bot API 封裝器

• Gemini Interactions API — Google 的統一 API，用於文字、音訊和多輪對話

• Gemini 3.1 Flash Lite — 快速、經濟高效的推理模型

• Gemini 3.1 Flash TTS — 具有自然語音效果的文字轉語音模型

• pydub + ffmpeg — 音訊格式轉換（PCM → OGG/Opus，適用於 Telegram）

先決條件

• Python 3.11+

• Telegram Bot Token （透過 @BotFather 建立機器人）

• Google AI API 金鑰

• ffmpeg已安裝（macOS 系統brew install ffmpeg ，Linux 系統apt-get install ffmpeg ）

https://youtu.be/J716eJOAnqE

專案設定

建立一個新目錄並設定基本資訊：

mkdir telegram-gemini-voice-bot && cd telegram-gemini-voice-bot

# Create a virtual environment
python -m venv .venv && source .venv/bin/activate

# Install dependencies
pip install 'python-telegram-bot[webhooks]~=21.11' 'google-genai>=1.55.0' 'pydub~=0.25'

建立一個包含您的憑證的.env檔案：

# .env
TELEGRAM_BOT_TOKEN=your-telegram-bot-token
GOOGLE_API_KEY=your-google-api-key
TELEGRAM_SECRET_TOKEN=generate-a-random-string-here
VOICE_ENABLED=true

第一步：骨架

建立bot.py ，並先匯入模組和配置：

import base64
import io
import logging
import os
import wave

from google import genai
from pydub import AudioSegment
from telegram import Update
from telegram.ext import (
    Application,
    CommandHandler,
    ContextTypes,
    MessageHandler,
    filters,
)

# Config
TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
GOOGLE_API_KEY = os.environ["GOOGLE_API_KEY"]
WEBHOOK_URL = os.environ.get("WEBHOOK_URL", "")
TELEGRAM_SECRET_TOKEN = os.environ.get("TELEGRAM_SECRET_TOKEN")
PORT = int(os.environ.get("PORT", "8080"))

REASONING_MODEL = "gemini-3.1-flash-lite-preview"
TTS_MODEL = "gemini-3.1-flash-tts-preview"
TTS_VOICE = "Kore"

logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    level=logging.INFO,
)
logger = logging.getLogger(__name__)

# Initialize the Gemini client
gemini_client = genai.Client(api_key=GOOGLE_API_KEY)

我們使用的是兩款 Gemini 模型：

• Flash Lite能夠理解文字和音訊——它是 Gemini 系列中速度最快、價格最便宜的型號，非常適合聊天機器人。

• Flash TTS用於產生語音回應－它能產生具有可設定語音的自然語音。

步驟 2：透過互動 API 了解音頻

Interactions API 是 Gemini 的統一介面。您無需再費力地呼叫generateContent並手動追蹤對話歷史記錄，只需呼叫interactions.create()並傳入previous_interaction_id即可實現多輪對話——伺服器會處理其餘部分。

以下是向 Gemini 發送文字或音訊的核心功能：

# Track conversation state (in-memory, resets on restart)
last_interaction_ids: dict[int, str] = {}  # chat_id → interaction ID

async def gemini_interact(
    chat_id: int,
    text: str | None = None,
    audio_bytes: bytes | None = None,
) -> str:
    """Send text or audio to Gemini, return the text response."""

    input_parts: list = []

    if audio_bytes is not None:
        # Encode audio as base64 for the API
        audio_b64 = base64.b64encode(audio_bytes).decode("utf-8")
        input_parts.append(
            {"type": "audio", "data": audio_b64, "mime_type": "audio/ogg"}
        )
        input_parts.append(
            {"type": "text", "text": "Listen to this voice message and respond helpfully."}
        )

    if text is not None:
        input_parts.append({"type": "text", "text": text})

    # Simplify input if it's just a single text part
    if len(input_parts) == 1 and input_parts[0]["type"] == "text":
        input_value = input_parts[0]["text"]
    else:
        input_value = input_parts

    kwargs = {
        "model": REASONING_MODEL,
        "input": input_value,
        "system_instruction": (
            "You are a helpful, concise AI assistant on Telegram. "
            "Keep responses short and informative. "
            "Always respond in the same language the user writes or speaks in."
        ),
    }

    # Chain to previous interaction for multi-turn context
    prev_id = last_interaction_ids.get(chat_id)
    if prev_id:
        kwargs["previous_interaction_id"] = prev_id

    interaction = gemini_client.interactions.create(**kwargs)

    # Store this interaction's ID for the next turn
    last_interaction_ids[chat_id] = interaction.id

    return interaction.outputs[-1].text or "(No response generated)"

這裡發生了什麼事：

1. 音訊輸入— 我們將語音訊息位元組進行 base64 編碼，並將其作為audio部分與文字提示一起傳遞，告訴模型該做什麼。

2. 多輪互動－我們儲存每次回應中的interaction.id ，並在下次呼叫時將其作為previous_interaction_id傳遞。伺服器會保留完整的對話歷史記錄——我們不需要這樣做。

3. 文字輸入— 對於純文字訊息，我們傳送一個簡單的字串，而不是一個多部分陣列。

步驟 3：使用 Gemini TTS 進行文字轉語音

Gemini 的 TTS 模型返回的是原始 PCM 音訊。 Telegram 語音訊息需要 OGG/Opus 格式。因此，我們需要一個轉換流程：

Text → Gemini TTS → raw PCM (24kHz, 16-bit, mono) → WAV → OGG/Opus → Telegram

以下是實作方法：

async def gemini_tts(text: str) -> bytes:
    """Convert text to OGG/Opus audio bytes via Gemini TTS."""
    interaction = gemini_client.interactions.create(
        model=TTS_MODEL,
        input=text,
        response_modalities=["AUDIO"],
        generation_config={
            "speech_config": {
                "voice": TTS_VOICE.lower(),
            }
        },
    )

    # Extract PCM audio from response
    pcm_audio = None
    for output in interaction.outputs:
        if output.type == "audio":
            pcm_audio = base64.b64decode(output.data)
            break

    if pcm_audio is None:
        raise RuntimeError("No audio output from TTS")

    # Convert raw PCM → WAV (pydub needs a container format)
    wav_buffer = io.BytesIO()
    with wave.open(wav_buffer, "wb") as wav_file:
        wav_file.setnchannels(1)        # mono
        wav_file.setsampwidth(2)        # 16-bit
        wav_file.setframerate(24000)    # 24kHz
        wav_file.writeframes(pcm_audio)

    wav_buffer.seek(0)
    audio_segment = AudioSegment.from_wav(wav_buffer)

    # WAV → OGG/Opus (Telegram's required format for voice messages)
    ogg_buffer = io.BytesIO()
    audio_segment.export(ogg_buffer, format="ogg", codec="libopus")
    ogg_buffer.seek(0)
    return ogg_buffer.read()

關鍵細節：Gemini TTS 傳回 24kHz、16 位元、單聲道的原始 PCM樣本。我們使用 Python 的wave模組將其封裝在 WAV 頭部，然後使用pydub （底層呼叫ffmpeg ）重新編碼為 OGG/Opus——這是 Telegram 的reply_voice()所期望的格式。

💡 內嵌音訊標籤： Gemini TTS 支援內嵌音訊標籤－您可以將其直接嵌入到文字稿中以控制語音效果的方括號修飾符。例如， [whispers] 、 [laughs] 、 [excited] 、 [sighs]或[shouting] 。您可以在傳遞給 TTS 的文字中使用這些標籤，使語音回應更具表現力：

>

\[笑\] 哦，這真是個好問題！ \[低聲說\] 讓我告訴你一個秘密…

>

沒有固定的清單－這個模型理解各種各樣的情緒和表達方式，例如[sarcastic] 、 [panicked] 、 [curious]等等。

請在此處尋找 Gemini TTS 提示指南：https://dev.to/googleai/how-to-prompt-gemini-31s-new-text-to-speech-model-24bb

第四步：Telegram 管理員

現在，我們需要將所有內容與 Telegram 的處理程序系統連接起來。我們需要兩個處理程序：一個用於文本，一個用於語音。

處理簡訊

async def handle_text(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Handle incoming text messages."""
    chat_id = update.effective_chat.id
    user_text = update.message.text

    logger.info("Text message from chat %s: %s", chat_id, user_text[:100])

    # Show typing indicator
    await update.message.chat.send_action("typing")

    # Get Gemini response
    response_text = await gemini_interact(chat_id, text=user_text)

    # Always send text
    await update.message.reply_text(response_text)

    # Also send voice reply
    try:
        await update.message.chat.send_action("record_voice")
        ogg_audio = await gemini_tts(response_text)
        await update.message.reply_voice(voice=ogg_audio)
    except Exception as e:
        logger.error("TTS failed: %s", e)

處理語音留言

async def handle_voice(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Handle incoming voice messages."""
    chat_id = update.effective_chat.id

    logger.info("Voice message from chat %s", chat_id)

    await update.message.chat.send_action("typing")

    # Download voice file from Telegram (already in OGG/Opus format)
    voice = update.message.voice
    voice_file = await voice.get_file()
    audio_bytes = await voice_file.download_as_bytearray()

    # Send audio directly to Gemini — it understands OGG natively
    response_text = await gemini_interact(chat_id, audio_bytes=bytes(audio_bytes))

    # Send text response
    await update.message.reply_text(response_text)

    # Send voice response
    try:
        await update.message.chat.send_action("record_voice")
        ogg_audio = await gemini_tts(response_text)
        await update.message.reply_voice(voice=ogg_audio)
    except Exception as e:
        logger.error("TTS failed: %s", e)

妙處在於： Telegram 語音訊息本身就是 OGG/Opus 格式，而 Gemini 可以直接辨識這種格式。輸入時無需轉碼——我們只需傳遞原始位元組即可。

步驟五：啟動機器人

最後，設定應用程式，使其同時支援輪詢（本地開發）和 webhook（生產環境）：

def main() -> None:
    """Start the bot."""
    app = Application.builder().token(TELEGRAM_BOT_TOKEN).build()

    # Register handlers
    app.add_handler(CommandHandler("start", start_command))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text))
    app.add_handler(MessageHandler(filters.VOICE, handle_voice))

    if WEBHOOK_URL:
        # Webhook mode (production / Cloud Run)
        logger.info("Starting webhook on port %s → %s", PORT, WEBHOOK_URL)
        app.run_webhook(
            listen="0.0.0.0",
            port=PORT,
            url_path="webhook",
            webhook_url=f"{WEBHOOK_URL}/webhook",
            secret_token=TELEGRAM_SECRET_TOKEN,
        )
    else:
        # Polling mode (local dev — no public URL needed)
        logger.info("Starting polling mode (no WEBHOOK_URL set)")
        app.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()

輪詢與 Webhook：

• 投票功能—該機器人會循環向 Telegram 詢問「是否有新訊息？」。簡單易用，隨時隨地可用。非常適合本地開發。

• Webhook ——Telegram 將訊息推送到您的 URL。效率更高，是無伺服器架構（Cloud Run）的必要項。 python python-telegram-bot函式庫透過run_webhook()自動處理 webhook 註冊。

本地執行

# Load environment variables
export $(cat .env | xargs)

# Start in polling mode (no WEBHOOK_URL = polling)
python bot.py

打開 Telegram，找到你的機器人，然後給它一個語音訊息。你應該會收到一則文字回覆和一則語音回覆。 🎉

部署至 Cloud Run

想讓它全天候運作並支援零容量縮減嗎？以下是 Dockerfile 檔案：

FROM python:3.12-slim

# Install ffmpeg for audio conversion (WAV → OGG/Opus)
RUN apt-get update && \
    apt-get install -y --no-install-recommends ffmpeg && \
    rm -rf /var/lib/apt/lists/*

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY bot.py .

ENV PORT=8080
EXPOSE 8080

CMD ["python", "bot.py"]

1. 初始化gcloud並啟用 API

首先，請確保您的gcloud CLI 已配置正確的專案：

gcloud init --skip-diagnostics

啟用所需的 API－Secret Manager 用於儲存憑證，Cloud Build 用於建置容器：

gcloud services enable secretmanager.googleapis.com
gcloud services enable cloudbuild.googleapis.com

2. 商店秘訣

切勿直接將 API 金鑰放入環境變數中。請使用密鑰管理器：

echo -n "$(grep TELEGRAM_BOT_TOKEN .env | cut -d '=' -f2)" | \
  gcloud secrets create TELEGRAM_BOT_TOKEN --data-file=-
echo -n "$(grep GOOGLE_API_KEY .env | cut -d '=' -f2)" | \
  gcloud secrets create GOOGLE_API_KEY --data-file=-
echo -n "$(openssl rand -base64 32)" | \
  gcloud secrets create TELEGRAM_SECRET_TOKEN --data-file=-

注意： echo -n標誌會移除結尾的換行符，使其不包含在儲存的金鑰中。如果在執行 echo 指令時看到輸出結尾有% ，那隻是 zsh 表示沒有結尾換行符，並非金鑰的一部份。

3. 授予身分和存取管理 (IAM) 權限

Cloud Run 來源部署使用預設的 Compute Engine 服務帳戶來建置和執行容器。此帳戶需要三個預設未授予的額外角色：

# Get your project number
PROJECT_NUMBER=$(gcloud projects describe $(gcloud config get-value project) \
  --format='value(projectNumber)')

# Allow the service account to build containers
gcloud projects add-iam-policy-binding $(gcloud config get-value project) \
  --member="serviceAccount:${PROJECT_NUMBER}[email protected]" \
  --role="roles/cloudbuild.builds.builder"

# Allow it to read uploaded source code from Cloud Storage
gcloud projects add-iam-policy-binding $(gcloud config get-value project) \
  --member="serviceAccount:${PROJECT_NUMBER}[email protected]" \
  --role="roles/storage.objectViewer"

# Allow it to access secrets at runtime
gcloud projects add-iam-policy-binding $(gcloud config get-value project) \
  --member="serviceAccount:${PROJECT_NUMBER}[email protected]" \
  --role="roles/secretmanager.secretAccessor"

為什麼需要這些權限？預設的 Compute Engine 服務帳戶擁有roles/editor角色，但 Editor 角色不包含 Cloud Build 執行權限、細粒度的 Cloud Storage 讀取權限或 Secret Manager 存取權限。每個專案只需進行一次設定。

4. 部署

gcloud run deploy telegram-gemini-bot \
  --source . \
  --region us-central1 \
  --allow-unauthenticated \
  --set-secrets="TELEGRAM_BOT_TOKEN=TELEGRAM_BOT_TOKEN:latest,GOOGLE_API_KEY=GOOGLE_API_KEY:latest,TELEGRAM_SECRET_TOKEN=TELEGRAM_SECRET_TOKEN:latest" \
  --no-cpu-throttling

關於--no-cpu-throttling說明：此參數指示 Cloud Run 即使在發送初始回應後也保持 CPU 活動。由於機器人需要在確認訊息後處理文字轉語音 (TTS) 並發送語音回复，因此此參數可防止 CPU 被限制，否則會導致語音回复延遲或卡頓，直到收到下一則訊息。

請注意，這裡沒有WEBHOOK_URL這沒關係。機器人會透過K_SERVICE環境變數（Cloud Run 會自動設定）自動偵測 Cloud Run，並在 8080 連接埠啟動 HTTP 伺服器。它只是暫時不會向 Telegram 註冊 webhook，因此要到步驟 5 才會收到訊息。

5. 設定真實的 Webhook URL

從部署輸出取得實際的服務 URL，然後更新服務：

gcloud run services update telegram-gemini-bot \
  --region us-central1 \
  --update-env-vars="WEBHOOK_URL=https://telegram-gemini-bot-xxxxx-uc.a.run.app"

Cloud Run 提供 HTTPS、自動擴展和零擴展功能——只有當有人實際向機器人發送訊息時，您才需要付費。

部署故障排除

錯誤 | 原因 | 解決方法 |

|---|---|---|

| PERMISSION_DENIED: Build failed because the default service account is missing required IAM permissions | Compute Engine 服務帳戶缺少 Cloud Build 權限 | 授予roles/cloudbuild.builds.builder和roles/storage.objectViewer權限（請參閱步驟 3） |

| Permission denied on secret | 服務帳戶無法存取金鑰管理員 | 授予roles/secretmanager.secretAccessor權限（請參閱步驟 3） |

| API [secretmanager.googleapis.com] not enabled | Secret Manager API 尚未開啟 | 執行gcloud services enable secretmanager.googleapis.com |

| API [cloudbuild.googleapis.com] not enabled | Cloud Build API 尚未開啟 | 出現提示時請說Y ，或執行gcloud services enable cloudbuild.googleapis.com |

Voice replies are slow or delayed | 文字回覆後 CPU 被限制 | 使用--no-cpu-throttling部署，以保持 CPU 活躍以執行後台任務 |

關鍵建築理念

1. 伺服器端會話內存

傳統的聊天機器人 API 需要您自行管理對話歷史記錄。每次請求都需要發送完整的對話歷史記錄，而且每次請求都會增加您的令牌成本。

Interactions API 則相反。你傳遞previous_interaction_id ，伺服器會保留上下文：

# Turn 1
i1 = client.interactions.create(model="gemini-3.1-flash-lite-preview", input="Hi, I'm Alex")

# Turn 2 — server remembers "Alex"
i2 = client.interactions.create(
    model="gemini-3.1-flash-lite-preview",
    input="What's my name?",
    previous_interaction_id=i1.id  # ← that's it
)

在我們的機器人中，我們透過chat_id來輸入此訊息，因此每個 Telegram 聊天都有自己的對話線程。

2. 無需轉錄的多模態輸入

Gemini 原生支援音訊處理。無需低聲細語，無需轉錄步驟，也無需中間文字。我們直接發送 OGG 位元組：

input_parts = [
    {"type": "audio", "data": audio_b64, "mime_type": "audio/ogg"},
    {"type": "text", "text": "Listen and respond helpfully."},
]

這意味著模型不僅能辨識單詞，還能辨識語氣、重音和語言本身。它可以用使用者所說的語言做出回應，區分疑問句和陳述句，並捕捉轉錄過程中會失去的細微差別。

3. 雙模型架構

我們針對兩種不同的工作使用兩種不同的模型：

| 工作 | 模式 | 為什麼 |

|---|---|---|

| 理解 + 推理 | gemini-3.1-flash-lite-preview | 最便宜、最快－聊天機器人的理想之選 |

| 文字轉語音 | gemini-3.1-flash-tts-preview | 專為自然語音合成而設計 |

這比使用單一型號同時處理這兩個任務更便宜、更好。 Flash Lite 負責思考，TTS 負責朗讀。

更進一步

完整的源程式碼在此基礎上進行了擴展：

• 模式切換－代理模式、轉錄模式和翻譯模式，均配有內建鍵盤

• 可配置語音開關—— /voice on|off ，用於控制 TTS 響應

• 語言選擇— /language Spanish以設定翻譯目標

• 不同模式的系統指令－每種模式都有客製化的提示

這些都是同一個gemini_interact()函數的不同變體，只是system_instruction值不同。核心語音管道保持不變。

簡而言之： Gemini 的互動 API 讓語音機器人的搭建變得異常簡單。音訊以 base64 編碼輸入，輸出文本，然後由 TTS 將其轉換回語音。伺服器會自動追蹤對話狀態，無需您操心。只需新增一個 Dockerfile，即可在 Cloud Run 上建立一個可用於生產環境的語音助理。

祝你破解愉快！ 🚀

原文出處：https://dev.to/googleai/build-a-voice-enabled-telegram-bot-with-the-gemini-interactions-api-nm5