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

Merge pull request #2974 from pipecat-ai/mb/language-mapping-improvements

Browse Source 
Improve language checking in STT and TTS services
This commit is contained in:
Mark Backman
2025-11-05 14:59:41 -05:00
committed by GitHub
parent 3db5ab9f23 f6d7b6ae5f
commit 11b101e8a6
26 changed files with 144 additions and 295 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGELOG.md
View File

@@ -53,6 +53,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Updated the `GoogleVertexLLMService` to use the `GoogleLLMService` as a base
class instead of the `OpenAILLMService`.
- Updated STT and TTS services to pass through unverified language codes with a
warning instead of returning None. This allows developers to use newly
supported languages before Pipecat's service classes are updated, while still
providing guidance on verified languages.
### Fixed
- Fixed an issue that would cause wrong user/assistant context ordering when

16
src/pipecat/services/asyncai/tts.py
View File

@@ -28,7 +28,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import InterruptibleTTSService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_tts
try:
@@ -50,7 +50,7 @@ def language_to_async_language(language: Language) -> Optional[str]:
Returns:
The corresponding Async language code, or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.EN: "en",
Language.FR: "fr",
Language.ES: "es",
@@ -58,17 +58,7 @@ def language_to_async_language(language: Language) -> Optional[str]:
Language.IT: "it",
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
# Convert enum value to string and get the base language part (e.g. en-En -> en)
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
# Look up the base code in our supported languages
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
class AsyncAITTSService(InterruptibleTTSService):

6
src/pipecat/services/aws/stt.py
View File

