はじめに

「周囲の会話を聞いて、状況に応じて話しかけてくれるAI」を作ってみました。

例えば、議論がヒートアップしてきたら「少し休憩しませんか？」と提案してくれるようなエージェントです。

本記事では、Azure Speech ServiceとAzure OpenAI（GPT-4）を組み合わせて、リアルタイムで会話をモニタリング・分析するシステムの作り方を紹介します。

完成イメージ

🎤 会話聞き取りエージェント 起動

💬 [14:23:01] 今日のミーティングどうだった？
💬 [14:23:05] あの案件、全然進んでないんだよね
💬 [14:23:08] え、また？先週も同じこと言ってたよね
💬 [14:23:12] だからさ、相手が返事くれないんだって
💬 [14:23:15] ちゃんと催促した？

🔍 分析中...
⚠️  緊張度: 6/10 - やや緊張した会話

アーキテクチャ

┌─────────────┐     ┌──────────────────┐     ┌─────────────┐
│   マイク    │ ──> │  Azure Speech    │ ──> │  テキスト   │
│             │     │  Service         │     │  蓄積       │
└─────────────┘     └──────────────────┘     └──────┬──────┘
                                                    │
                                            N発話ごとに分析
                                                    │
                                                    v
┌─────────────┐     ┌──────────────────┐     ┌─────────────┐
│  アラート   │ <── │  Azure OpenAI    │ <── │  会話履歴   │
│  or 介入    │     │  (GPT-4)         │     │  送信       │
└─────────────┘     └──────────────────┘     └─────────────┘

技術スタック

技術	用途
Azure Speech Service	リアルタイム音声認識（STT）
Azure OpenAI (GPT-4)	会話の雰囲気分析
Python	実装言語

なぜAzure Speechを選んだか

OpenAI Realtime APIなど他の選択肢もありますが、Azure Speechを選んだ理由：

項目	Azure Speech	OpenAI Realtime
VADパラメータ調整	✅ 可能	❌ 固定
日本語最適化	✅ 充実	△ 英語優先
ノイズ耐性	✅ 高い（Teams技術）	△ 一般的
エッジ展開	✅ コンテナ版あり	❌ クラウドのみ

特にVADパラメータの調整ができる点が重要です。短い発話の応酬も正確に拾えます。

# VAD調整例：無音300msで区切り
speech_config.set_property(
    speechsdk.PropertyId.Speech_SegmentationSilenceTimeoutMs, 
    "300"
)

事前準備

Azureリソースの作成

以下のリソースを作成してください：

リソース	参考リンク
Azure Speech Service	クイックスタート: 音声テキスト変換
Azure OpenAI Service	Azure OpenAI Serviceリソースを作成する

ライブラリのインストール

pip install azure-cognitiveservices-speech openai python-dotenv

各ライブラリの詳細：

環境変数（.env）

# Azure Speech
AZURE_SPEECH_KEY=xxxxxxxxxxxxxxxxxxxxx
AZURE_SPEECH_REGION=japaneast

# Azure OpenAI
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com/
AZURE_OPENAI_KEY=xxxxxxxxxxxxxxxxxxxxx
AZURE_OPENAI_DEPLOYMENT=gpt-4
AZURE_OPENAI_API_VERSION=2024-10-21

コード全体

"""
会話聞き取りエージェント
- Azure Speechで継続的に音声認識
- 発話を蓄積してGPT-4で分析
- 状況に応じてアラート
"""

import os
import time
import json
from datetime import datetime
from dotenv import load_dotenv

import azure.cognitiveservices.speech as speechsdk
from openai import AzureOpenAI

load_dotenv()

# ============== 設定 ==============
AZURE_SPEECH_KEY = os.getenv("AZURE_SPEECH_KEY")
AZURE_SPEECH_REGION = os.getenv("AZURE_SPEECH_REGION", "japaneast")

AZURE_OPENAI_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT")
AZURE_OPENAI_KEY = os.getenv("AZURE_OPENAI_KEY")
AZURE_OPENAI_DEPLOYMENT = os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4")
AZURE_OPENAI_API_VERSION = os.getenv("AZURE_OPENAI_API_VERSION", "2024-10-21")

# 分析設定
ANALYSIS_INTERVAL = 5        # N発話ごとに分析
ALERT_THRESHOLD = 7          # この緊張度以上でアラート
MAX_HISTORY = 30             # 保持する発話数


# ============== GPT分析 ==============
class ConversationAnalyzer:
    def __init__(self):
        self.client = AzureOpenAI(
            azure_endpoint=AZURE_OPENAI_ENDPOINT,
            api_key=AZURE_OPENAI_KEY,
            api_version=AZURE_OPENAI_API_VERSION
        )
        self.deployment = AZURE_OPENAI_DEPLOYMENT
    
    def analyze(self, utterances: list[str]) -> dict:
        """会話を分析して緊張度を返す"""
        conversation_text = "\n".join(utterances)
        
        try:
            response = self.client.chat.completions.create(
                model=self.deployment,
                messages=[
                    {
                        "role": "system",
                        "content": """あなたは会話の雰囲気を分析するアシスタントです。
会話を分析し、緊張度やヒートアップの度合いを判定してください。
必ずJSON形式で回答してください。"""
                    },
                    {
                        "role": "user", 
                        "content": f"""以下の会話を分析してください。

【会話】
{conversation_text}

【出力形式】JSON
- tension_level: 緊張度 1-10（1=穏やか、5=やや緊張、7=議論、10=激しい口論）
- mood: 雰囲気（穏やか/やや緊張/議論/口論）
- summary: 状況の要約（1文）"""
                    }
                ],
                response_format={"type": "json_object"},
                max_tokens=300,
                temperature=0.3
            )
            return json.loads(response.choices[0].message.content)
        except Exception as e:
            print(f"❌ GPT分析エラー: {e}")
            return {"tension_level": 0, "mood": "不明", "summary": "分析失敗"}


