LLMとAivis Cloud APIを連携させる方法｜リアルタイム対話システム（Voice Agent）構築の基本構成

皆さんこんにちは。
感情豊かな音声合成技術を誰もがかんたんに活用できる未来を目指す、Aivis Projectです✨

LLMを使ったチャットボットを作れるようになると、次に試したくなるのが回答の音声化です。

ユーザーの質問をOpenAIのLLMへ送り、生成された回答をAivis Cloud APIで音声に変換すれば、AIアシスタントや受付システム、電話応対、キャラクターとの対話などに活用できます。

ただし、単純に2つのAPIを順番に呼び出すだけでは、LLMの回答がすべて完成してから音声合成が始まります。
回答が長いほど、ユーザーは無音のまま待たされてしまいます。

そこで本記事では、OpenAI Responses APIのストリーミング出力を文単位で区切り、完成した文からAivis Cloud APIへ送る構成を解説します。

完成すると、ターミナルへ質問を入力するだけで、OpenAIのLLMが考えた回答をAivis Cloud APIの音声で聞けるようになります📚

【結論】2つのストリーミングをつなぐ

今回構築する処理は、次のような流れです。

ユーザーが質問を入力
        ↓
OpenAI Responses API
        ↓
回答をストリーミングで受信
        ↓
「。」「！」「？」などで文を区切る
        ↓
完成した文をAivis Cloud APIへ送信
        ↓
MP3データをストリーミングで受信
        ↓
受信済みの音声から順次再生

ポイントは、LLMの回答全文が完成するまで待たないことです。

たとえば、LLMが次の回答を生成するとします。

本日の東京は晴れる見込みです。最高気温は28度です。午後はにわか雨にご注意ください。

1文目の「本日の東京は晴れる見込みです。」まで生成された時点で、Aivis Cloud APIへ音声合成を依頼します。

その音声を再生している間も、OpenAI APIからは2文目、3文目が生成され続けます。これにより、文章生成と音声再生を一部並行して進められます。

【前提】この記事で作るのは「LLM＋TTS」の部分

一般的なVoice Agentは、次の4つの技術で構成されます。

VAD → STT → LLM → TTS

・VAD：人が話し始めた、または話し終わったことを検出する
・STT：ユーザーの音声を文字へ変換する
・LLM：質問に対する回答を生成する
・TTS：生成された回答を音声へ変換する

本記事で実装するのは、後半のLLMからTTSまでです。

最初はマイクではなく、ターミナルへ質問を入力します。

ターミナル入力 → OpenAIのLLM → Aivis Cloud API → 音声再生

この部分が動けば、あとからターミナル入力をSTTの認識結果に置き換えることで、マイクを使ったVoice Agentへ発展させられます。

参考記事：https://note.com/aivis_project/n/n7219d57ca8cb

【重要】Aivis Cloud APIとAivisSpeech Engineは別のAPI

実装を始める前に、混同しやすい点を整理しておきます。

Aivis Cloud APIの音声合成には、次のエンドポイントを使います。

POST https://api.aivis-project.com/v1/tts/synthesize

主なリクエストパラメータは次のとおりです。

{
  "model_uuid": "音声合成モデルのUUID",
  "text": "読み上げるテキスト",
  "output_format": "mp3"
}

一方、ローカル版のAivisSpeech Engineでは、/audio_queryや/synthesisなどのVOICEVOX API互換エンドポイントが使われます。

Aivis Cloud APIのURLに/audio_queryを付けても動きません。
エンドポイントだけでなく、話者やモデルを指定するパラメータも異なります。

⚠️ AivisSpeech Engine向けのコードは、接続先URLを変えるだけではAivis Cloud APIへ移行できません。

【準備】必要なもの

今回の実装では、次の環境を使います。

・Python 3.11以降
・OpenAI APIキー
・Aivis Cloud APIキー
・AivisHubで利用できる音声合成モデルのUUID
・FFmpegに含まれるffplay
・インターネット接続

APIキーは2種類あります

OpenAIとAivis Cloud APIでは、それぞれ別のAPIキーを使います。

OPENAI_API_KEY
└─ OpenAI Responses APIを呼び出すためのキー

AIVIS_API_KEY
└─ Aivis Cloud APIを呼び出すためのキー

さらに、Aivis Cloud APIでは音声合成モデルを指定するために、モデルUUIDまたは限定公開用アクセスキーが必要です。

AIVIS_MODEL_UUID
└─ 音声合成に使うモデルUUID
   または「ak_」から始まる限定公開用アクセスキー

APIキーとモデルUUIDは役割が違います。

・Aivis Cloud APIキーはAuthorizationヘッダーへ指定する
・モデルUUIDまたはアクセスキーはmodel_uuidへ指定する