@@ -30,7 +30,7 @@ from pipecat.frames.frames import (
)
from pipecat.services.aws.utils import build_event_message, decode_event, get_presigned_url
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
@@ -326,7 +326,7 @@ class AWSTranscribeSTTService(STTService):
Returns:
AWS Transcribe compatible language code, or None if unsupported.
"""
language_map = {
LANGUAGE_MAP = {
# Afrikaans
Language.AF: "af-ZA",
Language.AF_ZA: "af-ZA",
@@ -466,7 +466,7 @@ class AWSTranscribeSTTService(STTService):
Language.ZU_ZA: "zu-ZA",
}
return language_map.get(language)
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
@traced_stt
async def _handle_transcription(

6
src/pipecat/services/aws/tts.py
View File

@@ -26,7 +26,7 @@ from pipecat.frames.frames import (
TTSStoppedFrame,
)
from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_tts
try:
@@ -47,7 +47,7 @@ def language_to_aws_language(language: Language) -> Optional[str]:
Returns:
The corresponding AWS Polly language code, or None if not supported.
"""
language_map = {
LANGUAGE_MAP = {
# Arabic
Language.AR: "arb",
Language.AR_AE: "ar-AE",
@@ -119,7 +119,7 @@ def language_to_aws_language(language: Language) -> Optional[str]:
Language.CY_GB: "cy-GB",
}
return language_map.get(language)
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
class AWSPollyTTSService(TTSService):

8
src/pipecat/services/azure/common.py
View File

@@ -8,7 +8,9 @@
from typing import Optional
from pipecat.transcriptions.language import Language
from loguru import logger
from pipecat.transcriptions.language import Language, resolve_language
def language_to_azure_language(language: Language) -> Optional[str]:
@@ -20,7 +22,7 @@ def language_to_azure_language(language: Language) -> Optional[str]:
Returns:
The corresponding Azure language code, or None if not supported.
"""
language_map = {
LANGUAGE_MAP = {
# Afrikaans
Language.AF: "af-ZA",
Language.AF_ZA: "af-ZA",
@@ -341,4 +343,4 @@ def language_to_azure_language(language: Language) -> Optional[str]:
Language.ZU: "zu-ZA",
Language.ZU_ZA: "zu-ZA",
}
return language_map.get(language)
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)

16
src/pipecat/services/cartesia/tts.py
View File

@@ -28,7 +28,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import AudioContextWordTTSService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator
from pipecat.utils.tracing.service_decorators import traced_tts
@@ -77,7 +77,7 @@ def language_to_cartesia_language(language: Language) -> Optional[str]:
Returns:
The corresponding Cartesia language code, or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.AR: "ar",
Language.BG: "bg",
Language.BN: "bn",
@@ -122,17 +122,7 @@ def language_to_cartesia_language(language: Language) -> Optional[str]:
Language.ZH: "zh",
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
# Convert enum value to string and get the base language part (e.g. es-ES -> es)
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
# Look up the base code in our supported languages
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
class CartesiaTTSService(AudioContextWordTTSService):

14
src/pipecat/services/elevenlabs/stt.py
View File

@@ -20,7 +20,7 @@ from pydantic import BaseModel
from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
from pipecat.services.stt_service import SegmentedSTTService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
@@ -37,7 +37,7 @@ def language_to_elevenlabs_language(language: Language) -> Optional[str]:
Returns:
The corresponding ElevenLabs language code, or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.AF: "afr", # Afrikaans
Language.AM: "amh", # Amharic
Language.AR: "ara", # Arabic
@@ -139,15 +139,7 @@ def language_to_elevenlabs_language(language: Language) -> Optional[str]:
Language.ZU: "zul", # Zulu
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
class ElevenLabsSTTService(SegmentedSTTService):

16
src/pipecat/services/elevenlabs/tts.py
View File

@@ -37,7 +37,7 @@ from pipecat.services.tts_service import (
AudioContextWordTTSService,
WordTTSService,
)
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_tts
# See .env.example for ElevenLabs configuration needed
@@ -72,7 +72,7 @@ def language_to_elevenlabs_language(language: Language) -> Optional[str]:
Returns:
The corresponding ElevenLabs language code, or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.AR: "ar",
Language.BG: "bg",
Language.CS: "cs",
@@ -107,17 +107,7 @@ def language_to_elevenlabs_language(language: Language) -> Optional[str]:
Language.ZH: "zh",
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
# Convert enum value to string and get the base language part (e.g. es-ES -> es)
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
# Look up the base code in our supported languages
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
def output_format_from_sample_rate(sample_rate: int) -> str:

14
src/pipecat/services/fal/stt.py
View File

@@ -18,7 +18,7 @@ from pydantic import BaseModel
from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
from pipecat.services.stt_service import SegmentedSTTService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
@@ -41,7 +41,7 @@ def language_to_fal_language(language: Language) -> Optional[str]:
Returns:
The corresponding Fal Wizper language code, or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.AF: "af",
Language.AM: "am",
Language.AR: "ar",
@@ -142,15 +142,7 @@ def language_to_fal_language(language: Language) -> Optional[str]:
Language.ZH: "zh",
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
class FalSTTService(SegmentedSTTService):

16
src/pipecat/services/gladia/stt.py
View File

@@ -31,7 +31,7 @@ from pipecat.frames.frames import (
)
from pipecat.services.gladia.config import GladiaInputParams
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
@@ -54,7 +54,7 @@ def language_to_gladia_language(language: Language) -> Optional[str]:
Returns:
The Gladia language code string or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.AF: "af",
Language.AM: "am",
Language.AR: "ar",
@@ -156,17 +156,7 @@ def language_to_gladia_language(language: Language) -> Optional[str]:
Language.ZH: "zh",
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
# Convert enum value to string and get the base language part (e.g. es-ES -> es)
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
# Look up the base code in our supported languages
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
# Deprecation warning for nested InputParams

7
src/pipecat/services/google/gemini_live/llm.py
View File

@@ -74,7 +74,7 @@ from pipecat.services.openai.llm import (
OpenAIAssistantContextAggregator,
OpenAIUserContextAggregator,
)
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.string import match_endofsentence
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_gemini_live, traced_stt
@@ -133,7 +133,7 @@ def language_to_gemini_language(language: Language) -> Optional[str]:
Returns:
The Gemini language code string, or None if the language is not supported.
"""
language_map = {
LANGUAGE_MAP = {
# Arabic
Language.AR: "ar-XA",
# Bengali
@@ -214,7 +214,8 @@ def language_to_gemini_language(language: Language) -> Optional[str]:
Language.VI: "vi-VN",
Language.VI_VN: "vi-VN",
}
return language_map.get(language)
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
class GeminiLiveContext(OpenAILLMContext):

6
src/pipecat/services/google/stt.py
View File

@@ -36,7 +36,7 @@ from pipecat.frames.frames import (
TranscriptionFrame,
)
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.time import time_now_iso8601
try:
@@ -64,7 +64,7 @@ def language_to_google_stt_language(language: Language) -> Optional[str]:
Returns:
Optional[str]: Google STT language code or None if not supported.
"""
language_map = {
LANGUAGE_MAP = {
# Afrikaans
Language.AF: "af-ZA",
Language.AF_ZA: "af-ZA",
@@ -351,7 +351,7 @@ def language_to_google_stt_language(language: Language) -> Optional[str]:
Language.ZU_ZA: "zu-ZA",
}
return language_map.get(language)
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
class GoogleSTTService(STTService):

6
src/pipecat/services/google/tts.py
View File

@@ -36,7 +36,7 @@ from pipecat.frames.frames import (
TTSStoppedFrame,
)
from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
try:
from google.auth import default
@@ -70,7 +70,7 @@ def language_to_google_tts_language(language: Language) -> Optional[str]:
Returns:
The corresponding Google TTS language code, or None if not supported.
"""
language_map = {
LANGUAGE_MAP = {
# Afrikaans
Language.AF: "af-ZA",
Language.AF_ZA: "af-ZA",
@@ -226,7 +226,7 @@ def language_to_google_tts_language(language: Language) -> Optional[str]:
Language.VI_VN: "vi-VN",
}
return language_map.get(language)
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
class GoogleHttpTTSService(TTSService):

16
src/pipecat/services/lmnt/tts.py
View File

@@ -24,7 +24,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import InterruptibleTTSService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_tts
# See .env.example for LMNT configuration needed
@@ -46,7 +46,7 @@ def language_to_lmnt_language(language: Language) -> Optional[str]:
Returns:
The corresponding LMNT language code, or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.DE: "de",
Language.EN: "en",
Language.ES: "es",
@@ -68,17 +68,7 @@ def language_to_lmnt_language(language: Language) -> Optional[str]:
Language.ZH: "zh",
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
# Convert enum value to string and get the base language part (e.g. es-ES -> es)
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
# Look up the base code in our supported languages
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
class LmntTTSService(InterruptibleTTSService):

19
src/pipecat/services/minimax/tts.py
View File

@@ -26,7 +26,7 @@ from pipecat.frames.frames import (
TTSStoppedFrame,
)
from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_tts
@@ -39,7 +39,7 @@ def language_to_minimax_language(language: Language) -> Optional[str]:
Returns:
The corresponding MiniMax language name, or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.AR: "Arabic",
Language.CS: "Czech",
Language.DE: "German",
@@ -66,20 +66,7 @@ def language_to_minimax_language(language: Language) -> Optional[str]:
Language.ZH: "Chinese",
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
# Convert enum value to string and get the base language part (e.g. es-ES -> es)
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
# Find matching language
for code, name in BASE_LANGUAGES.items():
if str(code.value).lower().startswith(base_code):
result = name
break
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
class MiniMaxHttpTTSService(TTSService):

16
src/pipecat/services/neuphonic/tts.py
View File

@@ -35,7 +35,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import InterruptibleTTSService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_tts
try:
@@ -56,7 +56,7 @@ def language_to_neuphonic_lang_code(language: Language) -> Optional[str]:
Returns:
The corresponding Neuphonic language code, or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.DE: "de",
Language.EN: "en",
Language.ES: "es",
@@ -69,17 +69,7 @@ def language_to_neuphonic_lang_code(language: Language) -> Optional[str]:
Language.ZH: "zh",
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
# Convert enum value to string and get the base language part (e.g. es-ES -> es)
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
# Look up the base code in our supported languages
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
class NeuphonicTTSService(InterruptibleTTSService):

16
src/pipecat/services/playht/tts.py
View File

@@ -34,7 +34,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import InterruptibleTTSService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_tts
try:
@@ -55,7 +55,7 @@ def language_to_playht_language(language: Language) -> Optional[str]:
Returns:
The corresponding PlayHT language code, or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.AF: "afrikans",
Language.AM: "amharic",
Language.AR: "arabic",
@@ -95,17 +95,7 @@ def language_to_playht_language(language: Language) -> Optional[str]:
Language.ZH: "mandarin",
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
# Convert enum value to string and get the base language part (e.g. es-ES -> es)
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
# Look up the base code in our supported languages
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
class PlayHTTTSService(InterruptibleTTSService):

4
src/pipecat/services/rime/tts.py
View File

@@ -32,7 +32,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import AudioContextWordTTSService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator
from pipecat.utils.tracing.service_decorators import traced_tts
@@ -61,7 +61,7 @@ def language_to_rime_language(language: Language) -> str:
Language.EN: "eng",
Language.ES: "spa",
}
return LANGUAGE_MAP.get(language, "eng")
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
class RimeTTSService(AudioContextWordTTSService):

6
src/pipecat/services/riva/stt.py
View File

@@ -23,7 +23,7 @@ from pipecat.frames.frames import (
TranscriptionFrame,
)
from pipecat.services.stt_service import SegmentedSTTService, STTService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
@@ -48,7 +48,7 @@ def language_to_riva_language(language: Language) -> Optional[str]:
Returns:
Optional[str]: Riva language code or None if not supported.
"""
language_map = {
LANGUAGE_MAP = {
# Arabic
Language.AR: "ar-AR",
# English
@@ -85,7 +85,7 @@ def language_to_riva_language(language: Language) -> Optional[str]:
Language.ES_US: "es-US", # US Spanish
}
return language_map.get(language)
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
class RivaSTTService(STTService):

8
src/pipecat/services/sarvam/stt.py
View File

@@ -19,7 +19,7 @@ from pipecat.frames.frames import (
TranscriptionFrame,
)
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
@@ -43,7 +43,7 @@ def language_to_sarvam_language(language: Language) -> str:
The Sarvam language code string.
"""
# Mapping of pipecat Language enum to Sarvam language codes
SARVAM_LANGUAGES = {
LANGUAGE_MAP = {
Language.BN_IN: "bn-IN",
Language.GU_IN: "gu-IN",
Language.HI_IN: "hi-IN",
@@ -58,9 +58,7 @@ def language_to_sarvam_language(language: Language) -> str:
Language.AS_IN: "as-IN",
}
return SARVAM_LANGUAGES.get(
language, "unknown"
) # Default to unknown (Sarvam models auto-detect the language)
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
class SarvamSTTService(STTService):

4
src/pipecat/services/sarvam/tts.py
View File

@@ -29,7 +29,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import InterruptibleTTSService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_tts
try:
@@ -64,7 +64,7 @@ def language_to_sarvam_language(language: Language) -> Optional[str]:
Language.TE: "te-IN", # Telugu
}
return LANGUAGE_MAP.get(language)
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
class SarvamHttpTTSService(TTSService):

30
src/pipecat/services/speechmatics/stt.py
View File

@@ -31,7 +31,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_stt
try:
@@ -604,11 +604,21 @@ class SpeechmaticsSTTService(STTService):
Creates a transcription config object based on the service parameters. Aligns
with the Speechmatics RT API transcription config.
"""
# Convert language if it's a Language enum
language = self._params.language
if isinstance(language, Language):
language = _language_to_speechmatics_language(language)
# Convert output locale if it's a Language enum
output_locale = self._params.output_locale
if isinstance(output_locale, Language):
output_locale = _locale_to_speechmatics_locale(language, output_locale)
# Transcription config
transcription_config = TranscriptionConfig(
language=self._params.language,
language=language,
domain=self._params.domain,
output_locale=self._params.output_locale,
output_locale=output_locale,
operating_point=self._params.operating_point,
diarization="speaker" if self._params.enable_diarization else None,
enable_partials=self._params.enable_partials,
@@ -987,10 +997,10 @@ def _language_to_speechmatics_language(language: Language) -> str:
language: The Language enum to convert.
Returns:
str: The Speechmatics language code, if found.
str: The Speechmatics language code.
"""
# List of supported input languages
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.AR: "ar",
Language.BA: "ba",
Language.EU: "eu",
@@ -1047,15 +1057,7 @@ def _language_to_speechmatics_language(language: Language) -> str:
Language.CY: "cy",
}
# Get the language code
result = BASE_LANGUAGES.get(language)
# Fail if language is not supported
if not result:
raise ValueError(f"Unsupported language: {language}")
# Return the language code
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
def _locale_to_speechmatics_locale(language_code: str, locale: Language) -> str | None:

14
src/pipecat/services/whisper/base_stt.py
View File

@@ -18,7 +18,7 @@ from openai.types.audio import Transcription
from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
from pipecat.services.stt_service import SegmentedSTTService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
@@ -35,7 +35,7 @@ def language_to_whisper_language(language: Language) -> Optional[str]:
Returns:
str or None: The corresponding Whisper language code, or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.AF: "af",
Language.AR: "ar",
Language.HY: "hy",
@@ -95,15 +95,7 @@ def language_to_whisper_language(language: Language) -> Optional[str]:
Language.CY: "cy",
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
class BaseWhisperSTTService(SegmentedSTTService):

97
src/pipecat/services/whisper/stt.py
View File

@@ -20,7 +20,7 @@ from typing_extensions import TYPE_CHECKING, override
from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
from pipecat.services.stt_service import SegmentedSTTService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
@@ -106,159 +106,70 @@ def language_to_whisper_language(language: Language) -> Optional[str]:
Note:
Only includes languages officially supported by Whisper.
"""
language_map = {
LANGUAGE_MAP = {
# Arabic
Language.AR: "ar",
Language.AR_AE: "ar",
Language.AR_BH: "ar",
Language.AR_DZ: "ar",
Language.AR_EG: "ar",
Language.AR_IQ: "ar",
Language.AR_JO: "ar",
Language.AR_KW: "ar",
Language.AR_LB: "ar",
Language.AR_LY: "ar",
Language.AR_MA: "ar",
Language.AR_OM: "ar",
Language.AR_QA: "ar",
Language.AR_SA: "ar",
Language.AR_SY: "ar",
Language.AR_TN: "ar",
Language.AR_YE: "ar",
# Bengali
Language.BN: "bn",
Language.BN_BD: "bn",
Language.BN_IN: "bn",
# Czech
Language.CS: "cs",
Language.CS_CZ: "cs",
# Danish
Language.DA: "da",
Language.DA_DK: "da",
# German
Language.DE: "de",
Language.DE_AT: "de",
Language.DE_CH: "de",
Language.DE_DE: "de",
# Greek
Language.EL: "el",
Language.EL_GR: "el",
# English
Language.EN: "en",
Language.EN_AU: "en",
Language.EN_CA: "en",
Language.EN_GB: "en",
Language.EN_HK: "en",
Language.EN_IE: "en",
Language.EN_IN: "en",
Language.EN_KE: "en",
Language.EN_NG: "en",
Language.EN_NZ: "en",
Language.EN_PH: "en",
Language.EN_SG: "en",
Language.EN_TZ: "en",
Language.EN_US: "en",
Language.EN_ZA: "en",
# Spanish
Language.ES: "es",
Language.ES_AR: "es",
Language.ES_BO: "es",
Language.ES_CL: "es",
Language.ES_CO: "es",
Language.ES_CR: "es",
Language.ES_CU: "es",
Language.ES_DO: "es",
Language.ES_EC: "es",
Language.ES_ES: "es",
Language.ES_GQ: "es",
Language.ES_GT: "es",
Language.ES_HN: "es",
Language.ES_MX: "es",
Language.ES_NI: "es",
Language.ES_PA: "es",
Language.ES_PE: "es",
Language.ES_PR: "es",
Language.ES_PY: "es",
Language.ES_SV: "es",
Language.ES_US: "es",
Language.ES_UY: "es",
Language.ES_VE: "es",
# Persian
Language.FA: "fa",
Language.FA_IR: "fa",
# Finnish
Language.FI: "fi",
Language.FI_FI: "fi",
# French
Language.FR: "fr",
Language.FR_BE: "fr",
Language.FR_CA: "fr",
Language.FR_CH: "fr",
Language.FR_FR: "fr",
# Hindi
Language.HI: "hi",
Language.HI_IN: "hi",
# Hungarian
Language.HU: "hu",
Language.HU_HU: "hu",
# Indonesian
Language.ID: "id",
Language.ID_ID: "id",
# Italian
Language.IT: "it",
Language.IT_IT: "it",
# Japanese
Language.JA: "ja",
Language.JA_JP: "ja",
# Korean
Language.KO: "ko",
Language.KO_KR: "ko",
# Dutch
Language.NL: "nl",
Language.NL_BE: "nl",
Language.NL_NL: "nl",
# Polish
Language.PL: "pl",
Language.PL_PL: "pl",
# Portuguese
Language.PT: "pt",
Language.PT_BR: "pt",
Language.PT_PT: "pt",
# Romanian
Language.RO: "ro",
Language.RO_RO: "ro",
# Russian
Language.RU: "ru",
Language.RU_RU: "ru",
# Slovak
Language.SK: "sk",
Language.SK_SK: "sk",
# Swedish
Language.SV: "sv",
Language.SV_SE: "sv",
# Thai
Language.TH: "th",
Language.TH_TH: "th",
# Turkish
Language.TR: "tr",
Language.TR_TR: "tr",
# Ukrainian
Language.UK: "uk",
Language.UK_UA: "uk",
# Urdu
Language.UR: "ur",
Language.UR_IN: "ur",
Language.UR_PK: "ur",
# Vietnamese
Language.VI: "vi",
Language.VI_VN: "vi",
# Chinese
Language.ZH: "zh",
Language.ZH_CN: "zh",
Language.ZH_HK: "zh",
Language.ZH_TW: "zh",
}
return language_map.get(language)
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
class WhisperSTTService(SegmentedSTTService):

21
src/pipecat/services/xtts/tts.py
View File

@@ -25,7 +25,7 @@ from pipecat.frames.frames import (
TTSStoppedFrame,
)
from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language
from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_tts
# The server below can connect to XTTS through a local running docker
@@ -45,7 +45,7 @@ def language_to_xtts_language(language: Language) -> Optional[str]:
Returns:
The corresponding XTTS language code, or None if not supported.
"""
BASE_LANGUAGES = {
LANGUAGE_MAP = {
Language.CS: "cs",
Language.DE: "de",
Language.EN: "en",
@@ -65,22 +65,7 @@ def language_to_xtts_language(language: Language) -> Optional[str]:
Language.ZH: "zh-cn",
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
# Convert enum value to string and get the base language part (e.g. es-ES -> es)
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
# Special handling for Chinese variants
if base_code == "zh":
result = "zh-cn"
else:
# Look up the base code in our supported languages
result = base_code if base_code in BASE_LANGUAGES.values() else None
return result
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
class XTTSService(TTSService):

52
src/pipecat/transcriptions/language.py
View File

@@ -14,6 +14,8 @@ combinations for various speech and text processing services.
import sys
from enum import Enum
from loguru import logger
if sys.version_info < (3, 11):
class StrEnum(str, Enum):
@@ -569,3 +571,53 @@ class Language(StrEnum):
# Zulu
ZU = "zu"
ZU_ZA = "zu-ZA"
def resolve_language(
language: Language, language_map: dict[Language, str], use_base_code: bool = True
) -> str:
"""Resolve a Language enum to a service-specific language code.
Checks the language map first, then falls back to extracting the appropriate
code format with a warning if not found in the verified list.
Args:
language: The Language enum value to convert.
language_map: Dictionary mapping Language enums to service language codes.
use_base_code: If True, extracts base code (e.g., 'en' from 'en-US').
If False, uses full language code as-is.
Returns:
The resolved language code for the service.
Examples::
# Service expecting base codes (e.g., Cartesia)
>>> LANGUAGE_MAP = {Language.EN: "en", Language.ES: "es"}
>>> resolve_language(Language.EN_US, LANGUAGE_MAP, use_base_code=True)
# Logs: "Language en-US not verified. Using base code 'en'."
"en"
# Service expecting full codes (e.g., AWS)
>>> LANGUAGE_MAP = {Language.EN_US: "en-US", Language.ES_ES: "es-ES"}
>>> resolve_language(Language.EN_GB, LANGUAGE_MAP, use_base_code=False)
# Logs: "Language en-GB not verified. Using 'en-GB'."
"en-GB"
"""
# Check if language is in the verified map
result = language_map.get(language)
if result is not None:
return result
# Not in map - fall back with warning
lang_str = str(language.value)
if use_base_code:
# Extract base code (e.g., "en" from "en-US")
base_code = lang_str.split("-")[0].lower()
logger.warning(f"Language {language.value} not verified. Using base code '{base_code}'.")
return base_code
else:
logger.warning(f"Language {language.value} not verified. Using '{lang_str}'.")
return lang_str