はじめに
「周囲の会話を聞いて、状況に応じて話しかけてくれる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コンテナ版を使ってオフライン環境で動作させる