【準備1】OpenAI APIキーを用意する

OpenAI PlatformでAPIキーを作成します。

OpenAI APIは、ChatGPTの月額プランとは別のサービスです。
ChatGPT Plusなどへ加入していても、OpenAI APIの利用料金が自動的に含まれるわけではありません。

APIキーを作成したら、安全な場所へ保存してください。

⚠️ APIキーは第三者へ送ったり、GitHubへ公開したりしないでください。

【準備2】Aivis Cloud APIキーを用意する

Aivis Cloud APIダッシュボードからAPIキーを発行します。

Aivis Cloud APIキーは、次のようにBearerトークンとして送信します。

Authorization: Bearer YOUR_AIVIS_API_KEY

ブラウザからAivis Cloud APIを直接呼び出すこともできますが、公開WebサイトのJavaScriptへAPIキーを埋め込むと、閲覧者にキーを取得される可能性があります。
必ず第三者から閲覧されない場所へ、Aivis Cloud APIのAPIキーも保存してください。

【準備3】音声合成モデルを決める

AivisHubで、利用したい音声合成モデルを選びます。

公開モデルや、自分が所有する非公開モデルは、通常のモデルUUIDで指定できます。

限定公開モデルを投稿者以外が利用する場合は、モデルUUIDではなく、AivisHubのモデル詳細ページで確認できるak_から始まるアクセスキーを指定します。

AIVIS_MODEL_UUID=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx

限定公開用アクセスキーの場合は、次のようになります。

AIVIS_MODEL_UUID=ak_xxxxxxxxxxxxxxxxxxxx

【環境構築1】作業フォルダと仮想環境を作る

任意の場所に作業フォルダを作成します。

mkdir openai-aivis-voice-agent
cd openai-aivis-voice-agent

Pythonの仮想環境を作成します。

python3 -m venv .venv

macOSまたはLinuxでは、次のコマンドで有効化します。

source .venv/bin/activate

Windowsでは、次のコマンドを実行します。

.venv\Scripts\activate

【環境構築2】必要なライブラリをインストールする

次の3つをインストールします。

pip install -U openai python-dotenv "httpx[http2]"

それぞれの役割は次のとおりです。

・openai：OpenAI Responses APIを呼び出す
・python-dotenv：.envからAPIキーを読み込む
・httpx：Aivis Cloud APIの音声ストリームを受信する

asyncio、queue、threadingなどはPythonの標準ライブラリなので、個別にインストールする必要はありません。

【環境構築3】FFmpegをインストールする

今回のサンプルでは、Aivis Cloud APIから受信したMP3をffplayへ渡して再生します。

macOSでHomebrewを使っている場合は、次のコマンドでインストールできます。

brew install ffmpeg

Ubuntu系Linuxでは、次のコマンドを使います。

sudo apt install ffmpeg

Windowsでは、FFmpegをインストールし、ffplayへパスが通っている状態にしてください。

インストール後、次のコマンドが成功すれば準備完了です。

ffplay -version

【環境構築4】APIキーを.envへ保存する

作業フォルダの直下へ、.envという名前のファイルを作成します。

OPENAI_API_KEY=YOUR_OPENAI_API_KEY
AIVIS_API_KEY=YOUR_AIVIS_API_KEY
AIVIS_MODEL_UUID=YOUR_AIVIS_MODEL_UUID
OPENAI_MODEL=gpt-5.6-luna

.envをGitへ登録しないように、.gitignoreも作成します。

.env
.venv/
__pycache__/

⚠️ APIキーをPythonコードへ直接書かないでください。チーム開発や本番環境では、シークレットマネージャーの利用も検討してください。

【動作確認1】OpenAI APIだけを呼び出す

まず、音声合成を組み込む前にOpenAI APIとの接続を確認します。

openai_test.pyを作成し、次のコードを保存してください。

import os

from dotenv import load_dotenv
from openai import OpenAI


load_dotenv()

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

stream = client.responses.create(
    model="gpt-5.6-luna",
    input="日本語で短く自己紹介してください。",
    reasoning={"effort": "none"},
    text={"verbosity": "low"},
    stream=True,
)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)

print()

実行します。

python openai_test.py

文字が少しずつ表示されれば、OpenAI APIとの接続は成功です。

OpenAIのResponses APIでは、stream=Trueを指定すると、回答全体の完成を待たずにイベントを受け取れます。

テキストの差分は、次のイベントで届きます。

response.output_text.delta

今回の完成版でも、このイベントのdeltaを順番に連結します。

【動作確認2】Aivis Cloud APIだけを呼び出す

続いて、Aivis Cloud API単体で音声を生成します。

