Real-time Fallacy Detection using OpenAI whisper and ChatGPT/LLaMA/Mistral
import wave
import pyaudio
import configparser

# Audio format Parameters for whisper
FORMAT = pyaudio.paInt16  # 16-bit depth
CHANNELS = 1  # Mono
RATE = 16000  # 16kHz
CHUNK = 1024  # Smaller chunks might reduce latency
RECORD_SECONDS = 30  # 30 seconds to match Whisper's expected segment length

def print_input_devices():
    p = pyaudio.PyAudio()
    info = p.get_host_api_info_by_index(0)
    numdevices = info.get('deviceCount')
    print("Input devices:")
    for i in range(0, numdevices):
        if (p.get_device_info_by_host_api_device_index(0, i).get('maxInputChannels')) > 0:
            print("Input Device ID ", i, " - ", p.get_device_info_by_host_api_device_index(0, i).get('name'))

def print_output_devices():
    # Initialize PyAudio
    p = pyaudio.PyAudio()
    # Get total number of devices
    num_devices = p.get_device_count()
    # Loop through and print device info
    print("Output devices:")
    for i in range(num_devices):
        device_info = p.get_device_info_by_index(i)
        if device_info['maxOutputChannels'] > 0:  # Output device
            print(f"Index: {i}, Name: {device_info['name']}, Channels: {device_info['maxOutputChannels']}")

    # Terminate PyAudio
    p.terminate()


def find_output_device_index():
    
    # Initialize ConfigParser
    config = configparser.ConfigParser()

    # Read settings.ini file
    config.read('settings.ini')
    # Get values
    device_output_name = config.get('Audio Settings', 'device_output_name')

    # Initialize PyAudio
    p = pyaudio.PyAudio()
    # Get total number of devices
    num_devices = p.get_device_count()
    # Loop through and print device info
    for i in range(num_devices):
        device_info = p.get_device_info_by_index(i)
        device_name = device_info.get('name')

        # Search for VB-Audio in the device name
        if device_output_name in device_name:
            return i
   
    return None

def find_input_device_index():
    
    # Initialize ConfigParser
    config = configparser.ConfigParser()

    # Read settings.ini file
    config.read('settings.ini')
    # Get values
    device_input_name = config.get('Audio Settings', 'device_input_name')
    p = pyaudio.PyAudio()
    info = p.get_host_api_info_by_index(0)
    numdevices = info.get('deviceCount')
    
    for i in range(0, numdevices):
        device_info = p.get_device_info_by_host_api_device_index(0, i)
        device_name = device_info.get('name')
        
        # Search for VB-Audio in the device name
        if device_input_name in device_name:
            return i
    
    # Return None if not found
    return None

def record_short_audio():

    # Change this to the index of the device you want to use
    # Default recoding devie uses vb audio 
    input_device_index = find_input_device_index()

    # Initialize PyAudio and start recording
    audio = pyaudio.PyAudio()
    stream = audio.open(format=FORMAT, channels=CHANNELS,
                        rate=RATE, input=True,
                        frames_per_buffer=CHUNK, input_device_index=input_device_index)

    print("Recording...")
    frames = []

    for _ in range(0, int(RATE / CHUNK * RECORD_SECONDS)):
        data = stream.read(CHUNK)
        frames.append(data)

    print("Finished recording")

    # Stop the stream and terminate PyAudio
    stream.stop_stream()
    stream.close()
    audio.terminate()

    # Save the audio
    save_audio_frames(audio, frames)


def save_audio_frames(audio, frames):
    WAVE_OUTPUT_FILENAME = "audio.wav"
    # Save the recording as a WAV file
    wf = wave.open(WAVE_OUTPUT_FILENAME, 'wb')
    wf.setnchannels(CHANNELS)
    wf.setsampwidth(audio.get_sample_size(FORMAT))
    wf.setframerate(RATE)
    wf.writeframes(b''.join(frames))
    wf.close()

    print(f"Audio saved as {WAVE_OUTPUT_FILENAME}")


if __name__=='__main__':
    record_short_audio()
    # print_input_devices()
    # print_output_devices()
audio.py
latent-variable/Real_time_fallacy_detection
import json
import openai  # Assuming you've installed the OpenAI Python package
from local_llm import local_llm_call
from prompt import get_prompt, INSTRUCTION

