[docs]classAzureCogsSpeech2TextTool(BaseTool):# type: ignore[override]"""Tool that queries the Azure Cognitive Services Speech2Text API. In order to set this up, follow instructions at: https://learn.microsoft.com/en-us/azure/cognitive-services/speech-service/get-started-speech-to-text?pivots=programming-language-python """azure_cogs_key:str=""#: :meta private:azure_cogs_region:str=""#: :meta private:speech_language:str="en-US"#: :meta private:speech_config:Any#: :meta private:name:str="azure_cognitive_services_speech2text"description:str=("A wrapper around Azure Cognitive Services Speech2Text. ""Useful for when you need to transcribe audio to text. ""Input should be a url to an audio file.")@model_validator(mode="before")@classmethoddefvalidate_environment(cls,values:Dict)->Any:"""Validate that api key and endpoint exists in environment."""azure_cogs_key=get_from_dict_or_env(values,"azure_cogs_key","AZURE_COGS_KEY")azure_cogs_region=get_from_dict_or_env(values,"azure_cogs_region","AZURE_COGS_REGION")try:importazure.cognitiveservices.speechasspeechsdkvalues["speech_config"]=speechsdk.SpeechConfig(subscription=azure_cogs_key,region=azure_cogs_region)exceptImportError:raiseImportError("azure-cognitiveservices-speech is not installed. ""Run `pip install azure-cognitiveservices-speech` to install.")returnvaluesdef_continuous_recognize(self,speech_recognizer:Any)->str:done=Falsetext=""defstop_cb(evt:Any)->None:"""callback that stop continuous recognition"""speech_recognizer.stop_continuous_recognition_async()nonlocaldonedone=Truedefretrieve_cb(evt:Any)->None:"""callback that retrieves the intermediate recognition results"""nonlocaltexttext+=evt.result.text# retrieve text on recognized eventsspeech_recognizer.recognized.connect(retrieve_cb)# stop continuous recognition on either session stopped or canceled eventsspeech_recognizer.session_stopped.connect(stop_cb)speech_recognizer.canceled.connect(stop_cb)# Start continuous speech recognitionspeech_recognizer.start_continuous_recognition_async()whilenotdone:time.sleep(0.5)returntextdef_speech2text(self,audio_path:str,speech_language:str)->str:try:importazure.cognitiveservices.speechasspeechsdkexceptImportError:passaudio_src_type=detect_file_src_type(audio_path)ifaudio_src_type=="local":audio_config=speechsdk.AudioConfig(filename=audio_path)elifaudio_src_type=="remote":tmp_audio_path=download_audio_from_url(audio_path)audio_config=speechsdk.AudioConfig(filename=tmp_audio_path)else:raiseValueError(f"Invalid audio path: {audio_path}")self.speech_config.speech_recognition_language=speech_languagespeech_recognizer=speechsdk.SpeechRecognizer(self.speech_config,audio_config)returnself._continuous_recognize(speech_recognizer)def_run(self,query:str,run_manager:Optional[CallbackManagerForToolRun]=None,)->str:"""Use the tool."""try:text=self._speech2text(query,self.speech_language)returntextexceptExceptionase:raiseRuntimeError(f"Error while running AzureCogsSpeech2TextTool: {e}")