在香港 LiteLLM 接入四款 ZenMux ASR 对比路由

servers/bwh-hk-cmi/litellm/README.md

@@ -29,3 +29,21 @@
 - `PERSONAL_VOCABULARY_B64` 只存 Arcane Project 环境（Base64 不是加密）；init 将其解码至现有私有凭据卷 `/credentials/personal-vocabulary.json`，umask 077，LiteLLM 只读挂载，不进入 Git。
 - 更新词库时使用 server-management 的 `scripts/export-entity-dictation-vocabulary.py`，先审查纯实体名称与频次，替换 Arcane 中上述变量并重建 init；单改环境变量不会更新已生成文件。保留旧私有词库以便回滚。升级 LiteLLM 前重跑 `scripts/test-vertex-dictation-wire.py`，确认原生 SMART/热词和普通路线隔离。
 - 验收：OpenAI multipart 音频端点接受一秒静音及 Google 公开 Brooklyn Bridge 语音示例，json 均返回 200，公开样本正确输出 `How old is the Brooklyn Bridge?`。测试证明接口和原生参数有效，不代表中文口语及个人名词已进行识别质量对比。
+
+## 四款 ZenMux ASR 对比入口
+
+同一 Base URL `https://api.opfo.xyz/v1`、HK 现有密钥，Spokenly 选择自定义 OpenAI-compatible transcription，response_format=json，Prompt 留空，额外 AI Processing 关闭。只切换模型名：
+
+| 客户端模型名 | ZenMux 实际型号 |
+| --- | --- |
+| `asr-gpt` | `openai/gpt-transcribe` |
+| `asr-qwen` | `qwen/qwen-audio-3.1-asr-flash` |
+| `asr-doubao` | `bytedance/doubao-seed-asr-2.0` |
+| `asr-mimo` | `xiaomi/mimo-v2.5-asr` |
+
+- 四条路线只使用 ZenMux 订阅 key，不注入个人热词或 SMART、不加额外 LLM 改写，不自动切模型或 PAYG。供应商自身默认的标点和规范化未强制覆盖。
+- `zenmux_transcription.py` 将 multipart 文件转换成 ZenMux 要求的 JSON/base64，固定上游 `/api/v1/audio/transcriptions`；不新增服务、端口或音频存储。
+- 适配器只识别四个上游型号，校验固定 ZenMux API 地址，其他 OpenAI／Vertex 路线仍用原适配器。原 `gemini-transcribe-personal` 的 SMART 与个人词库不变，会议管线不动。
+- 本轮只验收非流式文件转写和 JSON 输出。不要请求 speaker labels、时间戳或 verbose_json，也不使用提示词来冒充润色能力。
+- ZenMux 没有返回细分 audio/text token 计量，适配器只转发其音频秒数，不编造 token 拆分；费用以 ZenMux 的实际用量记录为准。
+- 每次升级 LiteLLM，重跑 server-management 的 `scripts/test-zenmux-asr-wire.py`（同步／异步、原音频字节、JSON 转换、其他 provider 不变）和现有 Vertex dictation 回归测试。

servers/bwh-hk-cmi/litellm/compose.yaml

@@ -40,6 +40,7 @@ services:
       - ./provider_routes.py:/etc/litellm/provider_routes.py:ro
       - ./render_config.py:/etc/litellm/render_config.py:ro
       - ./vertex_dictation.py:/etc/litellm/vertex_dictation.py:ro
+      - ./zenmux_transcription.py:/etc/litellm/zenmux_transcription.py:ro
       - google-adc:/credentials:ro
     entrypoint: ["/bin/sh", "-c"]
     command:

servers/bwh-hk-cmi/litellm/render_config.py

@@ -12,6 +12,19 @@ def build_hk_config(catalog, env):
     config['general_settings'] = {'master_key': 'os.environ/LITELLM_MASTER_KEY'}
     routes = config['model_list']
     routes[:] = [route for route in routes if route['model_name'] != 'gemini-transcribe-personal']