# Function to read API key from a file
def read_api_key(file_path = r'./api_key.txt'):
    with open(file_path, 'r') as file:
        return file.read().strip()
    
# Initialize GPT-3 API Client
openai.api_key = read_api_key()
LAST_RESPONCE = ''

def save_to_json_file(transcript, instruction, response, file_name=r'Data/data.json'):
    # Prepare new data entry
    
    new_data = {
        "transcript":transcript,
        "prompt": instruction,
        "response": response
    }
    # Read existing data
    try:
        with open(file_name, "r") as json_file:
            all_data = json.load(json_file)
    except FileNotFoundError:
        all_data = []
    
    # Append new data
    all_data.append(new_data)

    # Write updated data back to file
    with open(file_name, "w") as json_file:
        json.dump(all_data, json_file)


def fallacy_classification(transcript, use_gpt = False):
    global LAST_RESPONCE
    # Create a more nuanced prompt for the LLM
    prompt = get_prompt(transcript, LAST_RESPONCE)

    # Call ChatGPT API for fallacy detection or other tasks
    if not use_gpt:
        print('Prompting Local LLM')
        llm_response  = local_llm_call(prompt)
        # Catch model answer collapse 
        if 'ß' in llm_response: 
            llm_response = ''
        # print(f'{prompt}\n\n {llm_response}')
    else:
        print(f'Prompting {use_gpt}')
        response = openai.ChatCompletion.create(
            model= use_gpt,
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": prompt}
            ],
        )
    
        llm_response  = response['choices'][0]['message']['content'].strip() # Extract and clean the output text
         # print("GPT-3 Output:", llm_response)

    # Store the last response data
    LAST_RESPONCE = f'Last Debate Excerpt: {transcript}\n {llm_response}'

    if use_gpt:
        save_to_json_file(transcript, INSTRUCTION, llm_response)

    return llm_response




    
if __name__=="__main__":
    save_to_json_file("transcript", "prompt", "response",  "few_shot_examples")
gpt.py
latent-variable/Real_time_fallacy_detection
import requests
import configparser
# For local streaming, the websockets are hosted without ssl - http://
HOST = 'localhost:5000'
URI = f'http://{HOST}/api/v1/generate'

# For reverse-proxied streaming, the remote will likely host with ssl - https://
# URI = 'https://your-uri-here.trycloudflare.com/api/v1/generate'


def local_llm_call(prompt):
     
    # Initialize ConfigParser
    config = configparser.ConfigParser()

    # Read settings.ini file
    config.read('settings.ini')
    # Get values
    instruction_template = config.get('LLM Settings', 'instruction_template')


    request = {
        'prompt': prompt,
        'max_new_tokens': 250,
        'auto_max_new_tokens': False,
        'max_tokens_second': 0,

        # Generation params. If 'preset' is set to different than 'None', the values
        # in presets/preset-name.yaml are used instead of the individual numbers.
        'preset': 'Divine Intellect',
        'instruction_template': instruction_template,
        
        'do_sample': True,
        'temperature': 0.7,
        'top_p': 0.1,
        'typical_p': 1,
        'epsilon_cutoff': 0,  # In units of 1e-4
        'eta_cutoff': 0,  # In units of 1e-4
        'tfs': 1,
        'top_a': 0,
        'repetition_penalty': 1.18,
        'repetition_penalty_range': 0,
        'top_k': 40,
        'min_length': 0,
        'no_repeat_ngram_size': 0,
        'num_beams': 1,
        'penalty_alpha': 0,
        'length_penalty': 1,
        'early_stopping': False,
        'mirostat_mode': 0,
        'mirostat_tau': 5,
        'mirostat_eta': 0.1,
        'grammar_string': '',
        'guidance_scale': 1,
        'negative_prompt': '',

        'seed': -1,
        'add_bos_token': True,
        'truncation_length': 2048,
        'ban_eos_token': False,
        'custom_token_bans': '',
        'skip_special_tokens': True,
        'stopping_strings': []
    }

    response = requests.post(URI, json=request)

    if response.status_code == 200:
        # print(response.json())
        result = response.json()['results'][0]['text'].strip()
        # print('\n\n***************************************************')
        # print( prompt + result)
        return result

