OpenAI Realtime API

Weave を使用して、OpenAI Realtime API への呼び出しを自動的にトレースします。

Weave の OpenAI Realtime API とのインテグレーションにより、アプリケーションで発生する音声対音声のやり取りをリアルタイムで自動的にトレースできます。これを使用すると、エージェントとユーザーの会話をキャプチャして、エージェントのパフォーマンスを確認・評価できます。

このガイドでは、アプリケーションで Realtime API のトレースを有効にする方法を説明し、ローカルで実行できる 2 つのエンドツーエンドの音声アシスタントの例を紹介します。

​

リアルタイムトレースを統合する

このセクションでは、あらゆるアプリケーションで OpenAI Realtime API の Weave トレースを有効にするために必要な最小限のコードを示します。

Weave は OpenAI Realtime API に自動的にパッチを適用するため、アプリケーションでのオーディオインタラクションのキャプチャを開始するには、数行のコードを追加するだけです。以下のコードでは、Weave と Realtime API のインテグレーションをインポートします。

import weave
from weave.integrations import patch_openai_realtime

weave.init("your-team-name/your-project-name")
patch_openai_realtime()

# アプリケーションのロジック

コードにimportして実行すると、Weave はユーザーと OpenAI Realtime API 間のやり取りを自動的にトレースします。

​

OpenAI Agents SDK を使用してリアルタイム音声アシスタントを実行する

この例では、マイクの音声を OpenAI の Realtime API にストリーミングし、AI の音声による応答をローカルマシンのスピーカーで再生するリアルタイム音声アシスタントを実行します。このアプリケーションは RealtimeAgent と RealtimeRunner を使用する OpenAI Agents SDK を利用し、patch_openai_realtime() を適用して Weave のトレースを有効にします。Realtime セッションの管理を上位レベルの Agents SDK に任せたい場合は、この方法を使用します。

例を実行するには、次の手順を実行します：

1. Python 環境を起動して、次のライブラリをインストールします:

   uv

   uv add weave openai-agents websockets pyaudio numpy
   

   pip

   pip install weave openai-agents websockets pyaudio numpy
   

