From 37c9cb8c0f8defb4063040b882f316e894f0beaf Mon Sep 17 00:00:00 2001 From: lucas Date: Sat, 30 May 2026 19:07:35 +0900 Subject: [PATCH] Add Fireworks transcription provider Signed-off-by: lucas --- .env.example | 3 + sapat/providers/fireworks.py | 96 ++++++++++++++++++++++++ tests/providers/test_fireworks.py | 119 ++++++++++++++++++++++++++++++ 3 files changed, 218 insertions(+) create mode 100644 sapat/providers/fireworks.py create mode 100644 tests/providers/test_fireworks.py diff --git a/.env.example b/.env.example index 3e70778..80cb928 100644 --- a/.env.example +++ b/.env.example @@ -37,6 +37,9 @@ LOCALAI_API_KEY= # --- ElevenLabs --- ELEVENLABS_API_KEY= +# --- Fireworks AI --- +FIREWORKS_API_KEY= + # --- Symbl.ai --- SYMBL_ACCESS_TOKEN= # Or use app credentials instead: diff --git a/sapat/providers/fireworks.py b/sapat/providers/fireworks.py new file mode 100644 index 0000000..52c0cf0 --- /dev/null +++ b/sapat/providers/fireworks.py @@ -0,0 +1,96 @@ +# ABOUTME: Fireworks AI transcription provider +# ABOUTME: Uses Fireworks Audio API directly without a provider SDK dependency + +import os +from typing import Optional + +import requests + +from sapat.providers import register +from sapat.providers.base import ( + AudioFormat, + ProviderConfig, + TranscriptionProvider, + TranscriptionResult, +) + + +@register +class FireworksProvider(TranscriptionProvider): + name = "fireworks" + config = ProviderConfig( + required_env_vars=["FIREWORKS_API_KEY"], + max_file_size_mb=1000.0, + preferred_format=AudioFormat.WAV, + supports_correction=False, + default_model="whisper-v3", + ) + + def resolve_model(self, model_alias: str) -> str: + aliases = { + "default": "whisper-v3", + "whisper": "whisper-v3", + "turbo": "whisper-v3-turbo", + } + return aliases.get(model_alias, model_alias) + + def transcribe( + self, + audio_file: str, + model: str, + language: str = "en", + prompt: Optional[str] = None, + temperature: float = 0, + **kwargs, + ) -> TranscriptionResult: + headers = {"Authorization": os.getenv("FIREWORKS_API_KEY", "")} + data = { + "model": model, + "response_format": "json", + "temperature": temperature, + } + if language: + data["language"] = language + if prompt: + data["prompt"] = prompt + + with open(audio_file, "rb") as f: + response = requests.post( + self._endpoint_for_model(model), + headers=headers, + data=data, + files={"file": f}, + ) + + if response.status_code != 200: + raise RuntimeError( + f"Fireworks transcription failed ({response.status_code}): {response.text}" + ) + + payload = response.json() + return TranscriptionResult( + text=self._extract_text(payload), + language=payload.get("language") if isinstance(payload, dict) else None, + raw_response=payload, + ) + + @staticmethod + def _endpoint_for_model(model: str) -> str: + if model == "whisper-v3-turbo": + host = "https://audio-turbo.api.fireworks.ai" + else: + host = "https://audio-prod.api.fireworks.ai" + return f"{host}/v1/audio/transcriptions" + + @staticmethod + def _extract_text(payload) -> str: + if isinstance(payload, str): + text = payload + elif isinstance(payload, dict): + text = payload.get("text", "") + else: + text = "" + + if not text: + raise RuntimeError("Fireworks response contained no transcript text") + return text diff --git a/tests/providers/test_fireworks.py b/tests/providers/test_fireworks.py new file mode 100644 index 0000000..aee9e85 --- /dev/null +++ b/tests/providers/test_fireworks.py @@ -0,0 +1,119 @@ +# ABOUTME: Tests for the Fireworks AI transcription provider +# ABOUTME: Verifies request shape, model routing, response parsing, and availability + +import os +from unittest.mock import patch + +import pytest + +from sapat.providers.base import TranscriptionResult + + +class FakeResponse: + def __init__(self, status_code=200, payload=None, text=""): + self.status_code = status_code + self._payload = payload if payload is not None else {} + self.text = text + + def json(self): + return self._payload + + +@pytest.fixture +def audio_file(tmp_path): + path = tmp_path / "sample.wav" + path.write_bytes(b"fake wav bytes") + return str(path) + + +class TestFireworksProvider: + @patch.dict(os.environ, {"FIREWORKS_API_KEY": "fw-test-key"}, clear=True) + def test_available_with_api_key(self): + from sapat.providers.fireworks import FireworksProvider + + assert FireworksProvider.is_available() is True + + @patch.dict(os.environ, {}, clear=True) + def test_not_available_without_api_key(self): + from sapat.providers.fireworks import FireworksProvider + + assert FireworksProvider.is_available() is False + + @patch.dict(os.environ, {"FIREWORKS_API_KEY": "fw-test-key"}, clear=True) + @patch("sapat.providers.fireworks.requests.post") + def test_transcribe_sends_prod_request(self, mock_post, audio_file): + mock_post.return_value = FakeResponse( + payload={"text": "hello fireworks", "language": "en"} + ) + + from sapat.providers.fireworks import FireworksProvider + + provider = FireworksProvider() + result = provider.transcribe( + audio_file, + model="whisper-v3", + language="en", + prompt="product names: Daytona, Sapat", + temperature=0.2, + ) + + assert isinstance(result, TranscriptionResult) + assert result.text == "hello fireworks" + assert result.language == "en" + + assert ( + mock_post.call_args.args[0] + == "https://audio-prod.api.fireworks.ai/v1/audio/transcriptions" + ) + kwargs = mock_post.call_args.kwargs + assert kwargs["headers"] == {"Authorization": "fw-test-key"} + assert kwargs["data"]["model"] == "whisper-v3" + assert kwargs["data"]["response_format"] == "json" + assert kwargs["data"]["language"] == "en" + assert kwargs["data"]["prompt"] == "product names: Daytona, Sapat" + assert kwargs["data"]["temperature"] == 0.2 + assert "file" in kwargs["files"] + + @patch.dict(os.environ, {"FIREWORKS_API_KEY": "fw-test-key"}, clear=True) + @patch("sapat.providers.fireworks.requests.post") + def test_transcribe_routes_turbo_model_to_turbo_endpoint( + self, mock_post, audio_file + ): + mock_post.return_value = FakeResponse(payload={"text": "fast transcript"}) + + from sapat.providers.fireworks import FireworksProvider + + provider = FireworksProvider() + result = provider.transcribe(audio_file, model="whisper-v3-turbo") + + assert result.text == "fast transcript" + assert ( + mock_post.call_args.args[0] + == "https://audio-turbo.api.fireworks.ai/v1/audio/transcriptions" + ) + + @patch.dict(os.environ, {"FIREWORKS_API_KEY": "fw-test-key"}, clear=True) + @patch("sapat.providers.fireworks.requests.post") + def test_raises_on_api_error(self, mock_post, audio_file): + mock_post.return_value = FakeResponse(status_code=401, text="unauthorized") + + from sapat.providers.fireworks import FireworksProvider + + provider = FireworksProvider() + with pytest.raises(RuntimeError, match="401"): + provider.transcribe(audio_file, model="whisper-v3") + + def test_raises_when_response_has_no_text(self): + from sapat.providers.fireworks import FireworksProvider + + with pytest.raises(RuntimeError, match="no transcript"): + FireworksProvider._extract_text({}) + + def test_model_aliases(self): + from sapat.providers.fireworks import FireworksProvider + + provider = FireworksProvider() + assert provider.resolve_model("default") == "whisper-v3" + assert provider.resolve_model("whisper") == "whisper-v3" + assert provider.resolve_model("turbo") == "whisper-v3-turbo" + assert provider.resolve_model("custom-model") == "custom-model"