OpenAI「Agents SDK」⑧音声エージェント

uv init -p 3.12 openai-voice-agents-work && cd $_

uv add 'openai-agents[voice]'

(snip)
 + openai-agents==0.3.0
(snip)

import asyncio
import random

import numpy as np
import sounddevice as sd

from agents import (
    Agent,
    function_tool,
    set_tracing_disabled,
)
from agents.voice import (
    AudioInput,
    SingleAgentVoiceWorkflow,
    VoicePipeline,
)
from agents.extensions.handoff_prompt import prompt_with_handoff_instructions


# ツールの定義
@function_tool
def get_weather(city: str) -> str:
    """指定された都市の天気を取得する"""
    print(f"[debug] get_weather が読み出されました。指定された都市名: {city}")
    choices = ["晴れ", "曇り", "雨", "雪"]
    return f"{city} の天気は {random.choice(choices)} です。"

# ハンドオフされるエージェント
spanish_agent = Agent(
    name="Spanish",
    handoff_description="スペイン語を話すエージェント",
    instructions=prompt_with_handoff_instructions(
        "あなたは人間と話しているので、丁寧で簡潔に話してください。スペイン語で話してください。",
    ),
    model="gpt-4o-mini",
)

# メインのエージェント
agent = Agent(
    name="Assistant",
    instructions=prompt_with_handoff_instructions(
        (
            "あなたは人間と話しているので、丁寧で簡潔に話してください。"
            "もしユーザがスペイン語を話す場合は、スペイン語を話すエージェントにハンドオフしてください。"
        )
    ),
    model="gpt-4o-mini",
    handoffs=[spanish_agent],  # ハンドオフするエージェントを追加
    tools=[get_weather],  # ツールを追加
)

async def main():
    # `VoicePipeline` でパイプラインを作成
    pipeline = VoicePipeline(workflow=SingleAgentVoiceWorkflow(agent))

    # シンプルな例として、3秒間の静音を生成する
    # 実際にはマイクから音声データを取得する
    buffer = np.zeros(24000 * 3, dtype=np.int16)
    audio_input = AudioInput(buffer=buffer)

    # パイプラインに音声データを渡して実行
    result = await pipeline.run(audio_input)

    # 音声の再生に `sounddevice` を使う
    player = sd.OutputStream(samplerate=24000, channels=1, dtype=np.int16)
    player.start()

    # 音声ストリームを再生
    async for event in result.stream():
        if event.type == "voice_stream_event_audio":
            player.write(event.data)


if __name__ == "__main__":
    asyncio.run(main())

export OPENAI_API_KEY=XXXXXXXXXX

uv run sample.py

ModuleNotFoundError: No module named 'sounddevice'

uv add sounddevice

import curses
import time

import numpy as np
import numpy.typing as npt
import sounddevice as sd

import asyncio
import random

import numpy as np

from agents import Agent, function_tool
from agents.extensions.handoff_prompt import prompt_with_handoff_instructions
from agents.voice import (
    AudioInput,
    SingleAgentVoiceWorkflow,
    SingleAgentWorkflowCallbacks,
    VoicePipeline,
)

"""
録音されたオーディオバッファを使用するシンプルな例。

1. You can record an audio clip in the terminal.
2. The pipeline automatically transcribes the audio.
3. The agent workflow is a simple one that starts at the Assistant agent.
4. The output of the agent is streamed to the audio player.

以下のような例を試すことができます:
- なにかジョークを話して （ジョークで応答する）
- 東京の天気は？ （`get_weather`ツールを呼び出して応答する）
- Hola, como estas? （スペイン語を話すエージェントにハンドオフされる）
"""


def _record_audio(screen: curses.window) -> npt.NDArray[np.float32]:
    """
    マイクを使って録音を行うヘルパー関数
    """
    screen.nodelay(True)  # ノンブロッキング入力
    screen.clear()
    screen.addstr(
        "<スペースキー>を押して録音を開始してください。再度<スペースキー>を押すと録音が終了します。\n"
    )
    screen.refresh()

    recording = False
    audio_buffer: list[npt.NDArray[np.float32]] = []

    def _audio_callback(indata, frames, time_info, status):
        if status:
            screen.addstr(f"Status: {status}\n")
            screen.refresh()
        if recording:
            audio_buffer.append(indata.copy())

    # コールバックで音声ストリームを開く
    with sd.InputStream(samplerate=24000, channels=1, dtype=np.float32, callback=_audio_callback):
        while True:
            key = screen.getch()
            if key == ord(" "):
                recording = not recording
                if recording:
                    screen.addstr("録音を開始しました...\n")
                else:
                    screen.addstr("録音を終了しました...\n")
                    break
                screen.refresh()
            time.sleep(0.01)

    # 録音した音声チャンクを結合する
    if audio_buffer:
        audio_data = np.concatenate(audio_buffer, axis=0)
    else:
        audio_data = np.empty((0,), dtype=np.float32)

    return audio_data