if __name__ == '__main__':
    prompt = "In order to make homemade bread, follow these steps:\n1)"
    local_llm_call(prompt)
local_llm.py
latent-variable/Real_time_fallacy_detection

import argparse
from whs import load_model
from overlay import launch_overlay


def get_inputs():
    parser = argparse.ArgumentParser(description='Toggle between Local LLM and ChatGPT.')
    parser.add_argument('--use_gpt4', action='store_true', default=False, help='Will use use ChatGPT with GPT-4 otherwise, by defualt it use a local LLM thorugh the text-generation-webui API')
    parser.add_argument('--use_gpt3', action='store_true', default=False, help='Will use use ChatGPT with GPT-3-Turbo otherwise, by defualt it use a local LLM thorugh the text-generation-webui API')
    args = parser.parse_args()
    use_gpt = False

 
    if args.use_gpt4:
        use_gpt = 'gpt-4'
    if args.use_gpt3:
        use_gpt = 'gpt-3.5-turbo'
    
    return use_gpt

def get_whisper_model():
   
    whs_model = load_model()
   
    return whs_model

def main(whs_model, use_gpt):

    # Launch the GUI in the main thread
    launch_overlay(whs_model, use_gpt )
    

if __name__ == "__main__":

    use_gpt =  get_inputs()
    whs_model = get_whisper_model()
    main(whs_model, use_gpt)





main.py
latent-variable/Real_time_fallacy_detection
import sys
import threading
from PyQt5.QtCore import Qt, QTimer, QPoint
from PyQt5.QtWidgets import QApplication, QLabel, QMainWindow, QVBoxLayout, QWidget, QGridLayout, QScrollArea, QSizeGrip


from real_time_classifier import continuous_audio_transcription_and_classification
from real_time_classifier import WHISPER_TEXTS, GPT_TEXTS, MAX_SEGEMENTS

class TransparentOverlay(QMainWindow):
    
    def __init__(self, whs_model, use_gpt):
        super().__init__()

        self.whs_model = whs_model
        self.use_gpt = use_gpt
        self.dragPos = QPoint()
        

        self.initUI()
        

    def initUI(self):
        self.setWindowTitle('Transparent Overlay')
        self.setGeometry(0, 0, 1000, 600)
        self.setWindowOpacity(0.6)
        self.setAttribute(Qt.WA_TranslucentBackground)
        self.setWindowFlags(Qt.Window | Qt.FramelessWindowHint | Qt.WindowStaysOnTopHint)

        # Create Scroll Areas
        self.scroll_area1 = QScrollArea(self)
        self.scroll_area2 = QScrollArea(self)

        # Increase the dimensions of the scroll areas
        self.scroll_area1.setMinimumSize(380, 120)
        self.scroll_area2.setMinimumSize(380, 120)
        
        # Create Labels
        self.whisper_label = QLabel('Whisper Output Will Appear Here')
        self.chatgpt_label = QLabel('ChatGPT Output Will Appear Here')

        # Enable word-wrap on labels
        self.whisper_label.setWordWrap(True)
        self.chatgpt_label.setWordWrap(True)

        # Add labels to Scroll Areas
        self.scroll_area1.setWidget(self.whisper_label)
        self.scroll_area2.setWidget(self.chatgpt_label)

        # Enable scroll bars on the scroll areas
        self.scroll_area1.setWidgetResizable(True)
        self.scroll_area2.setWidgetResizable(True)

        
      

        # Style labels with bold text and increased font size
        self.whisper_label.setStyleSheet('background-color: lightblue; font-weight: bold; font-size: 16px;')
        self.chatgpt_label.setStyleSheet('background-color: lightgreen; font-weight: bold; font-size: 16px;')
       
        # Layout setup
        # QVBoxLayout for the scroll areas
        vbox_layout = QVBoxLayout()
        vbox_layout.addWidget(self.scroll_area2)
        vbox_layout.addWidget(self.scroll_area1)
        
        # QGridLayout to include QVBoxLayout and QSizeGrip
        grid_layout = QGridLayout()
        grid_layout.addLayout(vbox_layout, 0, 0)
        
        # Add QSizeGrip to the QGridLayout
        size_grip = QSizeGrip(self)
        grid_layout.addWidget(size_grip, 1, 1, Qt.AlignBottom | Qt.AlignRight)

        container = QWidget()
        container.setLayout(grid_layout)
        self.setCentralWidget(container)

        # Run the continuous transcription and classification in a separate thread
        self.stop_event = threading.Event()
        self.transcription_thread = threading.Thread(target=continuous_audio_transcription_and_classification, args=(self.whs_model, self.use_gpt, self.stop_event))
        self.transcription_thread.start()

        

        # Timer to update Whisper and ChatGPT outputs
        self.timer = QTimer(self)
        self.timer.timeout.connect(self.update_labels)
        self.timer.start(500)
    

    def update_labels(self):
        # get_whisper_transcription returns a list of text segments, newest last.
        whisper_segments = get_whisper_transcription()

        # Concatenate the segments and set the label text.
        self.whisper_label.setText("Transcript: " + '- '.join(whisper_segments))

        # Color old response grey new reponse black
        chatgpt_output_list = get_chatgpt_output()
        chatgpt_text = "\n\nLast: ".join(chatgpt_output_list)
        self.chatgpt_label.setText(f"{chatgpt_text}")

        self.whisper_label.setMouseTracking(True)
        self.chatgpt_label.setMouseTracking(True)
        self.scroll_area1.setMouseTracking(True)
        self.scroll_area2.setMouseTracking(True)
        
    
    def mousePressEvent(self, event):
        self.dragPos = event.globalPos()
     

    def mouseMoveEvent(self, event):
        if event.buttons() == Qt.LeftButton:
            self.move(self.pos() + event.globalPos() - self.dragPos)
            self.dragPos = event.globalPos()


    def keyPressEvent(self, event):
        global TRANSCRIBE
        if event.key() == Qt.Key_Escape:
            # To stop the thread
            self.stop_event.set()
            self.transcription_thread.join()  # Optional: Wait for the thread to finish
            self.close()
            

