wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #421 from pipecat-ai/aleix/improve-multi-lingual-support

Browse Source 
improve multi lingual support
This commit is contained in:
Aleix Conchillo Flaqué
2024-08-29 13:19:40 -07:00
committed by GitHub
parent 84d72c0d5c c0ac5c6ae8
commit 05f0492a8d
12 changed files with 316 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
CHANGELOG.md
View File

@@ -5,6 +5,33 @@ All notable changes to **pipecat** will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [Unreleased]
### Added
- Added new `LmntTTSService` text-to-speech service.
(see https://www.lmnt.com/)
- Added `TTSModelUpdateFrame`, `TTSLanguageUpdateFrame`, `STTModelUpdateFrame`,
and `STTLanguageUpdateFrame` frames to allow you to switch models, language
and voices in TTS and STT services.
- Added new `transcriptions.Language` enum.
### Changed
- `DailyTransport.on_joined` event now returns the full session data instead of
just the participant.
- `CartesiaTTSService` is now a subclass of `TTSService`.
- `DeepgramSTTService` is now a subclass of `STTService`.
- `WhisperSTTService` is now a subclass of `SegmentedSTTService`. A
`SegmentedSTTService` is a `STTService` where the provided audio is given in a
big chunk (i.e. from when the user starts speaking until the user stops
speaking) instead of a continous stream.
## [0.0.41] - 2024-08-22
### Added

2
README.md
View File

@@ -38,7 +38,7 @@ pip install "pipecat-ai[option,...]"
Your project may or may not need these, so they're made available as optional requirements. Here is a list:
- **AI services**: `anthropic`, `azure`, `deepgram`, `gladia`, `google`, `fal`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper`, `xtts`
- **AI services**: `anthropic`, `azure`, `deepgram`, `gladia`, `google`, `fal`, `lmnt`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper`, `xtts`
- **Transports**: `local`, `websocket`, `daily`
## Code examples

2
examples/foundational/07k-interruptible-lmnt.py
View File

@@ -50,7 +50,7 @@ async def main():
tts = LmntTTSService(
api_key=os.getenv("LMNT_API_KEY"),
voice="morgan"
voice_id="morgan"
)
llm = OpenAILLMService(

39
src/pipecat/frames/frames.py
View File

@@ -8,6 +8,7 @@ from typing import Any, List, Mapping, Optional, Tuple
from dataclasses import dataclass, field
from pipecat.transcriptions.language import Language
from pipecat.utils.utils import obj_count, obj_id
from pipecat.vad.vad_analyzer import VADParams
@@ -131,9 +132,10 @@ class TranscriptionFrame(TextFrame):
"""
user_id: str
timestamp: str
language: Language | None = None
def __str__(self):
return f"{self.name}(user: {self.user_id}, text: {self.text}, timestamp: {self.timestamp})"
return f"{self.name}(user: {self.user_id}, text: {self.text}, language: {self.language}, timestamp: {self.timestamp})"
@dataclass
@@ -142,9 +144,10 @@ class InterimTranscriptionFrame(TextFrame):
the transport's receive queue when a participant speaks."""
user_id: str
timestamp: str
language: Language | None = None
def __str__(self):
return f"{self.name}(user: {self.user_id}, text: {self.text}, timestamp: {self.timestamp})"
return f"{self.name}(user: {self.user_id}, text: {self.text}, language: {self.language}, timestamp: {self.timestamp})"
@dataclass
@@ -433,6 +436,13 @@ class LLMModelUpdateFrame(ControlFrame):
model: str
@dataclass
class TTSModelUpdateFrame(ControlFrame):
"""A control frame containing a request to update the TTS model.
"""
model: str
@dataclass
class TTSVoiceUpdateFrame(ControlFrame):
"""A control frame containing a request to update to a new TTS voice.
@@ -440,6 +450,31 @@ class TTSVoiceUpdateFrame(ControlFrame):
voice: str
@dataclass
class TTSLanguageUpdateFrame(ControlFrame):
"""A control frame containing a request to update to a new TTS language and
optional voice.
"""
language: Language
@dataclass
class STTModelUpdateFrame(ControlFrame):
"""A control frame containing a request to update the STT model and optional
language.
"""
model: str
@dataclass
class STTLanguageUpdateFrame(ControlFrame):
"""A control frame containing a request to update to STT language.
"""
language: Language
@dataclass
class FunctionCallInProgressFrame(SystemFrame):
"""A frame signaling that a function call is in progress.

34
src/pipecat/processors/frameworks/rtvi.py
View File

@@ -81,11 +81,6 @@ class RTVIAction(BaseModel):
return super().model_post_init(__context)
#
# Client -> Pipecat messages.
#
class RTVIServiceOptionConfig(BaseModel):
name: str
value: Any
@@ -100,6 +95,16 @@ class RTVIConfig(BaseModel):
config: List[RTVIServiceConfig]
#
# Client -> Pipecat messages.
#
class RTVIUpdateConfig(BaseModel):
config: List[RTVIServiceConfig]
interrupt: bool = False
class RTVIActionRunArgument(BaseModel):
name: str
value: Any
@@ -489,8 +494,8 @@ class RTVIProcessor(FrameProcessor):
case "get-config":
await self._handle_get_config(message.id)
case "update-config":
config = RTVIConfig.model_validate(message.data)
await self._handle_update_config(message.id, config)
update_config = RTVIUpdateConfig.model_validate(message.data)
await self._handle_update_config(message.id, update_config)
case "action":
action = RTVIActionRun.model_validate(message.data)
await self._handle_action(message.id, action)
@@ -545,17 +550,14 @@ class RTVIProcessor(FrameProcessor):
await handler(self, service.name, option)
self._update_config_option(service.name, option)
async def _update_config(self, data: RTVIConfig):
async def _update_config(self, data: RTVIConfig, interrupt: bool):
if interrupt:
await self.interrupt_bot()
for service_config in data.config:
await self._update_service_config(service_config)
async def _handle_update_config(self, request_id: str, data: RTVIConfig):
# NOTE(aleix): The bot might be talking while we receive a new
# config. Let's interrupt it for now and update the config. Another
# solution is to wait until the bot stops speaking and then apply the
# config, but this definitely is more complicated to achieve.
await self.interrupt_bot()
await self._update_config(data)
async def _handle_update_config(self, request_id: str, data: RTVIUpdateConfig):
await self._update_config(RTVIConfig(config=data.config), data.interrupt)
await self._handle_get_config(request_id)
async def _handle_function_call_result(self, data):
@@ -583,7 +585,7 @@ class RTVIProcessor(FrameProcessor):
async def _maybe_send_bot_ready(self):
if self._pipeline_started and self._client_ready:
await self._send_bot_ready()
await self._update_config(self._config)
await self._update_config(self._config, False)
async def _send_bot_ready(self):
if not self._params.send_bot_ready:

98
src/pipecat/services/ai_services.py
View File

@@ -18,8 +18,12 @@ from pipecat.frames.frames import (
ErrorFrame,
Frame,
LLMFullResponseEndFrame,
STTLanguageUpdateFrame,
STTModelUpdateFrame,
StartFrame,
StartInterruptionFrame,
TTSLanguageUpdateFrame,
TTSModelUpdateFrame,
TTSSpeakFrame,
TTSStartedFrame,
TTSStoppedFrame,
@@ -30,6 +34,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.async_frame_processor import AsyncFrameProcessor
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.transcriptions.language import Language
from pipecat.utils.audio import calculate_audio_volume
from pipecat.utils.string import match_endofsentence
from pipecat.utils.utils import exp_smoothing
@@ -173,10 +178,18 @@ class TTSService(AIService):
self._stop_frame_queue: asyncio.Queue = asyncio.Queue()
self._current_sentence: str = ""
@abstractmethod
async def set_model(self, model: str):
pass
@abstractmethod
async def set_voice(self, voice: str):
pass
@abstractmethod
async def set_language(self, language: Language):
pass
# Converts the text to audio.
@abstractmethod
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
@@ -233,8 +246,12 @@ class TTSService(AIService):
await self.push_frame(frame, direction)
elif isinstance(frame, TTSSpeakFrame):
await self._push_tts_frames(frame.text, False)
elif isinstance(frame, TTSModelUpdateFrame):
await self.set_model(frame.model)
elif isinstance(frame, TTSVoiceUpdateFrame):
await self.set_voice(frame.voice)
elif isinstance(frame, TTSLanguageUpdateFrame):
await self.set_language(frame.language)
else:
await self.push_frame(frame, direction)
@@ -289,6 +306,47 @@ class TTSService(AIService):
class STTService(AIService):
"""STTService is a base class for speech-to-text services."""
def __init__(self, **kwargs):
super().__init__(**kwargs)
@abstractmethod
async def set_model(self, model: str):
pass
@abstractmethod
async def set_language(self, language: Language):
pass
@abstractmethod
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Returns transcript as a string"""
pass
async def process_audio_frame(self, frame: AudioRawFrame):
await self.process_generator(self.run_stt(frame.audio))
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Processes a frame of audio data, either buffering or transcribing it."""
await super().process_frame(frame, direction)
if isinstance(frame, AudioRawFrame):
# In this service we accumulate audio internally and at the end we
# push a TextFrame. We don't really want to push audio frames down.
await self.process_audio_frame(frame)
elif isinstance(frame, STTModelUpdateFrame):
await self.set_model(frame.model)
elif isinstance(frame, STTLanguageUpdateFrame):
await self.set_language(frame.language)
else:
await self.push_frame(frame, direction)
class SegmentedSTTService(STTService):
"""SegmentedSTTService is an STTService that will detect speech and will run
speech-to-text on speech segments only, instead of a continous stream.
"""
def __init__(self,
*,
min_volume: float = 0.6,
@@ -309,24 +367,7 @@ class STTService(AIService):
self._smoothing_factor = 0.2
self._prev_volume = 0
@abstractmethod
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Returns transcript as a string"""
pass
def _new_wave(self):
content = io.BytesIO()
ww = wave.open(content, "wb")
ww.setsampwidth(2)
ww.setnchannels(self._num_channels)
ww.setframerate(self._sample_rate)
return (content, ww)
def _get_smoothed_volume(self, frame: AudioRawFrame) -> float:
volume = calculate_audio_volume(frame.audio, frame.sample_rate)
return exp_smoothing(volume, self._prev_volume, self._smoothing_factor)
async def _append_audio(self, frame: AudioRawFrame):
async def process_audio_frame(self, frame: AudioRawFrame):
# Try to filter out empty background noise
volume = self._get_smoothed_volume(frame)
if volume >= self._min_volume:
@@ -346,9 +387,7 @@ class STTService(AIService):
self._silence_num_frames = 0
self._wave.close()
self._content.seek(0)
await self.start_processing_metrics()
await self.process_generator(self.run_stt(self._content.read()))
await self.stop_processing_metrics()
(self._content, self._wave) = self._new_wave()
async def stop(self, frame: EndFrame):
@@ -357,16 +396,17 @@ class STTService(AIService):
async def cancel(self, frame: CancelFrame):
self._wave.close()
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Processes a frame of audio data, either buffering or transcribing it."""
await super().process_frame(frame, direction)
def _new_wave(self):
content = io.BytesIO()
ww = wave.open(content, "wb")
ww.setsampwidth(2)
ww.setnchannels(self._num_channels)
ww.setframerate(self._sample_rate)
return (content, ww)
if isinstance(frame, AudioRawFrame):
# In this service we accumulate audio internally and at the end we
# push a TextFrame. We don't really want to push audio frames down.
await self._append_audio(frame)
else:
await self.push_frame(frame, direction)
def _get_smoothed_volume(self, frame: AudioRawFrame) -> float:
volume = calculate_audio_volume(frame.audio, frame.sample_rate)
return exp_smoothing(volume, self._prev_volume, self._smoothing_factor)
class ImageGenService(AIService):

32
src/pipecat/services/cartesia.py
View File

@@ -10,9 +10,8 @@ import base64
import asyncio
import time
from typing import AsyncGenerator
from typing import AsyncGenerator, Mapping
from pipecat.processors.frame_processor import FrameDirection
from pipecat.frames.frames import (
CancelFrame,
ErrorFrame,
@@ -26,6 +25,8 @@ from pipecat.frames.frames import (
TextFrame,
LLMFullResponseEndFrame
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.transcriptions.language import Language
from pipecat.services.ai_services import TTSService
from loguru import logger
@@ -40,6 +41,25 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_cartesia_language(language: Language) -> str | None:
match language:
case Language.DE:
return "de"
case Language.EN:
return "en"
case Language.ES:
return "es"
case Language.FR:
return "fr"
case Language.JA:
return "ja"
case Language.PT:
return "pt"
case Language.ZH:
return "zh"
return None
class CartesiaTTSService(TTSService):
def __init__(
@@ -90,10 +110,18 @@ class CartesiaTTSService(TTSService):
def can_generate_metrics(self) -> bool:
return True
async def set_model(self, model: str):
logger.debug(f"Switching TTS model to: [{model}]")
self._model_id = model
async def set_voice(self, voice: str):
logger.debug(f"Switching TTS voice to: [{voice}]")
self._voice_id = voice
async def set_language(self, language: Language):
logger.debug(f"Switching TTS language to: [{language}]")
self._language = language_to_cartesia_language(language)
async def start(self, frame: StartFrame):
await super().start(frame)
await self._connect()

71
src/pipecat/services/deepgram.py
View File

@@ -16,12 +16,11 @@ from pipecat.frames.frames import (
Frame,
InterimTranscriptionFrame,
StartFrame,
SystemFrame,
TTSStartedFrame,
TTSStoppedFrame,
TranscriptionFrame)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import AsyncAIService, TTSService
from pipecat.services.ai_services import STTService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from loguru import logger
@@ -30,10 +29,12 @@ from loguru import logger
# See .env.example for Deepgram configuration needed
try:
from deepgram import (
AsyncListenWebSocketClient,
DeepgramClient,
DeepgramClientOptions,
LiveTranscriptionEvents,
LiveOptions,
LiveResultResponse
)
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
@@ -107,7 +108,7 @@ class DeepgramTTSService(TTSService):
logger.exception(f"{self} exception: {e}")
class DeepgramSTTService(AsyncAIService):
class DeepgramSTTService(STTService):
def __init__(self,
*,
api_key: str,
@@ -120,6 +121,8 @@ class DeepgramSTTService(AsyncAIService):
channels=1,
interim_results=True,
smart_format=True,
punctuate=True,
profanity_filter=True,
),
**kwargs):
super().__init__(**kwargs)
@@ -128,40 +131,62 @@ class DeepgramSTTService(AsyncAIService):
self._client = DeepgramClient(
api_key, config=DeepgramClientOptions(url=url, options={"keepalive": "true"}))
self._connection = self._client.listen.asynclive.v("1")
self._connection: AsyncListenWebSocketClient = self._client.listen.asyncwebsocket.v("1")
self._connection.on(LiveTranscriptionEvents.Transcript, self._on_message)
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
async def set_model(self, model: str):
logger.debug(f"Switching STT model to: [{model}]")
self._live_options.model = model
await self._disconnect()
await self._connect()
if isinstance(frame, SystemFrame):
await self.push_frame(frame, direction)
elif isinstance(frame, AudioRawFrame):
await self._connection.send(frame.audio)
else:
await self.queue_frame(frame, direction)
async def set_language(self, language: Language):
logger.debug(f"Switching STT language to: [{language}]")
self._live_options.language = language
await self._disconnect()
await self._connect()
async def start(self, frame: StartFrame):
await super().start(frame)
await self._connect()
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._disconnect()
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
await self.start_processing_metrics()
await self._connection.send(audio)
yield None
await self.stop_processing_metrics()
async def _connect(self):
if await self._connection.start(self._live_options):
logger.debug(f"{self}: Connected to Deepgram")
else:
logger.error(f"{self}: Unable to connect to Deepgram")
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._connection.finish()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._connection.finish()
async def _disconnect(self):
if self._connection.is_connected:
await self._connection.finish()
logger.debug(f"{self}: Disconnected from Deepgram")
async def _on_message(self, *args, **kwargs):
result = kwargs["result"]
result: LiveResultResponse = kwargs["result"]
if len(result.channel.alternatives) == 0:
return
is_final = result.is_final
transcript = result.channel.alternatives[0].transcript
language = None
if result.channel.alternatives[0].languages:
language = result.channel.alternatives[0].languages[0]
language = Language(language)
if len(transcript) > 0:
if is_final:
await self.queue_frame(TranscriptionFrame(transcript, "", time_now_iso8601()))
await self.push_frame(TranscriptionFrame(transcript, "", time_now_iso8601(), language))
else:
await self.queue_frame(InterimTranscriptionFrame(transcript, "", time_now_iso8601()))
await self.push_frame(InterimTranscriptionFrame(transcript, "", time_now_iso8601(), language))

11
src/pipecat/services/lmnt.py
View File

@@ -4,23 +4,19 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import json
import uuid
import base64
import asyncio
import time
from typing import AsyncGenerator
from pipecat.processors.frame_processor import FrameDirection
from pipecat.frames.frames import (
AudioRawFrame,
CancelFrame,
EndFrame,
ErrorFrame,
Frame,
AudioRawFrame,
StartFrame,
StartInterruptionFrame,
EndFrame,
TTSStartedFrame,
TTSStoppedFrame,
)
@@ -95,7 +91,8 @@ class LmntTTSService(TTSService):
async def _connect(self):
try:
self._speech = Speech()
self._connection = await self._speech.synthesize_streaming(self._voice_id, format="raw", sample_rate=self._output_format["sample_rate"])
self._connection = await self._speech.synthesize_streaming(
self._voice_id, format="raw", sample_rate=self._output_format["sample_rate"])
self._receive_task = self.get_event_loop().create_task(self._receive_task_handler())
except Exception as e:
logger.exception(f"{self} initialization error: {e}")

9
src/pipecat/services/whisper.py
View File

@@ -14,7 +14,7 @@ from typing import AsyncGenerator
import numpy as np
from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
from pipecat.services.ai_services import STTService
from pipecat.services.ai_services import SegmentedSTTService
from pipecat.utils.time import time_now_iso8601
from loguru import logger
@@ -38,7 +38,7 @@ class Model(Enum):
DISTIL_MEDIUM_EN = "Systran/faster-distil-whisper-medium.en"
class WhisperSTTService(STTService):
class WhisperSTTService(SegmentedSTTService):
"""Class to transcribe audio with a locally-downloaded Whisper model"""
def __init__(self,
@@ -77,6 +77,7 @@ class WhisperSTTService(STTService):
yield ErrorFrame("Whisper model not available")
return
await self.start_processing_metrics()
await self.start_ttfb_metrics()
# Divide by 32768 because we have signed 16-bit data.
@@ -88,7 +89,9 @@ class WhisperSTTService(STTService):
if segment.no_speech_prob < self._no_speech_prob:
text += f"{segment.text} "
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
if text:
await self.stop_ttfb_metrics()
logger.debug(f"Transcription: [{text}]")
yield TranscriptionFrame(text, "", time_now_iso8601())

64
src/pipecat/transcriptions/language.py Normal file
View File

@@ -0,0 +1,64 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import sys
from enum import Enum
if sys.version_info < (3, 11):
class StrEnum(str, Enum):
def __new__(cls, value):
obj = str.__new__(cls, value)
obj._value_ = value
return obj
else:
from enum import StrEnum
class Language(StrEnum):
BG = "bg" # Bulgarian
CA = "ca" # Catalan
ZH = "zh" # Chinese simplified
ZH_TW = "zh-TW" # Chinese traditional
CS = "cs" # Czech
DA = "da" # Danish
NL = "nl" # Dutch
EN = "en" # English
EN_US = "en-US" # English (USA)
EN_AU = "en-AU" # English (Australia)
EN_GB = "en-GB" # English (Great Britain)
EN_NZ = "en-NZ" # English (New Zealand)
EN_IN = "en-IN" # English (India)
ET = "et" # Estonian
FI = "fi" # Finnish
NL_BE = "nl-BE" # Flemmish
FR = "fr" # French
FR_CA = "fr-CA" # French (Canada)
DE = "de" # German
DE_CH = "de-CH" # German (Switzerland)
EL = "el" # Greek
HI = "hi" # Hindi
HU = "hu" # Hungarian
ID = "id" # Indonesian
IT = "it" # Italian
JA = "ja" # Japanese
KO = "ko" # Korean
LV = "lv" # Latvian
LT = "lt" # Lithuanian
MS = "ms" # Malay
NO = "no" # Norwegian
PL = "pl" # Polish
PT = "pt" # Portuguese
PT_BR = "pt-BR" # Portuguese (Brazil)
RO = "ro" # Romanian
RU = "ru" # Russian
SK = "sk" # Slovak
ES = "es" # Spanish
SV = "sv" # Swedish
TH = "th" # Thai
TR = "tr" # Turkish
UK = "uk" # Ukrainian
VI = "vi" # Vietnamese

16
src/pipecat/transports/services/daily.py
View File

@@ -36,6 +36,7 @@ from pipecat.frames.frames import (
UserImageRawFrame,
UserImageRequestFrame)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.transcriptions.language import Language
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -305,7 +306,7 @@ class DailyTransportClient(EventHandler):
if self._token and self._params.transcription_enabled:
await self._start_transcription()
await self._callbacks.on_joined(data["participants"]["local"])
await self._callbacks.on_joined(data)
else:
error_msg = f"Error joining {self._room_url}: {error}"
logger.error(error_msg)
@@ -864,8 +865,8 @@ class DailyTransport(BaseTransport):
self._input.capture_participant_video(
participant_id, framerate, video_source, color_format)
async def _on_joined(self, participant):
await self._call_event_handler("on_joined", participant)
async def _on_joined(self, data):
await self._call_event_handler("on_joined", data)
async def _on_left(self):
await self._call_event_handler("on_left")
@@ -950,11 +951,16 @@ class DailyTransport(BaseTransport):
text = message["text"]
timestamp = message["timestamp"]
is_final = message["rawResponse"]["is_final"]
try:
language = message["rawResponse"]["channel"]["alternatives"][0]["languages"][0]
language = Language(language)
except KeyError:
language = None
if is_final:
frame = TranscriptionFrame(text, participant_id, timestamp)
frame = TranscriptionFrame(text, participant_id, timestamp, language)
logger.debug(f"Transcription (from: {participant_id}): [{text}]")
else:
frame = InterimTranscriptionFrame(text, participant_id, timestamp)
frame = InterimTranscriptionFrame(text, participant_id, timestamp, language)
if self._input:
await self._input.push_transcription_frame(frame)