def record_audio():
    # cursesを使用することで、以下を実現しつつ録音を行う
    # - macOSでアクセシビリティ権限を必要としない
    # - ターミナルをブロックしない
    audio_data = curses.wrapper(_record_audio)
    return audio_data


class AudioPlayer:
    """
    音声を再生するヘルパークラス
    """
    def __enter__(self):
        self.stream = sd.OutputStream(samplerate=24000, channels=1, dtype=np.int16)
        self.stream.start()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.stream.stop()  # ストリームが完了するのを待つ
        self.stream.close()

    def add_audio(self, audio_data: npt.NDArray[np.int16]):
        self.stream.write(audio_data)


# ツールの定義
@function_tool
def get_weather(city: str) -> str:
    """Get the weather for a given city."""
    print(f"[debug] get_weather called with city: {city}")
    choices = ["sunny", "cloudy", "rainy", "snowy"]
    return f"The weather in {city} is {random.choice(choices)}."

# ハンドオフされるエージェント
spanish_agent = Agent(
    name="Spanish",
    handoff_description="スペイン語を話すエージェント",
    instructions=prompt_with_handoff_instructions(
        "あなたは人間と話しているので、丁寧で簡潔に話してください。スペイン語で話してください。",
    ),
    model="gpt-5-mini",
)

# メインのエージェント
agent = Agent(
    name="Assistant",
    instructions=prompt_with_handoff_instructions(
        (
            "あなたは人間と話しているので、丁寧で簡潔に話してください。"
            "もしユーザがスペイン語を話す場合は、スペイン語を話すエージェントにハンドオフしてください。"
        )
    ),
    model="gpt-5-mini",
    handoffs=[spanish_agent],
    tools=[get_weather],
)

# ワークフローのコールバッククラスの定義
class WorkflowCallbacks(SingleAgentWorkflowCallbacks):
    def on_run(self, workflow: SingleAgentVoiceWorkflow, transcription: str) -> None:
        print(f"[debug] on_run called with transcription: {transcription}")


async def main():
    # パイプラインを定義
    pipeline = VoicePipeline(
        workflow=SingleAgentVoiceWorkflow(agent, callbacks=WorkflowCallbacks())
    )

    # 録音した音声バッファを生成
    audio_input = AudioInput(buffer=record_audio())

    # 音声をパイプラインに渡して実行
    result = await pipeline.run(audio_input)

    with AudioPlayer() as player:
        async for event in result.stream():
            if event.type == "voice_stream_event_audio":
                player.add_audio(event.data)
                print("オーディオを受信しました")
            elif event.type == "voice_stream_event_lifecycle":
                print(f"ライフサイクルイベントを受信しました: {event.event}")

        # 音声ストリームの最後に1秒間の静音を追加して、音声の最後が途切れないようにする
        player.add_audio(np.zeros(24000 * 1, dtype=np.int16))


if __name__ == "__main__":
    asyncio.run(main())

uv run sample2.py

[debug] on_run called with transcription: おはようございます。
ライフサイクルイベントを受信しました: turn_started
オーディオを受信しました
オーディオを受信しました
オーディオを受信しました
ライフサイクルイベントを受信しました: turn_ended
ライフサイクルイベントを受信しました: session_ended

[debug] on_run called with transcription: 東京の天気を教えて。
[debug] get_weather called with city: 東京
ライフサイクルイベントを受信しました: turn_started
オーディオを受信しました
オーディオを受信しました
オーディオを受信しました
ライフサイクルイベントを受信しました: turn_ended
ライフサイクルイベントを受信しました: session_ended

[debug] on_run called with transcription: Hola, ¿cómo estás?
ライフサイクルイベントを受信しました: turn_started
オーディオを受信しました
オーディオを受信しました
オーディオを受信しました
オーディオを受信しました
ライフサイクルイベントを受信しました: turn_ended
ライフサイクルイベントを受信しました: session_ended

(snip)
     # 音声をパイプラインに渡して実行
     result = await pipeline.run(audio_input)

     with AudioPlayer() as player:
         async for event in result.stream():
             if event.type == "voice_stream_event_audio":
                 player.add_audio(event.data)
                 print("オーディオを受信しました")
             elif event.type == "voice_stream_event_lifecycle":
                 print(f"ライフサイクルイベントを受信しました: {event.event}")

(snip)