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

🏗️ アーキテクチャの概要

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. マイクロサービス化への備え

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

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