diff --git a/CLAUDE.md b/CLAUDE.md index a5c99eb..5ffb79c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -42,7 +42,7 @@ aai.settings.api_key = "your-key" - `aai.AsyncTranscript` — What `AsyncTranscriber` returns. Same fields as `Transcript`, coroutine methods - `aai.AsyncClient` — Shared `httpx.AsyncClient` pool for one or more `AsyncTranscriber`s - `assemblyai.prerecorded.v2` — Canonical module for both transcribers, matching the `/v2/transcript` API. The top-level `aai.*` names re-export it -- `aai.TranscriptionConfig` — All transcription options: `speech_models`, `speaker_labels`, `sentiment_analysis`, `entity_detection`, `auto_chapters`, `content_safety`, `language_detection`, `summarization`, `word_boost`, `disfluencies` +- `aai.TranscriptionConfig` — All transcription options: `speech_models`, `speaker_labels`, `sentiment_analysis`, `entity_detection`, `auto_chapters`, `content_safety`, `language_detection`, `summarization`, `word_boost`, `disfluencies`, `speech_understanding` - `aai.Transcript` — Result object with `.text`, `.status`, `.utterances`, `.words`, `.chapters`, `.entities`, `.sentiment_analysis`. Methods: `get_sentences()`, `get_paragraphs()`, `export_subtitles_srt()`, `export_subtitles_vtt()` - `aai.SyncTranscriber` — Synchronous pre-recorded transcription: audio in, transcript out, one request (no polling). Methods: `transcribe()`, `transcribe_live()`, `open_live()`, `transcribe_async()`, `warm()` - `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()`, `transcribe_live()` and `warm()` are coroutines; `open_live()` returns an `AsyncLiveSession`. Owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one diff --git a/assemblyai/__init__.py b/assemblyai/__init__.py index f625c97..2ed4c3a 100644 --- a/assemblyai/__init__.py +++ b/assemblyai/__init__.py @@ -75,11 +75,13 @@ Sentiment, SentimentType, Settings, + SpeakerDescription, SpeakerIdentificationRequest, SpeakerIdentificationResponse, SpeakerOptions, SpeakerType, SpeechModel, + SpeechUnderstandingEffort, SpeechUnderstandingFeatureRequests, SpeechUnderstandingFeatureResponses, SpeechUnderstandingRequest, @@ -189,11 +191,13 @@ def __getattr__(name: str): "Sentiment", "SentimentType", "Settings", + "SpeakerDescription", "SpeakerIdentificationRequest", "SpeakerIdentificationResponse", "SpeakerOptions", "SpeakerType", "SpeechModel", + "SpeechUnderstandingEffort", "SpeechUnderstandingFeatureRequests", "SpeechUnderstandingFeatureResponses", "SpeechUnderstandingRequest", diff --git a/assemblyai/__version__.py b/assemblyai/__version__.py index 14d9d2f..3c1e9cb 100644 --- a/assemblyai/__version__.py +++ b/assemblyai/__version__.py @@ -1 +1 @@ -__version__ = "1.7.0" +__version__ = "1.7.1" diff --git a/assemblyai/types.py b/assemblyai/types.py index f722658..2c888e5 100644 --- a/assemblyai/types.py +++ b/assemblyai/types.py @@ -707,6 +707,73 @@ class SpeakerType(str, Enum): "Identify speakers by their name" +class SpeechUnderstandingEffort(str, Enum): + """ + How much processing power to spend on a speech understanding task. 'medium' produces + higher quality results on harder audio at a higher cost. + """ + + low = "low" + "Standard processing effort" + + medium = "medium" + "Higher quality processing effort for harder audio" + + +class SpeakerDescription(BaseModel): + """ + Speaker description for speaker identification with rich metadata + """ + + name: Optional[str] = None + "Speaker name (when speaker_type is 'name')" + + role: Optional[str] = None + "Speaker role (when speaker_type is 'role')" + + description: Optional[str] = None + "Description of the speaker's role or characteristics in the audio" + + if pydantic_v2: + model_config = ConfigDict(extra="allow") + else: + + class Config: + extra = "allow" + + def __getitem__(self, item: str) -> Any: + if hasattr(self, item): + val = getattr(self, item) + if val is not None: + return val + if ( + hasattr(self, "__pydantic_extra__") + and self.__pydantic_extra__ + and item in self.__pydantic_extra__ + ): + return self.__pydantic_extra__[item] + if hasattr(self, item): + return getattr(self, item) + raise KeyError(item) + + def __contains__(self, item: str) -> bool: + if hasattr(self, item) and getattr(self, item) is not None: + return True + if ( + hasattr(self, "__pydantic_extra__") + and self.__pydantic_extra__ + and item in self.__pydantic_extra__ + ): + return True + return False + + def get(self, item: str, default: Any = None) -> Any: + try: + return self[item] + except KeyError: + return default + + class SpeakerIdentificationRequest(BaseModel): """ Speaker identification configuration for speech understanding @@ -718,9 +785,12 @@ class SpeakerIdentificationRequest(BaseModel): known_values: Optional[List[str]] = None "Known speaker values (required when speaker_type is 'role')" - speakers: Optional[List[Dict[str, Any]]] = None + speakers: Optional[List[Union[SpeakerDescription, Dict[str, Any]]]] = None "Known speaker definitions with optional descriptions for improved accuracy" + effort: Optional[Union[SpeechUnderstandingEffort, str]] = None + "How much effort to spend on this task ('low' or 'medium')" + class TranslationRequest(BaseModel): """ @@ -736,6 +806,12 @@ class TranslationRequest(BaseModel): match_original_utterance: Optional[bool] = False "Whether to match the original utterance structure in translations (default: False)" + force_translation: Optional[bool] = False + "Whether to translate even when the detected source language equals the target language (default: False)" + + effort: Optional[Union[SpeechUnderstandingEffort, str]] = None + "How much effort to spend on this task ('low' or 'medium')" + class CustomFormattingRequest(BaseModel): """ @@ -751,6 +827,12 @@ class CustomFormattingRequest(BaseModel): email: Optional[str] = None "Custom email format pattern (e.g., 'username@domain.com')" + format_utterances: Optional[bool] = False + "Whether to also format utterance-level text (default: False)" + + effort: Optional[Union[SpeechUnderstandingEffort, str]] = None + "How much effort to spend on this task ('low' or 'medium')" + class SummarizationRequest(BaseModel): """ @@ -766,7 +848,9 @@ class SummarizationRequest(BaseModel): approximate_chapters: Optional[int] = None "Override: Approximate number of chapters to split the transcript into" - effort: Optional[Literal["low", "medium"]] = None + effort: Optional[ + Union[SpeechUnderstandingEffort, Literal["low", "medium"], str] + ] = None "How much effort to spend on this task (default: 'low')" @@ -883,7 +967,9 @@ class SummarizationResponse(BaseModel): summary_type: Optional[Literal["bullets", "paragraph"]] = None "The summary type that was used" - effort: Optional[Literal["low", "medium"]] = None + effort: Optional[ + Union[SpeechUnderstandingEffort, Literal["low", "medium"], str] + ] = None "The effort level that was used" diff --git a/tests/unit/test_speech_understanding.py b/tests/unit/test_speech_understanding.py new file mode 100644 index 0000000..a3d89e8 --- /dev/null +++ b/tests/unit/test_speech_understanding.py @@ -0,0 +1,239 @@ +import assemblyai as aai + + +def test_speech_understanding_effort_enum(): + assert aai.SpeechUnderstandingEffort.low == "low" + assert aai.SpeechUnderstandingEffort.medium == "medium" + assert aai.SpeechUnderstandingEffort.low.value == "low" + assert aai.SpeechUnderstandingEffort.medium.value == "medium" + + +def test_speaker_identification_request_effort_enum(): + req = aai.SpeakerIdentificationRequest( + speaker_type=aai.SpeakerType.name, + effort=aai.SpeechUnderstandingEffort.medium, + ) + assert req.speaker_type == aai.SpeakerType.name + assert req.effort == aai.SpeechUnderstandingEffort.medium + data = req.dict(exclude_none=True) + assert data["effort"] == "medium" + assert data["speaker_type"] == "name" + + +def test_speaker_identification_request_effort_str(): + req = aai.SpeakerIdentificationRequest( + speaker_type=aai.SpeakerType.role, + known_values=["Agent", "Customer"], + effort="low", + ) + assert req.speaker_type == aai.SpeakerType.role + assert req.known_values == ["Agent", "Customer"] + assert req.effort == "low" + data = req.dict(exclude_none=True) + assert data["effort"] == "low" + + +def test_speaker_identification_request_effort_omitted(): + req = aai.SpeakerIdentificationRequest( + speaker_type=aai.SpeakerType.name, + ) + assert req.effort is None + data = req.dict(exclude_none=True) + assert "effort" not in data + + +def test_translation_request_effort_and_force_translation(): + req = aai.TranslationRequest( + target_languages=["es", "fr"], + force_translation=True, + effort=aai.SpeechUnderstandingEffort.medium, + ) + assert req.target_languages == ["es", "fr"] + assert req.force_translation is True + assert req.effort == aai.SpeechUnderstandingEffort.medium + assert req.formal is False + assert req.match_original_utterance is False + + data = req.dict(exclude_none=True) + assert data["target_languages"] == ["es", "fr"] + assert data["force_translation"] is True + assert data["effort"] == "medium" + + +def test_translation_request_defaults(): + req = aai.TranslationRequest(target_languages=["de"]) + assert req.force_translation is False + assert req.effort is None + + data = req.dict(exclude_none=True) + assert data["target_languages"] == ["de"] + assert data["force_translation"] is False + assert "effort" not in data + + +def test_custom_formatting_request_effort(): + req = aai.CustomFormattingRequest( + date="mm/dd/yyyy", + phone_number="(xxx)xxx-xxxx", + effort=aai.SpeechUnderstandingEffort.low, + ) + assert req.date == "mm/dd/yyyy" + assert req.phone_number == "(xxx)xxx-xxxx" + assert req.effort == aai.SpeechUnderstandingEffort.low + + data = req.dict(exclude_none=True) + assert data["date"] == "mm/dd/yyyy" + assert data["phone_number"] == "(xxx)xxx-xxxx" + assert data["effort"] == "low" + assert "email" not in data + + +def test_speech_understanding_request_nested_effort(): + req = aai.SpeechUnderstandingRequest( + request=aai.SpeechUnderstandingFeatureRequests( + speaker_identification=aai.SpeakerIdentificationRequest( + speaker_type=aai.SpeakerType.name, + effort=aai.SpeechUnderstandingEffort.medium, + ), + translation=aai.TranslationRequest( + target_languages=["es"], + force_translation=True, + effort=aai.SpeechUnderstandingEffort.low, + ), + custom_formatting=aai.CustomFormattingRequest( + date="mm/dd/yyyy", + effort=aai.SpeechUnderstandingEffort.medium, + ), + ) + ) + + data = req.dict(exclude_none=True) + assert data["request"]["speaker_identification"]["effort"] == "medium" + assert data["request"]["translation"]["effort"] == "low" + assert data["request"]["translation"]["force_translation"] is True + assert data["request"]["custom_formatting"]["effort"] == "medium" + + +def test_transcription_config_speech_understanding_dict_with_effort_and_force_translation(): + config = aai.TranscriptionConfig() + config.speech_understanding = { + "request": { + "speaker_identification": { + "speaker_type": "name", + "effort": "medium", + }, + "translation": { + "target_languages": ["es"], + "force_translation": True, + "effort": "low", + }, + "custom_formatting": { + "date": "mm/dd/yyyy", + "effort": "medium", + }, + } + } + + assert isinstance(config.speech_understanding, aai.SpeechUnderstandingRequest) + features = config.speech_understanding.request + assert features.speaker_identification.speaker_type == aai.SpeakerType.name + assert features.speaker_identification.effort == "medium" + assert features.translation.target_languages == ["es"] + assert features.translation.force_translation is True + assert features.translation.effort == "low" + assert features.custom_formatting.date == "mm/dd/yyyy" + assert features.custom_formatting.effort == "medium" + + +def test_speech_understanding_response_parses_request_with_effort_and_force_translation(): + raw_response = { + "request": { + "speaker_identification": { + "speaker_type": "name", + "effort": "medium", + }, + "translation": { + "target_languages": ["es"], + "force_translation": True, + "effort": "low", + }, + "custom_formatting": { + "date": "mm/dd/yyyy", + "effort": "medium", + }, + }, + "response": { + "speaker_identification": { + "status": "success", + "mapping": {"Speaker A": "Alice"}, + }, + "translation": { + "status": "success", + }, + "custom_formatting": { + "status": "success", + "formatted_text": "Call me at (123)456-7890.", + }, + }, + } + + res = aai.SpeechUnderstandingResponse.parse_obj(raw_response) + assert res.request.speaker_identification.effort == "medium" + assert res.request.translation.effort == "low" + assert res.request.translation.force_translation is True + assert res.request.custom_formatting.effort == "medium" + assert res.response.speaker_identification.status == "success" + assert res.response.speaker_identification.mapping == {"Speaker A": "Alice"} + + +def test_speaker_description_and_speakers_list(): + sd1 = aai.SpeakerDescription( + name="Alice", description="Host of the podcast", custom_attr="lead" + ) + sd2 = aai.SpeakerDescription(role="Guest", description="AI researcher") + + assert sd1.name == "Alice" + assert sd1.description == "Host of the podcast" + assert sd1.custom_attr == "lead" + assert sd2.role == "Guest" + assert sd2.description == "AI researcher" + + req = aai.SpeakerIdentificationRequest( + speaker_type=aai.SpeakerType.name, + speakers=[sd1, {"name": "Bob", "description": "Co-host"}], + effort=aai.SpeechUnderstandingEffort.medium, + ) + assert req.speakers[0].name == "Alice" + assert req.speakers[1].name == "Bob" + + data = req.dict(exclude_none=True) + assert data["speaker_type"] == "name" + assert data["effort"] == "medium" + assert data["speakers"][0]["name"] == "Alice" + assert data["speakers"][0]["description"] == "Host of the podcast" + assert data["speakers"][1]["name"] == "Bob" + + +def test_custom_formatting_format_utterances(): + req = aai.CustomFormattingRequest( + date="mm/dd/yyyy", + phone_number="(xxx)xxx-xxxx", + format_utterances=True, + effort=aai.SpeechUnderstandingEffort.medium, + ) + assert req.format_utterances is True + data = req.dict(exclude_none=True) + assert data["format_utterances"] is True + assert data["effort"] == "medium" + + +def test_summarization_request_with_effort_enum(): + req = aai.SummarizationRequest( + summary_type="paragraph", + effort=aai.SpeechUnderstandingEffort.medium, + headline_length=10, + ) + assert req.effort == aai.SpeechUnderstandingEffort.medium + data = req.dict(exclude_none=True) + assert data["effort"] == "medium" + assert data["summary_type"] == "paragraph"