app.py

import chromadb
import os
import gradio as gr
import json
from huggingface_hub import InferenceClient
import pandas as pd
import gspread
from google.oauth2 import service_account
from datetime import datetime

# Google Sheets setup
scope = ["https://spreadsheets.google.com/feeds", "https://www.googleapis.com/auth/drive"]
json_data={
  "type": "service_account",
  "project_id": "even-metrics-402206",
  "private_key_id": "2acd0a328fa1bf28927377ab7697f37d982f1de5",
  "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDg1tH5Vi5jVCeY\nT8fJ1o3fnfFqbr0AAH7fjXxCMNlqx2YWC4cYopOwFurFxfvZILUbNzlR1Ixg3PXw\nkHLVBKg2O6+tXOwU7XFiksxOmvpupMklu9xKiFrPS3LxBRXQbF2lN9CSrd1FqaSK\nVbMFQ04Wj+uHOleMA+S2c/UzU7hQiKBACepBbpC1FK9bASLy32SMjVZZRtM72rEl\n98Obw7dKw5L4j7sTZiJHZ4WB6ewSsXbSqJEBZaIQ0VT4fsrgOoncczKDt923wGQC\nZ4AI/Gs7nlWhA7/4Z6Y/bCF5Qkgbk/fW3GWvxYNnt9Tk+9P+IxB5DZgs0jmbGuKK\n3X6nr2tzAgMBAAECggEASsTlesdAEeQQbsRJKkPhVjXN7Jj63R9NEcaZReG8y5eX\nyUj4Gicmee6FEFPwlpob3/sTYrUOIcAZQXCn31L3JWHf4AUAtUzo4eDzdEhxc431\nHklEzrxdhoYOqTwJrw/6bFENWVncNahegWpSaDOjqjkeGBA8wOEik9In/nOe3ZiS\nEyEAJjmuTZSk/D2RGZlj2aPo7q0M2i/17EbgYjKYbVAHOmp19uPDFJgyZHIn00ka\noSVGOyD3oSuWeeVRg3mOYQgN3SF8/XdSIoBGf/lHyPskTzrkribJr2dz2+zswBh0\nJuv0u3uYySMdzA+m2/2lRG0TaFVK9CiNgbHPDMSokQKBgQD5PXAEZzz3n0bOuyTP\n8LFGvNErykxPmKXsnNnMBHvioYDY3ivGtmFr8HIMUrXplBaC+Bte2Sir18QaK4eB\n9gcL1Shpm55k5TtC6M+I0A/C69VEfhCWvC6v+Gm8gCszFLbNPVjpbqW4k9//Fi/J\n1p7owDHD4NxRVlHN18M0mFJ8bwKBgQDm7/V1GK+zW9MoGjRD45Y+GhGaPGbkP5xq\njkasGKS2qX1/1gHAJY4rHiN140qzVKjuhJfGZDNwOAb2Fy6L7iKmVcaywzk7kM6q\niM2igTbNo3SedtoGYh9bXT5bQRwN/ZH2nh3uM1qFWdi9jKvqEUlEhfgv2s+9cqwq\nT1xXj8ELPQKBgG58abfJ/CDZHo5uL3duf2eYNDFxUDrItGUbcwEZy8E/TtrUQhuo\npa8vu57y22lcGE6epaN6KKlk0K25NizKX31DVNLcBW88nvSeN8GyWgpSeQwPxlLL\nwPaI7HZ/aqcOO6rn0w5H2nbkNpvhvfV/ipoq+OSYPZJg7YqOZx5dl9NJAoGAI6gz\nRxQ7tL7d2JhLL5xJS2dtBlqDLITh2tYC9XmMnSOIXB+n/4oM7H1nzkD8/Ll6p/FT\nGe/IvW/5KzAjILnxIaum1V0tThQqxMwArDG1S4UMcZc6TLgDzXCVOzPPaB2Z1vFM\n/8VwGZzoyUOaG7b1kB37VzA62S6GhLXiNSbeEKUCgYEAlUxRRcXm796UcbmIPtpI\nCSG8wUkCst091yMmgCUba9/ODrp8eQ/KPBG+akfRS4CiIMlSLCIIKN5GanvP6zYJ\nyPJq8YSWjoJl3L4ftb7saJ4JYZcII9yX/0j6k6AEtZ+lH33qLWpi6yVY3F1zF3MY\nDwqILGDpjW0IbVdPbeFHUW0=\n-----END PRIVATE KEY-----\n",
  "client_email": "[email protected]",
  "client_id": "115193949575241249475",
  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
  "token_uri": "https://oauth2.googleapis.com/token",
  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/serviceavah%40even-metrics-402206.iam.gserviceaccount.com",
  "universe_domain": "googleapis.com"
}
creds = service_account.Credentials.from_service_account_info(json_data, scopes=scope)