curl -X POST \
  "https://api.aivis-project.com/v1/tts/synthesize" \
  -H "Authorization: Bearer YOUR_AIVIS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model_uuid": "YOUR_AIVIS_MODEL_UUID",
    "text": "Aivis Cloud APIの音声合成テストです。",
    "use_ssml": false,
    "output_format": "mp3"
  }' \
  --output test.mp3

test.mp3が作成され、正常に再生できれば接続成功です。

use_ssmlをfalseにする理由

Aivis Cloud APIでは、SSMLの解釈がデフォルトで有効です。

LLMが生成した文章に<や>などが含まれていると、意図せずSSMLとして解釈される可能性があります。

今回はLLMにプレーンテキストを生成させるため、次のように明示します。

{
  "use_ssml": false
}

LLMからSSMLを生成させたい場合は、使用可能なタグと属性を制限し、サーバー側で検証してから送信してください。

【完成版】OpenAIの回答をAivisで読み上げる

ここから、2つのAPIを連携します。

voice_agent.pyを作成し、次のようなコードを作成・保存してください。

from __future__ import annotations

import os
import queue
import random
import shutil
import subprocess
import threading
import time
from typing import Any

import httpx
from dotenv import load_dotenv
from openai import OpenAI


load_dotenv()


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"環境変数 {name} が設定されていません。")
    return value


OPENAI_API_KEY = require_env("OPENAI_API_KEY")
AIVIS_API_KEY = require_env("AIVIS_API_KEY")
AIVIS_MODEL_UUID = require_env("AIVIS_MODEL_UUID")

OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-5.6-luna")
AIVIS_TTS_URL = "https://api.aivis-project.com/v1/tts/synthesize"

INSTRUCTIONS = """あなたは日本語の音声エージェントです。
結論から、自然な話し言葉で答えてください。
回答は原則3文以内にまとめてください。
Markdown、箇条書き、URL、絵文字、顔文字は使わないでください。
記号や英字の多用を避け、そのまま読み上げても理解できる文章にしてください。
不明なことは推測で断定せず、確認が必要だと伝えてください。"""

STOP = object()


def split_ready_sentences(buffer: str) -> tuple[list[str], str]:
    """生成途中の文字列から、読み上げを開始できる文を取り出す。"""
    ready: list[str] = []
    start = 0

    for index, char in enumerate(buffer):
        if char in "。！？\n":
            sentence = buffer[start : index + 1].strip()
            if sentence:
                ready.append(sentence)
            start = index + 1

    rest = buffer[start:]

    # 文末がなかなか来ない長文は、80文字以内にある読点で区切る
    while len(rest) >= 80:
        cut = max(
            rest.rfind("、", 30, 80),
            rest.rfind("，", 30, 80),
            rest.rfind(",", 30, 80),
        )
        if cut < 0:
            break

        sentence = rest[: cut + 1].strip()
        if sentence:
            ready.append(sentence)

        rest = rest[cut + 1 :]

    return ready, rest


def retry_delay(response: httpx.Response | None, attempt: int) -> float:
    """サーバーが待機時間を返した場合は、その値を優先する。"""
    if response is not None:
        for header_name in (
            "Retry-After",
            "X-Aivis-RateLimit-Requests-Reset",
        ):
            value = response.headers.get(header_name)
            if value:
                try:
                    return max(0.1, float(value))
                except ValueError:
                    pass

    return min(8.0, 2**attempt) + random.uniform(0.0, 0.25)


def play_aivis_stream(client: httpx.Client, text: str) -> None:
    """AivisのMP3レスポンスを受信しながらffplayへ渡す。"""
    payload = {
        "model_uuid": AIVIS_MODEL_UUID,
        "text": text,
        "use_ssml": False,
        "output_format": "mp3",
        "leading_silence_seconds": 0.0,
        "trailing_silence_seconds": 0.1,
        "speaking_rate": 1.0,
        "tempo_dynamics": 1.0,
    }

    retryable_statuses = {429, 500, 502, 503, 504}

    for attempt in range(4):
        response: httpx.Response | None = None

        try:
            with client.stream(
                "POST",
                AIVIS_TTS_URL,
                json=payload,
            ) as response:
                if response.status_code in retryable_statuses:
                    response.read()

                    if attempt == 3:
                        response.raise_for_status()

                    time.sleep(retry_delay(response, attempt))
                    continue

                if response.status_code >= 400:
                    detail = response.read().decode(
                        "utf-8",
                        errors="replace",
                    )
                    raise RuntimeError(
                        f"Aivis Cloud API: "
                        f"HTTP {response.status_code} {detail}"
                    )

                player = subprocess.Popen(
                    [
                        "ffplay",
                        "-nodisp",
                        "-autoexit",
                        "-loglevel",
                        "error",
                        "-i",
                        "pipe:0",
                    ],
                    stdin=subprocess.PIPE,
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                )

                received = False

                try:
                    assert player.stdin is not None

                    for audio_chunk in response.iter_bytes():
                        if not audio_chunk:
                            continue

                        received = True
                        player.stdin.write(audio_chunk)

                    player.stdin.close()
                    return_code = player.wait()

                except Exception:
                    if player.stdin and not player.stdin.closed:
                        player.stdin.close()

                    player.terminate()
                    player.wait()
                    raise

                if not received:
                    raise RuntimeError(
                        "Aivis Cloud APIから"
                        "音声データを受信できませんでした。"
                    )

                if return_code != 0:
                    raise RuntimeError(
                        f"ffplayが終了コード{return_code}で停止しました。"
                    )

                return

        except (httpx.ConnectError, httpx.TimeoutException):
            if attempt == 3:
                raise

            time.sleep(retry_delay(response, attempt))

    raise RuntimeError(
        "Aivis Cloud APIへのリクエストに失敗しました。"
    )


