リアルタイムマルチモーダル API

セマンティック カーネルの最初のリアルタイム API 統合が追加されました。現在、Python でのみ使用でき、試験段階と見なされています。 これは、基になるサービスがまだ開発中であり、変更の影響を受ける可能性があるためです。セマンティック カーネルの API に重大な変更を加える必要がある場合があります。これは、これを使用する方法や、これらの種類のモデルと API の他のプロバイダーを追加する際に、セマンティック カーネルから学習するためです。

リアルタイム クライアントの抽象化

異なるプロトコルを使用して、異なるベンダーからの異なるリアルタイム API をサポートするために、新しいクライアント抽象化がカーネルに追加されました。 このクライアントは、リアルタイム サービスに接続し、メッセージを送受信するために使用されます。 クライアントは、サービスへの接続の処理、メッセージの送信、メッセージの受信を担当します。 また、クライアントは、接続またはメッセージの送受信プロセス中に発生したエラーを処理する役割も担います。 これらのモデルの動作方法を考慮すると、通常のチャットの完了よりもエージェントと見なすことができます。したがって、システム メッセージではなく指示を受け取り、独自の内部状態を維持し、Microsoft に代わって作業を行うために呼び出すことができます。

リアルタイム API

リアルタイム クライアントは、次のメソッドを実装します。

Python の実装

セマンティック カーネルの Python バージョンでは、現在、次のリアルタイム クライアントがサポートされています。

はじめに

Realtime API の使用を開始するには、semantic-kernel パッケージを realtime の追加機能とともにインストールする必要があります。

pip install semantic-kernel[realtime]

オーディオの処理方法によっては、pyaudio や sounddeviceなど、スピーカーやマイクとのインターフェイスに追加のパッケージが必要になる場合があります。

Websocket クライアント

その後、カーネルを作成し、それにリアルタイム クライアントを追加できます。これは、AzureRealtimeWebsocket 接続でそれを行う方法を示しています。AzureRealtimeWebsocket を OpenAIRealtimeWebsocket に置き換えることができます。さらに変更を加える必要はありません。

from semantic_kernel.connectors.ai.open_ai import (
    AzureRealtimeWebsocket,
    AzureRealtimeExecutionSettings,
    ListenEvents,
)
from semantic_kernel.contents import RealtimeAudioEvent, RealtimeTextEvent

# this will use environment variables to get the api key, endpoint, api version and deployment name.
realtime_client = AzureRealtimeWebsocket()
settings = AzureRealtimeExecutionSettings(voice='alloy')
async with realtime_client(settings=settings, create_response=True):
    async for event in realtime_client.receive():
        match event:
            # receiving a piece of audio (and send it to a undefined audio player)
            case RealtimeAudioEvent():
                await audio_player.add_audio(event.audio)
            # receiving a piece of audio transcript
            case RealtimeTextEvent():
                # Semantic Kernel parses the transcript to a TextContent object captured in a RealtimeTextEvent
                print(event.text.text, end="")
            case _:
                # OpenAI Specific events
                if event.service_type == ListenEvents.SESSION_UPDATED:
                    print("Session updated")
                if event.service_type == ListenEvents.RESPONSE_CREATED:
                    print("\nMosscap (transcript): ", end="")

注意すべき重要な点が 2 つあります。1 つ目は、realtime_client が非同期コンテキスト マネージャーであるという点です。つまり、非同期関数で使用し、async with を使用してセッションを作成できます。 2 つ目は、receive メソッドが非同期ジェネレーターであるということです。つまり、for ループで使用して、メッセージの受信時にメッセージを受信できます。

WebRTC クライアント

WebRTC 接続のセットアップは少し複雑なので、クライアントを作成するときに追加のパラメーターが必要です。 このパラメーター audio_track、MediaStreamTrack パッケージの aiortc プロトコルを実装するオブジェクトである必要があります。これは、以下にリンクされているサンプルでも示されています。

WebRTC を使用するクライアントを作成するには、次の操作を行います。