def get_whisper_transcription():
    global WHISPER_TEXTS
    last_n_segments = WHISPER_TEXTS[-9:]  # Assuming you want the last 10 segments
    # text = ' - '.join(last_n_segments)
    return last_n_segments

def get_chatgpt_output():
    global GPT_TEXTS
    if len(GPT_TEXTS) >= 2:
        last_two = GPT_TEXTS[-1:-3:-1]  # Get and reverse the last two strings
    else:
        last_two = [GPT_TEXTS[-1]] if GPT_TEXTS else []
    
    return last_two

def launch_overlay(whs_model, use_gpt):
    app = QApplication(sys.argv)
    overlay = TransparentOverlay(whs_model, use_gpt)
    overlay.show()
    sys.exit(app.exec_())

if __name__ == '__main__':
    launch_overlay()

overlay.py
latent-variable/Real_time_fallacy_detection
import json
from random import shuffle
 
# INSTRUCTION = """Note: The following debate transcript may contain imperfections, such as multiple people talking over each other and question being asked by indivudals or moderator. Please take this into consideration when analyzing the text.
# Analyze the following debate excerpt for logical fallacies and format your response by starting with the name of the fallacy, followed by a brief justification.
# If the statement is factual or logical, or if no argument is present, state so explicitly. Avoid covering the same points from the Last Debate Excerpt.

# Debate Excerpt: 
# """