def audio_worker(
    sentence_queue: queue.Queue[Any],
    error_queue: queue.Queue[Exception],
) -> None:
    timeout = httpx.Timeout(
        connect=10.0,
        read=90.0,
        write=10.0,
        pool=10.0,
    )

    headers = {
        "Authorization": f"Bearer {AIVIS_API_KEY}",
        "Content-Type": "application/json",
    }

    with httpx.Client(
        headers=headers,
        timeout=timeout,
        http2=True,
    ) as client:
        while True:
            item = sentence_queue.get()

            try:
                if item is STOP:
                    return

                play_aivis_stream(client, str(item))

            except Exception as error:
                error_queue.put(error)

            finally:
                sentence_queue.task_done()


def stream_openai_response(
    client: OpenAI,
    history: list[dict[str, str]],
    sentence_queue: queue.Queue[Any],
) -> str:
    buffer = ""
    full_text: list[str] = []

    stream = client.responses.create(
        model=OPENAI_MODEL,
        instructions=INSTRUCTIONS,
        input=history,
        reasoning={"effort": "none"},
        text={"verbosity": "low"},
        max_output_tokens=300,
        store=False,
        stream=True,
    )

    for event in stream:
        if event.type == "response.output_text.delta":
            delta = event.delta

            print(delta, end="", flush=True)
            full_text.append(delta)
            buffer += delta

            ready, buffer = split_ready_sentences(buffer)

            for sentence in ready:
                sentence_queue.put(sentence)

        elif event.type in {"error", "response.failed"}:
            raise RuntimeError(
                f"OpenAI APIのストリームが失敗しました: {event}"
            )

    if buffer.strip():
        sentence_queue.put(buffer.strip())

    print()
    return "".join(full_text).strip()


def main() -> None:
    if shutil.which("ffplay") is None:
        raise RuntimeError(
            "ffplayが見つかりません。"
            "FFmpegをインストールしてから再実行してください。"
        )

    openai_client = OpenAI(api_key=OPENAI_API_KEY)

    sentence_queue: queue.Queue[Any] = queue.Queue(maxsize=8)
    error_queue: queue.Queue[Exception] = queue.Queue()

    worker = threading.Thread(
        target=audio_worker,
        args=(sentence_queue, error_queue),
        daemon=True,
    )
    worker.start()

    history: list[dict[str, str]] = []

    print(
        "日本語Voice Agentを開始します。"
        "終了するには exit と入力してください。"
    )

    try:
        while True:
            user_text = input("\nあなた: ").strip()

            if user_text.lower() in {"exit", "quit"}:
                break

            if not user_text:
                continue

            history.append({
                "role": "user",
                "content": user_text,
            })

            print("AI: ", end="", flush=True)

            try:
                answer = stream_openai_response(
                    openai_client,
                    history,
                    sentence_queue,
                )

                sentence_queue.join()

            except Exception as error:
                print(f"\n処理に失敗しました: {error}")
                history.pop()
                sentence_queue.join()
                continue

            if answer:
                history.append({
                    "role": "assistant",
                    "content": answer,
                })

                # 直近6往復分を残し、履歴が増え続けることを防ぐ
                history = history[-12:]

            while not error_queue.empty():
                print(
                    f"音声再生エラー: {error_queue.get()}"
                )

    finally:
        sentence_queue.put(STOP)
        sentence_queue.join()
        worker.join()


if __name__ == "__main__":
    main()

【実行】Voice Agentを起動する

次のコマンドを実行します。

python voice_agent.py

