U hhT@sDdZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z m Z m Z mZmZmZmZmZmZmZmZmZddlmZmZddlmZddlmZmZmZm Z m!Z!m"Z"ddl#m$Z$m%Z%e rddl&Z'ddl(m)m*m+Z,e-e.Z/d Z0d Z1d Z2d d ddZ3GdddZ4e4Z5e4Z6Gddde7e Z8GdddeZ9GdddeZ:GdddZ;ee7ede7ddd Z?edee7ddfd!d"d#Z@eeAe4fZBe7ZCGd$d%d%ZDeDZEe7ZFGd&d'd'e9e:e%eEeFfZGee7eefZHeAZIGd(d)d)e9e:e%eHeIfZJeGZKeJZLeDZMdS)*z*A common module for NVIDIA Riva Runnables.N)Enum) TYPE_CHECKINGAnyAsyncGenerator AsyncIteratorDict GeneratorIteratorListOptionalTupleUnioncast) AnyMessage BaseMessage) PromptValue) AnyHttpUrl BaseModelField parse_obj_asroot_validator validator)RunnableConfigRunnableSerializable?i) .!?¡¿z riva.clientreturnc Cs@z ddl}Wn,tk r8}ztd|W5d}~XYnX|jS)z5Import the riva client and raise an error on failure.rNziCould not import the NVIDIA Riva client library. Please install it with `pip install nvidia-riva-client`.) riva.client ImportErrorclient)rivaerrr(M/tmp/pip-unpacked-wheel-9gdii04g/langchain_community/utilities/nvidia_riva.py_import_riva_client1s r*c@seZdZdZdS) SentinelTzAn empty Sentinel type.N)__name__ __module__ __qualname____doc__r(r(r(r)r+>sr+c@sNeZdZdZdZdZdZdZdZdZ e e ddd d Z e d d d dZdS)RivaAudioEncodingaAn enum of the possible choices for Riva audio encoding. The list of types exposed by the Riva GRPC Protobuf files can be found with the following commands: ```python import riva.client print(riva.client.AudioEncoding.keys()) # noqa: T201 ``` ALAWENCODING_UNSPECIFIEDFLAC LINEAR_PCMMULAWOGGOPUS) format_coder"c CsPz|j|j|jd|WStk rJ}ztd||W5d}~XYnXdS)zReturn the audio encoding specified by the format code in the wave file. ref: https://mmsp.ece.mcgill.ca/Documents/AudioFormats/WAVE/WAVE.html )z>The following wave file format code is not supported by Riva: N)r4r1r5KeyErrorNotImplementedError)clsr7r'r(r(r)from_wave_format_codeXsz'RivaAudioEncoding.from_wave_format_codezriva.client.AudioEncodingr!cCst}t|j|S)z-Returns the Riva API object for the encoding.)r*getattrZ AudioEncodingself riva_clientr(r(r)riva_pb2fszRivaAudioEncoding.riva_pb2N)r,r-r.r/r1r2r3r4r5r6 classmethodintr>propertyrCr(r(r(r)r0Fs  r0c@seZdZUdZeeddddddgdZeeefe d<ed d d Z e ee d <e d dddZ eddddeeedddZd S) RivaAuthMixinzBConfiguration for the authentication to a Riva service connection.zhttp://localhost:50051http)schemez1The full URL where the Riva service can be found.z"https://user@pass:riva.example.com) descriptionZexamplesurlNz@A full path to the file where Riva's public ssl key can be read.rJssl_certzriva.client.Authr!cCsBt}tt|j}|jdk}t|jdd}|j|j||dS)z!Return a riva client auth object.https/)rMuse_ssluri) r*rrrKrIstrsplitZAuthrM)rArBrKrQZ url_no_schemer(r(r)authzs  zRivaAuthMixin.authT)preZ allow_reusevalr"cCs$t|trtttt|Stt|S)z:Do some initial conversations for the URL before checking.) isinstancerSrrr)r=rXr(r(r) _validate_urls zRivaAuthMixin._validate_url)r,r-r.r/rrrKr rS__annotations__rMr rFrUrrDrrZr(r(r(r)rGms   rGc@sPeZdZUdZeejddZeed<edddZ e ed<edd dZ e ed <d S) RivaCommonConfigMixinz%A collection of common Riva settings.z!The encoding on the audio stream.)defaultrJencodingi@z*The sample rate frequency of audio stream.sample_rate_hertzzen-USzaThe [BCP-47 language code](https://www.rfc-editor.org/rfc/bcp/bcp47.txt) for the target language. language_codeN) r,r-r.r/rr0r4r^r[r_rEr`rSr(r(r(r)r\s r\c@szeZdZUdZejed<ejed<ddddZdddd Z ddd d Z e dd d Z ddddZ ddddZdS)_Eventz3A combined event that is threadsafe and async safe._event_aeventNr!cCst|_t|_dS)zInitialize the event.N) threadingEventrbasynciorcrAr(r(r)__init__s z_Event.__init__cCs|j|jdSzSet the event.N)rbsetrcrgr(r(r)rjs z _Event.setcCs|j|jdSri)rbclearrcrgr(r(r)rks z _Event.clearcCs |jS)zIndicate if the event is set.)rbis_setrgr(r(r)rlsz _Event.is_setcCs|jdS)zWait for the event to be set.N)rbwaitrgr(r(r)rmsz _Event.waitcs|jIdHdS)z#Async wait for the event to be set.N)rcrmrgr(r(r) async_waitsz_Event.async_wait)r,r-r.r/rdrer[rfrhrjrkboolrlrmrnr(r(r(r)ras   ra)output_directory sample_rater"c Cs^|rZtjddd|d }|j}W5QRXt|d}|d|d||||fSdS) zECreate a new wave file and return the wave write object and filename.Zbxz.wavF)modesuffixdeletedirwbr8rP)NN)tempfileNamedTemporaryFilenamewaveopenZ setnchannelsZ setsampwidthZ setframerate)rprqf wav_file_namewav_filer(r(r) _mk_wave_files    r TTSInputTyperWcCs.t|tr|St|tr&t|jSt|S)zAttempt to coerce the input value to a string. This is particularly useful for converting LangChain message to strings. )rYrZ to_stringrrScontentrXr(r(r)_coerce_strings    r)inputsr"ccsd}|D]}t|}tD]0}||kr||d\}}|||Vd}qq||7}t|tkrtdt|tD]}|||dVqnd}q|r|VdS)z9Filter the input chunks are return strings ready for TTS.r8rN)r_SENTENCE_TERMINATORSrTlen_MAX_TEXT_LENGTHrange)rbufferchunk terminatorZ last_sentenceidxr(r(r)_process_chunkss rc@sJeZdZUdZejed<ejed<ejed<e ed<e ed<e ed<e ej ed<d)e d d d d Z eed d fdddZeedddZeedddZeedddZeedddZeedddZd*ee e d dddZd+ee e d dddZd,e e d d d!d"Zd-e e d d d#d$Zed%d d&d'd(Zd S). AudioStreamz%A message containing streaming audio. _put_lock_queueoutputhangup user_talking user_quiet_workerrN)maxsizer"cCsDt|_tj|d|_t|_t|_t|_ t|_ d|_ dS)zInitialize the queue.)rN) rdLockrqueueQueuerrrarrrr)rArr(r(r)rh s  zAudioStream.__init__r!ccsNz|jdt}Wntjk r,YqYnX|tkr8qJ|V|jqdS)zReturn an error.TN)rget_QUEUE_GET_TIMEOUTrEmptyHANGUP task_donerAZnext_valr(r(r)__iter__s zAudioStream.__iter__cCs^z"td|jjdtIdH}Wntjk r<YqYnX|tkrHqZ|V|j qdS)z4Iterate through all items in the queue until HANGUP.NT) rfget_event_looprun_in_executorrrrrrrrrr(r(r) __aiter__&s zAudioStream.__aiter__cCs |jS)z(Indicate if the audio stream has hungup.)rrlrgr(r(r)hungup9szAudioStream.hungupcCs |jS)z-Indicate in the input stream buffer is empty.)remptyrgr(r(r)r>szAudioStream.emptycCs4|jo |j}|jdk o*|j o*|j}|o2|S)z;Indicate if the audio stream has hungup and been processed.N)rrris_aliver)rAZ input_doneZ output_doner(r(r)completeCs   zAudioStream.completecCs|jr|jSdS)z&Indicate if the ASR stream is running.F)rrrgr(r(r)runningNs zAudioStream.running)itemtimeoutr"c CsF|j6|jrtd|tkr(|j|jj||dW5QRXdS)zPut a new item into the queue.z?The audio stream has already been hungup. Cannot put more data.rN)rr RuntimeErrorrrrjrput)rArrr(r(r)rUs zAudioStream.putcs*t}t|d|j||IdHdS)z$Async put a new item into the queue.N)rfrwait_forrr)rArrloopr(r(r)aput`szAudioStream.aput)rr"cCs|t|dS)zSend the hangup signal.N)rrrArr(r(r)closeeszAudioStream.closecs|t|IdHdS)zAsync send the hangup signal.N)rrrr(r(r)acloseiszAudioStream.aclosezrasr.StreamingRecognizeResponse) responsesr"cs^jrtdtjdddddfdd }tj|d _d j_jdS) zIDrain the responses from the provided iterator and put them into a queue.z,An ASR instance has already been registered.rPrrNr!csD]x}|jsq |jD]d}|js*q|jrdjjtt |jdj }j |qj sjjqq dS)zConsume the ASR Generator.rN)rmresultsZ alternativesis_finalrrkrrjrrS transcriptrrrl)responseresultrZ has_startedrrAr(r)workerts     z$AudioStream.register..worker)targetT) rrrdBarrierThreadrdaemonstartrm)rArrr(rr)registerms zAudioStream.register)r)N)N)N)N) r,r-r.r/rdrr[rrrar rrErhrbytesrrStreamInputTyperrFrorrrrrrrrr rr(r(r(r)rs0       rc@seZdZUdZdZeed<dZeed<edddZ e ed <ed d dZ e ed <ed d dZ e ed<ed deeeefeeefdddZeddddZddddZdeeeedddZdS)RivaASRzNA runnable that performs Automatic Speech Recognition (ASR) using NVIDIA Riva.Znvidia_riva_asrryzA Runnable for converting audio bytes to a string.This is useful for feeding an audio stream into a chain andpreprocessing that audio to create an LLM prompt.rJr8z7The number of audio channels in the input audio stream.rLaudio_channel_countTz\Controls whether or not Riva should attempt to filter profanity out of the transcribed text.profanity_filterz]Controls whether Riva should attempt to correct senetence puncuation in the transcribed text.enable_automatic_punctuationrVvaluesr"cCs t}|Sz4Validate the Python environment and input arguments.r*r=r_r(r(r)_validate_environmentszRivaASR._validate_environmentz&riva.client.StreamingRecognitionConfigr!c Cs4t}|jd|j|j|j|jd|j|j|jddS)z)Create and return the riva config object.Tr8)r^r_rZmax_alternativesrrr`)Zinterim_resultsconfig) r*ZStreamingRecognitionConfigZRecognitionConfigr^r_rrrr`r@r(r(r)rszRivaASR.configzriva.client.ASRServicec CsFt}z||jWStk r@}ztd|W5d}~XYnXdS);Connect to the riva service and return the a client object.z5Error raised while connecting to the Riva ASR server.N)r*Z ASRServicerU Exception ValueErrorrArBr'r(r(r) _get_serviceszRivaASR._get_serviceNinputrr"c Cs|js(|}|j||jd}||g}|js|jj|jjd}W5QRX|r,|j sz||j g7}Wnt j k rYqXYnX|j qXtdt|d|Sq,dS)z3Transcribe the audio bytes into a string with Riva.)Z audio_chunksZstreaming_configg?zRiva ASR returning: %s r)rrZstreaming_response_generatorrrrr not_emptyrmr get_nowaitrrr_LOGGERdebugreprjoinstrip)rArrservicerZ full_responsereadyr(r(r)invokes*     zRivaASR.invoke)N)r,r-r.r/ryrSr[rJrrrErrorrrDrrrrFrr ASRInputTyper r ASROutputTyperr(r(r(r)rs8   $ rc@seZdZUdZdZeed<dZeed<edddZ eed <ed d dZ e eed <e d de eeefeeefdddZed e eedddZddddZd!eeed fedddZd"eee ee eeedddZd#eee ee eeed fddd Zd S)$RivaTTSz?A runnable that performs Text-to-Speech (TTS) with NVIDIA Riva.Znvidia_riva_ttsryz_A tool for converting text to speech.This is useful for converting LLM output into audio bytes.rJzEnglish-US.Female-1zThe voice model in Riva to use for speech. Pre-trained models are documented in [the Riva documentation](https://docs.nvidia.com/deeplearning/riva/user-guide/docs/tts/tts-overview.html).rL voice_nameNzThe directory where all audio files should be saved. A null value indicates that wave files should not be saved. This is useful for debugging purposes.rpTrrcCs t}|Srrrr(r(r)rszRivaTTS._validate_environment)vr"cCs,|r(t|}|jdddt|S|S)NT)parentsexist_ok)pathlibPathmkdirrSabsolute)r=rdirpathr(r(r)_output_directory_validator&s   z#RivaTTS._output_directory_validatorz"riva.client.SpeechSynthesisServicer!c CsFt}z||jWStk r@}ztd|W5d}~XYnXdS)rz5Error raised while connecting to the Riva TTS server.N)r*ZSpeechSynthesisServicerUrrrr(r(r)r/szRivaTTS._get_servicercCsd|t|gS)zDPerform TTS by taking a string and outputting the entire audio file.)r transformiter)rArrr(r(r)r9szRivaTTS.invoke)rrkwargsr"c ks|}t|j|j\}}t|D]X}td||j||j|j |j j |jd}|D]$} t t | j} |rr|| | VqTq"|r|td|dS)zHPerform TTS by taking a stream of characters and streaming output bytes.zRiva TTS chunk: %s)textrr`r^Zsample_rate_hzzRiva TTS wrote file: %sN)rrrpr_rrrZsynthesize_onlinerr`r^rCrraudioZwriteframesrawr) rArrrrr}r~rrresprr(r(r)r?s,      zRivaTTS.transformc  stttddfdd }ttdfdd ddfdd ddfd d }|}|}ztd IdH}Wntj j k rYqYnX |t krq|Vq|IdH|IdHdS) zGIntercept async transforms and route them to the synchronous transform.Nr!cs,2z3dHW}|q6tdS)z#Produce input into the input queue.N) put_nowait_TRANSFORM_ENDr)r input_queuer(r) _producerqsz%RivaTTS.atransform.._producerc3sBzjdd}Wntjk r*YqYnX|tkr6q>|VqdS)zIterate over the input_queue.rrN)rrrrr)rr(r)_input_iteratorws z+RivaTTS.atransform.._input_iteratorcs*D]}|q tdS)z!Consume the input with transform.N)rrrr)r out_queuerAr(r) _consumers z%RivaTTS.atransform.._consumercsdIdHdS)z"Coroutine that wraps the consumer.N)rr()rrr(r)_consumer_corosz*RivaTTS.atransform.._consumer_coror) rfZget_running_looprrr rZ create_taskrr exceptions TimeoutErrorrr) rArrrrrZproducerZconsumerrXr()rrrrrrrAr) atransformfs&     zRivaTTS.atransform)N)N)N)r,r-r.r/ryrSr[rJrrrpr rrDrrrrrrrr r TTSOutputTyperr rrrrr(r(r(r)rsN     $    * r)Nr/rfloggingrrrwrdrzenumrtypingrrrrrrr r r r r rZlangchain_core.messagesrrZlangchain_core.prompt_valuesrZlangchain_core.pydantic_v1rrrrrrZlangchain_core.runnablesrrr#r&Zriva.client.proto.riva_asr_pb2r%protoZ riva_asr_pb2Zrasr getLoggerr,rrrrr*r+rrrSr0rGr\rafloatZ Wave_writerrrrrZStreamOutputTyperrrrrrrZ NVIDIARivaASRZ NVIDIARivaTTSZNVIDIARivaStreamr(r(r(r)sp 8    '!#     g  !