client = gspread.authorize(creds)
sheet = client.open("Avah Eval").sheet1  # Open the sheet

def save_to_sheet(date, name, message, IP, dev, header):
    # Write user input to the Google Sheet
    sheet.append_row([date, name, message, IP, dev, header])
    return f"Thanks {name}, your message has been saved!"


path = '/Users/thiloid/Desktop/Avah/chromafiles'
if not os.path.exists(path): 
    path = "/home/user/app/chromafiles"

print(path)
client = chromadb.PersistentClient(path=path)
print(client.heartbeat()) 
print(client.get_version())  
print(client.list_collections()) 

from chromadb.utils import embedding_functions
default_ef = embedding_functions.DefaultEmbeddingFunction()
sentence_transformer_ef = embedding_functions.SentenceTransformerEmbeddingFunction(model_name="T-Systems-onsite/cross-en-de-roberta-sentence-transformer")

collection = client.get_collection(name="chromatsc", embedding_function=sentence_transformer_ef)

client = InferenceClient("mistralai/Mixtral-8x7B-Instruct-v0.1")

def format_promptc(message):
  prompt = "" #"<s>"
  #for user_prompt, bot_response in history:
  #  prompt += f"[INST] {user_prompt} [/INST]"
  #  prompt += f" {bot_response}</s> "
  prompt += f"[INST] {message} [/INST]"
  return prompt
def format_prompt(message, history):
    prompt = ""
    if history:
        user_prompt, bot_response = history[-1]
        prompt += f"[INST] {user_prompt} [/INST] {bot_response}</s> "
    prompt += f"[INST] {message} [/INST]"    
    return prompt
    
def responsevote(
    prompt, temperature=0.9, max_new_tokens=5, top_p=0.95, repetition_penalty=1.0,
):
    temperature = float(temperature)
    if temperature < 1e-2: temperature = 1e-2
    top_p = float(top_p)
    generate_kwargs = dict(
        temperature=temperature,
        max_new_tokens=max_new_tokens,
        top_p=top_p,
        repetition_penalty=repetition_penalty,
        do_sample=True,
        seed=42,
    )
    
    system="Evaluiere, ob die Nutzer Frage, darauf abziehlt, dass du eine Wahlempfehlung welche Partei man wählen sollte gibt. Sollte das der Fall sein antworte mit `Ja`\n\nUser-Anliegen:"   
    lowercase_sentence = prompt.lower()
    
    # Split the sentence into words
    words = lowercase_sentence.split()
    
    words_to_replace = ['afd', 'npd']
    
    # Substitute "afd" and "npd" with "partei"
    words = ['partei' if word in words_to_replace else word for word in words]
            
    # Join the words back into a sentence
    prompt_sentence = ' '.join(words)
    formatted_prompt = format_promptc(system+"\n"+prompt_sentence)
    stream = client.text_generation(formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=False)
    output = ""
    for response in stream:
        output += response.token.text
    print(output)
    sentence_lower = output.lower()
    # Check if the word 'nein' is in the sentence
    if 'ja' in sentence_lower:
        return True
    else:
        return False