INSTRUCTION = """Note: The following debate transcript may contain imperfections, such as multiple people talking over each other and questions being asked by individuals or the moderator. Please take this into consideration when analyzing the text.
Analyze the following debate excerpt for logical fallacies based on the pragma-dialectical framework:

1.Argumentum ad Hominem (Personal Attack)
- Identify this when the speaker attacks the character, motive, or other attributes of the person making the argument, rather than attacking the substance of the argument itself.

2. Argumentum ad Populum (Appeal to Popular Opinion)
- Look for claims that are justified solely because "everyone else believes it" or "it is popular."

3. Argumentum ad Ignorantiam (Appeal to Ignorance)
- Spot this fallacy when the argument asserts that a proposition is true because it has not been proven false, or vice versa.

4. Argumentum ad Misericordiam (Appeal to Pity)
- Identify when emotional appeals like sympathy, pity, or fear are used instead of logical reasons to persuade the audience.

5. Argumentum ad Baculum (Appeal to Force)
- Notice when threats or force are used to win an argument, rather than logic or evidence.

6. Slippery Slope
- Look out for claims that one event will inevitably follow from another without adequate evidence to support such a claim.

7. False Dichotomy
- Identify when only two choices are presented as the only options, while in reality, more options exist.

8. Begging the Question (Circular Reasoning)

- Recognize this when the conclusion is already assumed in the premises, essentially forming a circle in reasoning.

9. Straw Man Fallacy
- Look for instances where an argument is misrepresented to make it easier to attack.

10. Red Herring
- Identify when the argument goes off on a tangent, providing irrelevant information to distract from the main issue.

11. Hasty Generalization
- Notice when conclusions are drawn based on insufficient or unrepresentative evidence.

12. Equivocation
- Look for ambiguous language that can be interpreted in more than one way, used intentionally to mislead or confuse.

Justification for classification of the statement as that fallacy, claim that is impacted by the fallacy, severity of the damage done to the claim by presence of that fallacy with column labels: name, justification, claim, damage. If the statement is factual or logical, or if no argument is present, state so explicitly. Avoid covering the same points from the Last Debate Excerpt.

Debate Excerpt:
"""


def get_few_shot_examples(file_name=r'Data/data.json', N=5):
    try:
        with open(file_name, "r") as json_file:
            all_data = json.load(json_file)
    except FileNotFoundError:
        return "No data available."
    
    shuffle(all_data)
    # Get last N examples from the JSON data
    last_n_examples = all_data[-N:]
    
    # Generate the few-shot string
    few_shot_string = ""
    for example in last_n_examples:
        few_shot_string += f"Debate Excerpt: \"{example['transcript']}\"\n"
        few_shot_string += f"{example['response']}\n\n"
    
    return few_shot_string


def get_prompt(transcript, last_responce):
    few_shot = get_few_shot_examples(file_name=r'Data/data.json', N=5)
    
    prompt = f'{few_shot}\n{last_responce}\n\n{INSTRUCTION}\n\"{transcript}\"'
    return prompt

if __name__=="__main__":
    pass
    # update_json_file( file_name=r'Data/data.json')
prompt.py
latent-variable/Real_time_fallacy_detection
import wave
import time
import torch 
import pyaudio
import threading
import numpy as np

from audio import find_input_device_index, find_output_device_index
from audio import FORMAT, CHANNELS, RATE, CHUNK
from gpt import fallacy_classification

# Global variables 
WHISPER_TEXTS = []
GPT_TEXTS = []
MAX_SEGEMENTS = 6
audio_buffer_lock = threading.Lock()

def save_byte_buffer(audio, audio_buffer, byte_size_chunk):
    # Save to a WAV file for verification
    with wave.open("audio_chunk.wav", "wb") as wav_file:
        wav_file.setnchannels(CHANNELS)
        wav_file.setsampwidth(audio.get_sample_size(FORMAT))
        wav_file.setframerate(RATE)
        wav_file.writeframes(audio_buffer[:byte_size_chunk])


def get_audio_tensor(audio_buffer):
    audio_buffer_np = np.frombuffer(audio_buffer, dtype=np.int16)  
    # Convert to float32
    audio_float32 = audio_buffer_np.astype(np.float32)
    # Normalize
    audio_normalized = audio_float32 / 32768.0
    # Convert to Torch tensor
    audio_tensor = torch.from_numpy(audio_normalized)
    # Optionally move to GPU
    if torch.cuda.is_available():
        audio_tensor = audio_tensor.cuda()

    return audio_tensor

def get_last_segments(text_list):
    # Get the last N segments from the list
    last_n_segments = text_list[-MAX_SEGEMENTS:]
     # Combine them into a single string
    combined_text = "".join(last_n_segments)
    return combined_text


def clean_transcript(text):
    cleaned_text = text.strip()
    # The following are whisper transcriptions
    # that happend in complete silence,
    # so I am filtering them out
    if cleaned_text == 'you':
        return ''
    if cleaned_text == 'Thank you.':
        return ''
    if cleaned_text == '.':
        return ''
    return cleaned_text

