Mohamed ElkholyandXubin Ren
ce5272c153
Complete the symmetry left by #3214: ChannelManager._resolve_transcription_base already resolves providers.openai.api_base, but BaseChannel.transcribe_audio instantiated OpenAITranscriptionProvider without forwarding it, and the provider __init__ did not accept the parameter. Self-hosted OpenAI-compatible Whisper endpoints (LiteLLM, vLLM, etc.) configured via config.json were therefore ignored for the OpenAI backend. - OpenAITranscriptionProvider.__init__ now accepts api_base with env fallback (OPENAI_TRANSCRIPTION_BASE_URL) matching the Groq pattern. - BaseChannel.transcribe_audio forwards self.transcription_api_base to OpenAI. - Tests mirror the existing Groq coverage: manager propagation for provider "openai", BaseChannel-to-provider argument passing, and provider default vs override for api_url. Fully backward-compatible: when api_base is None and the env var is unset, the default https://api.openai.com/v1/audio/transcriptions is used. Refs #3213, follow-up to #3214.
99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
"""Voice transcription providers (Groq and OpenAI Whisper)."""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from loguru import logger
|
|
|
|
|
|
class OpenAITranscriptionProvider:
|
|
"""Voice transcription provider using OpenAI's Whisper API."""
|
|
|
|
def __init__(self, api_key: str | None = None, api_base: str | None = None):
|
|
self.api_key = api_key or os.environ.get("OPENAI_API_KEY")
|
|
self.api_url = (
|
|
api_base
|
|
or os.environ.get("OPENAI_TRANSCRIPTION_BASE_URL")
|
|
or "https://api.openai.com/v1/audio/transcriptions"
|
|
)
|
|
|
|
async def transcribe(self, file_path: str | Path) -> str:
|
|
if not self.api_key:
|
|
logger.warning("OpenAI API key not configured for transcription")
|
|
return ""
|
|
path = Path(file_path)
|
|
if not path.exists():
|
|
logger.error("Audio file not found: {}", file_path)
|
|
return ""
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
with open(path, "rb") as f:
|
|
files = {"file": (path.name, f), "model": (None, "whisper-1")}
|
|
headers = {"Authorization": f"Bearer {self.api_key}"}
|
|
response = await client.post(
|
|
self.api_url, headers=headers, files=files, timeout=60.0,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json().get("text", "")
|
|
except Exception as e:
|
|
logger.error("OpenAI transcription error: {}", e)
|
|
return ""
|
|
|
|
|
|
class GroqTranscriptionProvider:
|
|
"""
|
|
Voice transcription provider using Groq's Whisper API.
|
|
|
|
Groq offers extremely fast transcription with a generous free tier.
|
|
"""
|
|
|
|
def __init__(self, api_key: str | None = None, api_base: str | None = None):
|
|
self.api_key = api_key or os.environ.get("GROQ_API_KEY")
|
|
self.api_url = api_base or os.environ.get("GROQ_BASE_URL") or "https://api.groq.com/openai/v1/audio/transcriptions"
|
|
|
|
async def transcribe(self, file_path: str | Path) -> str:
|
|
"""
|
|
Transcribe an audio file using Groq.
|
|
|
|
Args:
|
|
file_path: Path to the audio file.
|
|
|
|
Returns:
|
|
Transcribed text.
|
|
"""
|
|
if not self.api_key:
|
|
logger.warning("Groq API key not configured for transcription")
|
|
return ""
|
|
|
|
path = Path(file_path)
|
|
if not path.exists():
|
|
logger.error("Audio file not found: {}", file_path)
|
|
return ""
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
with open(path, "rb") as f:
|
|
files = {
|
|
"file": (path.name, f),
|
|
"model": (None, "whisper-large-v3"),
|
|
}
|
|
headers = {
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
}
|
|
|
|
response = await client.post(
|
|
self.api_url,
|
|
headers=headers,
|
|
files=files,
|
|
timeout=60.0
|
|
)
|
|
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return data.get("text", "")
|
|
|
|
except Exception as e:
|
|
logger.error("Groq transcription error: {}", e)
|
|
return ""
|