# ============== メインシステム ==============
class ConversationMonitor:
    def __init__(self):
        # Azure Speech設定
        self.speech_config = speechsdk.SpeechConfig(
            subscription=AZURE_SPEECH_KEY,
            region=AZURE_SPEECH_REGION
        )
        self.speech_config.speech_recognition_language = "ja-JP"
        
        self.audio_config = speechsdk.audio.AudioConfig(
            use_default_microphone=True
        )
        
        self.recognizer = speechsdk.SpeechRecognizer(
            speech_config=self.speech_config,
            audio_config=self.audio_config
        )
        
        # 分析器
        self.analyzer = ConversationAnalyzer()
        
        # 発話履歴
        self.utterances = []
        self.utterance_count = 0
        self.is_running = False
        
        # イベントハンドラ設定
        self.recognizer.recognized.connect(self._on_recognized)
        self.recognizer.session_stopped.connect(self._on_stopped)
        self.recognizer.canceled.connect(self._on_canceled)
    
    def _on_recognized(self, evt):
        """発話認識時のコールバック"""
        if evt.result.reason == speechsdk.ResultReason.RecognizedSpeech:
            text = evt.result.text.strip()
            if text:
                timestamp = datetime.now().strftime("%H:%M:%S")
                print(f"💬 [{timestamp}] {text}")
                
                self.utterances.append(text)
                self.utterance_count += 1
                
                # 履歴を制限
                if len(self.utterances) > MAX_HISTORY:
                    self.utterances.pop(0)
                
                # 定期分析
                if self.utterance_count % ANALYSIS_INTERVAL == 0:
                    self._run_analysis()
    
    def _run_analysis(self):
        """GPT分析を実行"""
        if len(self.utterances) < 2:
            return
            
        print("\n🔍 分析中...")
        result = self.analyzer.analyze(self.utterances)
        
        tension = result.get("tension_level", 0)
        mood = result.get("mood", "不明")
        summary = result.get("summary", "")
        
        # 緊張度に応じた表示
        if tension >= ALERT_THRESHOLD:
            print(f"🚨 【警告】緊張度: {tension}/10 - {mood}")
            print(f"   {summary}")
            self._alert(result)
        elif tension >= 5:
            print(f"⚠️  緊張度: {tension}/10 - {mood}")
            print(f"   {summary}")
        else:
            print(f"✅ 緊張度: {tension}/10 - {mood}")
        print()
    
    def _alert(self, result):
        """アラート発報"""
        print("\n" + "=" * 50)
        print("🔔 アラート: 会話がヒートアップしています！")
        print(f"   緊張度: {result.get('tension_level')}/10")
        print(f"   状況: {result.get('summary')}")
        print("=" * 50 + "\n")
        
        # TODO: ここで外部連携（Slack通知、音声発話など）
    
    def _on_stopped(self, evt):
        print("🛑 セッション終了")
        self.is_running = False
    
    def _on_canceled(self, evt):
        print(f"❌ キャンセル: {evt.result.cancellation_details.reason}")
        self.is_running = False
    
    def start(self):
        """モニタリング開始"""
        print("=" * 50)
        print("🎤 会話聞き取りエージェント 起動")
        print(f"   分析間隔: {ANALYSIS_INTERVAL}発話ごと")
        print(f"   警告閾値: 緊張度 {ALERT_THRESHOLD} 以上")
        print("   終了: Ctrl+C")
        print("=" * 50 + "\n")
        
        self.is_running = True
        self.recognizer.start_continuous_recognition()
        
        try:
            while self.is_running:
                time.sleep(0.5)
        except KeyboardInterrupt:
            print("\n⏹️  停止中...")
        finally:
            self.recognizer.stop_continuous_recognition()
            print("👋 終了しました")


# ============== 実行 ==============
if __name__ == "__main__":
    monitor = ConversationMonitor()
    monitor.start()

履歴のリセット戦略

会話履歴を永遠に保持すると、関係ない過去の会話も分析対象になってしまいます。

方式	説明
発話数制限	最新N件のみ保持（今回採用）
時間制限	直近N分の発話のみ
緊張度リセット	平穏が続いたらクリア

# 発話数制限（最新30件）
MAX_HISTORY = 30

if len(self.utterances) > MAX_HISTORY:
    self.utterances.pop(0)

応用例

このシステムは様々な場面に応用できます：

場面	活用方法
会議ファシリテーション	議論が白熱したら休憩を提案
カスタマーサポート研修	対応の緊張度をリアルタイム可視化
メンタルヘルス	ストレスの高い会話パターンを検出
音声日記	日々の会話の雰囲気を自動記録

今後の発展

話者分離: Azure Speaker Diarizationで「誰が」話したかを識別
エッジ展開: Azure Speechコンテナ版でオフライン対応
音声での介入: TTSを使ってエージェントが直接話しかける

まとめ

Azure Speech + GPT-4を組み合わせることで、リアルタイムで会話を分析するエージェントを簡単に構築できました。

ポイントは：

Azure Speechの継続認識（start_continuous_recognition）でストリーミング処理
発話を蓄積してバッチでGPT分析（コスト効率）
履歴のリセット戦略で関連性のある会話のみ分析

次にやってみたいこと

本記事では基本的な会話聞き取りエージェントを構築しました。今後は以下のテーマについてです：

話者分離: Azure Speaker Diarizationを使って「誰が」話したかを識別する
エッジ展開: Azure Speechコンテナ版を使ってオフライン環境で動作させる

参考リンク

公式ドキュメント

クイックスタート

SDK リファレンス