Hello together, (Deutsch siehe unten)

At the beginning there was

The Idea:

It is often a small group of people who disturb the rest. And we often know these people too. But so far we have not defended ourselves or defended ourselves too little. We try to discuss with them, but most of the time we end up arguing (laboriously and research-intensively) and the other side only responds with hate comments or tons of fake news.
In doing so, they disturb or destroy the discussion forum and leave behind irritated, angry or even frightened participants.
Censoring hate comments is difficult and questionable. But accepting this widespread bad habit is also wrong.
Our solution approach looks like this:

(Taken from: @achimmertens/telekom-human-centered-technology-treffen-in-bonn)

On the other hand we have people that do good things, but get only a little attention. A lot of charity work is done here on Hive and we don't recognize it. But we should, because it is a crucial and undervalued part of Hive, giving it more meaning.

So, my challenge was to create a prototype, that reads Hive content, pushes it to a trained AI and checks for "good and evil" in the text and responds in one of the three categories "charity", "Hatespeech" or "don't know".

The Hive Content Checker Tool

It contains two python scripts and a text file with training data.

getHivePosts.py

This script gets content from the Hive API and puts it into a file:

import requests
from requests.exceptions import RequestException
import datetime
import re
import json

# Set up the headers for the request
headers = {
    "Content-Type": "application/json",
}

# Erstellen Sie eine Liste, um alle bereinigten Posts aufzunehmen
cleaned_posts = []

def remove_text_inside_brackets(text):
    """
    This function removes any text inside brackets and any URLs.
    """
    # Remove text inside any brackets
    # The regex pattern finds text inside (), [] and {}
    no_brackets = re.sub(r"\[.*?\]|\(.*?\)|\{.*?\}", "", text)
    # Remove URLs
    # The regex pattern finds strings that look like URLs
    no_urls = re.sub(r'http[s]?://\S+', '', no_brackets)
    return no_urls.strip()