正常に起動すると、次のメッセージが表示されます。

日本語Voice Agentを開始します。終了するには exit と入力してください。

あなた:

質問を入力します。

あなた: Aivis Projectについて簡単に教えてください。

OpenAI APIから届いた文章がターミナルへ逐次表示され、文末まで生成された文章からAivis Cloud APIへ送られます。

終了する場合は、次のいずれかを入力します。

exit
quit

【コード解説1】OpenAIにはResponses APIを使う

今回のコードでは、OpenAIのResponses APIを使用しています。

stream = client.responses.create(
    model=OPENAI_MODEL,
    instructions=INSTRUCTIONS,
    input=history,
    reasoning={"effort": "none"},
    text={"verbosity": "low"},
    max_output_tokens=300,
    store=False,
    stream=True,
)

model

既定では、次のモデルを使用します。

gpt-5.6-luna

GPT-5.6 Lunaは、高頻度・コスト重視の処理向けに設計されているため、応答回数の多いVoice Agentの出発点として扱いやすいモデルです。

より高い回答品質が必要な場合は、.envだけを変更できます。

OPENAI_MODEL=gpt-5.6-terra

最高水準の能力を優先する場合は、GPT-5.6 Solも選択できます。

OPENAI_MODEL=gpt-5.6-sol

Voice Agentでは、モデル名だけでなく、実際の質問を使って品質・速度・コストを比較することが重要です。

reasoning

reasoning={"effort": "none"}

Voice Agentは応答速度が重要なので、今回の例では推論強度をnoneにしています。

複雑な判断が必要な業務では、lowなども比較してください。ただし、推論強度を上げるほど、回答開始までの時間が長くなる可能性があります。

text

text={"verbosity": "low"}

音声では、画面のように文章を読み飛ばせません。長い回答はユーザーの負担になります。

そのため、OpenAI側の出力を短めに設定し、プロンプトでも「3文以内」と指定しています。

store

store=False

これは、生成したResponseをAPIからあとで取得するために保存しない設定です。

⚠️ store=Falseは、OpenAI APIに関するすべてのデータ保持を無効化する設定や、Zero Data Retentionの代替ではありません。
データの取り扱い要件がある場合は、OpenAIの最新のデータ管理方針と契約内容をご確認ください。

【コード解説2】LLM向けではなく「読み上げ向け」に指示する

Voice Agentでは、通常のチャットボットとは異なるプロンプト設計が必要です。

INSTRUCTIONS = """あなたは日本語の音声エージェントです。
結論から、自然な話し言葉で答えてください。
回答は原則3文以内にまとめてください。
Markdown、箇条書き、URL、絵文字、顔文字は使わないでください。
記号や英字の多用を避け、そのまま読み上げても理解できる文章にしてください。
不明なことは推測で断定せず、確認が必要だと伝えてください。"""

チャット画面では便利なMarkdownやURLも、音声にすると聞き取りにくくなります。

たとえば、次の文章は画面では読みやすくても、音声には向きません。

詳細は https://example.com/docs をご覧ください。
1. 設定
2. 実行
3. 確認

音声向けには、次のような文章へ変換します。

詳しい手順は、公式ドキュメントから確認できます。まず設定を行い、そのあと実行結果を確認してください。

LLMの回答品質だけでなく、読み上げたときに理解できるかを評価してください。

【コード解説3】文末が来たら音声合成を始める

OpenAI APIからは、単語や文字の断片が順番に届きます。

if event.type == "response.output_text.delta":
    delta = event.delta

ただし、届いた文字を1文字ずつAivis Cloud APIへ送ることはできません。短く分割しすぎると、発音や感情表現が不自然になり、APIリクエスト数も増えてしまいます。

そこで、次の記号を文の区切りとして扱います。

。
！
？
改行

if char in "。！？\n":

さらに、文末記号のない長文が続いた場合は、30〜80文字の範囲にある読点で分割します。

while len(rest) >= 80:

この数値に絶対的な正解はありません。

・短くするほど、音声合成の開始は早くなる
・長くするほど、文脈と感情表現を保ちやすい
・細かく分けるほど、Aivis Cloud APIのリクエスト数が増える

Voice Agentでは、1〜2文程度のまとまりを出発点に、実際の会話を聞きながら調整してください。

【コード解説4】キューで再生順を守る

OpenAI APIから完成した文は、Queueへ追加します。

sentence_queue.put(sentence)

音声再生用のスレッドは、キューから文章を1つずつ取り出します。

item = sentence_queue.get()
play_aivis_stream(client, str(item))

この構成により、OpenAIが2文目や3文目を生成している間に、1文目をAivis Cloud APIで合成・再生できます。

