Markus Clauss DIRU Vetsuisse
Properly implement thinking states with both visual and text feedback
540fde3
"""
🇨🇭 Apertus Dialekt-Konsil mit Roundtable
Ein AI-Konsil mit Schweizerdeutsch, Bayrisch und Schwäbisch Sprechern
Mit visuellem Roundtable und Moderator - Angepasst an Consilium Format
"""
import gradio as gr
import os
from huggingface_hub import InferenceClient
import json
import time
from datetime import datetime
import uuid
from gradio_consilium_roundtable import consilium_roundtable
# Configuration
HF_TOKEN = os.environ.get('HF_TOKEN', '')
# Initialize Apertus client with publicai provider
client = None
if HF_TOKEN:
client = InferenceClient(
provider="publicai",
api_key=HF_TOKEN
)
# Avatar URLs für die Dialekt-Sprecher - Wappen und Flaggen
avatar_images = {
"Ueli": "https://upload.wikimedia.org/wikipedia/commons/7/7d/Wappen_Basel-Stadt_matt.svg", # Basel Wappen
"Sepp": "https://upload.wikimedia.org/wikipedia/commons/d/d2/Bayern_Wappen.svg", # Bayern Wappen
"Karl": "https://upload.wikimedia.org/wikipedia/commons/0/0f/Lesser_coat_of_arms_of_Baden-W%C3%BCrttemberg.svg", # Baden-Württemberg Wappen
"Dr. Müller": "https://upload.wikimedia.org/wikipedia/commons/thumb/0/08/Flag_of_Switzerland_%28Pantone%29.svg/100px-Flag_of_Switzerland_%28Pantone%29.svg.png" # Schweiz Flagge für neutralen Moderator
}
# Dialekt-Persönlichkeiten
COUNCIL_MEMBERS = {
"Ueli": {
"full_name": "Ueli (Baseldytsch)",
"emoji": "🇨🇭",
"dialect": "Baseldytsch",
"location": "Basel",
"personality": "kultiviert, humorvoll, fasnachts-begeistert",
"thinking_message": "dänggt nooch...", # Baseldytsch für "denkt nach"
"system_prompt": """Du bist Ueli aus Basel und sprichst Baseldytsch.
Du bist kultiviert, humorvoll und liebst die Basler Fasnacht.
Verwende typische Basler Ausdrücke und Redewendungen.
Beispiele: 'Sali zämme', 'Das isch e gueti Sach, gäll', 'Mir Baasler wisse was guet isch', 'Das brucht e bitzeli Köpfli'"""
},
"Sepp": {
"full_name": "Sepp (Bayrisch)",
"emoji": "🥨",
"dialect": "Bayrisch",
"location": "München",
"personality": "gesellig, direkt, traditionsbewusst",
"thinking_message": "überlegt si...", # Bayrisch für "überlegt"
"system_prompt": """Du bist Sepp aus München und sprichst AUSSCHLIESSLICH BAYRISCH!
Du bist gesellig, direkt und traditionsbewusst.
ABSOLUT WICHTIG: Verwende NUR bayrische Ausdrücke, NIEMALS Schweizerdeutsch oder Schwäbisch!
Typisch bayrisch: 'Servus mitanand', 'Des is fei scho recht', 'I moan', 'Des passt scho', 'Mia san mia', 'Des is hoid so', 'Ja mei'
VERBOTEN: Schweizer Wörter wie 'isch', 'gäll', 'zämme' oder schwäbische wie 'schaffe', 'koscht'
NIEMALS 'isch' verwenden - immer 'is' oder 'ist'!"""
},
"Karl": {
"full_name": "Karl (Schwäbisch)",
"emoji": "🏰",
"dialect": "Schwäbisch",
"location": "Stuttgart",
"personality": "sparsam, fleißig, erfinderisch",
"thinking_message": "grübelt grad...", # Schwäbisch für "grübelt gerade"
"system_prompt": """Du bist Karl aus Stuttgart und sprichst SCHWÄBISCH (nicht bayrisch!).
Du bist sparsam, fleißig und erfinderisch.
WICHTIG: Verwende NUR schwäbische Ausdrücke, NICHT bayrische oder andere Dialekte!
Typisch schwäbisch: 'Grüß Gott mitanand', 'Des isch halt so', 'Mir müsset schaffe', 'Des koscht Geld', 'Schwätz net so viel', 'No ja, des goht scho'
NIEMALS verwenden: bayrische Wörter wie 'fei', 'miassn', 'hoid', 'mia'"""
}
}
# Moderator
MODERATOR = {
"name": "Dr. Müller",
"full_name": "Dr. Müller (Moderator)",
"emoji": "🎓",
"system_prompt": """Du bist Dr. Müller, ein neutraler Moderator für das Dialekt-Konsil.
Deine Aufgaben:
1. Fasse jede Diskussionsrunde zusammen
2. Bewerte den Konsensgrad auf einer Skala von 1-10
3. Identifiziere Gemeinsamkeiten und Unterschiede
4. Schlage Kompromisse vor wenn nötig
Sei neutral, professionell und präzise in deinen Bewertungen."""
}
# Session storage für isolierte Diskussionen
sessions = {}
class DialektKonsil:
def __init__(self, session_id=None):
self.session_id = session_id or str(uuid.uuid4())
self.conversation_history = []
self.current_topic = ""
self.consensus_score = 0
self.round_number = 0
# Roundtable state im Consilium Format
self.roundtable_state = {
"participants": ["Ueli", "Sepp", "Karl", "Dr. Müller"],
"messages": [],
"currentSpeaker": None,
"thinking": [],
"showBubbles": [],
"avatarImages": avatar_images
}
def generate_response(self, member_name, topic, previous_responses):
"""Generate response for a council member"""
member = COUNCIL_MEMBERS[member_name]
# Build context but isolate dialects better
context = f"Thema der Diskussion: {topic}\n\n"
if previous_responses:
context += "Bisherige Beiträge anderer Sprecher:\n"
for resp in previous_responses:
# Kürze die Responses drastisch um Kopieren zu vermeiden
context += f"- {resp['speaker']}: {resp['text'][:100]}...\n"
# Verstärke die Dialekt-Isolation
prompt = f"""{member['system_prompt']}
{context}
ABSOLUT WICHTIG: Du bist {member_name} aus {member['location']} und sprichst {member['dialect']}!
AUFTRAG: Gib deine EIGENE, EINZIGARTIGE Meinung zum Thema.
- Verwende AUSSCHLIESSLICH {member['dialect']} Dialekt
- Maximal 2 kurze Sätze
- KOPIERE NIEMALS andere Sprecher
- Entwickle DEINE EIGENE Perspektive
- Verwende {member_name}'s typische Ausdrücke
Antworte jetzt als {member_name} in {member['dialect']}:"""
try:
# Generate response using Apertus via publicai
completion = client.chat.completions.create(
model="swiss-ai/Apertus-8B-Instruct-2509",
messages=[
{"role": "system", "content": member['system_prompt']},
{"role": "user", "content": prompt}
],
max_tokens=80, # Noch kürzer um Abschneiden zu vermeiden
temperature=0.9, # Höhere Variabilität für unterschiedliche Antworten
top_p=0.95,
frequency_penalty=0.5, # Reduziere Wiederholungen
presence_penalty=0.3 # Ermutige neue Inhalte
)
response = completion.choices[0].message.content.strip()
# Aggressive Response-Cleanup um Kopieren zu verhindern
# Entferne Namen-Präfixe aller Sprecher
for name in ["Ueli", "Sepp", "Karl", "Dr.", "Müller"]:
if response.startswith(f"{name}:"):
response = response[len(name)+1:].strip()
if response.startswith(f"{name} "):
response = response[len(name)+1:].strip()
# Entferne Prompt-Wiederholungen
cleanup_phrases = ["Deine Antwort", "Antworte", "Als " + member_name,
"Ich bin " + member_name, member['dialect']]
for phrase in cleanup_phrases:
if phrase in response:
response = response.split(phrase)[0].strip()
# Stelle sicher dass Antwort kurz bleibt und vollständig ist
sentences = response.split('. ')
if len(sentences) > 2:
response = '. '.join(sentences[:2])
if not response.endswith('.'):
response += '.'
# Entferne unvollständige Sätze am Ende
if response.endswith(', will\'s') or response.endswith(', dass'):
words = response.split()
response = ' '.join(words[:-2]) + '.'
return response
except Exception as e:
# Fallback-Antworten je nach Sprecher
fallbacks = {
"Ueli": "Jo, das isch e schwierigi Frog, gäll. Mir müesse do guet überlegge.",
"Sepp": "Mei, des is fei a schwierige Sach. Da muas ma gscheid drüber nachdenka.",
"Karl": "Ha, des isch halt so a Sach. Do muss mr gründlich überlege."
}
return fallbacks.get(member_name, f"*hüstel* ({str(e)[:50]}...)")
def generate_moderator_summary(self, round_responses):
"""Generate moderator summary and consensus score - mit besserem Modell"""
context = "Diskussionsrunde abgeschlossen. Folgende Beiträge:\n"
for resp in round_responses:
context += f"- {resp['speaker']}: {resp['text']}\n"
prompt = f"""{MODERATOR['system_prompt']}
{context}
Aufgabe als neutraler Moderator:
1. Fasse die Hauptpunkte präzise zusammen (2-3 Sätze)
2. Bewerte den Konsensgrad objektiv von 1-10 (1=völlig uneinig, 10=vollständige Einigung)
3. Identifiziere klar die Gemeinsamkeiten
4. Nenne konkrete Unterschiede falls vorhanden
Format (GENAU befolgen):
ZUSAMMENFASSUNG: [Präzise Zusammenfassung der Diskussion]
KONSENSGRAD: [Zahl 1-10]
GEMEINSAMKEITEN: [Was alle Sprecher teilen]
UNTERSCHIEDE: [Konkrete Meinungsunterschiede]"""
try:
# Verwende das stärkere Apertus 70B Modell für bessere Moderation
completion = client.chat.completions.create(
model="swiss-ai/Apertus-70B-Instruct-2509", # Apertus 70B für bessere Moderation
messages=[
{"role": "system", "content": MODERATOR['system_prompt']},
{"role": "user", "content": prompt}
],
max_tokens=300,
temperature=0.5 # Niedrigere Temperatur für konsistentere Analyse
)
response = completion.choices[0].message.content.strip()
# Parse consensus score mit besserer Logik
consensus_score = 5 # Default
if "KONSENSGRAD:" in response:
try:
score_line = [l for l in response.split('\n') if 'KONSENSGRAD:' in l][0]
# Extrahiere alle Zahlen und nimm die erste zwischen 1-10
import re
scores = re.findall(r'\b([1-9]|10)\b', score_line)
if scores:
consensus_score = int(scores[0])
consensus_score = max(1, min(10, consensus_score))
except Exception as parse_error:
print(f"⚠️ Konsensgrad-Parsing Fehler: {parse_error}")
return response, consensus_score
except Exception as e:
print(f"❌ Moderator-Fehler: {e}")
# Fallback: Verwende Apertus falls Llama nicht verfügbar
try:
fallback_completion = client.chat.completions.create(
model="swiss-ai/Apertus-8B-Instruct-2509",
messages=[
{"role": "system", "content": MODERATOR['system_prompt']},
{"role": "user", "content": prompt}
],
max_tokens=250,
temperature=0.7
)
response = fallback_completion.choices[0].message.content.strip()
consensus_score = 5
if "KONSENSGRAD:" in response:
try:
score_line = [l for l in response.split('\n') if 'KONSENSGRAD:' in l][0]
consensus_score = int(''.join(filter(str.isdigit, score_line)))
consensus_score = max(1, min(10, consensus_score))
except:
pass
return response, consensus_score
except Exception as fallback_error:
return f"Runde abgeschlossen. Konsens wird noch analysiert. ({str(fallback_error)[:50]}...)", 5
def update_visual_state(self, **kwargs):
"""Update the visual roundtable state"""
self.roundtable_state.update(kwargs)
return json.dumps(self.roundtable_state)
def run_discussion_round(self):
"""Run one round of discussion with visual updates"""
self.round_number += 1
responses = []
# Reset messages for new round but keep participants
self.roundtable_state["messages"] = []
# Each member speaks once per round
for member_name in COUNCIL_MEMBERS.keys():
member = COUNCIL_MEMBERS[member_name]
thinking_msg = member.get("thinking_message", "denkt nach...")
# Show thinking state with visual indicator
self.roundtable_state["thinking"] = [member_name]
self.roundtable_state["currentSpeaker"] = None
# Also add dialect-specific thinking message
self.roundtable_state["messages"].append({
"speaker": member_name,
"text": f"{member['emoji']} *{thinking_msg}*"
})
yield self.update_visual_state()
time.sleep(1)
# Generate response
response = self.generate_response(member_name, self.current_topic, responses)
# Clear thinking, set as current speaker, update message
self.roundtable_state["thinking"] = []
self.roundtable_state["currentSpeaker"] = member_name
self.roundtable_state["messages"][-1] = {
"speaker": member_name,
"text": f"{member['emoji']} {response}"
}
responses.append({
"speaker": member_name,
"text": response,
"emoji": COUNCIL_MEMBERS[member_name]["emoji"],
"timestamp": datetime.now().strftime("%H:%M:%S"),
"round": self.round_number
})
# Show speech bubble
if member_name not in self.roundtable_state["showBubbles"]:
self.roundtable_state["showBubbles"].append(member_name)
yield self.update_visual_state()
# Add to conversation history
self.conversation_history.append(responses[-1])
time.sleep(2)
# Generate moderator summary
# Show moderator thinking with visual indicator
self.roundtable_state["thinking"] = ["Dr. Müller"]
self.roundtable_state["currentSpeaker"] = None
self.roundtable_state["messages"].append({
"speaker": "Dr. Müller",
"text": "🎓 *analysiert die Diskussion...*"
})
yield self.update_visual_state()
time.sleep(1)
moderator_summary, self.consensus_score = self.generate_moderator_summary(responses)
# Clear thinking, set as speaker, update message
self.roundtable_state["thinking"] = []
self.roundtable_state["currentSpeaker"] = "Dr. Müller"
self.roundtable_state["messages"][-1] = {
"speaker": "Dr. Müller",
"text": f"🎓 {moderator_summary}"
}
if "Dr. Müller" not in self.roundtable_state["showBubbles"]:
self.roundtable_state["showBubbles"].append("Dr. Müller")
self.conversation_history.append({
"speaker": MODERATOR["name"],
"text": moderator_summary,
"emoji": MODERATOR["emoji"],
"timestamp": datetime.now().strftime("%H:%M:%S"),
"round": self.round_number,
"is_moderator": True
})
yield self.update_visual_state()
def format_history(self):
"""Format conversation history for display"""
if not self.conversation_history:
return "*Keine Diskussion gestartet*"
output = f"## 🗣️ Dialekt-Konsil: {self.current_topic}\n\n"
current_round = 0
for entry in self.conversation_history:
if entry.get("round", 0) > current_round:
current_round = entry["round"]
output += f"\n### 🔄 Runde {current_round}\n\n"
if entry.get("is_moderator"):
output += f"**{entry['emoji']} {entry['speaker']}** - Zusammenfassung:\n"
output += f"```\n{entry['text']}\n```\n"
output += f"**Konsensgrad: {self.consensus_score}/10**\n\n"
else:
full_name = COUNCIL_MEMBERS[entry['speaker']]['full_name']
output += f"**{entry['emoji']} {full_name}** ({entry['timestamp']}):\n"
output += f"> {entry['text']}\n\n"
return output
def get_or_create_session(session_id=None):
"""Get existing session or create new one"""
if not session_id:
session_id = str(uuid.uuid4())
if session_id not in sessions:
sessions[session_id] = DialektKonsil(session_id)
return sessions[session_id], session_id
def start_new_discussion(topic):
"""Start a new discussion with a fresh session"""
if not topic:
initial_state = json.dumps({
"participants": ["Ueli", "Sepp", "Karl", "Dr. Müller"],
"messages": [],
"currentSpeaker": None,
"thinking": [],
"showBubbles": [],
"avatarImages": avatar_images
})
return initial_state, None, "*Keine Diskussion gestartet*", "❌ Bitte geben Sie ein Thema ein!"
if not client:
initial_state = json.dumps({
"participants": ["Ueli", "Sepp", "Karl", "Dr. Müller"],
"messages": [],
"currentSpeaker": None,
"thinking": [],
"showBubbles": [],
"avatarImages": avatar_images
})
return initial_state, None, "*Keine Diskussion gestartet*", "❌ HF_TOKEN nicht konfiguriert. Bitte in Space Settings setzen."
# Create new session
konsil, session_id = get_or_create_session()
konsil.current_topic = topic
konsil.conversation_history = []
konsil.round_number = 0
konsil.consensus_score = 0
# Reset roundtable state
konsil.roundtable_state = {
"participants": ["Ueli", "Sepp", "Karl", "Dr. Müller"],
"messages": [],
"currentSpeaker": None,
"thinking": [],
"showBubbles": [],
"avatarImages": avatar_images
}
return (
json.dumps(konsil.roundtable_state),
session_id,
konsil.format_history(),
f"✅ Diskussion gestartet: {topic}\n\n➡️ Klicken Sie auf 'Nächste Runde' um die erste Runde zu beginnen."
)
def run_next_round(session_id):
"""Run the next discussion round with streaming updates"""
if not session_id or session_id not in sessions:
empty_state = json.dumps({
"participants": ["Ueli", "Sepp", "Karl", "Dr. Müller"],
"messages": [],
"currentSpeaker": None,
"thinking": [],
"showBubbles": [],
"avatarImages": avatar_images
})
return empty_state, "*Keine Diskussion gestartet*", "❌ Keine aktive Session. Bitte starten Sie eine neue Diskussion."
konsil = sessions[session_id]
if not konsil.current_topic:
return json.dumps(konsil.roundtable_state), "*Keine Diskussion gestartet*", "❌ Kein Thema gesetzt. Bitte starten Sie eine neue Diskussion."
if konsil.consensus_score >= 8:
return json.dumps(konsil.roundtable_state), konsil.format_history(), f"✅ Konsens bereits erreicht (Score: {konsil.consensus_score}/10)"
# Run discussion round with generator for updates
final_state = None
for state in konsil.run_discussion_round():
final_state = state
history = konsil.format_history()
if konsil.consensus_score >= 8:
status = f"🎉 Konsens erreicht! (Score: {konsil.consensus_score}/10)"
elif konsil.round_number >= 5:
status = f"🤝 Maximale Rundenzahl erreicht. Finaler Konsensgrad: {konsil.consensus_score}/10"
else:
status = f"Runde {konsil.round_number} abgeschlossen. Konsensgrad: {konsil.consensus_score}/10"
return final_state, history, status
def create_interface():
"""Create Gradio interface with roundtable visualization"""
with gr.Blocks(
title="Apertus Dialekt-Konsil",
theme=gr.themes.Soft(),
css="""
.roundtable-container {
max-width: 800px;
margin: 0 auto;
}
"""
) as demo:
gr.Markdown("""
# 🇨🇭🥨🏰 Apertus Dialekt-Konsil mit Roundtable
Ein KI-gestütztes Konsil mit drei süddeutschen Dialekt-Sprechern und Moderator:
- 🇨🇭 **Ueli** aus Basel (Baseldytsch)
- 🥨 **Sepp** aus München (Bayrisch)
- 🏰 **Karl** aus Stuttgart (Schwäbisch)
- 🎓 **Dr. Müller** als neutraler Moderator
Geben Sie ein Thema ein und beobachten Sie die Diskussion am runden Tisch!
""")
# Hidden session ID
session_id = gr.State()
with gr.Row():
with gr.Column(scale=2):
topic_input = gr.Textbox(
label="💭 Diskussionsthema",
placeholder="z.B. 'Wie sollten wir mit KI in der Bildung umgehen?' oder 'Was ist das beste Essen?'",
lines=2
)
with gr.Row():
start_btn = gr.Button(
"🚀 Neue Diskussion starten",
variant="primary",
size="lg"
)
next_round_btn = gr.Button(
"➡️ Nächste Runde",
variant="secondary",
size="lg"
)
# Status display
status_display = gr.Markdown(value="*Bereit für neue Diskussion*")
# Roundtable visualization
with gr.Row():
with gr.Column(scale=1, elem_classes="roundtable-container"):
# Initial state im richtigen Format
initial_roundtable_state = json.dumps({
"participants": ["Ueli", "Sepp", "Karl", "Dr. Müller"],
"messages": [],
"currentSpeaker": None,
"thinking": [],
"showBubbles": [],
"avatarImages": avatar_images
})
roundtable = consilium_roundtable(
value=initial_roundtable_state,
label="Diskussionsrunde",
min_width=600
)
# Consensus score display
with gr.Row():
consensus_display = gr.Markdown(
value="**Konsensgrad:** 0/10 ⚪"
)
# Discussion history
with gr.Row():
with gr.Column():
history_display = gr.Markdown(
label="Diskussionsverlauf",
value="*Warten auf Diskussionsstart...*"
)
# Example topics
gr.Examples(
examples=[
["Wie können wir die Umwelt besser schützen?"],
["Was ist wichtiger: Tradition oder Innovation?"],
["Sollte KI in Schulen eingesetzt werden?"],
["Was macht eine gute Nachbarschaft aus?"],
["Wie finden wir Work-Life-Balance?"],
["Brauchen wir mehr oder weniger Regeln?"]
],
inputs=topic_input
)
# Event handlers
start_btn.click(
start_new_discussion,
inputs=[topic_input],
outputs=[roundtable, session_id, history_display, status_display]
)
next_round_btn.click(
run_next_round,
inputs=[session_id],
outputs=[roundtable, history_display, status_display]
)
# Update consensus display when roundtable updates
def update_consensus(roundtable_state):
if roundtable_state:
try:
state = json.loads(roundtable_state) if isinstance(roundtable_state, str) else roundtable_state
# Try to extract consensus from moderator messages
messages = state.get("messages", [])
for msg in reversed(messages):
if msg.get("speaker") == "Dr. Müller" and "KONSENSGRAD:" in msg.get("text", ""):
import re
match = re.search(r'KONSENSGRAD:\s*(\d+)', msg["text"])
if match:
score = int(match.group(1))
return f"**Konsensgrad:** {score}/10 {'🟢' if score >= 8 else '🟡' if score >= 5 else '🔴'}"
except:
pass
return "**Konsensgrad:** Bewertung läuft..."
roundtable.change(
update_consensus,
inputs=[roundtable],
outputs=[consensus_display]
)
# Auto-refresh für bessere Visualisierung während der Diskussion
def refresh_roundtable(session_id):
if session_id and session_id in sessions:
return json.dumps(sessions[session_id].roundtable_state)
return json.dumps({
"participants": ["Ueli", "Sepp", "Karl", "Dr. Müller"],
"messages": [],
"currentSpeaker": None,
"thinking": [],
"showBubbles": [],
"avatarImages": avatar_images
})
# Timer für Auto-Update (alle 1 Sekunde)
gr.Timer(1.0).tick(refresh_roundtable, inputs=[session_id], outputs=[roundtable])
gr.Markdown("""
---
### 📌 Hinweise
- Das Konsil verwendet das **Apertus-8B** Modell für authentische Dialekt-Generierung
- Der Moderator bewertet nach jeder Runde den Konsensgrad (1-10)
- Die Diskussion endet bei Konsensgrad ≥ 8 oder nach 5 Runden
- Benötigt HF_TOKEN in den Space Settings
### 🎯 Features
- **Visueller Roundtable**: Sehen Sie die Teilnehmer am runden Tisch
- **Authentische Dialekte**: Schweizerdeutsch, Bayrisch, Schwäbisch
- **Neutraler Moderator**: Fasst zusammen und bewertet Konsens
- **Interaktive Runden**: Verfolgen Sie jede Diskussionsrunde einzeln
- **Speech Bubbles**: Live-Anzeige der Diskussionsbeiträge
*Powered by Apertus Swiss AI* 🇨🇭
""")
return demo
# Launch the app
if __name__ == "__main__":
print("🇨🇭🥨🏰 Apertus Dialekt-Konsil mit Roundtable")
print(f"HF Token configured: {bool(HF_TOKEN)}")
demo = create_interface()
demo.launch()