# The payload for the POST request
# The before_date is set to tomorrow to ensure we get the latest posts
before_date = (datetime.datetime.utcnow() + datetime.timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%S')

payload = {
    "jsonrpc": "2.0",
    "method": "condenser_api.get_discussions_by_author_before_date",
    "params": {
        "author": "achimmertens",
        "start_permlink": "",
        "before_date": before_date,
        "limit": "3"
    },
    "id": 1
}

# The URL for the POST request
url = "https://api.hive.blog"

# Perform the POST request
try:
    response = requests.post(url, json=payload, proxies=proxies, headers=headers)

    # Check if the request was successful
    if response.status_code == 200:
        print("Request successful!")
        # show response.json title and body content:
        content = response.json()["result"]
        for post in content:
            print(f"Title: {post['title']}")
            body = post['body']
            # remove carriage returns and newlines in body
            body = body.replace("\r", "").replace("\n", "")
            # safe body to file with the name textToAnalyze.txt
            # remove all
            clean_text = remove_text_inside_brackets(body)
            # Fügen Sie den bereinigten Post zur Liste hinzu
            cleaned_posts.append({"title": post['title'], "body": clean_text})
            print(f"Body: {clean_text[:200]}...")  # Truncate body for display purposes
            print()
        # Speichern Sie die bereinigten Posts im JSON-Format in einer Datei
        with open("textToAnalyze.json", "w", encoding='utf-8') as file:
            json.dump(cleaned_posts, file, ensure_ascii=False, indent=4)
    else:
        print(f"Request failed with status code: {response.status_code}")
except RequestException as e:
    print(f"An error occurred: {e}")

As a result we get a text file "textToAnalyze.json" with title and consolidated body:

ask_openai.py

The second script reads the "textToAnalyze.json" file. It enables a session and creates a GPT Assistant in OpenAI.com. It defines how to react. Therefore it reads the file knowledge.txt, which contains examples.
The GPT assistant answers in a defined way and this result is written into a "analyzedText.json" file:

Details:

First I needed an API Key on openAI. We have to pay for it. I have put 10 USD on it and my tests costed me 0.84 USD the last two days:

The Training Data

Now we have to train the assistant. Therefore I created a simple text file, which contains some information and examples, what we are looking for.
I have done some research before and therefore have 8 examples of charity posts:

I also put in the sentence "The creator of this document is Achim." This is to test the AI, because when I ask for the creator, it can give me the name only, if it has read the text. (Teaser: it works 😉)

The Python Script

The following script (which I created with the help of AI) opens a session with api.openai.com. It creates an assistant, gives it some intructions, uploads the training data, uploads the text that should be analyzed and closes the session:

import openai
import configparser
import time
import json

config = configparser.ConfigParser()
config.read('../setup.conf')

OPENAI_API_KEY = config['api']['key']
openai.api_key = OPENAI_API_KEY

client = openai.OpenAI(api_key=OPENAI_API_KEY)

# Texte aus der JSON-Datei einlesen
with open("textToAnalyze.json", "r", encoding='utf-8') as file:
    posts = json.load(file)

# Datei für die Wissensdatenbank hochladen (wenn benötigt)
file = client.files.create(
    file=open("knowledge.txt", "rb"),
    purpose='assistants'
)

# Assistant erstellen
assistant = client.beta.assistants.create(
    name="Content Detector",
    instructions="You are an officer that checks, if there is charity or hate in the text. For examples look into knowledge.txt.",
    tools=[{"type": "retrieval"}],
    model="gpt-4-turbo-preview",
    file_ids=[file.id]
)

# Thread für die Konversation erstellen
thread = client.beta.threads.create()

for post in posts:
    title = post['title']
    body = post['body']

    # Nachricht an den Assistant senden
    message = client.beta.threads.messages.create(
        thread_id=thread.id,
        role="user",
        content="Does the following text contain hate speech? If yes, answer with 'This text contains hate speech'. If you find charity, based on the examples, use the phrase '!CHARY, this text is about charity'. If you are not sure, use 'I am not sure about to categorize this content.' Only use these phrases to answer. Here is the text to analyze: Title: "+title+" Body: "+body,
    )

    run = client.beta.threads.runs.create(
        thread_id=thread.id,
        assistant_id=assistant.id,
        instructions="Please address the user as Jane Doe. The user has a premium account.",
    )

    print("Checking assistant status for post: ", title)
    while True:
        run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)

        if run.status == "completed":
            print("Done for post: ", title)
            messages = client.beta.threads.messages.list(thread_id=thread.id)

            # Nehmen Sie die letzte Nachricht des Assistants (die Antwort)
            assistant_message = next(m for m in messages if m.role == 'assistant').content[0].text.value
            post['assistant_message'] = assistant_message  # Antwort zum Post hinzufügen


            print("Messages for post: ", title)
            for message in messages:
                assert message.content[0].type == "text"
                print({"role": message.role, "message": message.content[0].text.value})

            break
        else:
            print("In progress...")
            time.sleep(5)

# Den Assistant am Ende löschen
client.beta.assistants.delete(assistant.id)
with open("analyzedText.json", "w", encoding='utf-8') as file:
    json.dump(posts, file, ensure_ascii=False, indent=4)

Yes, this script can be optimized, i.e. to keep the assistant and use it not only once each time I start it. But hey, this is a proof of concept, and it works 😀

The Future

Now that we have a tool to check the content from some given authors on hive, we can use it to finetune it and if wanted, to give some comments to that posts.
We can also adapt it easily to other platforms like twitter, facebook, linkedin, etc.
When it works fine, we can also install a local GPT or other LLM to lower the costs.
So, there is still some work left to do 😉

Achim Mertens

---

Deutsch

Hallo zusammen,

Am Anfang gab es

Die Idee:

Oft ist es eine kleine Gruppe von Menschen, die den Rest stört. Und oft kennen wir diese Leute auch. Aber bisher haben wir uns nicht oder zu wenig gewehrt. Wir versuchen, mit ihnen zu diskutieren, aber meistens streiten wir (mühsam und rechercheintensiv) und die Gegenseite antwortet nur mit Hasskommentaren oder massenhaft Fake News.
Dadurch stören oder zerstören sie das Diskussionsforum und hinterlassen verärgerte, irritierte oder sogar verängstigte Teilnehmer.
Hasskommentare zu zensieren ist schwierig und fragwürdig. Aber es ist auch falsch, diese weit verbreitete Unsitte zu akzeptieren.
Unser Lösungsansatz sieht so aus:

(Entnommen aus: @achimmertens/telekom-human-centered-technology-treffen-in-bonn)

