{"record":{"id":"7bd164fa64e70550","repo":"microsoft/VibeVoice","slug":"vibevoicestreamingprocessor-call-is-not-implem","errorCode":null,"errorMessage":"VibeVoiceStreamingProcessor.__call__ is not implemented. Use process_input_with_cached_prompt for streaming inputs.","messagePattern":"VibeVoiceStreamingProcessor\\.__call__ is not implemented\\. Use process_input_with_cached_prompt for streaming inputs\\.","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"vibevoice/processor/vibevoice_streaming_processor.py","lineNumber":165,"sourceCode":"                \"normalize_audio\": getattr(self.audio_processor, 'normalize_audio', True),\n                \"target_dB_FS\": getattr(self.audio_processor, 'target_dB_FS', -25),\n                \"eps\": getattr(self.audio_processor, 'eps', 1e-6),\n            }\n        }\n        \n        config_path = os.path.join(save_directory, \"preprocessor_config.json\")\n        with open(config_path, 'w') as f:\n            json.dump(processor_config, f, indent=2)\n        \n        logger.info(f\"Processor configuration saved in {config_path}\")\n    \n    def __call__(self) -> BatchEncoding:\n        \"\"\"\n        Note:\n            This method is intentionally not implemented in the streaming processor.\n            Use `process_input_with_cached_prompt` for streaming use cases.\n        \"\"\"\n        raise NotImplementedError(\n            \"VibeVoiceStreamingProcessor.__call__ is not implemented. \"\n            \"Use process_input_with_cached_prompt for streaming inputs.\"\n        )\n\n    def process_input_with_cached_prompt(\n        self,\n        text: Optional[str] = None,\n        cached_prompt: Optional[Dict[str, Any]] = None,\n        padding: Union[bool, str, PaddingStrategy] = True,\n        truncation: Union[bool, str, TruncationStrategy] = False,\n        max_length: Optional[int] = None,\n        return_tensors: Optional[Union[str, TensorType]] = None,\n        return_attention_mask: bool = True,\n        **kwargs,\n    ) -> BatchEncoding:\n        \"\"\"\n        Main method to process one text script based on cached prompt. The function currently only supports single examples.\n","sourceCodeStart":147,"sourceCodeEnd":183,"githubUrl":"https://github.com/microsoft/VibeVoice/blob/94da20d98b2fa7688e9cbfaf7692ddb4954f7600/vibevoice/processor/vibevoice_streaming_processor.py#L147-L183","documentation":"VibeVoiceStreamingProcessor deliberately overrides ProcessorMixin.__call__ to raise NotImplementedError. Streaming inference maintains a cached prompt/kv state between chunks, so the stateless one-shot __call__ contract cannot be honored correctly; callers are redirected to process_input_with_cached_prompt. This is an API-design decision, not a bug.","triggerScenarios":"Calling the processor instance directly — processor(text=..., audio=...) — after loading VibeVoiceStreamingProcessor, usually because generic example code (written for VibeVoiceProcessor) was reused verbatim.","commonSituations":"Copy-pasting quickstart snippets that use the non-streaming processor; frameworks like inference servers that introspect and invoke __call__ uniformly across processor types.","solutions":["Use process_input_with_cached_prompt(text, cached_prompt, ...) for every streaming chunk, threading its returned cached_prompt into the next call.","If you do not need streaming, switch to VibeVoiceProcessor, whose __call__ works normally.","Wrap the streaming processor behind your own facade so generic callers never hit __call__."],"exampleFix":"# before\nenc = processor(text=prompt, audio=voice)  # NotImplementedError\n\n# after\nenc, cached = None, None\nresult = processor.process_input_with_cached_prompt(text=prompt, cached_prompt=None)\n# next chunk:\nresult = processor.process_input_with_cached_prompt(text=next_text,\n    cached_prompt=result.cached_prompt)","handlingStrategy":"validation","validationCode":"from vibevoice.processor import VibeVoiceStreamingProcessor\n# route by capability, not by name\nif isinstance(processor, VibeVoiceStreamingProcessor):\n    result = processor.process_input_with_cached_prompt(text=text, cached_prompt=cached)\nelse:\n    result = processor(text=text)","typeGuard":"def is_streaming_processor(p) -> bool:\n    from vibevoice.processor import VibeVoiceStreamingProcessor\n    return isinstance(p, VibeVoiceStreamingProcessor)","tryCatchPattern":"try:\n    enc = processor(text=text)\nexcept NotImplementedError as e:\n    if 'process_input_with_cached_prompt' in str(e):\n        enc = processor.process_input_with_cached_prompt(text=text, cached_prompt=None)\n    else:\n        raise","preventionTips":["Call process_input_with_cached_prompt explicitly for streaming processors; never __call__.","Use VibeVoiceProcessor when you want one-shot __call__ semantics.","Gate generic inference-server code on processor type before invoking."],"tags":["streaming","api-misuse","not-implemented","processor"],"backgroundTag":null,"analyzedSha":"94da20d98b2fa7688e9cbfaf7692ddb4954f7600","analyzedAt":"2026-08-15T04:12:07.418Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}