一方で、複数の音声を無制限に並列生成すると、処理の完了順が前後する可能性があります。

今回のサンプルは、低遅延よりも会話の順序を壊さないことを優先し、音声を1つずつ再生します。

sentence_queue: queue.Queue[Any] = queue.Queue(maxsize=8)

maxsize=8を指定しているため、音声処理が追いつかなくなった場合も、キューが無制限に増え続けません。

【コード解説5】Aivisの音声を受信しながら再生する

Aivis Cloud APIへのリクエストには、client.stream()を使っています。

with client.stream(
    "POST",
    AIVIS_TTS_URL,
    json=payload,
) as response:

通常のclient.post()でレスポンス全体を取得してから再生すると、音声データが完成するまで待つ必要があります。

client.stream()を使うと、受信した音声データを順次処理できます。

for audio_chunk in response.iter_bytes():
    player.stdin.write(audio_chunk)

受信したMP3データは、ffplayの標準入力へそのまま渡します。

player = subprocess.Popen(
    [
        "ffplay",
        "-nodisp",
        "-autoexit",
        "-loglevel",
        "error",
        "-i",
        "pipe:0",
    ],
    stdin=subprocess.PIPE,
)

これにより、すべてのMP3データを受信する前から再生を開始できます。

MP3を使う理由

Aivis Cloud APIは、次の出力形式に対応しています。

・WAV
・FLAC
・MP3
・AAC
・Opus

今回のサンプルでは、互換性が高く、低遅延ストリーミング再生に利用しやすいMP3を選んでいます。

WAVは無圧縮で高品質ですが、ストリーミング中はRIFFヘッダーのデータサイズが未設定になるため、厳密な一部のデコーダーでは正常に再生できない場合があります。

【コード解説6】Aivisのパラメータを理解する

今回指定しているリクエストは次のとおりです。

payload = {
    "model_uuid": AIVIS_MODEL_UUID,
    "text": text,
    "use_ssml": False,
    "output_format": "mp3",
    "leading_silence_seconds": 0.0,
    "trailing_silence_seconds": 0.1,
    "speaking_rate": 1.0,
    "tempo_dynamics": 1.0,
}

leading_silence_seconds

"leading_silence_seconds": 0.0

音声の先頭へ追加する無音時間です。既定値は0.1秒ですが、ストリーミング再生では0.0にすると、再生開始までの待ち時間を削減できます。

trailing_silence_seconds

"trailing_silence_seconds": 0.1

音声末尾の無音時間です。

文ごとに音声合成する場合、短くしすぎると文と文が詰まって聞こえる可能性があります。最初は既定値の0.1秒から試してください。

speaking_rate

"speaking_rate": 1.0

全体の話速です。0.5〜2.0の範囲で指定できます。

tempo_dynamics

"tempo_dynamics": 1.0

話す速さの緩急を0.0〜2.0で指定します。

「速さは適切なのに棒読みに聞こえる」という場合は、speaking_rateではなくtempo_dynamicsを調整すると改善することがあります。

【会話履歴】前の会話を覚えさせる

今回のコードでは、ユーザーとAIの会話をhistoryへ保存しています。

history.append({
    "role": "user",
    "content": user_text,
})

回答後は、AIの文章も追加します。

history.append({
    "role": "assistant",
    "content": answer,
})

次回のリクエストで、この履歴全体をOpenAI APIへ渡します。

input=history

これにより、次のような会話が成立します。

あなた: AivisSpeechについて教えてください。
AI: AivisSpeechは、感情豊かな日本語音声を作成できる無料の音声合成ソフトです。

あなた: 対応しているOSは？
AI: WindowsとmacOSに対応しています。

ただし、履歴を無制限に送り続けると、入力トークン数と処理時間が増加します。

今回のサンプルでは、直近12メッセージ、つまりおおむね6往復分に制限しています。

history = history[-12:]

本番環境では、古い会話を要約する仕組みや、ユーザーごとの会話状態を保存するデータベースを検討してください。

【エラー処理】再試行してよいエラーを分ける

Aivis Cloud APIでは、主に次のHTTPステータスが返されます。

・401：APIキーが無効
・402：クレジット残高不足
・404：指定した音声合成モデルが見つからない
・422：リクエストパラメータが不正
・429：レート制限に到達
・500 / 502 / 503 / 504：サーバー側または接続上の一時的な問題

このうち、時間を空けて再試行できる可能性があるのは、主に次のエラーです。

retryable_statuses = {429, 500, 502, 503, 504}

401、402、404、422は、同じリクエストを繰り返しても基本的には直りません。

また、429の場合は、レスポンスヘッダーの情報を優先して待機します。

"Retry-After"
"X-Aivis-RateLimit-Requests-Reset"