Auf der anderen Seite gibt es Menschen, die Gutes tun, aber nur wenig Aufmerksamkeit bekommen. Hier auf Hive wird viel Wohltätigkeitsarbeit geleistet und wir erkennen dies nicht an. Aber wir sollten es tun, denn es ist ein entscheidender und unterbewerteter Teil von Hive, der ihm mehr Bedeutung verleiht.

Meine Herausforderung bestand also darin, einen Prototyp zu erstellen, der Hive-Inhalte liest, an eine trainierte KI weiterleitet, den Text auf „Gut und Böse“ prüft und in einer der drei Kategorien „Wohltätigkeit“, „Hassrede“ oder „Ich weiß es nicht“ antwortet.

Das Hive-Content-Checker-Tool

Es enthält zwei Python-Skripte und eine Textdatei mit Trainingsdaten.

getHivePosts.py

Dieses Skript ruft Inhalte von der Hive-API ab und fügt sie in eine Datei ein:

import requests
from requests.exceptions import RequestException
import datetime
import re
import json

# Set up the headers for the request
headers = {
    "Content-Type": "application/json",
}

# Erstellen Sie eine Liste, um alle bereinigten Posts aufzunehmen
cleaned_posts = []

def remove_text_inside_brackets(text):
    """
    This function removes any text inside brackets and any URLs.
    """
    # Remove text inside any brackets
    # The regex pattern finds text inside (), [] and {}
    no_brackets = re.sub(r"\[.*?\]|\(.*?\)|\{.*?\}", "", text)
    # Remove URLs
    # The regex pattern finds strings that look like URLs
    no_urls = re.sub(r'http[s]?://\S+', '', no_brackets)
    return no_urls.strip()

