# Copyright 2024 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Module for interacting with Agent Assist Backend using dialogflow_v2beta1 API version Reference: https://cloud.google.com/python/docs/reference/dialogflow/latest/google.cloud.dialogflow_v2beta1 """ import logging import re import time import google.auth import redis from google.api_core.client_options import ClientOptions from google.api_core.exceptions import FailedPrecondition, OutOfRange, ResourceExhausted from google.cloud import dialogflow_v2beta1 as dialogflow from audio_stream import Stream from audiohook_config import config # Wait for 2 units of 0.5 second for the redis client to set conversation name AWAIT_REDIS_COUNTER = 2 AWAIT_REDIS_SECOND_PER_COUNTER = 0.5 LOCATION_ID_REGEX = r"^projects\/[^/]+\/locations\/([^/]+)" credentials, project = google.auth.default() redis_client = redis.StrictRedis( host=config.redis_host, port=config.redis_port) try: location_id = re.match( LOCATION_ID_REGEX, config.conversation_profile_name)[1] except Exception as e: raise ValueError( "Conversation profile name is not in correct format") from e def determine_dialogflow_api_endpoint(location: str) -> str: """Get Dialogflow api endpoint Reference: https://cloud.google.com/dialogflow/es/docs/reference/rest/v2-overview#service-endpoint """ dialogflow_endpoint = "dialogflow.googleapis.com" if location != 'global': dialogflow_endpoint = f"{location}-dialogflow.googleapis.com" logging.debug("Dialogflow api endpoint %s", dialogflow_endpoint) return dialogflow_endpoint def create_conversation_name(conversation_id: str, location_id: str, project: str) -> str: """Set conversation name for the object """ return f"projects/{project}/locations/{location_id}/conversations/{conversation_id}" def find_participant_by_role(role: dialogflow.Participant.Role, participants_list: list[dialogflow.Participant]) -> dialogflow.Participant | None: for participant in participants_list: if participant.role == role: logging.debug("the active participant is %s:%s ", participant.role, participant.name) return participant return None class DialogflowAPI: """Class for interacting with the Dialogflow API """ def __init__(self) -> None: self.api_endpoint = determine_dialogflow_api_endpoint( location_id) self.participants_client = dialogflow.ParticipantsClient( credentials=credentials, client_options=ClientOptions( api_endpoint=self.api_endpoint) ) self.conversations_client = dialogflow.ConversationsClient( credentials=credentials, client_options=ClientOptions( api_endpoint=self.api_endpoint ) ) self.conversation_profiles_client = dialogflow.ConversationProfilesClient( credentials=credentials, client_options=ClientOptions( api_endpoint=self.api_endpoint)) def get_conversation_profile( self, conversation_profile_name: str) -> dialogflow.ConversationProfile: """Load conversation profile """ logging.debug("Getting conversation profile for %s ", conversation_profile_name) return self.conversation_profiles_client.get_conversation_profile( request=dialogflow.GetConversationProfileRequest( name=conversation_profile_name ) ) def create_conversation( self, conversation_profile: dialogflow.ConversationProfile, conversation_id: str, ) -> dialogflow.Conversation: """Create conversation using conversation_id """ conversation = dialogflow.Conversation( conversation_profile=conversation_profile.name ) project_path = self.conversations_client.common_location_path( project, location_id) conversation_request = dialogflow.CreateConversationRequest( parent=project_path, conversation=conversation, conversation_id=conversation_id, ) conversation = self.conversations_client.create_conversation( request=conversation_request) logging.info( "Created conversation %s for project path %s", conversation.name, project_path) return conversation def get_conversation( self, conversation_name: str) -> dialogflow.Conversation: """Get conversation using the conversation_name from dialogflow """ get_conversation_request = dialogflow.GetConversationRequest( name=conversation_name, ) conversation = self.conversations_client.get_conversation( request=get_conversation_request) return conversation def list_participant(self, conversation_name: str) -> list[dialogflow.Participant]: """List existing participant for Human agent and End user Args: conversation_name (str): conversation name Returns: _type_: _description_ """ participants_pagers = self.participants_client.list_participants( dialogflow.ListParticipantsRequest(parent=conversation_name)) participant_list = list(participants_pagers.__iter__()) logging.debug("participant list %s, type ", participant_list) return participant_list def create_participant( self, conversation_name: str, role: str): """Create both the agent and customer participant for the conversation Args: conversation_id (str): conversation id from the websocket message parameters """ participant = dialogflow.Participant() participant.role = role participant = self.participants_client.create_participant( parent=conversation_name, participant=participant) logging.debug("Creating new participant %s:%s", role, participant) return participant def maintained_streaming_analyze_content( self, audio_stream: Stream, participant: dialogflow.Participant, audio_config: dialogflow.InputAudioConfig): """While the stream is not closed or terminated, maintain a steady call to streaming analyze content API endpoint """ logging.debug("Call streaming analyze content %s, %s", audio_stream.closed, audio_stream.is_final) while not audio_stream.terminate: # while not audio_stream.is_final and not audio_stream.closed: while not audio_stream.closed: self.streaming_analyze_content( audio_stream, participant, audio_config) def streaming_analyze_content( self, audio_stream: Stream, participant: dialogflow.Participant, audio_config: dialogflow.InputAudioConfig): """Call dialogflow backend StreamingAnalyzeContent endpoint, and send the audio binary stream from Audiohook. """ try: logging.debug("call streaming analyze content for %s", participant) responses = self.participants_client.streaming_analyze_content( requests=self.generator_streaming_analyze_content_request( audio_config, participant, audio_stream)) except OutOfRange as e: audio_stream.closed = True logging.warning( "The single audio stream last more than 120 second %s ", e) return except FailedPrecondition as e: audio_stream.closed = True logging.warning( "Failed the precondition check for StreamingAnalyzeContent %s ", e) return except ResourceExhausted as e: audio_stream.closed = True logging.warning( "Exceed quota for calling streaming analyze content %s ", e) return for response in responses: audio_stream.speech_end_offset = response.recognition_result.speech_end_offset.seconds * 1000 logging.debug(response) if response.recognition_result.is_final: audio_stream.is_final = True logging.debug( "Final transcript for %s: %s, and is final offset", participant.role.name, response.recognition_result.transcript, ) offset = response.recognition_result.speech_end_offset audio_stream.is_final_offset = int( offset.seconds * 1000 + offset.microseconds / 1000 ) if response.recognition_result: logging.debug( "Role %s: Interim response recognition result transcript: %s, time %s", participant.role.name, response.recognition_result.transcript, response.recognition_result.speech_end_offset) def complete_conversation(self, conversation_name: str): """Send complete conversation request to Dialogflow """ self.conversations_client.complete_conversation( name=conversation_name ) logging.debug("Call complete conversation for %s", conversation_name) def generator_streaming_analyze_content_request( self, audio_config: dialogflow.InputAudioConfig, participant: dialogflow.Participant, audio_stream: Stream): """Generates and return an iterator for StreamingAnalyzeContentRequest, The first request should only include the input_audio_config And the following request contains the audio chunks as input_audio. The last request does not have any input_audio or input_audio_config and indicates that the server side is half-closing the streaming Args: audio_config (dialogflow.InputAudioConfig): Input for Speech Recognizer https://cloud.google.com/dialogflow/es/docs/reference/rest/v2beta1/InputAudioConfig participant (dialogflow.Participant): Participant for the Dialogflow API call audio_queue (asyncio.Queue): Queue to store the audio binary stream Yields: _type_: first filed the audio config, and then yield the binary data. """ # Sending audio_config for participant enable_debugging_info = config.log_level.upper() == "DEBUG" generator = audio_stream.generator() yield dialogflow.StreamingAnalyzeContentRequest( participant=participant.name, audio_config=audio_config, enable_debugging_info=enable_debugging_info, ) for content in generator: # next audio chunks to streaming_analyze_content yield dialogflow.StreamingAnalyzeContentRequest( input_audio=content, enable_debugging_info=enable_debugging_info, ) logging.info( "Participant: %s, streaming analyze content request, end streaming yield an empty request ", participant.name) yield dialogflow.StreamingAnalyzeContentRequest( enable_debugging_info=enable_debugging_info, ) logging.debug( "Ending the current audio stream session, start new session") def await_redis(conversation_name: str) -> bool: """Check if the redis memory store connection has been established or not """ conversation_name = determine_conversation_name_without_location( conversation_name) # give the user bit time to accept the call and wait for the Agent Assist Backend # to create the redis memory store counter = AWAIT_REDIS_COUNTER redis_exists = redis_client.exists(conversation_name) != 0 while not redis_exists and counter > 0: time.sleep(AWAIT_REDIS_SECOND_PER_COUNTER) redis_exists = redis_client.exists(conversation_name) != 0 counter = counter - 1 logging.debug("return to send resume message redis client exist %s and final counter %s ", redis_exists, counter) return redis_exists def determine_conversation_name_without_location(conversation_name: str): """Returns a conversation name without its location id.""" conversation_name_without_location = conversation_name if '/locations/' in conversation_name: name_array = conversation_name.split('/') conversation_name_without_location = '/'.join( name_array[i] for i in [0, 1, -2, -1]) return conversation_name_without_location