Gemini 2.0 FlashのマルチモーダルライブAPIを試す

!pip install -U -q google-genai

from google.colab import userdata
import os

os.environ['GOOGLE_API_KEY'] = userdata.get('GOOGLE_API_KEY')

from google import genai
from google import genai

# notebookの場合は必要
import nest_asyncio
nest_asyncio.apply()

# クライアント初期化。`v1alpha`の指定が必要。
client = genai.Client(http_options={'api_version': 'v1alpha'})

# テキストでのレスポンスを指定
config={
    "generation_config": {
        "response_modalities": ["TEXT"]
    }
}

async def main():
    # 非同期クライアントを使って`live.connect()`で接続
    async with client.aio.live.connect(
        model="gemini-2.0-flash-exp",
        config=config
    ) as session:
        message = "こんにちは！あなたのお名前は？"
        print("User:", message, "\n")
        await session.send(message, end_of_turn=True)

        # テキストレスポンスの場合、モデルのターンが終わるとループから抜ける
        turn = session.receive()
        print("Assitant:", end="")
        async for chunk in turn:
            if chunk.text is not None:
                print(chunk.text)

asyncio.run(main())

User: こんにちは！あなたのお名前は？ 

Assitant:こんにちは
！私はGoogleによってトレーニングされた、大規模言語モデルのGeminiです。

from google import genai
import asyncio
import contextlib
import wave
from IPython.display import Audio

import nest_asyncio
nest_asyncio.apply()

client = genai.Client(http_options={'api_version': 'v1alpha'})

# 音声でのレスポンスを指定
config={
    "generation_config": {
        "response_modalities": ["AUDIO"]
    }
}


# 非同期でenumerateする
async def async_enumerate(it):
  n = 0
  async for item in it:
    yield n, item
    n +=1


# WAVファイルへの出力を行う関数
@contextlib.contextmanager
def wave_file(filename, channels=1, rate=24000, sample_width=2):
    with wave.open(filename, "wb") as wf:
        wf.setnchannels(channels)
        wf.setsampwidth(sample_width)
        wf.setframerate(rate)
        yield wf

async def main():
    async with client.aio.live.connect(
        model="gemini-2.0-flash-exp",
        config=config
    ) as session:
        file_name = 'audio.wav'
        with wave_file(file_name) as wav:
            message = "こんにちは！あなたのお名前は？"
            print("> ", message, "\n")
            await session.send(message, end_of_turn=True)

            turn = session.receive()
            async for n,response in async_enumerate(turn):
                if response.data is not None:
                    wav.writeframes(response.data)

                if n==0:
                    print(response.server_content.model_turn.parts[0].inline_data.mime_type)
                print('.', end='')

        display(Audio(file_name, autoplay=True))

asyncio.run(main())

!pip install loguru

from google import genai
import asyncio
import contextlib
import wave
from IPython.display import Audio
from loguru import logger
import sys

# notebookの場合は必要
import nest_asyncio
nest_asyncio.apply()

# DEBUGにすればより詳細に出力される
logger.remove()
logger.add(sys.stderr, level="INFO")

# 非同期でenumerateする
async def async_enumerate(it):
  n = 0
  async for item in it:
    yield n, item
    n +=1


# WAVファイルへの出力を行う関数
@contextlib.contextmanager
def wave_file(filename, channels=1, rate=24000, sample_width=2):
    with wave.open(filename, "wb") as wf:
        wf.setnchannels(channels)
        wf.setsampwidth(sample_width)
        wf.setframerate(rate)
        yield wf

class AudioLoop:
    def __init__(self, client, turns=None, config=None):
        self.client = client
        self.session = None
        self.index = 0
        self.turns = turns
        if config is None:
            config={
                "generation_config": {
                    "response_modalities": ["AUDIO"]
                }
            }
        self.config = config

    async def run(self):
        """
        メインのループ処理を行うメソッド
        - Live API に接続し`websocket` を開く
        - 最初の `setup` メソッドを呼び出す
        - メインループに入って、`send`が`False`を返すまで`send`と`recv`を交互に繰り返す
        """
        logger.debug('connect')
        async with self.client.aio.live.connect(
            model="gemini-2.0-flash-exp",
            config=self.config
        ) as session:
            self.session = session

            async for sent in self.send():
                # 実際には、sendとrecvは別々のタスクになる
                await self.recv()

    async def _iter(self):
        """
        非同期でユーザーからの入力を取得してループ内で処理するメソッド
        - turnsが与えられていたらそれを順に処理する
        - turnがなければユーザからの入力を受け取る
        """
        if self.turns:
            for text in self.turns:
                print("message >", text)
                yield text
        else:
            print("Type 'q' to quit")
            while True:
                text = await asyncio.to_thread(input, "message > ")

                # `q`が入力されたら終了
                if text.lower() == 'q':
                    break

                yield text

    async def send(self):
        """
        入力テキストをAPIに送信するメソッド
        - ユーザの入力を`client_content`メッセージにラップしてモデルに送信
            - `client_content`メッセージは`BidiGenerateContentClientContent`のインスタンス
        - ユーザーが`q`を入力したら`False`を返す
        """
        async for text in self._iter():
            logger.debug('send')

            # モデルにメッセージを送信
            await self.session.send(text, end_of_turn=True)
            logger.debug('sent')
            yield text

    async def recv(self):
        """
        APIからオーディオを受け取って再生するメソッド
        - オーディオチャンクをループで収集し`.wav` ファイルに書き込む
        - モデルが`turn_complete`メソッドを送るとループから抜け出し、音声を再生
        注: この例ではシンプルに、「すべての」オーディオを受け取ってから再生している
        """
        # WAVファイルを新規作成
        file_name = f"audio_{self.index}.wav"
        with wave_file(file_name) as wav:
            self.index += 1

            logger.debug('receive')

            # ソケットからチャンクを読み込み
            turn = self.session.receive()
            async for n, response in async_enumerate(turn):
                logger.debug(f'got chunk: {str(response)}')

                if response.data is None:
                    logger.debug(f'Unhandled server message! - {response}')
                else:
                    wav.writeframes(response.data)
                    if n == 0:
                        print(response.server_content.model_turn.parts[0].inline_data.mime_type)
                    print('.', end='')

            print('\n')

        display(Audio(file_name, autoplay=True))
        await asyncio.sleep(2)

async def main():
    client = genai.Client(http_options={'api_version': 'v1alpha'})
    await AudioLoop(client, ['こんにちは', "あなたの名前を教えて。"]).run()

asyncio.run(main())