2. weave_voice_assistant.py という名前のファイルを作成し、次のコードを追加します。

   ハイライトされた行は、アプリケーション内での Weave のインテグレーションを示しています。残りのコードは、基本的な音声アシスタントアプリを作成するものです。

   weave_voice_assistant.py

   import argparse
   import asyncio
   import queue
   import sys
   import termios
   import threading
   import tty
   import weave
   import pyaudio
   import numpy as np
   from weave.integrations import patch_openai_realtime
   from agents.realtime import RealtimeAgent, RealtimeRunner
   
   DEFAULT_WEAVE_PROJECT = "<your-team-name/your-project-name>"
   
   FORMAT = pyaudio.paInt16
   RATE = 24000  # Required by the OpenAI Realtime API.
   CHUNK = 1024
   MAX_INPUT_CHANNELS = 1
   MAX_OUTPUT_CHANNELS = 1
   
   INP_DEV_IDX = None
   OUT_DEV_IDX = None
   
   # Weave プロジェクト名とオーディオデバイス選択のための CLI 引数を解析します。
   def parse_args():
       parser = argparse.ArgumentParser(description="Realtime agent with Weave logging")
       parser.add_argument(
           "--weave-project",
           default=DEFAULT_WEAVE_PROJECT,
           help=f"Weave project name (default: {DEFAULT_WEAVE_PROJECT})",
           dest="weave_project"
       )
       parser.add_argument(
           "--input-device",
           type=int,
           default=None,
           help="PyAudio input (mic) device index. Defaults to system default. Run mic_detect.py to list devices.",
           dest="input_device"
       )
       parser.add_argument(
           "--output-device",
           type=int,
           default=None,
           help="PyAudio output (speaker) device index. Defaults to system default. Run mic_detect.py to list devices.",
           dest="output_device"
       )
       return parser.parse_args()
   
   
   # Weave を初期化し、トレース用に OpenAI Realtime API にパッチを適用します。
   def init_weave(project_name: str | None = None) -> None:
       name = project_name or DEFAULT_WEAVE_PROJECT
       weave.init(name)
       patch_openai_realtime()  # Enables automatic tracing of Realtime API sessions.
   
   
   mic_enabled = True
   
   # 't' キーを監視してマイクのオン/オフを切り替えます。デーモンスレッドで実行されます。
   def start_keylistener():
       global mic_enabled
       fd = sys.stdin.fileno()
       old_settings = termios.tcgetattr(fd)
       try:
           tty.setcbreak(fd)
           while True:
               ch = sys.stdin.read(1)
               if ch.lower() == 't':
                   mic_enabled = not mic_enabled
                   state = "ON" if mic_enabled else "OFF"
                   print(f"\n🎙  Mic {state} (press t to toggle)")
               elif ch == '\x03':  # Ctrl-C
                   break
       finally:
           termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
   
   
   # バックグラウンドスレッドでオーディオキューを消費し、スピーカーに書き込みます。
   def play_audio(output_stream: pyaudio.Stream, audio_output_queue: queue.Queue):
       while True:
           data = audio_output_queue.get()
           if data is None:
               break
           output_stream.write(data)
   
   
   # オーディオストリームを開き、Realtime セッションを開始して、送受信ループを実行します。
   async def main(*, input_device_index: int | None = None, output_device_index: int | None = None):
       p = pyaudio.PyAudio()
   
       if input_device_index is None:
           input_device_index = int(p.get_default_input_device_info()['index'])
       if output_device_index is None:
           output_device_index = int(p.get_default_output_device_info()['index'])
   
       # pyaudio エラーを避けるため、チャンネル数をデバイスの上限に制限します。
       input_info = p.get_device_info_by_index(input_device_index)
       output_info = p.get_device_info_by_index(output_device_index)
   
       input_channels = min(int(input_info['maxInputChannels']), MAX_INPUT_CHANNELS)
       output_channels = min(int(output_info['maxOutputChannels']), MAX_OUTPUT_CHANNELS)
   
       mic = p.open(
           format=FORMAT,
           channels=input_channels,
           rate=RATE,
           input=True,
           output=False,
           frames_per_buffer=CHUNK,
           input_device_index=input_device_index,
           start=False,
       )
       speaker = p.open(
           format=FORMAT,
           channels=output_channels,
           rate=RATE,
           input=False,
           output=True,
           frames_per_buffer=CHUNK,
           output_device_index=output_device_index,
           start=False,
       )
       mic.start_stream()
       speaker.start_stream()
   
       # 割り込み時に再生中のオーディオをフラッシュできるよう、キューを介してバッファリングします。
       audio_output_queue = queue.Queue()
       threading.Thread(
           target=play_audio, args=(speaker, audio_output_queue), daemon=True
       ).start()
   
       s_agent = RealtimeAgent(
           name="Speech Assistant",
           instructions="You are a tool using AI. Use tools to accomplish a task whenever possible"
       )
   
       s_runner = RealtimeRunner(s_agent, config={
           "model_settings": {
               "model_name": "gpt-realtime",
               "modalities": ["audio"],
               "output_modalities": ["audio"],
               "input_audio_format": "pcm16",
               "output_audio_format": "pcm16",
               "speed": 1.2,
               "turn_detection": {
                   "prefix_padding_ms": 100,
                   "silence_duration_ms": 100,
                   "type": "server_vad",
                   "interrupt_response": True,
                   "create_response": True,
               },
           }
       })
       print("--- Session Active (Speak into mic) ---")
       print("🎙  Mic ON (press t to toggle)")
   
       threading.Thread(target=start_keylistener, daemon=True).start()
   
       async with await s_runner.run() as session:
           # マイク入力を Realtime API にストリーミングし、ミュート時は無音を送信します。
           async def send_mic_audio():
               silence = b'\x00' * CHUNK * 2  # サンプルあたり 2 バイト（16 ビット PCM）。
               try:
                   while True:
                       raw_data = mic.read(CHUNK, exception_on_overflow=False)
   
                       if mic_enabled:
                           audio_data = np.frombuffer(raw_data, dtype=np.int16).astype(np.float64)
                           rms = np.sqrt(np.mean(audio_data**2))
                           meter = int(min(rms / 50, 50))
                           print(f"Mic Level: {'█' * meter}{' ' * (50-meter)} | 🎙 ON ", end="\r")
                           await session.send_audio(raw_data)
                       else:
                           print(f"Mic Level: {' ' * 50} | 🎙 OFF", end="\r")
                           await session.send_audio(silence)
   
                       await asyncio.sleep(0)  # 読み取りの合間にイベントループへ制御を返します。
               except Exception:
                   pass
   
           # セッションからイベントを受信し、オーディオをスピーカーに転送します。
           async def handle_events():
               async for event in session:
                   if event.type == "audio":
                       audio_output_queue.put(event.audio.data)
                   elif event.type == "audio_interrupted":
                       # キューに溜まった AI のオーディオをフラッシュして、ユーザーの発話に被らないようにします。
                       while not audio_output_queue.empty():
                           try:
                               audio_output_queue.get_nowait()
                           except queue.Empty:
                               break
   
           mic_task = asyncio.create_task(send_mic_audio())
           try:
               await handle_events()
           finally:
               mic_task.cancel()
   
       # クリーンアップ
       audio_output_queue.put(None)  # 再生スレッドに終了を通知します。
       mic.close()
       speaker.close()
       p.terminate()
   
   if __name__ == "__main__":
       args = parse_args()
       init_weave(args.weave_project)
       fd = sys.stdin.fileno()
       old_settings = termios.tcgetattr(fd)
       try:
           asyncio.run(main(input_device_index=args.input_device, output_device_index=args.output_device))
       finally:
           termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
   