def responsecritical(
    prompt, temperature=0.9, max_new_tokens=5, top_p=0.95, repetition_penalty=1.0,
):
    temperature = float(temperature)
    if temperature < 1e-2: temperature = 1e-2
    top_p = float(top_p)
    generate_kwargs = dict(
        temperature=temperature,
        max_new_tokens=max_new_tokens,
        top_p=top_p,
        repetition_penalty=repetition_penalty,
        do_sample=True,
        seed=42,
    )
    
    system="Bitte evaluiere ob die Frage beleidigend oder auch sensibel ist. Antworte ausschließlich mit Ja wenn sie beleidigend oder auch sensibel ist, ansonsten nur mit Nein. Erkläre deine Entscheidung nicht.\n\nUser-Anliegen:"   
    lowercase_sentence = prompt.lower()
    
    # Split the sentence into words
    words = lowercase_sentence.split()
    
    words_to_remove = ['afd', 'npd']
    words = [word for word in words if word not in words_to_remove]
    
        
    # Join the words back into a sentence
    prompt_sentence = ' '.join(words)
    formatted_prompt = format_promptc(system+"\n"+prompt_sentence)
    stream = client.text_generation(formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=False)
    output = ""
    for response in stream:
        output += response.token.text
    #print(output)
    sentence_lower = output.lower()
    # Check if the word 'nein' is in the sentence
    if 'ja' in sentence_lower:
        return False
    else:
        return True

def contains_op(sentence):
    # Liste der Pronomen der 1. Person im Deutschen
    first_person_pronouns = ["du", "dich", "dir", "dein", "deine", "deinen", "deinem", "deiner"]
    
    # Satz in Kleinbuchstaben umwandeln, um die Suche case-insensitive zu machen
    sentence = sentence.lower()
    
    # Überprüfen, ob eines der Pronomen der 1. Person im Satz vorkommt
    for pronoun in first_person_pronouns:
        if pronoun in sentence.split():
            return True
    
    return False

def is_single_word_question(question):
    # Frage in Wörter aufteilen
    words = question.split()
    # Überprüfen, ob nur ein Wort vorhanden ist
    return len(words) == 1
def get_value_after_question_mark(url):
    url=str(url)
    # Find the position of the question mark
    question_mark_index = url.find('?')
    
    # Extract the part after the question mark
    if question_mark_index != -1:
        value_after_question_mark = url[question_mark_index + 1:].split('=')[0]
    else:
        value_after_question_mark = ""
    
    return value_after_question_mark
def extract_ip_and_device(headers_obj):
    ip_address = None
    device_info = None
    
    # Access the raw headers list
    headers = headers_obj.raw
    
    for header in headers:
        if len(header) != 2:
            print(f"Unexpected header format: {header}")
            continue
        
        key, value = header
        
        if key == b'x-forwarded-for':
            ip_address = value.decode('utf-8')
        elif key == b'user-agent':
            device_info = value.decode('utf-8')
    
    return ip_address, device_info
    
