Openclaw, eine kleine Antwort

Ich konnte der Versuchung nicht widerstehen, trotz entsprechender Warnungen, dass es die gefährlichste Software der Welt sei, diese ein wenig auszutesten. Vorweg kann ich schonmal sagen, die Kosten sind ernüchternd. Binnen von 4 Tagen habe ich anscheinend das Kontingent von OpenAI go aufgebraucht. Anscheinend war es das Verfügbare innerhalb von 10 Tagen. Dann habe ich 1x 5 USDT aufgeladen, wobei die Kosten pro Antwort in etwa 1/2 ct. war.

Oh, da muss ich sagen das war eigentlich schon das günstigere Deepseek V3.2. Bei meiner Art der Verwendung war mir das viel zu teuer und ich bin auf gpt4 mini umgestiegen. Kostenfaktor 8x günstiger, aber beschränkteres Modell. Viele berichten von über 100 USD am Tag, mit großen Modellen. Da bin ich mit meinen 1,5 USD eh froh, die ich in den paar Tagen verbraucht habe.

Was habe ich damit gemacht?

Ich habe getestet
-dass er Dateien schreibt
-bestimmte Dateien liest
-als Sprachmodell
-Programme öffnet
-als Analysetool

Am allermeisten hat mich beeindruckt, dass ich das KI-Tool Openvino endlich installieren konnte. Da hatte ich schon etliche Versuche hinter mir, auch mit KI, das zu machen. Openclaw hat mein System untersucht und mir Hilfe bei der Installation angeboten.

Openclaw läuft bei mir über einen Bottoken und BenutzerID in Telegram.

---

Nachdem ich mich aber nicht auf Dauer mit Kosten beschäftigen möchte, habe ich mithilfe von ChatGPT selbst ein Tool entwickelt, das eingeschränkt bestimmte Dinge tun kann. Es läuft unter anderem mit einem lokalen KI Modell, das bei Openclaw im Gegensatz ein nicht unbeachtliches Problem mit dem Datenschutz war. Eine lokale AI, etwas das Openclaw so nicht erlaubt. Damit kann ich nun in Telegram

-erlaubte Verzeichnisse mit list auflisten
-mit send kann man sich Dateien schicken lassen.
-ausgewählte Befehle wie df, free, whoami, top und uptime ausführen und nach Telegram übertragen. Diese werden mit cmd getriggert.
-Mit analyze kann man Dateien, zB logs, analysieren lassen. Das funktioniert mit einem KI Fallback. Alles andere ist eine Zuweisung per Programmierung.
-mit show kann man sich Textdateien anzeigen lassen.

ChatGPT hat mich davor gewarnt, zu viele Befehle einzubauen, weil ich ansonsten bald eine Remoteshell habe, diesen Rat habe ich beherzigt.

Als KI Modell habe ich ein kleines Qwen2.5-7B-Instruct-Q4_K_M, damit es nicht zu lange dauert. Die Geschwindigkeit ist in Ordnung und hängt an Vulcan, weil SYCL, das besser geeignet wäre, nicht funktioniert. Man kann jetzt nicht allzuviel davon erwarten, aber es erfüllt seinen Zweck.

Derzeit ist es so eingestellt, dass es sich innerhalb der selben Session an Nachrichten erinnern kann.

Zum Verständnis, das kann ich auch vom Smartphone aus machen. Es ist eine Spielerei, die durchaus noch ausbaufähig ist. Momentan fehlt mir aber ein wenig Fantasie, um da mehr daraus zu machen.

https://odysee.com/@jeyf123:d/Minibot:c

import requests
import subprocess
import os
from telegram import Update
from telegram.ext import ApplicationBuilder, MessageHandler, ContextTypes, filters

# ==============================
# KONFIGURATION
# ==============================

TELEGRAM_BOT_TOKEN = "xxx"
ALLOWED_USER_ID = xxx

LLAMA_URL = "http://127.0.0.1:8080/v1/chat/completions"
MODEL_NAME = "Qwen2.5-7B-Instruct-Q4_K_M.gguf"

BASE_DIR = "/home/gerald"

ALLOWED_COMMANDS = {
    "df": ["df", "-h"],
    "free": ["free", "-h"],
    "uptime": ["uptime"],
    "whoami": ["whoami"],
    "top": ["top", "-b", "-n", "2"]
}


# ==============================
# SICHERHEIT
# ==============================

def is_safe_path(path):
    real = os.path.realpath(path)

    if not real.startswith(BASE_DIR):
        return False

    if "/." in real:
        return False

    if os.path.islink(real):
        return False

    return True