再生途中のエラーは安易に再試行しない

ストリーミングでは、音声の一部がすでに再生されたあとに通信が切れる可能性があります。

その文章を先頭から再試行すると、ユーザーには同じ内容が途中から二重に聞こえます。

今回のコードでは、HTTPステータスの段階で判明した一時エラーは再試行しますが、音声の受信・再生開始後に発生したエラーは自動で先頭から再生し直しません。

本番環境では、次のいずれかを選びます。

・その文をスキップして次へ進む
・ユーザーへ「音声の再生に失敗しました」と伝える
・再試行する前に、再生済みの位置を管理する
・文章を短くして、失敗時の影響範囲を小さくする

【低遅延化】どこに時間がかかっているか測る

Voice Agentの待ち時間は、単一のAPIだけで決まりません。

OpenAI APIが最初の文字を返すまで
        ＋
最初の文末が生成されるまで
        ＋
Aivis Cloud APIが最初の音声を返すまで
        ＋
デコーダーが再生を開始するまで

特に見落とされやすいのが、LLMが最初の文を完成させるまでの時間です。

OpenAI APIがすぐに文字を返していても、句点がなければ音声合成へ渡せません。

改善するには、次の方法があります。

回答を短くする

回答は原則3文以内にまとめてください。

結論から話させる

結論から、自然な話し言葉で答えてください。

一文を長くしすぎない

一文を短くし、適切に句点を入れてください。

必要以上に高い推論強度を使わない

音声対話では、複雑な推論より応答速度が重要な場面があります。まずnoneを基準にし、品質が不足する場合にlowなどを比較してください。

TTSリクエストを細かくしすぎない

1文ごとにAivis Cloud APIへ送ると再生開始は早くなりますが、リクエスト数が増えます。

定額プランのレート制限や追加使用量の設定も踏まえて、実際の会話量に合った分割単位を選んでください。

【本番運用】サンプルから追加すべき設計

今回のコードは、仕組みを理解しやすくするため、ターミナル上で動く構成にしています。

本番環境では、次の対応も必要です。

APIキーをクライアントへ渡さない

WebアプリやスマートフォンアプリへOpenAI APIキーやAivis Cloud APIキーを直接埋め込まないでください。

ブラウザ・アプリ
        ↓
自社バックエンド
        ↓
OpenAI API / Aivis Cloud API

APIキーはバックエンドで管理し、ユーザー認証、利用回数制限、監査ログなどを追加します。

タイムアウトを分ける

接続タイムアウトと読み取りタイムアウトは役割が違います。

今回のコードでは、次のように個別設定しています。

timeout = httpx.Timeout(
    connect=10.0,
    read=90.0,
    write=10.0,
    pool=10.0,
)

音声合成は文章の長さによって処理時間が変わるため、読み取りタイムアウトには余裕を持たせます。

接続を再利用する

リクエストごとにHTTPクライアントを作り直すと、接続確立の時間が毎回発生します。

今回のコードでは、httpx.Clientを音声再生スレッド内で使い回しています。

with httpx.Client(
    headers=headers,
    timeout=timeout,
    http2=True,
) as client:

個人情報をログへ残しすぎない

Voice Agentでは、ユーザーの質問に氏名、住所、問い合わせ内容などが含まれる可能性があります。

エラー調査のために全文をログへ保存すると、別の情報管理リスクが生まれます。

・本文をマスキングする
・リクエストIDだけを残す
・保存期間を定める
・閲覧権限を制限する

といった設計を検討してください。

【発展】マイク入力へ対応する

マイクを使ったVoice Agentへ発展させる場合は、今回のinput()をSTTの認識結果へ置き換えます。

現在は次の構成です。

user_text = input("\nあなた: ").strip()

発展後は、次のようになります。

マイク
  ↓
VADで発話終了を検出
  ↓
STTで文字起こし
  ↓
user_textへ代入
  ↓
今回実装したOpenAI＋Aivisの処理

OpenAIの音声認識機能や、ほかのSTTサービスと組み合わせることもできます。

STTとTTSを同じ事業者へ揃える必要はありません。
聞き取り精度、音声品質、遅延、利用条件などから、それぞれ選択できます。

【発展】ユーザーの割り込みに対応する

自然なVoice Agentでは、AIが話している途中にユーザーが話し始めることがあります。

この動作は、バージインや割り込みと呼ばれます。

割り込みを実装する場合は、次の処理が必要です。

ユーザーの発話開始をVADで検出
        ↓
再生中のffplayを停止
        ↓
再生待ちキューを空にする
        ↓
OpenAIの生成ストリームを中断
        ↓
新しい発話をSTTへ送る

