실시간 다중 모드 API

의미 체계 커널에 대한 첫 번째 실시간 API 통합이 추가되었으며, 현재 Python에서만 사용할 수 있으며 실험적인 것으로 간주됩니다. 이는 기본 서비스가 아직 개발 중이고 변경될 수 있기 때문입니다. 고객으로부터 사용 방법을 배우고 이러한 종류의 모델 및 API에 다른 공급자를 추가함에 따라, 의미 체계 커널에서 API의 호환성이 손상될 수 있는 변경 사항을 해야 할 필요가 있을 수 있습니다.

실시간 클라이언트 추상화

서로 다른 프로토콜을 사용하여 다른 공급업체의 다른 실시간 API를 지원하기 위해 새 클라이언트 추상화가 커널에 추가되었습니다. 이 클라이언트는 실시간 서비스에 연결하고 메시지를 보내고 받는 데 사용됩니다. 클라이언트는 서비스에 대한 연결을 처리하고, 메시지를 보내고, 메시지를 수신할 책임이 있습니다. 또한 클라이언트는 연결 또는 메시지 송신/수신 프로세스 중에 발생하는 모든 오류를 처리해야 합니다. 이러한 모델의 작동 방식을 고려할 때 일반 채팅 완료보다 에이전트로 간주될 수 있으므로 시스템 메시지 대신 지침을 받아 자체 내부 상태를 유지하고 대신 작업을 수행하도록 호출할 수 있습니다.

실시간 API

모든 실시간 클라이언트는 다음 메서드를 구현합니다.

Python 구현

의미 체계 커널의 Python 버전은 현재 다음과 같은 실시간 클라이언트를 지원합니다.

시작

실시간 API를 사용하려면 패키지 semantic-kernel를 설치해야 합니다.

pip install semantic-kernel

오디오를 처리하는 방법에 따라 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="")

주의해야 할 두 가지 중요한 사항이 있습니다. 첫 번째는 realtime_client 비동기 컨텍스트 관리자라는 것입니다. 즉, 비동기 함수에서 사용하고 async with 사용하여 세션을 만들 수 있습니다. 두 번째는 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 개체에 전달합니다.

참고: Azure OpenAI WebRTC 클라이언트에는 추가 매개 변수 region가 있습니다. 이는 OpenAI 서비스가 호스트되는 Azure 지역을 나타냅니다. 현재는 eastus2와 swedencentral만 지원됩니다.

오디오 출력 콜백

이 옆에는 audio_output_callback 메서드 및 클래스 생성에 receive 매개 변수가 있습니다. 이 콜백은 오디오를 더 이상 처리하기 전에 먼저 호출되고 오디오 데이터의 numpy 배열을 가져오는 대신 AudioContent로 구문 분석되어 처리할 수 있는 RealtimeAudioEvent로 반환됩니다. 이는 위에서 일어나는 일입니다. 이는 들어오는 오디오 데이터와 플레이어에게 제공되는 오디오 데이터 사이에 오버헤드가 적기 때문에 더 부드러운 오디오 출력을 제공하는 것으로 나타났습니다.

이 예제에서는 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="")

샘플

리포지토리 네 가지 샘플이 있으며, websocket과 WebRTC를 모두 사용하는 기본 사항과 함수 호출을 포함한 보다 복잡한 설정을 모두 다룹니다. 마지막으로 Azure Communication Services 사용하여 의미 체계 커널 향상된 실시간 API를 호출할 수 있도록 하는 더 복잡한 데모 있습니다.

곧 출시 예정

자세한 내용은 곧 제공될 예정입니다.

곧 출시 예정

자세한 내용은 곧 제공될 예정입니다.