3. DEFAULT_WEAVE_PROJECT の値をチーム名とプロジェクト名に更新します。

4. OPENAI_API_KEY 環境変数を設定します。

5. 音声アシスタントを起動します:

   python weave_voice_assistant.py
   

アシスタントが起動したら、キーボードの T キーを押してマイクのミュートを切り替えます。アシスタントは、発話のターン管理と割り込みの処理に、サーバー側の音声アクティビティ検出を使用します。

アシスタントに話しかけると、Weave がセッションのオーディオを含むトレースを記録し、それを Weave UI で確認できます。

​

WebSockets を使用してリアルタイム音声アシスタントを実行する

次の例では、WebSockets 経由で OpenAI Realtime API に直接接続します。マイクのオーディオを API にストリーミングし、音声応答を再生し、ツール呼び出し (天気の検索、数式の評価、コードの実行、ファイルの書き込み) をサポートします。Weave は weave.init() と patch_openai_realtime() を使用してセッションをトレースします。Agents SDK に依存せずに Realtime セッションを完全に制御したい場合は、この方法を使用してください。

例を実行するには、次の手順を完了してください:

1. Python 環境を起動し、次のライブラリをインストールしてください。

   uv

   uv add weave websockets pyaudio numpy
   

   pip

   pip install weave websockets pyaudio numpy
   

2. tool_definitions.py という名前のファイルを作成し、次のツール定義を追加します。メインのアプリケーションはこのモジュールをインポートします。

   tool_definitions.py

   import json
   import subprocess
   import tempfile
   from pathlib import Path
   import weave
   
   
   # @function_tool
   @weave.op
   def get_weather(city: str) -> str:
       """指定した都市の現在の天気を取得します。
   
       引数:
           city: 天気を取得する都市名。
       """
       return json.dumps({"city": city, "temperature": "72°F", "condition": "sunny"})
   
   
   @weave.op
   def calculate(expression: str) -> str:
       """数式を評価して結果を返します。
   
       引数:
           expression: 評価する数式（例: '2 + 2'）。
       """
       try:
           result = eval(expression)
           return str(result)
       except Exception as e:
           return f"Error: {e}"
   
   
   @weave.op
   def run_python_code(code: str) -> str:
       """Python スクリプトを書き出して実行し、その stdout/stderr を返します。
   
       引数:
           code: 実行する Python ソースコード。
       """
       with tempfile.NamedTemporaryFile(
           mode="w", suffix=".py", dir=tempfile.gettempdir(), delete=False
       ) as f:
           f.write(code)
           script_path = Path(f.name)
   
       try:
           result = subprocess.run(
               ["python", str(script_path)],
               capture_output=True,
               text=True,
               timeout=30,
           )
           output = result.stdout
           if result.stderr:
               output += f"\nSTDERR:\n{result.stderr}"
           if result.returncode != 0:
               output += f"\n(exit code {result.returncode})"
           return output or "(no output)"
       except subprocess.TimeoutExpired:
           return "Error: script timed out after 30 seconds."
       finally:
           script_path.unlink(missing_ok=True)
   
   
   @weave.op
   async def write_file(file_path: str, content: str) -> str:
       """ディスク上のファイルにコンテンツを書き込みます。
   
       引数:
           file_path: ファイルの書き込み先パス。
           content: ファイルに書き込むコンテンツ。
       """
       try:
           path = Path(file_path)
           path.parent.mkdir(parents=True, exist_ok=True)
           path.write_text(content)
           return f"Wrote {len(content)} bytes to {file_path}"
       except Exception as e:
           return f"Error writing file: {e}"
   

