システムアーキテクチャとデータフロー

本構成では、HTTPリクエストの処理、ローカル大規模言語モデル（LLM）との推論連携、およびテキスト音声変換（TTS）を分離されたモジュールで実行します。クライアントからPOSTリクエストが到達すると、FastAPIがバックエンド推論エンジンへクエリを転送し、生成されたテキストを即座にAPIレスポンスとして返却します。同時に、受信したテキストをバックグラウンドスレッドに渡すことで、メインスレッドのブロッキングを回避しつつ、ローカルオーディオデバイスから合成音声を再生します。

音声合成エンジンの実装

LLMの出力を受け取り、波形データを生成および再生する独立したコンポーネントを定義します。ハードウェアリソースの自動検出、モデルの初期化、およびオーディオストリームの制御をクラスベースでカプセル化しています。

# tts_processor.py
import torch
import sounddevice as sd
import ChatTTS

class AudioSynthesizer:
    def __init__(self):
        self.compute_unit = "cuda" if torch.cuda.is_available() else "cpu"
        self.tts_core = ChatTTS.Chat()
        self.tts_core.load(compile=False, device=self.compute_unit)

    def generate_and_emit(self, input_content: str) -> None:
        if not input_content or input_content.isspace():
            return

        # 音声特性の再現性を確保するための固定値
        torch.manual_seed(999)

        try:
            audio_tensors = self.tts_core.infer([input_content], use_decoder=True)
            if not audio_tensors:
                return

            raw_waveform = audio_tensors[0]
            # 固定サンプリングレートで再生し、ストリーム終了まで同期待機
            sd.play(raw_waveform, samplerate=24000)
            sd.wait()
            print("[Audio] Playback sequence completed.")
        except Exception as error:
            print(f"[TTS Error] Synthesis failed: {error}")

APIゲートウェイの制御ロジック

FastAPIを用いてHTTPエンドポイントを構築します。非同期HTTPクライアントを経由してOllamaへリクエストを送信し、正常なレスポンスを受信した後、前述の合成クラスインスタンスをスレッドとして起動します。

# app_server.py
import asyncio
import traceback
import threading
import httpx
import uvicorn
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from tts_processor import AudioSynthesizer

gateway = FastAPI(title="AI Voice Response Router")
voice_engine = AudioSynthesizer()

class QuerySchema(BaseModel):
    prompt: str
    backend_model: str = "llama3:8b"
    streaming_mode: bool = False

@gateway.post("/api/infer")
async def route_inference(req: QuerySchema):
    try:
        async with httpx.AsyncClient(timeout=100.0) as session:
            llm_resp = await session.post(
                url="http://127.0.0.1:11434/api/generate",
                json={
                    "model": req.backend_model,
                    "prompt": req.prompt,
                    "stream": False
                }
            )

            if llm_resp.status_code != 200:
                raise HTTPException(status_code=502, detail="Model backend unavailable")

            payload = llm_resp.json()
            output_string = payload.get("response", "")

            # HTTP応答をブロックせずにバックグラウンドで音声処理を実行
            if output_string:
                audio_worker = threading.Thread(
                    target=voice_engine.generate_and_emit,
                    args=(output_string,),
                )
                audio_worker.start()

            return {
                "status": "delivered",
                "text": output_string,
                "engine": req.backend_model,
                "is_complete": True
            }
    except Exception as exc:
        error_stack = traceback.format_exc()
        print(f"[System Crash] {error_stack}")
        raise HTTPException(status_code=500, detail=f"Request processing failed: {exc}")

if __name__ == "__main__":
    uvicorn.run(
        "app_server:gateway",
        host="127.0.0.1",
        port=8080,
        reload=True
    )

実行手順

依存ライブラリをインストールした後、ローカル推論サーバー（Ollama）を起動状態に保ちます。FastAPIスクリプトを実行し、クライアントから指定されたエンドポイントへJSON形式でリクエストを送信すると、テキスト応答の返却と同時にシステムスピーカーから音声が出力されます。