feat(transcription): add AssemblyAI as transcription provider

Add AssemblyAI as a third transcription provider option alongside
OpenAI and Groq. AssemblyAI offers better accuracy for certain
audio types (distant voices, noisy environments) and serves as a
reliable fallback when other providers struggle.

Changes:
- Add AssemblyAITranscriptionProvider class in providers/transcription.py
- Add 'assemblyai' option in base channel's transcribe_audio()
- Per-channel configuration via transcriptionProvider in config

Usage:
  Set transcriptionProvider: 'assemblyai' and provide an AssemblyAI
  API key via transcriptionApiKey in the channel config.
This commit is contained in:
comadreja
2026-06-09 05:33:18 +08:00
committed by Xubin Ren
parent f183b37542
commit f3eb2aa08b
17 changed files with 780 additions and 113 deletions
+15
View File
@@ -245,3 +245,18 @@ def test_match_provider_routes_forced_novita_model_api_models() -> None:
assert config.get_provider_name() == "novita"
assert config.get_api_base() == "https://api.novita.ai/openai"
def test_transcription_only_provider_is_not_chat_fallback() -> None:
config = Config.model_validate({
"providers": {
"assemblyai": {"apiKey": "aai-test"},
},
"agents": {
"defaults": {
"model": "assemblyai/universal-3-pro",
}
},
})
assert config.get_provider_name() is None
+196
View File
@@ -14,8 +14,14 @@ from nanobot.audio.transcription import (
resolve_transcription_config,
transcribe_audio_file,
)
from nanobot.audio.transcription_registry import (
get_transcription_provider,
resolve_transcription_provider,
transcription_provider_names,
)
from nanobot.config.schema import Config
from nanobot.providers.transcription import (
AssemblyAITranscriptionProvider,
GroqTranscriptionProvider,
OpenAITranscriptionProvider,
OpenRouterTranscriptionProvider,
@@ -44,6 +50,17 @@ def _raw_response(status: int, content: bytes) -> httpx.Response:
return httpx.Response(status_code=status, content=content, request=request)
def _json_response(
status: int,
payload: dict[str, object],
*,
method: str = "POST",
url: str = "https://example.test/audio/transcriptions",
) -> httpx.Response:
request = httpx.Request(method, url)
return httpx.Response(status_code=status, json=payload, request=request)
def test_resolver_uses_legacy_channel_provider_when_top_level_is_unset() -> None:
config = Config()
config.channels.transcription_provider = "openai"
@@ -128,6 +145,29 @@ def test_resolver_accepts_legacy_xiaomi_transcription_alias() -> None:
assert resolved.api_key == "mimo-test"
def test_transcription_registry_lists_providers_and_aliases() -> None:
assert "assemblyai" in transcription_provider_names()
assert get_transcription_provider("assemblyai").default_model == "universal-3-pro,universal-2"
assert resolve_transcription_provider("mimo").name == "xiaomi_mimo"
def test_resolver_supports_assemblyai_provider_config() -> None:
config = Config()
config.transcription.provider = "assemblyai"
config.transcription.model = "universal-3-pro"
config.transcription.language = "en"
config.providers.assemblyai.api_key = "aai-test"
config.providers.assemblyai.api_base = "https://assembly.example/v2"
resolved = resolve_transcription_config(config)
assert resolved.provider == "assemblyai"
assert resolved.model == "universal-3-pro"
assert resolved.language == "en"
assert resolved.api_key == "aai-test"
assert resolved.api_base == "https://assembly.example/v2"
@pytest.mark.asyncio
async def test_transcribe_audio_file_routes_openrouter_provider(audio_file: Path) -> None:
captured: dict[str, object] = {}
@@ -200,6 +240,42 @@ async def test_transcribe_audio_file_routes_xiaomi_mimo_provider(audio_file: Pat
}
@pytest.mark.asyncio
async def test_transcribe_audio_file_routes_assemblyai_provider(audio_file: Path) -> None:
captured: dict[str, object] = {}
class StubAssemblyAI:
def __init__(self, **kwargs):
captured.update(kwargs)
async def transcribe(self, file_path: str | Path) -> str:
captured["file_path"] = Path(file_path)
return "assembly ok"
config = EffectiveTranscriptionConfig(
enabled=True,
provider="assemblyai",
model="universal-3-pro",
language="en",
api_key="aai-test",
api_base="https://assembly.example/v2",
max_duration_sec=120,
max_upload_mb=25,
)
with patch("nanobot.providers.transcription.AssemblyAITranscriptionProvider", StubAssemblyAI):
result = await transcribe_audio_file(audio_file, config)
assert result == "assembly ok"
assert captured == {
"api_key": "aai-test",
"api_base": "https://assembly.example/v2",
"language": "en",
"model": "universal-3-pro",
"file_path": audio_file,
}
def test_resolved_transcription_repr_hides_api_key() -> None:
config = Config()
config.providers.groq.api_key = "gsk-secret"
@@ -628,6 +704,126 @@ async def test_xiaomi_mimo_shares_retry_contract(audio_file: Path) -> None:
assert post.await_count == 2
def test_assemblyai_defaults_and_base_normalization() -> None:
provider = AssemblyAITranscriptionProvider(api_key="aai-test")
assert provider.upload_url == "https://api.assemblyai.com/v2/upload"
assert provider.transcript_url == "https://api.assemblyai.com/v2/transcript"
assert provider.model == "universal-3-pro,universal-2"
custom = AssemblyAITranscriptionProvider(
api_key="aai-test",
api_base="https://assembly.example/v2",
model="universal-3-pro",
)
assert custom.upload_url == "https://assembly.example/v2/upload"
assert custom.transcript_url == "https://assembly.example/v2/transcript"
assert custom.model == "universal-3-pro"
@pytest.mark.asyncio
async def test_assemblyai_uploads_creates_and_polls(audio_file: Path) -> None:
provider = AssemblyAITranscriptionProvider(
api_key="aai-test",
api_base="https://assembly.example/v2",
language="en",
model="universal-3-pro,universal-2",
)
post = AsyncMock(
side_effect=[
_json_response(200, {"upload_url": "https://cdn.example/audio"}, url=provider.upload_url),
_json_response(200, {"id": "tr_123"}, url=provider.transcript_url),
]
)
get = AsyncMock(
return_value=_json_response(
200,
{"status": "completed", "text": "assembly ok"},
method="GET",
url=f"{provider.transcript_url}/tr_123",
)
)
with patch("httpx.AsyncClient.post", post), patch("httpx.AsyncClient.get", get), patch(
"asyncio.sleep", AsyncMock()
):
result = await provider.transcribe(audio_file)
assert result == "assembly ok"
assert post.await_count == 2
assert get.await_count == 1
upload_call, create_call = post.await_args_list
assert upload_call.args == ("https://assembly.example/v2/upload",)
assert upload_call.kwargs["headers"]["Authorization"] == "aai-test"
assert upload_call.kwargs["headers"]["Content-Type"] == "application/octet-stream"
assert upload_call.kwargs["content"] == audio_file.read_bytes()
assert create_call.args == ("https://assembly.example/v2/transcript",)
assert create_call.kwargs["json"] == {
"audio_url": "https://cdn.example/audio",
"speech_models": ["universal-3-pro", "universal-2"],
"language_code": "en",
}
assert get.await_args.args == ("https://assembly.example/v2/transcript/tr_123",)
@pytest.mark.asyncio
async def test_assemblyai_polls_until_completed(audio_file: Path) -> None:
provider = AssemblyAITranscriptionProvider(api_key="aai-test")
post = AsyncMock(
side_effect=[
_json_response(200, {"upload_url": "https://cdn.example/audio"}, url=provider.upload_url),
_json_response(200, {"id": "tr_123"}, url=provider.transcript_url),
]
)
get = AsyncMock(
side_effect=[
_json_response(200, {"status": "processing"}, method="GET"),
_json_response(200, {"status": "completed", "text": "done"}, method="GET"),
]
)
sleep = AsyncMock()
with patch("httpx.AsyncClient.post", post), patch("httpx.AsyncClient.get", get), patch(
"asyncio.sleep", sleep
):
assert await provider.transcribe(audio_file) == "done"
assert get.await_count == 2
assert sleep.await_count == 1
@pytest.mark.asyncio
async def test_assemblyai_returns_empty_on_failed_transcript(audio_file: Path) -> None:
provider = AssemblyAITranscriptionProvider(api_key="aai-test")
post = AsyncMock(
side_effect=[
_json_response(200, {"upload_url": "https://cdn.example/audio"}, url=provider.upload_url),
_json_response(200, {"id": "tr_123"}, url=provider.transcript_url),
]
)
get = AsyncMock(
return_value=_json_response(
200,
{"status": "error", "error": "bad audio"},
method="GET",
)
)
with patch("httpx.AsyncClient.post", post), patch("httpx.AsyncClient.get", get), patch(
"asyncio.sleep", AsyncMock()
):
assert await provider.transcribe(audio_file) == ""
@pytest.mark.asyncio
async def test_assemblyai_missing_api_key_short_circuits(audio_file: Path) -> None:
with patch.dict("os.environ", {}, clear=True):
provider = AssemblyAITranscriptionProvider(api_key=None)
post = AsyncMock()
with patch("httpx.AsyncClient.post", post):
assert await provider.transcribe(audio_file) == ""
assert post.await_count == 0
@pytest.mark.parametrize("status", [408, 429, 500, 502, 503, 504])
@pytest.mark.asyncio
async def test_retries_on_every_advertised_transient_status(
+68
View File
@@ -299,6 +299,50 @@ def test_settings_payload_exposes_xiaomi_mimo_transcription_provider(
assert providers["xiaomi_mimo"]["configured"] is True
def test_settings_payload_exposes_assemblyai_transcription_provider(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
config_path = tmp_path / "config.json"
config = Config()
config.transcription.provider = "assemblyai"
config.providers.assemblyai.api_key = "aai-test"
save_config(config, config_path)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
payload = settings_payload()
assert payload["transcription"]["provider"] == "assemblyai"
assert payload["transcription"]["provider_configured"] is True
providers = {provider["name"]: provider for provider in payload["transcription"]["providers"]}
assert providers["assemblyai"]["label"] == "AssemblyAI"
assert providers["assemblyai"]["configured"] is True
assert providers["assemblyai"]["default_api_base"] == "https://api.assemblyai.com/v2"
provider_rows = {provider["name"]: provider for provider in payload["providers"]}
assert provider_rows["assemblyai"]["configured"] is True
assert provider_rows["assemblyai"]["model_selectable"] is False
def test_model_configuration_rejects_transcription_only_provider(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
config_path = tmp_path / "config.json"
config = Config()
config.providers.assemblyai.api_key = "aai-test"
save_config(config, config_path)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
with pytest.raises(WebUISettingsError, match="does not support chat models"):
create_model_configuration(
{
"label": ["Voice only"],
"provider": ["assemblyai"],
"model": ["universal-3-pro"],
}
)
def test_update_transcription_settings_writes_top_level_only(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
@@ -385,6 +429,30 @@ def test_update_transcription_settings_accepts_xiaomi_mimo(
assert payload["transcription"]["provider_configured"] is True
def test_update_transcription_settings_accepts_assemblyai(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
config_path = tmp_path / "config.json"
config = Config()
config.providers.assemblyai.api_key = "aai-test"
save_config(config, config_path)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
payload = update_transcription_settings(
{
"provider": ["assemblyai"],
"model": ["universal-3-pro"],
}
)
saved = load_config(config_path)
assert saved.transcription.provider == "assemblyai"
assert saved.transcription.model == "universal-3-pro"
assert payload["transcription"]["provider"] == "assemblyai"
assert payload["transcription"]["provider_configured"] is True
def test_update_transcription_settings_validates_language(
tmp_path,
monkeypatch: pytest.MonkeyPatch,