はじめに
下記GitHubにGemini 2.0のMulti Modal APIのサンプルがありました。
私の端末ではうまく動かなかったです。
原因は2つで、「パッケージのインストール(よく読んだらインストールせよと書いてあった)」と「APIキーの設定」でした。
パッケージのインストール
Windows PCの場合、下記コマンドプロンプトでインストールができます。
pip install google-genai opencv-python pyaudio pillow
APIキーの設定
いろいろやり方がありますが、今回はPythonのコード上にべた書きしました。
os.environ['GOOGLE_API_KEY'] = "XXX"
試しに実行してみる
下記ソースコードで試しに動かしてみました。
ソースコード
test.py
import asyncio
import base64
import io
import os
import sys
import traceback
import cv2
import pyaudio
import PIL.Image
from google import genai
# APIキーの準備
os.environ['GOOGLE_API_KEY'] = "XXX"
if sys.version_info < (3, 11, 0):
import taskgroup, exceptiongroup
asyncio.TaskGroup = taskgroup.TaskGroup
asyncio.ExceptionGroup = exceptiongroup.ExceptionGroup
FORMAT = pyaudio.paInt16
CHANNELS = 1
SEND_SAMPLE_RATE = 16000
RECEIVE_SAMPLE_RATE = 24000
CHUNK_SIZE = 1024
MODEL = "models/gemini-2.0-flash-exp"
client = genai.Client(http_options={"api_version": "v1alpha"})
CONFIG = {"generation_config": {"response_modalities": ["AUDIO"]}}
pya = pyaudio.PyAudio()
class AudioLoop:
def __init__(self):
self.audio_in_queue = None
self.audio_out_queue = None
self.video_out_queue = None
self.session = None
self.send_text_task = None
self.receive_audio_task = None
self.play_audio_task = None
async def send_text(self):
while True:
text = await asyncio.to_thread(
input,
"message > ",
)
if text.lower() == "q":
break
await self.session.send(text or ".", end_of_turn=True)
def _get_frame(self, cap):
# Read the frameq
ret, frame = cap.read()
# Check if the frame was read successfully
if not ret:
return None
# Fix: Convert BGR to RGB color space
# OpenCV captures in BGR but PIL expects RGB format
# This prevents the blue tint in the video feed
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
img = PIL.Image.fromarray(frame_rgb) # Now using RGB frame
img.thumbnail([1024, 1024])
image_io = io.BytesIO()
img.save(image_io, format="jpeg")
image_io.seek(0)
mime_type = "image/jpeg"
image_bytes = image_io.read()
return {"mime_type": mime_type, "data": base64.b64encode(image_bytes).decode()}
async def get_frames(self):
# This takes about a second, and will block the whole program
# causing the audio pipeline to overflow if you don't to_thread it.
cap = await asyncio.to_thread(
cv2.VideoCapture, 0
) # 0 represents the default camera
while True:
frame = await asyncio.to_thread(self._get_frame, cap)
if frame is None:
break
await asyncio.sleep(1.0)
await self.out_queue.put(frame)
# Release the VideoCapture object
cap.release()
async def send_realtime(self):
while True:
msg = await self.out_queue.get()
await self.session.send(msg)
async def listen_audio(self):
mic_info = pya.get_default_input_device_info()
self.audio_stream = await asyncio.to_thread(
pya.open,
format=FORMAT,
channels=CHANNELS,
rate=SEND_SAMPLE_RATE,
input=True,
input_device_index=mic_info["index"],
frames_per_buffer=CHUNK_SIZE,
)
if __debug__:
kwargs = {"exception_on_overflow": False}
else:
kwargs = {}
while True:
data = await asyncio.to_thread(self.audio_stream.read, CHUNK_SIZE, **kwargs)
await self.out_queue.put({"data": data, "mime_type": "audio/pcm"})
async def receive_audio(self):
"Background task to reads from the websocket and write pcm chunks to the output queue"
while True:
turn = self.session.receive()
async for response in turn:
if data := response.data:
self.audio_in_queue.put_nowait(data)
continue
if text := response.text:
print(text, end="")
# If you interrupt the model, it sends a turn_complete.
# For interruptions to work, we need to stop playback.
# So empty out the audio queue because it may have loaded
# much more audio than has played yet.
while not self.audio_in_queue.empty():
self.audio_in_queue.get_nowait()
async def play_audio(self):
stream = await asyncio.to_thread(
pya.open,
format=FORMAT,
channels=CHANNELS,
rate=RECEIVE_SAMPLE_RATE,
output=True,
)
while True:
bytestream = await self.audio_in_queue.get()
await asyncio.to_thread(stream.write, bytestream)
async def run(self):
try:
async with (
client.aio.live.connect(model=MODEL, config=CONFIG) as session,
asyncio.TaskGroup() as tg,
):
self.session = session
self.audio_in_queue = asyncio.Queue()
self.out_queue = asyncio.Queue(maxsize=5)
send_text_task = tg.create_task(self.send_text())
tg.create_task(self.send_realtime())
tg.create_task(self.listen_audio())
tg.create_task(self.get_frames())
tg.create_task(self.receive_audio())
tg.create_task(self.play_audio())
await send_text_task
raise asyncio.CancelledError("User requested exit")
except asyncio.CancelledError:
pass
except ExceptionGroup as EG:
self.audio_stream.close()
traceback.print_exception(EG)
if __name__ == "__main__":
main = AudioLoop()
asyncio.run(main.run())
結果ですが、使い方のせいなのか英語でしか会話ができませんでした。
また、カメラをオンにした状態で実行したところ、何が見える？に対して、メガネやジャケットが見えると話しており、私を特定しているようでした。
自分のカメラを表示させながら会話できるようにする
ClaudeやChatGPTに聞きながら、カメラを表示するように改良しました！
また、どうやらカメラに最初に映った画像のみしかGeminiへ連携されていないようだったので、フレームレートを設定して定期的にGeminiに映っている画像を送るように変更しました。
ソースコード
test_camera.py
import asyncio
import base64
import io
import os
import sys
import traceback
import threading
import queue
import tkinter as tk
from tkinter import ttk, scrolledtext
import cv2
import pyaudio
import PIL.Image
import PIL.ImageTk
from google import genai
# APIキーの準備（必要に応じて設定）
os.environ['GOOGLE_API_KEY'] = "XXX"
if sys.version_info < (3, 11, 0):
import taskgroup, exceptiongroup
asyncio.TaskGroup = taskgroup.TaskGroup
asyncio.ExceptionGroup = exceptiongroup.ExceptionGroup
FORMAT = pyaudio.paInt16
CHANNELS = 1
SEND_SAMPLE_RATE = 16000
RECEIVE_SAMPLE_RATE = 24000
CHUNK_SIZE = 1024
MODEL = "models/gemini-2.0-flash-exp"
client = genai.Client(http_options={"api_version": "v1alpha"})
CONFIG = {"generation_config": {"response_modalities": ["AUDIO"]}}
pya = pyaudio.PyAudio()
class AudioLoop:
def __init__(self, ui):
self.ui = ui
self.audio_in_queue = None
self.send_queue = None
self.session = None
self.audio_stream = None
# video frame用
self.frame_queue = asyncio.Queue()
# Text I/O用
self.input_text_queue = asyncio.Queue() # ユーザーがGUIで入力したテキスト
self.stop_flag = False
async def send_text(self):
# GUIからの入力を待機する
while True:
if self.stop_flag:
break
text = await self.input_text_queue.get()
if text.lower() == "q":
break
await self.session.send(text or ".", end_of_turn=True)
def _get_frame(self, cap):
ret, frame = cap.read()
if not ret:
return None, None
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
img = PIL.Image.fromarray(frame_rgb)
img.thumbnail([640, 480]) # サイズ調整
image_io = io.BytesIO()
img.save(image_io, format="jpeg")
image_io.seek(0)
image_bytes = image_io.read()
mime_type = "image/jpeg"
send_data = {"mime_type": mime_type, "data": base64.b64encode(image_bytes).decode()}
return img, send_data
async def get_frames(self):
cap = await asyncio.to_thread(cv2.VideoCapture, 0)
while True:
if self.stop_flag:
break
img, frame_data = await asyncio.to_thread(self._get_frame, cap)
if img is None:
break
# UI表示用のasync queueへ画像を送る
await self.frame_queue.put(img)
# セッションへも送信（フレーム画像をモデルに送る場合）
await self.send_queue.put(frame_data)
await asyncio.sleep(0.5)
cap.release()
async def send_realtime(self):
while True:
if self.stop_flag:
break
msg = await self.send_queue.get()
await self.session.send(msg)
async def listen_audio(self):
mic_info = pya.get_default_input_device_info()
self.audio_stream = await asyncio.to_thread(
pya.open,
format=FORMAT,
channels=CHANNELS,
rate=SEND_SAMPLE_RATE,
input=True,
input_device_index=mic_info["index"],
frames_per_buffer=CHUNK_SIZE,
)
if __debug__:
kwargs = {"exception_on_overflow": False}
else:
kwargs = {}
while True:
if self.stop_flag:
break
data = await asyncio.to_thread(self.audio_stream.read, CHUNK_SIZE, **kwargs)
await self.send_queue.put({"data": data, "mime_type": "audio/pcm"})
async def receive_audio(self):
while True:
if self.stop_flag:
break
turn = self.session.receive()
async for response in turn:
if data := response.data:
self.audio_in_queue.put_nowait(data)
continue
if text := response.text:
# ここでUIテキストに書き込み
self.ui.append_text("Model: " + text)
# ターン完了後、残っているオーディオデータがあればクリア
while not self.audio_in_queue.empty():
self.audio_in_queue.get_nowait()
async def play_audio(self):
stream = await asyncio.to_thread(
pya.open,
format=FORMAT,
channels=CHANNELS,
rate=RECEIVE_SAMPLE_RATE,
output=True,
)
while True:
if self.stop_flag:
break
bytestream = await self.audio_in_queue.get()
await asyncio.to_thread(stream.write, bytestream)
async def run(self):
try:
async with (
client.aio.live.connect(model=MODEL, config=CONFIG) as session,
asyncio.TaskGroup() as tg,
):
self.session = session
self.audio_in_queue = asyncio.Queue()
self.send_queue = asyncio.Queue(maxsize=5)
send_text_task = tg.create_task(self.send_text())
tg.create_task(self.send_realtime())
tg.create_task(self.listen_audio())
tg.create_task(self.get_frames())
tg.create_task(self.receive_audio())
tg.create_task(self.play_audio())
await send_text_task
raise asyncio.CancelledError("User requested exit")
except asyncio.CancelledError:
pass
except ExceptionGroup as EG:
if self.audio_stream:
self.audio_stream.close()
traceback.print_exception(EG)
class UI:
def __init__(self):
self.root = tk.Tk()
self.root.title("Real-Time Camera & Voice UI")
# カメラ表示ラベル
self.image_label = tk.Label(self.root)
self.image_label.pack()
self.loop = AudioLoop(self)
# 非同期から同期に受け渡すためのキュー
self.sync_frame_queue = queue.Queue()
# 映像表示用の定期アップデート
self.root.after(100, self.update_frame)
def append_text(self, message):
self.text_area.insert(tk.END, message + "\n")
self.text_area.see(tk.END)
def send_text(self):
text = self.input_entry.get()
self.input_entry.delete(0, tk.END)
if text.strip():
# ユーザー入力をasyncioキューへ送る
asyncio.run_coroutine_threadsafe(self.loop.input_text_queue.put(text), self.async_loop)
self.append_text("You: " + text)
def update_frame(self):
# 同期キューから画像を取り出し、表示
try:
while True:
img = self.sync_frame_queue.get_nowait()
tk_image = PIL.ImageTk.PhotoImage(img)
self.image_label.config(image=tk_image)
self.image_label.image = tk_image # 参照保持が必要
except queue.Empty:
pass
# 次回呼び出し
self.root.after(100, self.update_frame)
def start_asyncio_tasks(self):
# 別スレッドでasyncio.runを起動
def run_loop():
self.async_loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.async_loop)
async def frame_updater():
while not self.loop.stop_flag:
img = await self.loop.frame_queue.get()
# 非同期キューから取得した画像を同期キューへ送る
self.sync_frame_queue.put(img)
# frame_updaterを起動
self.async_loop.create_task(frame_updater())
self.async_loop.run_until_complete(self.loop.run())
self.thread = threading.Thread(target=run_loop, daemon=True)
self.thread.start()
def run(self):
self.start_asyncio_tasks()
self.root.mainloop()
# 終了時フラグ
self.loop.stop_flag = True
if __name__ == "__main__":
ui = UI()
ui.run()
上記Pythonを動かすと下記動画のようにカメラの様子が映されます。
で、カメラごしにGeminiと会話できました！
例えば、「私の着ている服の色は？」「私が手に持っているものは？」「私はどこにいますか？」といった問いかけに的確に答えてくれました。
さらにNintendo Switchを手に持って「手に持っているのはなに？」と問いかけると「Nintendo Switchです」と回答してくれました。
ただし、日本語で回答してくれる時と、英語の時があります。
・・・めちゃ楽しい😄
さいごに
今回、Gemini 2.0のMulti Modal APIを試してみましたが、OpenAIのAdvanced Voice ModeをAPIを使ってPythonで体験できた点、感動しました！
ChatGPTが話題になって以降、コールセンター業務の品質向上で、会話内容を文字お越しして生成AIに回答案を出力させるなど取り組みがなされていましたが、このAPIを応用するとより高速かつ的確な回答ができるのではと思いました。(一方で、顧客情報や機密情報の観点でクリアしなければならない壁がありそうですが。)
アクセンチュア株式会社所属の社員有志による運営です。アクセンチュアの社員による様々な発信をまとめています。なお、投稿内容は社員個人の見解であり、所属する組織を代表するものではありません。
Discussion