from semantic_kernel.connectors.ai.open_ai import (
    ListenEvents,
    OpenAIRealtimeExecutionSettings,
    OpenAIRealtimeWebRTC,
)
from aiortc.mediastreams import MediaStreamTrack

class AudioRecorderWebRTC(MediaStreamTrack):
    # implement the MediaStreamTrack methods.

realtime_client = OpenAIRealtimeWebRTC(audio_track=AudioRecorderWebRTC())
# Create the settings for the session
settings = OpenAIRealtimeExecutionSettings(
    instructions="""
You are a chat bot. Your name is Mosscap and
you have one goal: figure out what people need.
Your full name, should you need to know it, is
Splendid Speckled Mosscap. You communicate
effectively, but you tend to answer with long
flowery prose.
""",
    voice="shimmer",
)
audio_player = AudioPlayer
async with realtime_client(settings=settings, create_response=True):
    async for event in realtime_client.receive():
        match event.event_type:
            # receiving a piece of audio (and send it to a undefined audio player)
            case "audio":
                await audio_player.add_audio(event.audio)
            case "text":
                # the model returns both audio and transcript of the audio, which we will print
                print(event.text.text, end="")
            case "service":
                # OpenAI Specific events
                if event.service_type == ListenEvents.SESSION_UPDATED:
                    print("Session updated")
                if event.service_type == ListenEvents.RESPONSE_CREATED:
                    print("\nMosscap (transcript): ", end="")

どちらのサンプルも、オーディオを RealtimeAudioEvent として受け取り、指定されていないaudio_player オブジェクトに渡します。

オーディオ出力コールバック

この横には、audio_output_callback メソッドとクラスの作成に receive というパラメーターがあります。 このコールバックは、オーディオをさらに処理する前に最初に呼び出され、オーディオ データが AudioContent に解析されて RealtimeAudioEvent として返されるのではなく、オーディオ データの numpy 配列を取得します。これは上記で行われます。 これは、入ってくるオーディオ データとプレーヤーに渡されるオーディオ データの間のオーバーヘッドが少ないため、よりスムーズなオーディオ出力を提供することが示されています。

この例では、audio_output_callbackを定義して使用する方法を示します。

from semantic_kernel.connectors.ai.open_ai import (
    ListenEvents,
    OpenAIRealtimeExecutionSettings,
    OpenAIRealtimeWebRTC,
)
from aiortc.mediastreams import MediaStreamTrack

class AudioRecorderWebRTC(MediaStreamTrack):
    # implement the MediaStreamTrack methods.

class AudioPlayer:
    async def play_audio(self, content: np.ndarray):
        # implement the audio player

realtime_client = OpenAIRealtimeWebRTC(audio_track=AudioRecorderWebRTC())
# Create the settings for the session
settings = OpenAIRealtimeExecutionSettings(
    instructions="""
You are a chat bot. Your name is Mosscap and
you have one goal: figure out what people need.
Your full name, should you need to know it, is
Splendid Speckled Mosscap. You communicate
effectively, but you tend to answer with long
flowery prose.
""",
    voice="shimmer",
)
audio_player = AudioPlayer
async with realtime_client(settings=settings, create_response=True):
    async for event in realtime_client.receive(audio_output_callback=audio_player.play_audio):
        match event.event_type:
            # no need to handle case: "audio"
            case "text":
                # the model returns both audio and transcript of the audio, which we will print
                print(event.text.text, end="")
            case "service":
                # OpenAI Specific events
                if event.service_type == ListenEvents.SESSION_UPDATED:
                    print("Session updated")
                if event.service_type == ListenEvents.RESPONSE_CREATED:
                    print("\nMosscap (transcript): ", end="")

サンプル

リポジトリ には 4 つのサンプルがあり、websocket と WebRTC の両方を使用した基本と、関数呼び出しを含むより複雑なセットアップの両方について説明します。 最後に、Azure Communication Services を使用してセマンティック カーネル拡張リアルタイム API を呼び出せるようにする、より 複雑なデモ があります。