+    if env.get('ZENMUX_SUBSCRIPTION_API_KEY'):
+        asr_models = {
+            'asr-gpt': 'openai/gpt-transcribe',
+            'asr-qwen': 'qwen/qwen-audio-3.1-asr-flash',
+            'asr-doubao': 'bytedance/doubao-seed-asr-2.0',
+            'asr-mimo': 'xiaomi/mimo-v2.5-asr',
+        }
+        for alias, model in asr_models.items():
+            routes.append({'model_name': alias, 'litellm_params': {
+                'model': 'openai/' + model,
+                'api_key': 'os.environ/ZENMUX_SUBSCRIPTION_API_KEY',
+                'api_base': 'https://zenmux.ai/api/v1',
+                'max_retries': 0}})
     if env.get('GOOGLE_AI_STUDIO_API_KEY'):
         routes.append({'model_name': 'gemini-transcribe', 'litellm_params': {
             'model': 'gemini/gemini-3.5-transcribe',

servers/bwh-hk-cmi/litellm/test_config.py

@@ -21,6 +21,13 @@ class RoutingTests(unittest.TestCase):
         self.assertEqual(routes['gemini-transcribe-personal']['vertex_location'], 'global')
         self.assertNotIn('grok-latest', routes)
         self.assertNotIn('gpt-sol-latest', routes)
+        self.assertEqual(routes['asr-gpt']['model'], 'openai/openai/gpt-transcribe')
+        self.assertEqual(routes['asr-qwen']['model'], 'openai/qwen/qwen-audio-3.1-asr-flash')
+        self.assertEqual(routes['asr-doubao']['model'], 'openai/bytedance/doubao-seed-asr-2.0')
+        self.assertEqual(routes['asr-mimo']['model'], 'openai/xiaomi/mimo-v2.5-asr')
+        for alias in ['asr-gpt', 'asr-qwen', 'asr-doubao', 'asr-mimo']:
+            self.assertNotIn('personal_dictation', routes[alias])
+            self.assertEqual(routes[alias]['api_key'], 'os.environ/ZENMUX_SUBSCRIPTION_API_KEY')
         self.assertEqual(routes['gemini-transcribe']['model'], 'gemini/gemini-3.5-transcribe')
         self.assertEqual(routes['gemini-3.6-flash']['model'], 'vertex_ai/gemini-3.8-flash')
         self.assertEqual(len(result['router_settings']['fallbacks']), 6)

servers/bwh-hk-cmi/litellm/vertex_dictation.py

@@ -45,6 +45,8 @@ def install():
 if __name__ == '__main__':
     import sys
     from litellm import run_server
+    from zenmux_transcription import install as install_zenmux_asr
+    install_zenmux_asr()
     install()
     sys.argv[0] = 'litellm'
     sys.exit(run_server())

servers/bwh-hk-cmi/litellm/zenmux_transcription.py

@@ -0,0 +1,91 @@
+"""Adapt four allowlisted ZenMux JSON ASR models to LiteLLM's audio endpoint."""
+import base64
+
+from litellm.llms.base_llm.audio_transcription.transformation import AudioTranscriptionRequestData
+from litellm.llms.openai.transcriptions.whisper_transformation import OpenAIWhisperAudioTranscriptionConfig
+from litellm.litellm_core_utils.audio_utils.utils import process_audio_file
+from litellm.types.utils import TranscriptionResponse, TranscriptionUsageDurationObject
+
+MODELS = frozenset({'openai/gpt-transcribe', 'qwen/qwen-audio-3.1-asr-flash',
+    'bytedance/doubao-seed-asr-2.0', 'xiaomi/mimo-v2.5-asr'})
+API_BASE = 'https://zenmux.ai/api/v1'
+
+
+class ZenMuxTranscriptionConfig(OpenAIWhisperAudioTranscriptionConfig):
+    @property
+    def has_native_transcription_endpoint(self):
+        return True
+
+    def get_supported_openai_params(self, model):
+        return ['language', 'response_format', 'prompt']
+
+    def map_openai_params(self, non_default_params, optional_params, model, drop_params):
+        params = super().map_openai_params(non_default_params, optional_params, model, drop_params)
+        if params.get('prompt'):
+            raise ValueError('ZenMux comparison routes do not accept prompts; leave prompt empty')
+        if params.get('response_format') not in (None, 'json', 'text'):
+            raise ValueError('ZenMux comparison routes support json/text only')
+        return params
+
+    def validate_environment(self, headers, model, messages, optional_params,
+                             litellm_params, api_key=None, api_base=None):
+        if not api_key:
+            raise ValueError('ZenMux subscription API key is required')
+        return {**headers, 'Authorization': 'Bearer ' + api_key, 'Content-Type': 'application/json'}
+
+    def get_complete_url(self, api_base, api_key, model, optional_params, litellm_params, stream=None):
+        if (api_base or '').rstrip('/') != API_BASE or model not in MODELS:
+            raise ValueError('ZenMux ASR adapter is restricted to the configured endpoint and model allowlist')
+        if stream:
+            raise ValueError('Streaming is not enabled for comparison dictation routes')
+        return API_BASE + '/audio/transcriptions'
+
+    def transform_audio_transcription_request(self, model, audio_file, optional_params, litellm_params):
+        if model not in MODELS:
+            raise ValueError('Unknown ZenMux ASR model')
+        if optional_params.get('timestamp_granularities') or optional_params.get('prompt'):
+            raise ValueError('Timestamps and prompt injection are disabled on comparison routes')
+        audio = process_audio_file(audio_file)
+        formats = {'audio/wav': 'wav', 'audio/x-wav': 'wav', 'audio/wave': 'wav',
+            'audio/mpeg': 'mp3', 'audio/mp3': 'mp3', 'audio/flac': 'flac',
+            'audio/x-flac': 'flac', 'audio/m4a': 'm4a', 'audio/mp4': 'm4a',
+            'audio/x-m4a': 'm4a', 'audio/ogg': 'ogg', 'audio/webm': 'webm',
+            'video/webm': 'webm', 'audio/aac': 'aac'}
+        mime = audio.content_type.split(';', 1)[0].lower()
+        if mime not in formats:
+            raise ValueError('Unsupported audio MIME type for ZenMux ASR')
+        body = {'model': model, 'input_audio': {
+            'data': base64.b64encode(audio.file_content).decode(), 'format': formats[mime]},
+            'stream': False}
+        language = optional_params.get('language')
+        if language:
+            body['language'] = language
+        return AudioTranscriptionRequestData(data=body)
+
+    def transform_audio_transcription_response(self, raw_response):
+        body = raw_response.json()
+        if not isinstance(body.get('text'), str):
+            raise ValueError('ZenMux ASR returned no transcript text')
+        response = TranscriptionResponse(text=body['text'])
+        usage = body.get('usage') or {}
+        if usage.get('seconds') is not None:
+            # ZenMux has no audio/text token breakdown; report duration without inventing one.
+            response.usage = TranscriptionUsageDurationObject(type='duration', seconds=usage['seconds'])
+            response['duration'] = usage['seconds']
+        return response
+
+
+def install():
+    import litellm
+    from litellm.utils import ProviderConfigManager
+    original = ProviderConfigManager.get_provider_audio_transcription_config
+    if getattr(original, '_zenmux_asr_adapter', False):
+        return
+
+    def select(model, provider):
+        if provider == litellm.LlmProviders.OPENAI and model in MODELS:
+            return ZenMuxTranscriptionConfig()
+        return original(model, provider)
+
+    select._zenmux_asr_adapter = True
+    ProviderConfigManager.get_provider_audio_transcription_config = staticmethod(select)