Einen Eigenen Chatbot im Talk ohne App hier meinen Code den ich gerne teilen möchte:
- Teil Script erstellen
# Python zuerst prüfen wenn nicht vorhanden dann installieren
# @python3 --version
# Wenn nicht vorhanden
# @sudo apt update
# @sudo apt install python3 -y
#
# Beschreibung
# Der Bot als Systembenutzer im Nextcloud mit Wetter und Chatverlauf.
# Wenn dieses Script verwendet wird kann Wetter abgefragt werden, so:
# wie ist das wetter, wie ist morgen das wetter, wie ist das wetter in Zürich.
# Wenn der Ort nicht erwähnt wird, wird der standard Ort im Script verwendet.
# Zusätzlich gibt es eien Chatverlauf, Gedächnis für die KI:
# Das Gedächnis wir in die Datei die automatisch erstellt wir gespeichert:
# /opt/chat-bridge/chat_history.json
# Um Gedächnis zurückzusetzen, Datei löschen, Dienst neu starten:
# sudo systemctl restart chat-bridge.service
# Datei Inhalt 3. Version Wetter und Chatverlauf Gedächnis START
# @mkdir /opt/chat-bridge/
# @nano /opt/chat-bridge/chat_bridge.py
# Datei Inhalt START
import json
import urllib.request
import urllib.error
import urllib.parse
import ssl
import hmac
import hashlib
import secrets
import re
import os
from http.server import BaseHTTPRequestHandler, HTTPServer
# ==========================================
# KONFIGURATION
# ==========================================
NEXTCLOUD_URL = "https://LAN-IP"
BOT_SECRET = "Mein_40_Zeichen_Passwort"
LOCALAI_URL = "http://192.168.1.231:8080/v1/chat/completions"
LOCALAI_API_KEY = "Mein_Localai_Key"
LOCALAI_MODEL = "fusechat-llama-3.1-8b-instruct"
WEATHER_DEFAULT_LOCATION = "Aarau"
# --- HISTORIE KONFIGURATION ---
HISTORY_FILE = "/opt/chat-bridge/chat_history.json"
# Wie viele Nachrichten-Paare (Frage & Antwort) soll sich die KI maximal pro Raum merken?
MAX_CHAT_PAIRS = 10
# ==========================================
# ==========================================
# PRO-MODE: GLOBALER SPEICHER (Sauber getrennt)
# ==========================================
CHAT_HISTORIES = {}
def load_history():
"""Lädt den Chatverlauf sauber beim Serverstart."""
if os.path.exists(HISTORY_FILE):
try:
with open(HISTORY_FILE, 'r', encoding='utf-8') as f:
print("✅ Chatverlauf erfolgreich von Festplatte geladen.", flush=True)
return json.load(f)
except Exception as e:
print(f"⚠️ Fehler beim Laden der Historie: {e}", flush=True)
return {}
def save_history():
"""Speichert den Chatverlauf persistent auf der Festplatte."""
try:
with open(HISTORY_FILE, 'w', encoding='utf-8') as f:
json.dump(CHAT_HISTORIES, f, ensure_ascii=False, indent=4)
except Exception as e:
print(f"⚠️ Fehler beim Speichern der Historie: {e}", flush=True)
class TalkBridge(BaseHTTPRequestHandler):
last_reply = ""
def do_GET(self):
self.send_response(200)
self.send_header('Content-Type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(b"200 OK - LocalAI Talk Bridge is running!\n")
def do_POST(self):
if self.path == '/hook':
length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(length)
try:
payload = json.loads(body)
target = payload.get('target', {})
room_token = target.get('id')
# --- HIER IST DIE EINZIGE ÄNDERUNG: DER BENUTZERNAME ---
actor = payload.get('actor', {})
actor_name = actor.get('name', actor.get('id', 'Ein Benutzer'))
# -------------------------------------------------------
obj = payload.get('object', {})
content_raw = obj.get('content', '')
user_msg = content_raw
try:
if isinstance(content_raw, str) and content_raw.startswith('{'):
content_json = json.loads(content_raw)
if isinstance(content_json, dict):
user_msg = content_json.get('message', content_raw)
except Exception:
pass
if not user_msg or not room_token:
return
# Verhindert Loop, falls die Bridge auf sich selbst antwortet
if user_msg == TalkBridge.last_reply:
self.send_response(200)
self.end_headers()
return
if "deleted a message" in user_msg or "KI_FEHLER" in user_msg:
self.send_response(200)
self.end_headers()
return
# Bereinigung der Nachricht
words_list = user_msg.split()
clean_for_history = " ".join([w for w in words_list if not w.startswith("@")]).strip()
if not clean_for_history:
clean_for_history = user_msg
# Raum-Historie initialisieren
if room_token not in CHAT_HISTORIES:
CHAT_HISTORIES[room_token] = []
room_history = CHAT_HISTORIES[room_token]
max_messages = MAX_CHAT_PAIRS * 2
# Prüfen, ob der Bot getaggt wurde
is_bot_tagged = "@" in user_msg.lower()
if not is_bot_tagged:
# PASSIVES MITHÖREN MIT NAMEN (Geändert für weniger Rollenspiel-Risiko)
print(f"Passives Mithören: [{actor_name}] schreibt in Raum {room_token}", flush=True)
room_history.append({"role": "user", "content": f"Nachricht von {actor_name}: {clean_for_history}"})
if len(room_history) > max_messages:
CHAT_HISTORIES[room_token] = room_history[-max_messages:]
save_history()
self.send_response(200)
self.end_headers()
return
# Ab hier: Der Bot WURDE getaggt!
print(f"--- ECHTE BOT-ANFRAGE EMPFANGEN ---", flush=True)
print(f"Von: {actor_name} | Raum: {room_token} | Nachricht: {user_msg}", flush=True)
clean_msg = clean_for_history
if not clean_msg:
clean_msg = "hallo"
lower_msg = clean_msg.lower()
# =========================================================
# STRIKTER WETTER-TÜRSTEHER MIT ABSTANDS-MESSUNG
# =========================================================
direct_match = bool(re.search(r'\b(wetter|temperatur)\s+in\b', lower_msg))
words = [w for w in re.split(r'[^\w\ä\ö\ü\ß]+', lower_msg) if w]
wie_index = words.index('wie') if 'wie' in words else -1
wetter_index = -1
for i, w in enumerate(words):
if w in ['wetter', 'temperatur']:
wetter_index = i
break
proximity_match = False
if wie_index != -1 and wetter_index != -1:
if abs(wie_index - wetter_index) <= 6:
proximity_match = True
is_weather_request = direct_match or proximity_match
# Basis-Prompt für diese Runde vorbereiten
if is_weather_request:
print("Wetter-Anfrage erkannt! Hole Daten...", flush=True)
location = WEATHER_DEFAULT_LOCATION
match = re.search(r'\bin\s+([a-zäöüß\.\-]+(?:\s+[a-zäöüß\.\-]+)?)', lower_msg)
if match:
extracted_loc = match.group(1)
for time_word in ["heute", "morgen", "übermorgen", "uebermorgen", "wird", "ist", "bitte"]:
extracted_loc = re.sub(rf'\b{time_word}\b', '', extracted_loc).strip()
if extracted_loc:
location = extracted_loc.title()
day_filter = "heute"
if "übermorgen" in lower_msg or "uebermorgen" in lower_msg:
day_filter = "übermorgen"
elif "morgen" in lower_msg:
day_filter = "morgen"
weather_data_text = self.fetch_weather_data(location, day_filter)
if weather_data_text.startswith("FEHLER"):
current_turn_prompt = f"Anfrage an dich von {actor_name}: {clean_msg}\nEs gab jedoch einen Fehler beim Wetter-Abruf: {weather_data_text}. Bitte teile dem Nutzer höflich mit, dass du das Wetter gerade nicht abrufen kannst."
else:
current_turn_prompt = (
f"Anfrage an dich von {actor_name}: {clean_msg}\n\n"
f"WETTERDATEN (Rohdaten):\n{weather_data_text}\n\n"
"SYSTEM INSTRUCTION: Du bist ein präziser Wetter-Assistent. "
"Formuliere aus den oben genannten Rohdaten einen freundlichen, flüssigen Wetterbericht in 2 bis 3 zusammenhängenden Sätzen. "
"Erwähne zwingend die Temperatur, die Wetterlage und das Regenrisiko. "
"WICHTIG: Nutze AUSSCHLIESSLICH Fließtext! Tabellen, Listen, oder Bullet-Points sind strengstens verboten."
)
else:
current_turn_prompt = f"Anfrage an dich von {actor_name}: {clean_msg}"
# KI anfunken (inklusive Historie)
ai_reply = self.ask_localai(current_turn_prompt, room_history)
if ai_reply.startswith("KI_FEHLER:"):
print(ai_reply, flush=True)
else:
# Anfrage speichern
room_history.append({"role": "user", "content": f"Anfrage an dich von {actor_name}: {clean_msg}"})
room_history.append({"role": "assistant", "content": ai_reply})
if len(room_history) > max_messages:
CHAT_HISTORIES[room_token] = room_history[-max_messages:]
save_history()
TalkBridge.last_reply = ai_reply
self.send_to_nextcloud(room_token, ai_reply)
except Exception as e:
print(f"Verarbeitungsfehler in do_POST: {e}", flush=True)
self.send_response(200)
self.end_headers()
else:
self.send_response(404)
self.end_headers()
def fetch_weather_data(self, location, day_filter):
try:
safe_location = urllib.parse.quote(location)
geo_url = f"https://geocoding-api.open-meteo.com/v1/search?name={safe_location}&count=1&language=de&format=json"
req_geo = urllib.request.Request(geo_url, headers={'User-Agent': 'LocalAI-Talk-Bridge/1.0'})
with urllib.request.urlopen(req_geo, timeout=5) as response:
geo_data = json.loads(response.read().decode('utf-8'))
if not geo_data.get('results'):
return f"FEHLER: Der Ort '{location}' wurde nicht gefunden."
lat = geo_data['results'][0]['latitude']
lon = geo_data['results'][0]['longitude']
real_name = geo_data['results'][0]['name']
weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}¤t=temperature_2m,relative_humidity_2m,surface_pressure,wind_speed_10m,weather_code&daily=weather_code,temperature_2m_max,temperature_2m_min,precipitation_probability_max,wind_speed_10m_max&timezone=auto&forecast_days=3"
req_weather = urllib.request.Request(weather_url, headers={'User-Agent': 'LocalAI-Talk-Bridge/1.0'})
with urllib.request.urlopen(req_weather, timeout=5) as response:
w_data = json.loads(response.read().decode('utf-8'))
wmo = {0:"Sonnig/Klar",1:"Leicht bewölkt",2:"Wolkig",3:"Bedeckt",45:"Nebel",48:"Nebel",51:"Nieselregen",53:"Nieselregen",55:"Nieselregen",61:"Regen",63:"Regen",65:"Starker Regen",71:"Schnee",73:"Schnee",75:"Starker Schneefall",95:"Gewitter"}
day_idx = 0
if day_filter == "morgen": day_idx = 1
elif day_filter == "übermorgen": day_idx = 2
daily = w_data['daily']
d_max = daily['temperature_2m_max'][day_idx]
d_min = daily['temperature_2m_min'][day_idx]
d_rain = daily['precipitation_probability_max'][day_idx]
d_wind = daily['wind_speed_10m_max'][day_idx]
d_code = wmo.get(daily['weather_code'][day_idx], "Unbekannt")
result = f"Ort: {real_name}\nGewünschter Zeitraum: {day_filter}\n"
if day_idx == 0:
cur = w_data['current']
c_temp = cur['temperature_2m']
c_code = wmo.get(cur['weather_code'], "Unbekannt")
result += f"Jetzt gerade: {c_temp}°C, {c_code}\n"
result += f"Vorhersage für {day_filter}: Max {d_max}°C, Min {d_min}°C, {d_code}, Regenrisiko {d_rain}%, Max Wind {d_wind} km/h"
return result
except Exception as e:
return f"FEHLER: Server-Verbindung zur Wetter-API fehlgeschlagen ({e})"
def ask_localai(self, prompt, history):
try:
# Völlig überarbeiteter System-Prompt, der der KI ganz klare Grenzen setzt!
system_prompt = (
"Du bist 'andiki', ein freundlicher, höflicher, hochqualifizierter und hilfsbereiter Assistent. "
"Du bist tierlieb und hast zwei Mudi-Hunde, Bogi und Mino. "
"\n\nWICHTIGE REGELN FÜR DEN GRUPPENCHAT:\n"
"1. Du bist hier der KI-Assistent. Dein Name ist andiki.\n"
"2. Du befindest dich in einem Chatraum mit mehreren menschlichen Benutzern.\n"
"3. Die Nachrichten der Benutzer werden dir in diesem Format übergeben: 'Nachricht von Name: Text' oder 'Anfrage an dich von Name: Text'.\n"
"4. Du bist IMMER andiki. Schlüpfe NIEMALS in die Rolle eines anderen Benutzers! Beginne deine Antworten NIEMALS mit '[Name antwortet]:' oder ähnlichem.\n"
"5. Wenn dich Benutzer A bittet, Benutzer B etwas zu sagen (z.B. 'admin bittet dich: Sag Andreas, dass...'), dann antworte als andiki und sprich Benutzer B direkt an (z.B. 'Hallo Andreas, admin hat recht, weil...').\n"
"6. Antworte flüssig, natürlich und hilfreich. Nutze Emojis wie 😊, 💙, 👍."
)
messages = [{"role": "system", "content": system_prompt}]
messages.extend(history)
messages.append({"role": "user", "content": prompt})
data = {
"model": LOCALAI_MODEL,
"messages": messages,
"temperature": 0.8
}
body = json.dumps(data).encode('utf-8')
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {LOCALAI_API_KEY}'
}
req = urllib.request.Request(LOCALAI_URL, data=body, headers=headers, method='POST')
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
with urllib.request.urlopen(req, context=context, timeout=120) as response:
resp_body = response.read().decode('utf-8')
resp_json = json.loads(resp_body)
reply = resp_json.get('choices', [{}])[0].get('message', {}).get('content', '')
return reply.strip()
except urllib.error.HTTPError as e:
error_details = e.read().decode('utf-8', errors='ignore')
print(f"LOCALAI HTTP FEHLER (Code {e.code}): {error_details}", flush=True)
return f"KI_FEHLER: LocalAI API error: {error_details}"
except Exception as e:
print(f"ALLGEMEINER LOCALAI FEHLER: {e}", flush=True)
return f"KI_FEHLER: {e}"
def send_to_nextcloud(self, token, message):
url = f"{NEXTCLOUD_URL}/ocs/v2.php/apps/spreed/api/v1/bot/{token}/message"
payload_dict = {"message": message, "replyTo": 0}
body_bytes = json.dumps(payload_dict, ensure_ascii=False).encode('utf-8')
random_string = secrets.token_hex(16)
data_to_sign = random_string + message
signature = hmac.new(
BOT_SECRET.encode('utf-8'),
data_to_sign.encode('utf-8'),
hashlib.sha256
).hexdigest()
headers = {
'Content-Type': 'application/json',
'OCS-APIRequest': 'true',
'Accept': 'application/json',
'X-Nextcloud-Talk-Bot-Random': random_string,
'X-Nextcloud-Talk-Bot-Signature': signature
}
req = urllib.request.Request(url, data=body_bytes, headers=headers, method='POST')
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
try:
with urllib.request.urlopen(req, context=context) as response:
print(f"Antwort erfolgreich an Nextcloud gesendet. Status: {response.getcode()}", flush=True)
except urllib.error.HTTPError as e:
error_details = e.read().decode('utf-8', errors='ignore')
print(f"Nextcloud HTTP {e.code} Fehlerdetails: {error_details}", flush=True)
except Exception as e:
print(f"Netzwerkfehler beim Senden an Nextcloud: {e}", flush=True)
if __name__ == '__main__':
# PRO-MODE: Laden der Historie beim allerersten Start des Dienstes
CHAT_HISTORIES.update(load_history())
server = HTTPServer(('0.0.0.0', 8788), TalkBridge)
print("Die LocalAI Chat-Bridge lauscht auf Port 8788...", flush=True)
server.serve_forever()
# Datei Inhalt ENDE
- Teil Dienst erstellen, aktivieren, starten und im Nextcloud aktivieren
# Dienst erstellen und aktivieren
@nano /etc/systemd/system/chat-bridge.service
# Datei Inhalt chat-bridge.service
[Unit]
Description=LocalAI Nextcloud Talk Bridge
After=network.target
[Service]
Type=simple
User=root
ExecStart=/usr/bin/python3 /opt/chat-bridge/chat_bridge.py
Restart=always
RestartSec=5
[Install]
WantedBy=multi-user.target
# Datei Inhalt chat-bridge.service ENDE
@systemctl daemon-reload
@systemctl enable chat-bridge.service
@systemctl start chat-bridge.service
#
# Bot im Nextcloud aktivieren
# Im Terminal den Chatbot im Nextcloud hinzufügen.
# andiki ist der KI (Bot) Name und kann nach Wunsch angepasst werden, über dieser Name wird der Bot im Chat angesprochen.
# Wenn die 1. Version von chat_bridge.py verwendet wird ist das Passwort nur ein dummy und kann frei gewählt werden,
# da Nextcloud ein passwort im Setup für Chatbot Installation erwartet.
# Wenn die 2. Version von chat_bridge.py verwendet wird muss dieses Passwort im # chat_bridge.py unter BOT_SECRET eingetragen werden.
#
@sudo -u www-data php /var/www/html/nextcloud/occ talk:bot:install "andiki" "Mein_40_Zeichen_Passwort" "http://127.0.0.1:8788/hook" -f webhook -f response
#
# Testen ob Bot verfügbar ist
@sudo -u www-data php /var/www/html/nextcloud/occ talk:bot:list
# Jetzt im Talk in einer Unterhaltung im drei Punktmenu
# den Bot unter Bot aktivieren und testen @andiki hallo
- Teil Wartung
# Wartung
# Wenn gewünscht wird den chatbot zu deinstallieren.
# 2 ist install id. id kann mit dem Befehl occ talk:bot:list ausgelesen werden.
# Vorher den Bot im Talk Unterhaltungseinstellungen wieder deaktivieren
# ID auslesen
@sudo -u www-data php /var/www/html/nextcloud/occ talk:bot:list
# Bot deinstallieren
@sudo -u www-data php /var/www/html/nextcloud/occ talk:bot:delete 2
# Status des Dienst anzeigen
@systemctl status chat-bridge.service
# Protokoll anzeigen
@sudo journalctl -u chat-bridge.service -f
Sollte es gewünscht werden werde ich den Code erweiter so, dass auch Internetsuche geht über KI Anfrage dazu wird ein searxing benötigt ist aber in 2 Minuten installiert wenn Docker vorhanden ist.