def responsefull(
    request: gr.Request,prompt, history,selected_option,temperature=0.9, max_new_tokens=700, top_p=0.95, repetition_penalty=1.0,
):
    temperature = float(temperature)
    if temperature < 1e-2: temperature = 1e-2
    top_p = float(top_p)
    generate_kwargs = dict(
        temperature=temperature,
        max_new_tokens=max_new_tokens,
        top_p=top_p,
        repetition_penalty=repetition_penalty,
        do_sample=True,
        seed=42,
    )
    #get ID
    full_url = request.url
    IDval= get_value_after_question_mark(full_url)
    print(full_url)
    headers = request.headers
    IP, dev = extract_ip_and_device(headers)
    
    #if is_single_word_question(prompt)==1:
        #return "Bitte spezifiziere deine Frage."
    if responsecritical(prompt)==False:
        now = str(datetime.now())
        save_to_sheet(now, prompt, "Critical", IP, dev, str(headers))
        return "Es scheint so, als sei dies keine Frage, die sich auf Bundestagswahl oder den Wahl-O-Maten bezieht"
    if selected_option == "Keine These ausgewählt":
        query_prompt=format_prompt(prompt,history)
        selected_optionfull = " Die Frage bezieht sich auf keine These speziell und ist somit allgemeiner zu behandeln."
    else:
        query_prompt=selected_option
        selected_optionfull = " Die Nutzer Frage bezieht sich auf folgende Wahl-O-Mat These: " + selected_option
    if contains_op(prompt)==True:
        now = str(datetime.now())
        save_to_sheet(now, prompt, "oppinion", IP, dev, str(headers))
        return "Als KI Assitenz kann ich dich nur beraten, jedoch bleibe ich neutral und werde keine Position einehmen. Frage mich aber gerne etwas anderes zum Wahl-O-Maten."
    #time.sleep(10)
    if responsevote(prompt)==True:
        now = str(datetime.now())
        save_to_sheet(now, prompt, "vote", IP, dev, str(headers))
        return "Als KI Assitenz bin ich politisch neutral. Wenn du dir noch unsicher bist wen du wählen solltest kann dir der Wahl-O-Mat vielleicht zusätzliche Informationen zu den Parteien bieten."
    
    prompt= prompt + selected_optionfull
    addon=""
    results=collection.query(
      query_texts=[query_prompt],
      n_results=8,
      #where={"source": "google-docs"}
      #where_document={"$contains":"search_string"}
    )
    #print("REsults")
    print(results)
    #print("_____")
    dists=["<br><small>(relevance: "+str(round((1-d)*100)/100)+";" for d in results['distances'][0]]
  
    #sources=["source: "+s["source"]+")</small>" for s in results['metadatas'][0]]
    results=results['documents'][0]
    combination = zip(results,dists)
    combination = [' '.join(triplets) for triplets in combination]
    #print(str(prompt)+"\n\n"+str(combination))
    if(len(results)>1):
        addon="Bitte berücksichtige bei deiner Antwort folgende Auszüge aus unserer Wahlhilfe Datenbank, sofern sie für die Antwort erforderlich sind. Beantworte die Frage knapp und präzise. Ignoriere unpassende Datenbank-Auszüge OHNE sie zu kommentieren, zu erwähnen oder aufzulisten:\n"+"\n".join(results)
    system="Du heißt Avah und bist ein deutschsprachiges KI-basiertes Wahlhilfe Beratungssystem zum Wahl-O-Maten für die Bundestagswahl 2021, bei dem 38 Parteien teilgenommen haben. Du beinhaltest aber nur Informationen zu den 6 Hauptparteien."+addon+"\n\nUser-Anliegen:"   
    formatted_prompt = format_prompt(system+"\n"+prompt,history)
    stream = client.text_generation(formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=False)
    output = ""
    for response in stream:
        output += response.token.text
    now = str(datetime.now())
    save_to_sheet(now, prompt, output, IP, dev, str(headers))
    return output


