class RailPunk:
def __init__(self, limit=5, db: AbsDictionaryDB | None = None):
self.ec = EventChatV2(limit) # the memory: key -> weighted replies (limit = replies kept per key)
self.context = "stand by" # the last line the bot is "on"; monolog continues from here
self.db: AbsDictionaryDB | None = db # optional storage; None = memory only
# DB wrapper exists only if a DB was given, this is what auto-picks DB or no DB
self.eliza_wrapper: ElizaDBWrapper | None = ElizaDBWrapper() if db is not None else None
self.funcs: dict[str, PopulatorFunc] = {} # populators: small helpers that deduce extra knowledge from input
self.add_populator(KeysFunnel()) # default populator: indexes lines by their canonical key
self.removables: set[str] = Tokenizer.exclusions # filler words dropped when building canonical keys
self.skip = False # True after a reply(), makes the next monolog take an extra step
# ---------- populators ----------
def add_populator(self, func: PopulatorFunc):
# register a populator under its name (empty names are ignored)
if len(func.regex) > 0:
self.funcs[func.regex] = func
def _populate(self, str1: str):
# run every populator on the input
for func in self.funcs.values():
func.populate(self, str1)
def set_context(self, context):
# manually set the current context (empty values are ignored)
if not context:
return
self.context = context
# ---------- replying ----------
def _fetch(self, key: str, latest: bool) -> str:
# the ONLY place that cares about DB or no DB
if self.eliza_wrapper is None:
# no DB: read from memory only
return self.ec.response_latest(key) if latest else self.ec.response(key)
# with DB: wrapper loads the key from the DB first (once), then reads memory
if latest:
return self.eliza_wrapper.respond_latest(key, self.ec, self.db)
return self.eliza_wrapper.respond(key, self.ec, self.db)
def reply(self, ear: str, latest: bool = False) -> str:
# answer an input; latest=True gives the newest stored reply instead of a weighted random one
self.skip = True
result = self._fetch(ear, latest) # try the exact input first
if result:
return result
# fallback: try the canonical key (lowercase, sorted, filler words removed)
return self._fetch(Tokenizer.canonical_key(ear, self.removables), latest)
def reply_monolog(self, latest: bool = False) -> str:
# the bot talks to itself, following the chain from self.context
passes = 2 if self.skip else 1 # after a reply() take one extra step first
self.skip = False
result = ""
for _ in range(passes):
if not self.context:
return ""
result = self._fetch(self.context, latest)
if result:
self.context = result # move forward along the chain
return result
# ---------- learning ----------
def learn(self, ear):
# full learning: deduce + link the previous line to this one
if not ear or ear == self.context: # ignore empty or repeated line
return
self._populate(ear) # populators deduce extra knowledge
self.ec.add_key_value(self.context, ear) # previous line -> this line
self.context = ear # this line becomes the new context
def learn_only_deduce(self, ear: str):
# populators only: deduce knowledge but do NOT link previous line -> this line
if not ear:
return
self._populate(ear)
self.context = ear # still move context so monolog and the next learn() continue from here
def learn_key_value(self, context, reply):
# teach one key -> reply pair directly
self.ec.add_key_value(context, reply)
def feed_key_value_pairs(self, kv_list: list[AXKeyValuePair]):
# teach many key -> reply pairs at once
if not kv_list:
return
for kv in kv_list:
self.learn_key_value(kv.get_key(), kv.get_value())
# ---------- saving / clearing ----------
def save_learned_data(self):
# write everything learned since the last save to the DB (does nothing without a DB)
if self.eliza_wrapper is None:
return
self.eliza_wrapper.sleep_n_save(self.ec, self.db)
def clear(self):
"""Wipe in-memory state, and DB-backed state if this instance has a DB."""
self.ec.clear() # forget all learned pairs
self.context = "stand by" # reset context
self.skip = False
if self.eliza_wrapper is not None:
self.eliza_wrapper.clear() # forget which keys were already loaded from the DB
if self.db is not None:
self.db.clear() # wipe the DB too
