システムアーキテクチャ設計の詳細

️ アーキテクチャ概要

ASR-LLM-TTS 音声対話アシスタントのバックエンドは、イベント駆動型の非同期アーキテクチャに基づいており、WebSocket によるリアルタイム双方向通信を行い、ストリーミングデータ処理と並行タスク実行に対応しています。

設計原則

1. 単一責任の原則

各モジュールは特定の機能領域に専念します。

• server.py：WebSocket 接続管理とメッセージのルーティング
• turn_manager.py：対話フローの調整
• llm/client.py：LLM API との通信
• tts/client.py：TTS サービスとの通信
• asr/asr_service.py：音声認識処理

2. 非同期優先

• すべての I/O 処理で async/await を使用します
• ネットワークリクエスト、ファイル操作、音声処理はすべてノンブロッキングです
• asyncio のイベントループで並行タスクを管理します

3. ストリーミング処理

• LLM の応答をストリーミングで返し、最初の文字が表示されるまでの遅延を短縮します
• TTS の双方向ストリーミング合成により、生成しながら再生します
• 音声をストリーミング処理し、リアルタイムのバッファ管理に対応します

4. フォールトトレラント設計

• 多層的なエラー処理と復旧の仕組み
• コンポーネント単体の障害がサービス全体に影響しない
• 自動リトライと機能縮退（デグレード）のポリシー

コアコンポーネントのアーキテクチャ

WebSocket サーバー（server.py）

# 架构特点
- FastAPI异步Web框架
- 单一WebSocket端点 /ws/chat
- 消息类型路由器
- 组件生命周期管理
- 全局状态维护

主な役割：

1. 接続管理：アクティブな WebSocket 接続のリストを保持します
2. メッセージのルーティング：メッセージタイプに応じて対応するハンドラーに振り分けます
3. コンポーネントの初期化：起動時にすべてのサービスコンポーネントを初期化します
4. リソースのクリーンアップ：サービス終了時にリソースを適切にクリーンアップします

メッセージフロー：

Client WebSocket Message → server.py → Message Router → turn_manager.py
                          ↑                                      ↓
Client ← WebSocket Response ← Callback Function ← Component Response

対話マネージャー（app/turn_manager.py）

# 架构特点
- 单轮对话状态机
- 多轮工具调用循环 (最多5轮)
- 并发任务协调器
- 流式数据管道

コアプロセス：

1. ユーザー入力の受信 → メッセージ履歴を構築
2. LLM の呼び出し → ストリーミング応答
3. ツール呼び出しの検出 → Function Calling を実行
4. ツール結果の処理 → LLM に返して対話を継続
5. TTS の開始 → 並行して音声合成
6. 状態の同期 → フロントエンドの状態をリアルタイムで更新

並行処理の制御：

# 三路并发处理
async def handle_turn():
    # 任务1: LLM流式输出
    llm_task = asyncio.create_task(stream_llm_response())

    # 任务2: TTS合成与播放  
    tts_task = asyncio.create_task(stream_tts_and_play())

    # 任务3: 工具调用处理
    tool_task = asyncio.create_task(execute_function_calls())

    # 协调任务执行
    await asyncio.gather(llm_task, tts_task, tool_task)

LLM クライアント（llm/client.py）

# 架构特点
- HTTP/2连接池复用
- 流式响应处理器
- 错误重试策略
- Function Calling解析

ストリーミング処理の仕組み：

async def stream_chat(self, messages):
    async with httpx.AsyncClient() as client:
        async with client.stream("POST", self.api_url, json=payload) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    chunk = json.loads(line[6:])
                    yield self._process_chunk(chunk)

エラー処理の階層：

1. HTTP 層のエラー：ステータスコード、接続タイムアウト
2. API 層のエラー：業務ロジックのエラー、レート制限
3. データ層のエラー：JSON の解析、フォーマットの検証
4. アプリケーション層のエラー：Function Calling の解析

TTS クライアント（tts/client.py）

# 架构特点  
- WebSocket双向流式通信
- 会话状态管理
- 音频数据流处理
- 连接重试机制

双方向ストリーミングアーキテクチャ：

Text Queue → TTS Client → WebSocket → Volcengine TTS
     ↑                        ↓
TTS Player ← Audio Stream ← WebSocket Response

主要技術の実装：

async def synthesize_stream_bidi(self, text_queue):
    # 发送任务: 从队列读取文本发送
    send_task = asyncio.create_task(self._send_text_stream(text_queue))

    # 接收任务: 监听音频数据流
    async for audio_chunk in self._receive_audio_stream():
        yield audio_chunk  # 实时返回音频数据

