(
runtime: StreamingRuntime,
request: StreamingRequest,
)
| 325 | |
| 326 | |
| 327 | def build_processor_inputs( |
| 328 | runtime: StreamingRuntime, |
| 329 | request: StreamingRequest, |
| 330 | ) -> tuple[dict[str, torch.Tensor], Optional[torch.Tensor], str]: |
| 331 | processor = runtime.processor |
| 332 | mode = str(request.mode or "continuation").strip().lower() |
| 333 | if mode not in {"continuation", "voice_clone"}: |
| 334 | raise ValueError("mode must be continuation or voice_clone.") |
| 335 | |
| 336 | prompt_audio_path = (request.prompt_audio_path or "").strip() or None |
| 337 | prompt_text = request.prompt_text or "" |
| 338 | if mode == "voice_clone" and not prompt_audio_path: |
| 339 | mode = "continuation" |
| 340 | prompt_fields = _build_prompt_fields(request) |
| 341 | user_kwargs = dict( |
| 342 | text=request.text, |
| 343 | instruction=prompt_fields.get("instruction"), |
| 344 | tokens=prompt_fields.get("tokens"), |
| 345 | quality=prompt_fields.get("quality"), |
| 346 | sound_event=prompt_fields.get("sound_event"), |
| 347 | ambient_sound=prompt_fields.get("ambient_sound"), |
| 348 | language=prompt_fields.get("language"), |
| 349 | ) |
| 350 | |
| 351 | prompt_audio_codes: Optional[torch.Tensor] = None |
| 352 | if mode == "voice_clone": |
| 353 | conversation = [ |
| 354 | processor.build_user_message( |
| 355 | reference=[prompt_audio_path], |
| 356 | **user_kwargs, |
| 357 | ) |
| 358 | ] |
| 359 | processor_mode = "generation" |
| 360 | elif prompt_audio_path: |
| 361 | prompt_audio_codes = processor.encode_audios_from_path(prompt_audio_path, n_vq=runtime.n_vq)[0] |
| 362 | continuation_text = prompt_text + request.text if prompt_text.strip() else request.text |
| 363 | conversation = [ |
| 364 | processor.build_user_message( |
| 365 | text=continuation_text, |
| 366 | instruction=prompt_fields.get("instruction"), |
| 367 | tokens=prompt_fields.get("tokens"), |
| 368 | quality=prompt_fields.get("quality"), |
| 369 | sound_event=prompt_fields.get("sound_event"), |
| 370 | ambient_sound=prompt_fields.get("ambient_sound"), |
| 371 | language=prompt_fields.get("language"), |
| 372 | ), |
| 373 | processor.build_assistant_message(audio_codes_list=[prompt_audio_codes]), |
| 374 | ] |
| 375 | processor_mode = "continuation" |
| 376 | else: |
| 377 | # No-prompt continuation degenerates to direct TTS generation. This is |
| 378 | # trained for TACv5 and avoids forcing users to provide a reference. |
| 379 | conversation = [processor.build_user_message(**user_kwargs)] |
| 380 | processor_mode = "generation" |
| 381 | |
| 382 | batch = processor(conversation, mode=processor_mode, n_vq=runtime.n_vq) |
| 383 | return _move_batch_to_device(batch, runtime.tts_device), prompt_audio_codes, processor_mode |
| 384 |
no test coverage detected