Compare commits

...

2 commits

2 changed files with 50 additions and 7 deletions

54
app.py
View file

@ -2,6 +2,7 @@ import os
import select
import sys
import time
import unicodedata
from collections import deque
from importlib import import_module
from pathlib import Path
@ -53,7 +54,9 @@ TOOL_REGISTRY = {
"module": "intelligent_tool",
"class": "IntelligentCommunicationTool",
"kwargs": {
"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small")
"model": os.environ.get(
"MISTLE_LLM_MODEL", "mistral/mistral-small"
)
},
"description": "Uses an LLM to craft a polite reply to private tells.",
},
@ -61,7 +64,9 @@ TOOL_REGISTRY = {
"module": "intelligent_tool",
"class": "IntelligentCommunicationTool",
"kwargs": {
"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small")
"model": os.environ.get(
"MISTLE_LLM_MODEL", "mistral/mistral-small"
)
},
"description": "Alias of 'intelligent'. Uses an LLM to craft a polite reply to private tells.",
},
@ -74,6 +79,38 @@ TOOL_DESCRIPTIONS = {
from telnetclient import TelnetClient
_UMLAUT_TRANSLATION = str.maketrans(
{
"ä": "ae",
"ö": "oe",
"ü": "ue",
"Ä": "Ae",
"Ö": "Oe",
"Ü": "Ue",
"ß": "ss",
}
)
def sanitize_for_mud(text: str) -> str:
"""Return ASCII-only text suitable for Silberland's input parser."""
if not text:
return ""
replaced = text.translate(_UMLAUT_TRANSLATION)
normalized = unicodedata.normalize("NFKD", replaced)
cleaned: list[str] = []
for ch in normalized:
if unicodedata.combining(ch):
continue
code = ord(ch)
if 32 <= code <= 126:
cleaned.append(ch)
else:
cleaned.append("?")
return "".join(cleaned)
class SessionState:
"""Share Telnet session state safely across threads."""
@ -113,8 +150,9 @@ class SessionState:
self._listeners_lock = Lock()
def send(self, client: TelnetClient, message: str) -> None:
sanitized = sanitize_for_mud(message)
with self._send_lock:
client.send(message)
client.send(sanitized)
def tool_send(
self,
@ -125,12 +163,14 @@ class SessionState:
stop_event: Event,
) -> bool:
"""Send on behalf of the tool while respecting a minimum cadence."""
sanitized = sanitize_for_mud(message)
while not stop_event.is_set():
with self._send_lock:
now = time.time()
elapsed = now - self._last_tool_send
if elapsed >= min_interval:
client.send(message)
client.send(sanitized)
self._last_tool_send = now
return True
wait_time = min_interval - elapsed
@ -340,10 +380,10 @@ def login(
state.update_output(banner)
if user:
client.send(user)
client.send(sanitize_for_mud(user))
time.sleep(0.2)
if password:
client.send(password)
client.send(sanitize_for_mud(password))
response = client.receive(timeout=response_timeout)
if response:
@ -414,7 +454,7 @@ def graceful_shutdown(
if state:
state.send(client, exit_command)
else:
client.send(exit_command)
client.send(sanitize_for_mud(exit_command))
farewell = client.receive(timeout=2.0)
if farewell:
print(farewell, end="" if farewell.endswith("\n") else "\n")

View file

@ -22,6 +22,9 @@ class IntelligentCommunicationTool(Tool):
system_prompt: str = (
"Du bist Mistle, ein hilfsbereiter MUD-Bot. "
"Antworte freundlich und knapp in deutscher Sprache."
"Sprich informell und freundlich."
"Antworte immer auf Deutsch."
"Verwende emoticons, wenn es angebracht ist: :) ;) 8) :( ;( :P X) XD :D"
)
temperature: float = 0.7
max_output_tokens: int = 120