3. 同じディレクトリに weave_ws_voice_assistant.py というファイルを作成し、次のコードを追加します。

   weave_ws_voice_assistant.py

   import asyncio
   import base64
   import json
   import os
   import queue
   import threading
   from typing import Any, Callable
   
   import numpy as np
   import pyaudio
   import websockets
   
   import weave
   weave.init("<your-team-name/your-project-name>")
   from weave.integrations import patch_openai_realtime
   patch_openai_realtime()
   
   from tool_definitions import (
       calculate,
       get_weather,
       run_python_code,
       write_file,
   )
   
   # オーディオフォーマット（Realtime API では PCM16 を使用する必要があります）。
   FORMAT = pyaudio.paInt16
   RATE = 24000
   CHUNK = 1024
   MAX_INPUT_CHANNELS = 2
   MAX_OUTPUT_CHANNELS = 2
   
   OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
   REALTIME_URL = "wss://api.openai.com/v1/realtime?model=gpt-realtime"
   
   DEBUG_WRITE_LOG = False
   
   # ツール名 -> 関数呼び出しディスパッチ用の callable のマッピング。
   TOOL_REGISTRY: dict[str, Callable[..., Any]] = {
       "get_weather": get_weather,
       "calculate": calculate,
       "run_python_code": run_python_code,
       "write_file": write_file,
   }
   
   # Realtime API セッション設定用の生のツール定義。
   TOOL_DEFINITIONS = [
       {
           "type": "function",
           "name": "get_weather",
           "description": "Get the current weather for a city.",
           "parameters": {
               "type": "object",
               "properties": {
                   "city": {
                       "type": "string",
                       "description": "The city name to get weather for.",
                   }
               },
               "required": ["city"],
           },
       },
       {
           "type": "function",
           "name": "calculate",
           "description": "Evaluate a math expression and return the result.",
           "parameters": {
               "type": "object",
               "properties": {
                   "expression": {
                       "type": "string",
                       "description": "A math expression to evaluate, e.g. '2 + 2'.",
                   }
               },
               "required": ["expression"],
           },
       },
       {
           "type": "function",
           "name": "run_python_code",
           "description": "Write and execute a Python script, returning its stdout/stderr.",
           "parameters": {
               "type": "object",
               "properties": {
                   "code": {
                       "type": "string",
                       "description": "The Python source code to execute.",
                   }
               },
               "required": ["code"],
           },
       },
       {
           "type": "function",
           "name": "write_file",
           "description": "Write content to a file on disk.",
           "parameters": {
               "type": "object",
               "properties": {
                   "file_path": {
                       "type": "string",
                       "description": "The path to write the file to.",
                   },
                   "content": {
                       "type": "string",
                       "description": "The content to write into the file.",
                   },
               },
               "required": ["file_path", "content"],
           },
       },
   ]
   
   
   async def send_event(ws, event: dict) -> None:
       await ws.send(json.dumps(event))
   
   
   async def configure_session(ws) -> None:
       event = {
           "type": "session.update",
           "session": {
               "type": "realtime",
               "model": "gpt-realtime",
               "output_modalities": ["audio"],
               "instructions": (
                   "You are a helpful AI assistant with access to tools. "
                   "Use tools to accomplish tasks whenever possible. "
                   "Speak clearly and briefly."
               ),
               "tools": TOOL_DEFINITIONS,
               "tool_choice": "auto",
               "audio": {
                   "input": {
                       "format": {"type": "audio/pcm", "rate": 24000},
                       "transcription": {"model": "gpt-4o-transcribe"},
                       "turn_detection": {
                           "type": "server_vad",
                           "threshold": 0.5,
                           "prefix_padding_ms": 300,
                           "silence_duration_ms": 500,
                       },
                   },
                   "output": {
                       "format": {"type": "audio/pcm", "rate": 24000},
                   },
               },
           },
       }
       await send_event(ws, event)
       print("Session configured.")
   
   
   async def handle_function_call(ws, call_id: str, name: str, arguments: str) -> None:
       if not name:
           raise Exception("Did not get a function name")
   
       print(f"\n[Function Call] {name}({arguments})")
       tool_fn = TOOL_REGISTRY.get(name)
       if tool_fn is None:
           result = json.dumps({"error": f"Unknown function: {name}"})
       else:
           try:
               args = json.loads(arguments)
               result = tool_fn(**args)
               if asyncio.iscoroutine(result):
                   result = await result
           except Exception as e:
               result = json.dumps({"error": str(e)})
   
       print(f"[Function Result] {result}")
   
       # 関数呼び出しの出力をモデルに送り返す。
       await send_event(ws, {
           "type": "conversation.item.create",
           "item": {
               "type": "function_call_output",
               "call_id": call_id,
               "output": result if isinstance(result, str) else json.dumps(result),
           },
       })
   
       # モデルが関数の結果を反映できるよう、新しいレスポンスをトリガーする。
       await send_event(ws, {"type": "response.create"})
   
   
   def play_audio(output_stream: pyaudio.Stream, audio_output_queue: queue.Queue):
       """pyaudio の write() はサウンドカードがサンプルを消費するまでブロックするため、
       別スレッドで実行します。再生を非同期イベントループから切り離すことで、
       処理中の書き込みの完了を待たずに割り込み時にキューをフラッシュできます。"""
       while True:
           data = audio_output_queue.get()
           if data is None:
               break
           output_stream.write(data)
   
   
   async def send_mic_audio(ws, mic) -> None:
       try:
           while True:
               raw_data = mic.read(CHUNK, exception_on_overflow=False)
   
               # ビジュアルボリュームメーター。
               audio_data = np.frombuffer(raw_data, dtype=np.int16).astype(np.float64)
               rms = np.sqrt(np.mean(audio_data**2))
               meter = int(min(rms / 50, 50))
               print(f"Mic Level: {'█' * meter}{' ' * (50 - meter)} |", end="\r")
   
               # Base64 エンコードしてオーディオチャンクを送信する。
               b64_audio = base64.b64encode(raw_data).decode("utf-8")
               await send_event(ws, {
                   "type": "input_audio_buffer.append",
                   "audio": b64_audio,
               })
   
               await asyncio.sleep(0)
       except asyncio.CancelledError:
           pass
   
   
   async def receive_events(ws, audio_output_queue: queue.Queue) -> None:
       # デルタイベントをまたいで関数呼び出しの引数を蓄積する。
       pending_calls: dict[str, dict] = {}
       async for raw_message in ws:
           if DEBUG_WRITE_LOG:
               with open("data.jsonl", "a", encoding="utf-8") as f:
                   f.write(json.dumps(raw_message) + "\n")
   
           event = json.loads(raw_message)
           event_type = event.get("type", "")
   
           if event_type == "session.created":
               print(raw_message)
   
           elif event_type == "session.updated":
               print(raw_message)
   
           elif event_type == "error":
               print(f"\n[Error] {event}")
   
           elif event_type == "input_audio_buffer.speech_started":
               # キューに溜まった AI オーディオをフラッシュして、ユーザーの発話に被らないようにする。
               while not audio_output_queue.empty():
                   try:
                       audio_output_queue.get_nowait()
                   except queue.Empty:
                       break
   
           elif event_type == "input_audio_buffer.speech_stopped":
               pass
   
           elif event_type == "input_audio_buffer.committed":
               pass
   
           elif event_type == "response.created":
               pass
   
           elif event_type == "response.output_text.delta":
               pass
   
           elif event_type == "response.output_text.done":
               pass
   
           # オーディオ出力デルタ - 再生用にキューに追加する。
           elif event_type == "response.output_audio.delta":
               audio_bytes = base64.b64decode(event.get("delta", ""))
               audio_output_queue.put(audio_bytes)
   
           elif event_type == "response.output_audio_transcript.delta":
               pass
   
           elif event_type == "response.output_audio_transcript.done":
               pass
   
           # 関数呼び出し開始 - 保留中の呼び出しを初期化する。
           elif event_type == "response.output_item.added":
               item = event.get("item", {})
               if item.get("type") == "function_call" and item.get("status") == "in_progress":
                   item_id = item.get("id", "")
                   pending_calls[item_id] = {
                       "call_id": item.get("call_id", ""),
                       "name": item.get("name", ""),
                       "arguments": "",
                   }
                   print(f"\n[Function Call Started] {item.get('name', '')}")
   
           # 関数呼び出しの引数デルタ - 蓄積する。
           elif event_type == "response.function_call_arguments.delta":
               item_id = event.get("item_id", "")
               if item_id in pending_calls:
                   pending_calls[item_id]["arguments"] += event.get("delta", "")
   
           elif event_type == "response.function_call_arguments.done":
               item_id = event.get("item_id", "")
               call_info = pending_calls.pop(item_id, None)
               if call_info is None:
                   # Fallback: use data directly from the done event.
                   call_info = {
                       "call_id": event.get("call_id"),
                       "name": event.get("name"),
                       "arguments": event.get("arguments"),
                   }
               try:
                   await handle_function_call(
                       ws,
                       call_info["call_id"],
                       call_info["name"],
                       call_info["arguments"],
                   )
               except Exception as e:
                   print(f"Failed to call function for message {call_info}: error - {e}")
   
           elif event_type == "response.done":
               pass
   
           elif event_type == "rate_limits.updated":
               pass
   
           else:
               print(f"\n[Event: {event_type}]")
   
   
   async def main():
       if not OPENAI_API_KEY:
           print("Error: OPENAI_API_KEY environment variable not set")
           return
   
       p = pyaudio.PyAudio()
   
       input_device_index = int(p.get_default_input_device_info()['index'])
       output_device_index = int(p.get_default_output_device_info()['index'])
   
       # Channel count must match the device's capabilities or pyaudio errors upon open.
       input_info = p.get_device_info_by_index(input_device_index)
       output_info = p.get_device_info_by_index(output_device_index)
       input_channels = min(int(input_info['maxInputChannels']), 1)
       output_channels = min(int(output_info['maxOutputChannels']), 1)
   
       mic = p.open(
           format=FORMAT,
           channels=input_channels,
           rate=RATE,
           input=True,
           output=False,
           frames_per_buffer=CHUNK,
           input_device_index=input_device_index,
           start=False,
       )
       speaker = p.open(
           format=FORMAT,
           channels=output_channels,
           rate=RATE,
           input=False,
           output=True,
           frames_per_buffer=CHUNK,
           output_device_index=output_device_index,
           start=False,
       )
       mic.start_stream()
       speaker.start_stream()
   
       # オーディオをキュー経由にすることで、ユーザーが割り込んだときにフラッシュできます。
       # スピーカーに直接書き込むと、再生中のオーディオをキャンセルできなくなります。
       audio_output_queue = queue.Queue()
       threading.Thread(
           target=play_audio, args=(speaker, audio_output_queue), daemon=True
       ).start()
   
       headers = {
           "Authorization": f"Bearer {OPENAI_API_KEY}",
       }
   
       print("Connecting to OpenAI Realtime API...")
   
       async with websockets.connect(
           REALTIME_URL,
           additional_headers=headers,
       ) as ws:
           print("Connected! Configuring session...")
           await configure_session(ws)
   
           print("--- Session Active (Speak into mic) ---")
   
           mic_task = asyncio.create_task(send_mic_audio(ws, mic))
           try:
               await receive_events(ws, audio_output_queue)
           finally:
               mic_task.cancel()
               try:
                   await mic_task
               except asyncio.CancelledError:
                   pass
   
       # Cleanup
       audio_output_queue.put(None)  # Signal playback thread to exit.
       mic.close()
       speaker.close()
       p.terminate()
       print("\nSession ended.")
   
   
   if __name__ == "__main__":
       asyncio.run(main())
   

4. weave.init() の呼び出しを、チーム名とプロジェクト名に合わせて更新します。

5. OPENAI_API_KEY 環境変数を設定してください。

6. 音声アシスタントを起動します:

   python weave_ws_voice_assistant.py
   

アシスタントの実行中は、キーボードの T キーを押してマイクのミュートを切り替えます。アシスタントは、発話のターン管理と割り込みの処理に、サーバー側の音声アクティビティ検出を使用します。

アシスタントに話しかけると、Weave がセッションのオーディオを含むトレースを記録し、それを Weave UI で確認できます。