今回のサンプルは、回答を最後まで順番に再生する構成です。バージインを追加する場合は、再生プロセスと生成ストリームの両方をキャンセルできるように設計してください。

【日本語対策】固有名詞はユーザー辞書へ登録する

LLMが正しい文字列を生成しても、固有名詞の読み方が自動的に正しく決まるとは限りません。

たとえば、次のような語は事前の調整が必要です。

・社名
・商品名
・担当者名
・地名
・略語
・業界固有の専門用語

Aivis Cloud APIのユーザー辞書を作成し、音声合成時にuser_dictionary_uuidを指定できます。

payload = {
    "model_uuid": AIVIS_MODEL_UUID,
    "user_dictionary_uuid": os.environ["AIVIS_USER_DICTIONARY_UUID"],
    "text": text,
    "use_ssml": False,
    "output_format": "mp3",
}

Voice Agentでは、よく使う固有名詞を事前に洗い出し、実際の会話文で発音とアクセントを確認してください。

よくある質問（FAQ）

Q. ChatGPTの有料プランへ加入していれば、OpenAI APIも使えますか？

ChatGPTとOpenAI APIは別のサービスです。OpenAI APIを利用するには、OpenAI Platform側でAPIキーと支払い設定をご用意ください。

Q. Chat Completions APIではなくResponses APIを使う理由は何ですか？

現在のOpenAI APIでは、ストリーミング、推論、ツール利用、複数ターンの処理などにResponses APIが推奨されています。本記事ではresponse.output_text.deltaイベントから文章を取得しています。

Q. OpenAIから文章は届くのに、音声が再生されません。

Aivis Cloud APIキー、モデルUUID、FFmpegのインストール状況を確認してください。ffplay -versionが実行できることも確認します。

Q. 404エラーが返ります。

モデルUUIDや限定公開用アクセスキーが正しいか確認してください。モデルをアップロードした直後は、音声合成バックエンドへ反映されるまで最大5分程度かかることがあります。

Q. 422エラーが返ります。

リクエストパラメータの形式が正しくありません。モデルに存在しない話者やスタイルを指定していないか、各数値が許容範囲内かを確認してください。

Q. 429エラーが増えました。

LLMの回答を細かく分けすぎると、1回の会話で複数のTTSリクエストが発生します。1〜2文をまとめる、回答を短くする、追加使用量の設定を確認するなど、運用に合わせて調整してください。

Q. 回答がすべて完成してから音声が始まります。

LLMが句点を出しているか確認してください。response.output_text.deltaを全文連結してからTTSへ渡す実装では、ストリーミングの効果が得られません。

Q. 読み上げが途中で細かく途切れます。

分割単位が短すぎる可能性があります。単語や短い節ではなく、1文前後をまとめてAivis Cloud APIへ送ってください。

Q. SSMLを使うことはできますか？

利用できます。ただし、LLMが生成したSSMLを無検証で使用するのではなく、対応タグと属性を限定し、検証後にuse_ssmlをtrueへ変更してください。

Q. ブラウザだけで実装できますか？

Aivis Cloud APIの音声合成エンドポイントは、ブラウザからのクロスオリジンアクセスに対応しています。ただし、公開ページへAPIキーを埋め込むと取得される可能性があるため、本番サービスではバックエンド経由の構成を推奨します。

Q. AivisSpeechへ切り替える場合は、URLだけ変えればよいですか？

いいえ。Aivis Cloud APIとAivisSpeech Engineでは、エンドポイントとリクエストパラメータが異なります。それぞれのAPI仕様に合わせた実装が必要です。

まとめ

Voice Agentの体感速度は、LLMと音声合成APIそれぞれの速度だけでは決まりません。

LLMが生成した文章を、いつ音声合成へ渡すか。生成された音声を、いつ再生し始めるか。
この2つの接続方法が、会話体験を大きく左右します✨

まずは今回のターミナル版を動かし、文章の分割単位、プロンプト、音声合成パラメータを調整してみてください。

以上で、この記事はおしまいです。

Aivis ProjectやAivis Cloud APIに関してご不明点やご相談がある場合には、お問い合わせフォームよりお気軽にご連絡ください。

🔗 関連リンク
・Aivis Cloud API：https://aivis-project.com/cloud-api/
・Aivis Cloud APIドキュメント：https://api.aivis-project.com/v1/docs
・リアルタイム音声合成デモ：https://api.aivis-project.com/v1/demo/realtime-streaming
・AivisHub：https://hub.aivis-project.com
・OpenAI APIドキュメント：https://developers.openai.com/api/docs
・OpenAI Responses APIのストリーミング：https://developers.openai.com/api/docs/guides/streaming-responses