# ==============================
# HAUPTFUNKTION
# ==============================

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):

    user_id = update.effective_user.id
    if user_id != ALLOWED_USER_ID:
        return

    user_text = update.message.text.strip()
    print("DEBUG:", repr(user_text))

    # ==============================
    # ANALYZE DATEI
    # ==============================

    if user_text.lower().startswith("analyze "):
        relative_path = user_text[8:].strip()
        candidate = os.path.join(BASE_DIR, relative_path)

        if not is_safe_path(candidate):
            await update.message.reply_text("Pfad nicht erlaubt.")
            return

        if not os.path.exists(candidate):
            await update.message.reply_text("Datei existiert nicht.")
            return

        with open(candidate, "r", encoding="utf-8", errors="ignore") as f:
            content = f.read(16000)

        # Datei im User-Kontext speichern
        context.user_data["last_file_content"] = content
        context.user_data["last_file_name"] = relative_path

        try:
            response = requests.post(
                LLAMA_URL,
                json={
                    "model": MODEL_NAME,
                    "messages": [
                        {
                            "role": "system",
                            "content": "Du analysierst Dateiinhalt sachlich und strukturiert. Erfinde nichts."
                        },
                        {
                            "role": "user",
                            "content": content
                        }
                    ],
                    "temperature": 0.2,
                    "max_tokens": 2048
                },
                timeout=300
            )

            result = response.json()

            if "choices" not in result:
                await update.message.reply_text(f"Server-Fehler:\n{result}")
                return

            answer = result["choices"][0]["message"]["content"].strip()
            await update.message.reply_text(answer[:3500])

        except Exception as e:
            await update.message.reply_text(f"Analyse-Fehler: {e}")

        return


    # ==============================
    # CMD WHITELIST
    # ==============================

    if user_text.lower().startswith("cmd "):
        cmd_name = user_text[4:].strip().lower()

        if cmd_name not in ALLOWED_COMMANDS:
            await update.message.reply_text("Befehl nicht erlaubt.")
            return

        try:
            result = subprocess.run(
                ALLOWED_COMMANDS[cmd_name],
                capture_output=True,
                text=True,
                timeout=10
            )

            output = result.stdout or result.stderr
            if not output:
                output = "Keine Ausgabe."

            await update.message.reply_text(
                f"```\n{output[:3500]}\n```",
                parse_mode="Markdown"
            )

        except Exception as e:
            await update.message.reply_text(f"Fehler: {e}")

        return


        # ==============================
    # DATEI SENDEN
    # ==============================

    if user_text.lower().startswith("send "):
        relative_path = user_text[5:].strip()
        candidate = os.path.join(BASE_DIR, relative_path)

        if not is_safe_path(candidate):
            await update.message.reply_text("Pfad nicht erlaubt.")
            return

        if not os.path.exists(candidate):
            await update.message.reply_text("Datei existiert nicht.")
            return

        if os.path.getsize(candidate) > 20 * 1024 * 1024:
            await update.message.reply_text("Datei zu groß.")
            return

        try:
            with open(candidate, "rb") as f:
                await update.message.reply_document(f)
        except Exception as e:
            await update.message.reply_text(f"Sende-Fehler: {e}")

        return

        # ==============================
    # DATEIEN AUFLISTEN
    # ==============================

    if user_text.lower().startswith("list"):
        relative_path = user_text[4:].strip() or "."
        candidate = os.path.join(BASE_DIR, relative_path)

        if not is_safe_path(candidate):
            await update.message.reply_text("Pfad nicht erlaubt.")
            return

        if not os.path.isdir(candidate):
            await update.message.reply_text("Kein Verzeichnis.")
            return

        try:
            entries = [
                e for e in os.listdir(candidate)
                if not e.startswith(".")
            ]

            if not entries:
                await update.message.reply_text("Verzeichnis ist leer.")
                return

            dirs = sorted(
                e for e in entries
                if os.path.isdir(os.path.join(candidate, e))
            )

            files = sorted(
                e for e in entries
                if not os.path.isdir(os.path.join(candidate, e))
            )

            output_lines = []

            for d in dirs:
                output_lines.append(f"<DIR>  {d}")

            for f in files:
                output_lines.append(f"       {f}")

            output = "\n".join(output_lines)

            await update.message.reply_text(
                f"```\n{output[:3500]}\n```",
                parse_mode="Markdown"
            )

        except Exception as e:
            await update.message.reply_text(f"Fehler: {e}")

        return

        # ==============================
    # DATEI ANZEIGEN (TEXT)
    # ==============================

    if user_text.lower().startswith("show "):
        relative_path = user_text[5:].strip()
        candidate = os.path.join(BASE_DIR, relative_path)

        ALLOWED_SHOW_EXT = (".txt", ".log", ".md", ".conf")

        if not is_safe_path(candidate):
            await update.message.reply_text("Pfad nicht erlaubt.")
            return

        if not os.path.exists(candidate):
            await update.message.reply_text("Datei existiert nicht.")
            return

        if not candidate.lower().endswith(ALLOWED_SHOW_EXT):
            await update.message.reply_text("Dateityp nicht erlaubt.")
            return

        try:
            with open(candidate, "r", encoding="utf-8", errors="ignore") as f:
                content = f.read(4000)

            if not content.strip():
                await update.message.reply_text("Datei ist leer.")
                return

            await update.message.reply_text(
                f"```\n{content}\n```",
                parse_mode="Markdown"
            )

        except Exception as e:
            await update.message.reply_text(f"Fehler: {e}")

        return


    # ==============================
    # KI FALLBACK MIT DATEIKONTEXT
    # ==============================

    last_content = context.user_data.get("last_file_content")
    last_name = context.user_data.get("last_file_name")

    try:
        response = requests.post(
            LLAMA_URL,
            json={
                "model": MODEL_NAME,
                "messages": [
                    {
                        "role": "system",
                        "content": "Du bist ein hilfreicher Linux- und Technik-Assistent."
                    },
                    {
                        "role": "user",
                        "content": f"""Aktuelle Anfrage:
{user_text}

Falls sich die Anfrage auf eine zuvor analysierte Datei bezieht:

Dateiname: {last_name}
Dateiinhalt:
{last_content}
"""
                    }
                ],
                "temperature": 0.7,
                "max_tokens": 2048
            },
            timeout=300
        )

        result = response.json()

        if "choices" not in result:
            await update.message.reply_text(f"Server-Fehler:\n{result}")
            return

        answer = result["choices"][0]["message"]["content"].strip()
        await update.message.reply_text(answer[:3500])

    except Exception as e:
        await update.message.reply_text(f"KI-Fehler: {e}")


# ==============================
# START
# ==============================

if __name__ == "__main__":
    app = ApplicationBuilder().token(TELEGRAM_BOT_TOKEN).build()
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    print("Bot läuft...")
    app.run_polling()