ASR サービス（asr/asr_service.py）

# 架构特点
- 多进程隔离架构
- VAD端点检测集成  
- 实时音频流处理
- 进程间队列通信

マルチプロセスアーキテクチャの設計：

Main Process                    ASR Process
    │                              │
    ├─ WebSocket Server            ├─ Audio Capture
    ├─ Turn Manager               ├─ VAD Detection  
    ├─ TTS Player                 ├─ ASR Recognition
    │                              │
    └─ Queue Consumer ←──Queue──← └─ Result Publisher

VAD による発話区間検出の流れ：

1. 音声の取得 → マイクからのリアルタイムデータストリーム
2. VAD 検出 → silero-vad モデルで発話の有無を判定
3. 音声の分割 → 無音区間に基づいて発話区間を切り出し
4. ASR 認識 → sherpa-onnx で発話区間を処理
5. 結果の配信 → キュー経由で認識結果を送信

通信プロトコルの設計

WebSocket メッセージプロトコル

メッセージ形式：

interface WebSocketMessage {
    type: string;           // 消息类型标识
    payload: {              // 消息载荷
        [key: string]: any;
    };
    timestamp?: number;     // 可选时间戳
}

メッセージタイプの分類：

ストリーミングデータパイプライン

User Input → WebSocket → Turn Manager
                            ↓
LLM API ← HTTP Stream ← LLM Client
    ↓                        ↓  
LLM Chunks → WebSocket → Frontend
    ↓
Text Queue → TTS Client → Audio Stream
                ↓
Audio Player → sounddevice → Hardware

Function Calling のアーキテクチャ

ツールの登録の仕組み

class FunctionRegistry:
    _registry: Dict[str, Dict[str, Any]] = {}

    @classmethod
    def register_function(cls, name: str, description: str = ""):
        def decorator(func: Callable):
            cls._registry[name] = {
                'func': func,
                'description': description,
                'is_async': inspect.iscoroutinefunction(func)
            }
            return func
        return decorator

MCP 統合アーキテクチャ

Local Function Registry ←─┐
                          │
                          ├─ MCPClient (统一客户端)
                          │    │
                          │    ├─ MCPConnection 1 → vision-arm-control
                          │    ├─ MCPConnection 2 → face-tracker
                          │    └─ MCPConnection N → amap-maps
                          │
Enhanced Function Call ←─┘

ツール呼び出しの流れ：

1. LLM がツール呼び出しを生成 → Function Calling 形式
2. MCPClient がルーティングを判定 → ローカルツールか MCP ツールか
3. ツール関数を実行 → 同期/非同期で実行
4. 結果の整形 → 統一された戻り値形式
5. LLM に返して処理を継続 → 複数ラウンドのツール呼び出しに対応

パフォーマンス最適化戦略

1. コネクションプールの管理

• HTTP クライアントのコネクションプールを再利用
• WebSocket 接続を常時維持
• データベースのコネクションプール（必要な場合）

2. メモリの最適化

• ストリーミング処理により大きなオブジェクトを回避
• 音声バッファを適時解放
• LRU キャッシュでホットデータを保持

3. 並行処理の制御

• 同時対話数の上限を設定
• ツール呼び出しのタイムアウト制御
• 非同期タスクの優先度スケジューリング

4. キャッシュ戦略

• TTS 音声クリップのキャッシュ
• ASR モデルのホットロード
• 設定ファイルのメモリキャッシュ

️ セキュリティ上の考慮事項

1. 入力の検証

• WebSocket メッセージ形式の検証
• Function Calling パラメーターの検証
• ファイルパスのセキュリティチェック

2. リソースの制限

• 音声ファイルのサイズ制限
• 対話ラウンド数の制限
• メモリ使用量の監視

3. エラーの分離

• コンポーネントの例外が他のモジュールに影響しない
• 悪意のある入力によってシステムがクラッシュしない
• 機密情報をログに記録しない

拡張性の設計

1. プラグインアーキテクチャ

• Function Calling ツールのプラグイン化
• 音声コーデックのプラグイン化
• LLM プロバイダーの抽象化レイヤー
• MCP サーバーの動的な接続 - 設定ファイルで管理

2. 設定駆動

• 実行時パラメーターを設定で変更可能
• 複数環境の設定に対応
• 設定の動的ホットアップデート

3. マイクロサービス化への備え

• モジュール間の疎結合な設計
• 標準化されたインターフェース定義
• 分散デプロイへの対応

このアーキテクチャ設計により、システムの高性能、高可用性、拡張の容易さが確保され、音声対話アシスタントの堅固な技術基盤となっています。