Gemini Live API で感情認識音声アプリを構築する

# 必要なライブラリのインストール
pip install google-generativeai pyaudio websockets numpy
 
# 環境変数の設定
export GEMINI_API_KEY="YOUR_GEMINI_API_KEY"

import google.generativeai as genai
import os
 
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
 
model = genai.GenerativeModel("gemini-3.1-flash-live")
 
# セッション設定
session_config = {
    "response_modalities": ["AUDIO", "TEXT"],
    "speech_config": {
        "voice_config": {
            "prebuilt_voice_config": {
                "voice_name": "Aoede"
            }
        }
    },
    "system_instruction": """
    あなたは感情に配慮した対話アシスタントです。
    ユーザーの声のトーンや話し方から感情を推定し、
    それに応じた口調と内容で応答してください。
    焦っている場合は簡潔に、落ち込んでいる場合は
    優しく励ますように応答します。
    """
}
 
session = model.start_live_session(config=session_config)
print("Session established")
# 出力: Session established

import pyaudio
import numpy as np
import threading
 
# オーディオ設定
RATE = 16000       # サンプリングレート（16kHz）
CHUNK = 1024       # チャンクサイズ
FORMAT = pyaudio.paInt16
CHANNELS = 1
 
audio = pyaudio.PyAudio()
 
# マイク入力ストリーム
input_stream = audio.open(
    format=FORMAT,
    channels=CHANNELS,
    rate=RATE,
    input=True,
    frames_per_buffer=CHUNK
)
 
# スピーカー出力ストリーム
output_stream = audio.open(
    format=FORMAT,
    channels=CHANNELS,
    rate=RATE,
    output=True,
    frames_per_buffer=CHUNK
)
 
def send_audio(session):
    """マイクからの音声をLive APIに送信"""
    while True:
        data = input_stream.read(CHUNK, exception_on_overflow=False)
        audio_array = np.frombuffer(data, dtype=np.int16)
 
        # 無音検出（閾値以下はスキップ）
        if np.abs(audio_array).mean() < 100:
            continue
 
        session.send_audio(data)
 
def receive_response(session):
    """Live APIからの応答を受信・再生"""
    for response in session.receive():
        if response.audio:
            output_stream.write(response.audio)
        if response.text:
            print(f"[Gemini] {response.text}")
 
# 送受信を別スレッドで並行実行
send_thread = threading.Thread(target=send_audio, args=(session,))
recv_thread = threading.Thread(target=receive_response, args=(session,))
send_thread.start()
recv_thread.start()

EMOTION_AWARE_PROMPT = """
あなたは共感力の高い対話アシスタントです。
以下のルールに従って応答してください:
 
1. ユーザーが早口で緊張している場合:
   - 落ち着いたトーンで応答する
   - 要点を簡潔に伝える
   - 「大丈夫ですよ」のような安心感を与える表現を使う
 
2. ユーザーが落ち込んでいる場合:
   - 共感を示してから情報を提供する
   - 「つらいですよね」のような受容的な表現を使う
   - 解決策は押し付けず、選択肢として提示する
 
3. ユーザーが楽しそうな場合:
   - テンションを合わせて明るく応答する
   - ユーモアを交えた表現を使う
 
4. ユーザーが怒っている場合:
   - まず謝罪と共感を示す
   - 問題解決に集中した簡潔な応答をする
   - 弁解ではなく具体的な解決策を提示する
"""

METADATA_EXTRACTION_PROMPT = """
ユーザーの音声に対して、以下の2つを返してください:
 
1. JSON形式の感情メタデータ（テキスト応答の先頭に配置）:
{
  "emotion": "happy|sad|angry|anxious|neutral",
  "confidence": 0.0-1.0,
  "energy_level": "high|medium|low"
}
 
2. その後に通常の応答テキスト
 
例:
{"emotion": "anxious", "confidence": 0.8, "energy_level": "high"}
お急ぎのようですね。ご質問にすぐお答えします。
"""
 
import json
 
def process_response_with_emotion(response_text):
    """応答テキストから感情メタデータを抽出"""
    lines = response_text.strip().split("\n", 1)
 
    try:
        emotion_data = json.loads(lines[0])
        reply_text = lines[1] if len(lines) > 1 else ""
        return emotion_data, reply_text
    except json.JSONDecodeError:
        return {"emotion": "neutral", "confidence": 0.5}, response_text
 
# 使用例
emotion, reply = process_response_with_emotion(response.text)
print(f"Detected emotion: {emotion['emotion']} ({emotion['confidence']:.0%})")
print(f"Reply: {reply}")
# 出力例:
# Detected emotion: anxious (80%)
# Reply: お急ぎのようですね。ご質問にすぐお答えします。

from collections import deque
from datetime import datetime
 
class EmotionTracker:
    """会話中の感情推移を追跡するクラス"""
 
    def __init__(self, window_size=10):
        self.history = deque(maxlen=window_size)
 
    def add(self, emotion_data):
        self.history.append({
            "timestamp": datetime.now().isoformat(),
            **emotion_data
        })
 
    def get_trend(self):
        """感情のトレンドを分析"""
        if len(self.history) < 3:
            return "insufficient_data"
 
        recent = list(self.history)[-3:]
        emotions = [e["emotion"] for e in recent]
 
        # 一貫してネガティブなら、特別な対応が必要
        if all(e in ("sad", "angry", "anxious") for e in emotions):
            return "persistent_negative"
 
        # 改善傾向
        if emotions[-1] in ("happy", "neutral") and emotions[0] in ("sad", "angry"):
            return "improving"
 
        return "stable"
 
    def should_escalate(self):
        """エスカレーション判定"""
        trend = self.get_trend()
        return trend == "persistent_negative"
 
# 使用例
tracker = EmotionTracker()
tracker.add({"emotion": "anxious", "confidence": 0.8, "energy_level": "high"})
tracker.add({"emotion": "angry", "confidence": 0.7, "energy_level": "high"})
tracker.add({"emotion": "angry", "confidence": 0.9, "energy_level": "high"})
 
if tracker.should_escalate():
    print("Escalation recommended: persistent negative emotion detected")
# 出力: Escalation recommended: persistent negative emotion detected

# ノイズゲート: 環境ノイズを除去
def apply_noise_gate(audio_chunk, threshold=200):
    """閾値以下の音声を無音に置換"""
    audio_array = np.frombuffer(audio_chunk, dtype=np.int16)
    mask = np.abs(audio_array) < threshold
    audio_array[mask] = 0
    return audio_array.tobytes()
 
# 音量正規化: 入力レベルを一定に保つ
def normalize_volume(audio_chunk, target_rms=3000):
    """RMS値を基準に音量を正規化"""
    audio_array = np.frombuffer(audio_chunk, dtype=np.int16).astype(np.float32)
    current_rms = np.sqrt(np.mean(audio_array ** 2))
    if current_rms > 0:
        gain = target_rms / current_rms
        audio_array = np.clip(audio_array * gain, -32768, 32767)
    return audio_array.astype(np.int16).tobytes()