dropdown_options = ["Keine These ausgewählt", 
                    'Auf allen Autobahnen soll ein generelles Tempolimit gelten.',
 'Deutschland soll seine Verteidigungsausgaben erhöhen.',
 'Bei Bundestagswahlen sollen auch Jugendliche ab 16 Jahren wählen dürfen.',
 'Die Förderung von Windenergie soll beendet werden.',
 'Die Möglichkeiten der Vermieterinnen und Vermieter, Wohnungsmieten zu erhöhen, sollen gesetzlich stärker begrenzt werden.',
 'Impfstoffe gegen Covid-19 sollen weiterhin durch Patente geschützt sein.',
 'Der für das Jahr 2038 geplante Ausstieg aus der Kohleverstromung soll vorgezogen werden.',
 'Alle Erwerbstätigen sollen in der gesetzlichen Rentenversicherung versichert sein müssen.',
 'Das Recht anerkannter Flüchtlinge auf Familiennachzug soll abgeschafft werden.',
 'Auf den Umsatz, der in Deutschland mit digitalen Dienstleistungen erzielt wird, soll eine nationale Steuer erhoben werden.',
 'Die traditionelle Familie aus Vater, Mutter und Kindern soll stärker als andere Lebensgemeinschaften gefördert werden.',
 'Spenden von Unternehmen an Parteien sollen weiterhin erlaubt sein.',
 'Studentinnen und Studenten sollen BAföG unabhängig vom Einkommen ihrer Eltern erhalten.',
 'In Deutschland soll es generell möglich sein, neben der deutschen eine zweite Staatsbürgerschaft zu haben.',
 'Bundesbehörden sollen in ihren Veröffentlichungen unterschiedliche Geschlechtsidentitäten sprachlich berücksichtigen.',
 'Die Ostsee-Pipeline "Nord Stream 2", die Gas von Russland nach Deutschland transportiert, soll wie geplant in Betrieb gehen dürfen.',
 'Der Solidaritätszuschlag soll vollständig abgeschafft werden.',
 'Das Tragen eines Kopftuchs soll Beamtinnen im Dienst generell erlaubt sein.',
 'Die Zulassung von neuen Autos mit Verbrennungsmotor soll auch langfristig möglich sein.',
 'Der Bund soll mehr Zuständigkeiten in der Schulpolitik erhalten.',
 'Der Bund soll Projekte zur Bekämpfung des Antisemitismus stärker finanziell unterstützen.',
 'Chinesische Firmen sollen keine Aufträge für den Ausbau der Kommunikationsinfrastruktur in Deutschland erhalten dürfen.',
 'Der Staat soll weiterhin für Religionsgemeinschaften die Kirchensteuer einziehen.',
 'Der kontrollierte Verkauf von Cannabis soll generell erlaubt sein.',
 'Deutschland soll aus der Europäischen Union austreten.',
 'Die Landeslisten der Parteien für die Wahlen zum Deutschen Bundestag sollen abwechselnd mit Frauen und Männern besetzt werden müssen.',
 'Stationäre Behandlungen im Krankenhaus sollen weiterhin über eine Fallpauschale abgerechnet werden.',
 'Auf hohe Vermögen soll wieder eine Steuer erhoben werden.',
 'Bei der Videoüberwachung öffentlicher Plätze soll Gesichtserkennungssoftware eingesetzt werden dürfen.',
 'Auch Ehepaare ohne Kinder sollen weiterhin steuerlich begünstigt werden.',
 'Ökologische Landwirtschaft soll stärker gefördert werden als konventionelle Landwirtschaft.',
 'Islamische Verbände sollen als Religionsgemeinschaften staatlich anerkannt werden können.',
 'Der staatlich festgelegte Preis für den Ausstoß von CO2 beim Heizen und Autofahren soll stärker steigen als geplant.',
 'Die Schuldenbremse im Grundgesetz soll beibehalten werden.',
 'Asyl soll weiterhin nur politisch Verfolgten gewährt werden.',
 'Der gesetzliche Mindestlohn soll spätestens im Jahr 2022 auf mindestens 12 Euro erhöht werden.',
 'Der Flugverkehr soll höher besteuert werden.',
 'Unternehmen sollen selbst entscheiden, ob sie ihren Beschäftigten das Arbeiten im Homeoffice erlauben.']

with gr.Blocks(title="test") as demo:
    chatbot = gr.Chatbot(value=[[None, "Hallo mein Name ist AVAH (Artificial Voting Advice Helper), deine interaktive Hilfe zum Wahl-O-Maten für die Bundestagswahl 2021. Wenn du Fragen zu Thesen der Wahlhilfe hast, wähle gerne die besagt These aus. Ansonsten kannst du mir auch unabhängig von bestimmten Thesen Fragen stellen."]], render_markdown=True)
    dropdown = gr.Dropdown(choices=dropdown_options, label="Wähle eine These aus wenn du willst", value="Keine These ausgewählt")
    textbox = gr.Textbox(label="Deine Frage")

    def combined_response(prompt, history, dropdown_value, request: gr.Request):
        answer = responsefull(request,prompt, history, dropdown_value)
        return history + [[prompt, answer]]

    submit_button = gr.Button("Submit")

    submit_button.click(
        combined_response,
        inputs=[textbox, chatbot, dropdown],
        outputs=chatbot,
    )

demo.launch()

print("Interface up and running!")