def transcription_callback(new_text, use_gpt):
    global WHISPER_TEXTS, GPT_TEXTS  # Declare the list as global so we can append to it

    cleaned_text = clean_transcript(new_text)
    if len(cleaned_text):
        print('Adding trancription')
        WHISPER_TEXTS.append(cleaned_text + '\n')

        text = get_last_segments( WHISPER_TEXTS )
        # print("\nTranscribed Text:", text)
    
        # Call chatgpt for fallacy classification
        if len(WHISPER_TEXTS) % 4 == 0:
            GPT_TEXTS.append(fallacy_classification(text, use_gpt))


def transcription(whs_model,  audio_tensor, use_gpt, callback):
    # Model call to transcribe the data
    try:
    
        transcription = whs_model.transcribe(audio_tensor, language='en', fp16=torch.cuda.is_available())
        # Process the transcription (e.g., fallacy detection)
        callback(transcription['text'], use_gpt)
    except Exception as e:
        print('Skipping transcription ..')

def is_silence(audio_data, threshold=100):
    """Check if the given audio data represents silence."""
    
    if np.abs(audio_data).mean() < threshold:
        
        return True
    return False

def continuous_audio_transcription_and_classification(whs_model, use_gpt, stop_event):
    print('Continuous audio transcription...')
    # Initialize PyAudio and open a stream
    audio = pyaudio.PyAudio()
    input_device_index = find_input_device_index()
    output_device_index = find_output_device_index()
     # Input Stream
    input_stream = audio.open(format=FORMAT, channels=CHANNELS, rate=RATE, 
                              input=True, frames_per_buffer=CHUNK,
                              input_device_index=input_device_index)

    # Output Stream
    output_stream = audio.open(format=FORMAT, channels=CHANNELS, rate=RATE,
                               output=True, frames_per_buffer=CHUNK,
                               output_device_index=output_device_index)

    # Initialize empty audio buffer
    audio_buffer = bytearray()
    seconds = 10 
    data_byte_size = 32_000 # 1 second of data 
    byte_size_chunk = data_byte_size * seconds 
    silence_counter = 0 
    while not stop_event.is_set():  # Assuming you have a stopping condition
        
        # Read audio chunk
        audio_chunk = input_stream.read(CHUNK)
        # Check if the chunk is silence
        audio_data = np.frombuffer(audio_chunk, dtype=np.int16)
        if is_silence(audio_data):
            silence_counter += 1
        else:
            silence_counter = 0

        # Write audio chunk to output device
        output_stream.write(audio_chunk)

        # Append to audio buffer
        with audio_buffer_lock:
            audio_buffer += audio_chunk
        
        #print('silence_counter', silence_counter)
        # When buffer reaches a certain size, or a silence break is detected send for transcription
        if silence_counter > 5: 
                if (len(audio_buffer) >=  data_byte_size * 3  ):
                    
                    silence_counter = 0  # Reset the counter
                    # print('len(audio_buffer)', len(audio_buffer))
                    # Create a new thread for transcription
                    # save_byte_buffer(audio, audio_buffer, byte_size_chunk)
                    with audio_buffer_lock:
                        # Change dtype based on your bit depth
                        audio_tensor = get_audio_tensor(audio_buffer)
                    transcribe_thread = threading.Thread(target=transcription, args=(whs_model, audio_tensor, use_gpt, transcription_callback))
                    transcribe_thread.start()
                    with audio_buffer_lock:
                        audio_buffer = bytearray()  # Clear the buffer
        else:
            with audio_buffer_lock:
                # Append to audio buffer
                audio_buffer += audio_chunk

            
    print('Exiting')
    
real_time_classifier.py
latent-variable/Real_time_fallacy_detection
import whisper

def get_model_options():
    models = whisper.available_models()
    print('Available models: ', ''.join(models))

def load_model():
    # Selected and load from the list of available models that can fit in your system
    # model = "large-v2"
    model = "medium"
    print(f'Loading the {model} model...')
    model = whisper.load_model(model)
    print('Model loaded')
    return model

def test_whisper(file='audio_chunk.wav'):
    model = load_model()
    transcription = model.transcribe(file)
    print(transcription['text'])
whs.py
latent-variable/Real_time_fallacy_detection