# The payload for the POST request
# The before_date is set to tomorrow to ensure we get the latest posts
before_date = (datetime.datetime.utcnow() + datetime.timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%S')

payload = {
    "jsonrpc": "2.0",
    "method": "condenser_api.get_discussions_by_author_before_date",
    "params": {
        "author": "achimmertens",
        "start_permlink": "",
        "before_date": before_date,
        "limit": "3"
    },
    "id": 1
}

# The URL for the POST request
url = "https://api.hive.blog"

# Perform the POST request
try:
    response = requests.post(url, json=payload, proxies=proxies, headers=headers)

    # Check if the request was successful
    if response.status_code == 200:
        print("Request successful!")
        # show response.json title and body content:
        content = response.json()["result"]
        for post in content:
            print(f"Title: {post['title']}")
            body = post['body']
            # remove carriage returns and newlines in body
            body = body.replace("\r", "").replace("\n", "")
            # safe body to file with the name textToAnalyze.txt
            # remove all
            clean_text = remove_text_inside_brackets(body)
            # Fügen Sie den bereinigten Post zur Liste hinzu
            cleaned_posts.append({"title": post['title'], "body": clean_text})
            print(f"Body: {clean_text[:200]}...")  # Truncate body for display purposes
            print()
        # Speichern Sie die bereinigten Posts im JSON-Format in einer Datei
        with open("textToAnalyze.json", "w", encoding='utf-8') as file:
            json.dump(cleaned_posts, file, ensure_ascii=False, indent=4)
    else:
        print(f"Request failed with status code: {response.status_code}")
except RequestException as e:
    print(f"An error occurred: {e}")

Als Ergebnis erhalten wir eine Textdatei „textToAnalyze.json“ mit Titel und konsolidiertem Text:

ask_openai.py

Das zweite Skript liest die Datei „textToAnalyze.json“. Es aktiviert eine Sitzung und erstellt einen GPT-Assistenten in OpenAI.com. Es definiert, wie man reagiert. Dazu liest es die Datei Knowledge.txt, die Beispiele enthält.
Der GPT-Assistent antwortet definiert und dieses Ergebnis wird in eine „analyzedText.json“-Datei geschrieben:

Details:

Zuerst brauchte ich einen API-Schlüssel für openAI. Wir müssen dafür bezahlen. Ich habe 10 USD darauf gesetzt und meine Tests haben mich in den letzten beiden Tagen 0,84 USD gekostet:

Die Trainingsdaten

Jetzt müssen wir den Assistenten schulen. Deshalb habe ich eine einfache Textdatei erstellt, die einige Informationen und Beispiele enthält, wonach wir suchen.
Ich habe bereits einige Recherchen durchgeführt und habe daher 8 Beispiele für Wohltätigkeitsbeiträge:

Ich habe auch den Satz „Der Ersteller dieses Dokuments ist Achim“ eingefügt. Dies dient dazu, die KI zu testen, denn wenn ich nach dem Ersteller frage, kann sie mir den Namen nur nennen, wenn sie den Text gelesen hat. (Teaser: Es funktioniert 😉)

Das Python-Skript

Das folgende Skript (das ich mit Hilfe von KI erstellt habe) öffnet eine Sitzung mit api.openai.com. Es erstellt einen Assistenten, gibt ihm einige Anweisungen, lädt die Trainingsdaten hoch, lädt den zu analysierenden Text hoch und schließt die Sitzung:

import openai
import configparser
import time
import json

config = configparser.ConfigParser()
config.read('../setup.conf')

OPENAI_API_KEY = config['api']['key']
openai.api_key = OPENAI_API_KEY

client = openai.OpenAI(api_key=OPENAI_API_KEY)

# Texte aus der JSON-Datei einlesen
with open("textToAnalyze.json", "r", encoding='utf-8') as file:
    posts = json.load(file)

# Datei für die Wissensdatenbank hochladen (wenn benötigt)
file = client.files.create(
    file=open("knowledge.txt", "rb"),
    purpose='assistants'
)

# Assistant erstellen
assistant = client.beta.assistants.create(
    name="Content Detector",
    instructions="You are an officer that checks, if there is charity or hate in the text. For examples look into knowledge.txt.",
    tools=[{"type": "retrieval"}],
    model="gpt-4-turbo-preview",
    file_ids=[file.id]
)

# Thread für die Konversation erstellen
thread = client.beta.threads.create()

for post in posts:
    title = post['title']
    body = post['body']

    # Nachricht an den Assistant senden
    message = client.beta.threads.messages.create(
        thread_id=thread.id,
        role="user",
        content="Does the following text contain hate speech? If yes, answer with 'This text contains hate speech'. If you find charity, based on the examples, use the phrase '!CHARY, this text is about charity'. If you are not sure, use 'I am not sure about to categorize this content.' Only use these phrases to answer. Here is the text to analyze: Title: "+title+" Body: "+body,
    )

    run = client.beta.threads.runs.create(
        thread_id=thread.id,
        assistant_id=assistant.id,
        instructions="Please address the user as Jane Doe. The user has a premium account.",
    )

    print("Checking assistant status for post: ", title)
    while True:
        run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)

        if run.status == "completed":
            print("Done for post: ", title)
            messages = client.beta.threads.messages.list(thread_id=thread.id)

            # Nehmen Sie die letzte Nachricht des Assistants (die Antwort)
            assistant_message = next(m for m in messages if m.role == 'assistant').content[0].text.value
            post['assistant_message'] = assistant_message  # Antwort zum Post hinzufügen


            print("Messages for post: ", title)
            for message in messages:
                assert message.content[0].type == "text"
                print({"role": message.role, "message": message.content[0].text.value})

            break
        else:
            print("In progress...")
            time.sleep(5)

# Den Assistant am Ende löschen
client.beta.assistants.delete(assistant.id)
with open("analyzedText.json", "w", encoding='utf-8') as file:
    json.dump(posts, file, ensure_ascii=False, indent=4)

Ja, dieses Skript kann optimiert werden, z. B. um den Assistenten beizubehalten und ihn nicht nur einmal bei jedem Start zu verwenden. Aber hey, das ist ein Proof of Concept und es funktioniert 😀

Die Zukunft

Da wir nun über ein Tool verfügen, mit dem wir den Inhalt einiger bestimmter Autoren auf Hive überprüfen können, können wir ihn zur Feinabstimmung verwenden und bei Bedarf einige Kommentare zu diesen Beiträgen abgeben.
Wir können es auch problemlos an andere Plattformen wie Twitter, Facebook, LinkedIn usw. anpassen.
Wenn es gut funktioniert, können wir auch ein lokales GPT oder ein anderes LLM installieren, um die Kosten zu senken.
Es gibt also noch einiges zu tun 😉

Achim Mertens