242 lines
8.8 KiB
Python
Executable file
242 lines
8.8 KiB
Python
Executable file
#!/usr/bin/env python
|
|
|
|
import anthropic
|
|
import json
|
|
import requests
|
|
import sys
|
|
|
|
from os import listdir, path
|
|
# from os.path import join, isfile
|
|
from pathlib import Path
|
|
from random import choice
|
|
from re import match, search, split, sub
|
|
from time import sleep
|
|
|
|
from evennia.utils import logger, delay
|
|
from evennia.utils.search import search_object
|
|
|
|
from typeclasses.scripts import Script
|
|
from typeclasses.puppets import Puppet
|
|
|
|
|
|
personality_dir = "personalities"
|
|
|
|
def fix_paragraph(paragraph):
|
|
"""
|
|
Because the number of tokens is small, a response may end mid-sentence.
|
|
Seems like displaying everything but the last fragment is sufficient.
|
|
"""
|
|
|
|
# Split the paragraph into sentences
|
|
sentences = split(r'(?<=[.!?])\s+', paragraph)
|
|
|
|
# Remove the last sentence if it doesn't end with punctuation
|
|
if not search(r"[.!?]$", sentences[-1]):
|
|
sentences.pop()
|
|
|
|
return ' '.join(sentences)
|
|
|
|
|
|
class ChatBot(Puppet):
|
|
"""
|
|
|
|
py me.search("squirrel").backstory("squirrel")
|
|
"""
|
|
|
|
short_history = []
|
|
|
|
def pop_recent_events(self):
|
|
history = "\n\n".join(self.short_history) if self.short_history else None
|
|
self.short_history = []
|
|
return history
|
|
|
|
def backstory(self, personality=None):
|
|
"""
|
|
Read a file that includes a character's name and knowledge.
|
|
"""
|
|
if not personality:
|
|
files = listdir(personality_dir)
|
|
personalities = [f for f in files
|
|
if path.isfile(path.join(personality_dir, f))]
|
|
personality_file = choice(personalities)
|
|
else:
|
|
personality_file = personality + ".md"
|
|
|
|
filename = Path(path.join(personality_dir, personality_file))
|
|
if filename.exists():
|
|
personality = filename.stem
|
|
else:
|
|
logger.error(f"Chatbot Identity, {personality}, doesn't exist: {filename}")
|
|
|
|
with open(filename, "r") as ids:
|
|
details = ids.read()
|
|
|
|
# Create some paragraphs, and take the second:
|
|
content = details.split('\n\n')[1]
|
|
lines = content.splitlines()
|
|
desc = []
|
|
|
|
# Find name and description
|
|
for line in lines:
|
|
m = match(r"([A-z]+): +(.*)", line)
|
|
if m:
|
|
key = m.group(1).lower()
|
|
value = m.group(2).strip()
|
|
if key == "name":
|
|
self.aliases.remove()
|
|
self.aliases.add(value)
|
|
elif key == "description":
|
|
self.sdesc.add(value)
|
|
elif key == "gender":
|
|
self.db.gender = value
|
|
elif key == "pose":
|
|
self.db.pose = value
|
|
else:
|
|
desc.append(line.strip())
|
|
|
|
self.db.personality = personality
|
|
self.db.desc = ' '.join(desc)
|
|
self.db.personality_file = filename
|
|
return details
|
|
|
|
def setting_and_backstory(self, speaker):
|
|
logger.info(f"Reading {self.db.personality_file}")
|
|
system_prompt = Path(self.db.personality_file).read_text()
|
|
system_prompt += "\n\n"
|
|
system_prompt += "You are currently in " + speaker.location.key + ". "
|
|
if speaker.location.key == "Cozy House":
|
|
system_prompt += "This is the dwelling of the gnome, Dabbler."
|
|
if speaker.location.key == "Homey Hut":
|
|
system_prompt += "This is the dwelling of the witch, Trampoli."
|
|
system_prompt += "Described as " + speaker.location.desc
|
|
system_prompt += "\n\n"
|
|
system_prompt += "You are talking to a "
|
|
system_prompt += speaker.db.gender + " " + speaker.sdesc.get() + ". "
|
|
system_prompt += "Described as " + speaker.db.desc
|
|
# logger.info(f"Prompt: {system_prompt}")
|
|
return system_prompt
|
|
|
|
def history_file(self, speaker):
|
|
combo_name = f".{self.db.personality}-{speaker}.json".lower()
|
|
filename = path.join(personality_dir, combo_name)
|
|
logger.info(f"Chatbot history_file: {filename}")
|
|
return Path(filename)
|
|
|
|
def history(self, speaker):
|
|
history_file = self.history_file(speaker)
|
|
return json.loads(history_file.read_text()) if history_file.exists() else []
|
|
|
|
def update_history(self, speaker, messages, reply):
|
|
history_file = self.history_file(speaker)
|
|
messages.append({"role": "assistant", "content": reply})
|
|
history_file.write_text(json.dumps(messages, indent=2))
|
|
|
|
def think(self, speaker, speech):
|
|
system_prompt = self.setting_and_backstory(speaker)
|
|
messages = self.history(speaker)
|
|
recent_events = self.pop_recent_events()
|
|
if recent_events:
|
|
speech = f"{recent_events}{speaker.key}: {speech}"
|
|
messages.append({"role": "user", "content": speech})
|
|
|
|
# Get reply
|
|
client = anthropic.Anthropic()
|
|
response = client.messages.create(
|
|
model="claude-haiku-4-5",
|
|
max_tokens=240,
|
|
system=system_prompt,
|
|
messages=messages,
|
|
)
|
|
reply = response.content[0].text
|
|
|
|
# Write reply
|
|
self.update_history(speaker, messages, reply)
|
|
return reply
|
|
|
|
def other_say(self, speaker, speech):
|
|
logger.info(f"Chatbot hears: '{speech}' from {speaker}.")
|
|
logger.info(f"Characters: {self.characters_here()}")
|
|
if len(self.characters_here()) == 1:
|
|
self.other_sayto(speaker, speech)
|
|
|
|
def other_sayto(self, speaker, speech):
|
|
logger.info(f"Direct Chatbot hears: '{speech}' from {speaker}.")
|
|
if speech:
|
|
logger.info("Starting to think of a reply")
|
|
reply = self.think(speaker, speech)
|
|
logger.info(f"My reply will be: {reply}")
|
|
paragraphs = reply.split('\n\n')
|
|
for idx, paragraph in enumerate(paragraphs):
|
|
delay(6 * idx, self.location.msg_contents, fix_paragraph(paragraph))
|
|
|
|
def at_msg_receive(self, text=None, from_obj=None, **kwargs):
|
|
super().at_msg_receive(text, from_obj=from_obj, **kwargs)
|
|
|
|
msg = text if isinstance(text, str) else text[0]
|
|
# Strip out the colored formatting (we will only strip the simple stuff):
|
|
msg = sub(r'\|[a-zA-Z]', '', msg) if msg else msg
|
|
|
|
if from_obj:
|
|
if hasattr(from_obj, 'sdesc') and from_obj.sdesc.get():
|
|
name = from_obj.sdesc.get()
|
|
else:
|
|
name = from_obj.key
|
|
self.short_history.append(f"{name}: {msg}")
|
|
else:
|
|
self.short_history.append(msg)
|
|
|
|
# We don't want to store _all_ the events:
|
|
self.short_history = self.short_history[-10:]
|
|
return True
|
|
|
|
class Witch(ChatBot):
|
|
"""
|
|
@update Trampoli = typeclasses.chatbots.Witch
|
|
@set/delete Trampoli/arrive
|
|
"""
|
|
|
|
def visit(self):
|
|
self.execute_cmd("pose reset")
|
|
|
|
hut = self.location
|
|
delay(1, hut.msg_contents,
|
|
"The old lady says, \"I must be off now, dearie, to visit an old friend.\"")
|
|
delay(5, hut.msg_contents,
|
|
"She grabs and old broom, and flies out the door!")
|
|
|
|
dest = self.search("mp04", global_search=True)
|
|
delay(6, self.move_to, dest, quiet=True)
|
|
|
|
delay(7, dest.msg_contents,
|
|
"An old lady flies in on a broom.")
|
|
delay(10, dest.msg_contents,
|
|
"The old lady says, \"Hello Dearie, I'm just here to visit an old friend.\"")
|
|
|
|
knocker = self.search("knocker", global_search=True, location=dest)
|
|
delay(14, knocker.do_knock, self)
|
|
delay(20, dest.msg_contents,
|
|
"\"I think I heard a 'come in',\" says the old lady. \"You heard it too, right?\"")
|
|
|
|
home = self.search("mp03", global_search=True)
|
|
delay(24, self.move_to, home)
|
|
|
|
delay(30, home.msg_contents,
|
|
"\"Now this looks cheery,\" says the old lady, as she places her broom by the door. \"I do believe we are in for a spell of rain.\"")
|
|
|
|
def leave(self):
|
|
self.location.msg_contents("The old lady wraps her shawl tightly around her, and grabs her broom. \"I must be getting home now, dear,\" she says.")
|
|
|
|
dest = self.search("mp04", global_search=True)
|
|
delay(2, self.move_to, dest)
|
|
delay(4, dest.msg_contents,
|
|
"The old lady hops on her broomstick and flies away through the trees.\"")
|
|
|
|
hut = self.search("mp09", global_search=True)
|
|
delay(5, self.move_to, hut, quiet=True)
|
|
delay(6, hut.msg_contents,
|
|
"And old lady flies through the door on her broomstick! \"Hello dearies!\" she exclaims to everyone and everything in her homey hut. \"I have returned from visiting my friend. What have I missed while I was away?\"")
|
|
|
|
sleep_pose = self.attributes.get("pose_sleep")
|
|
if sleep_pose:
|
|
self.execute_cmd(f"pose {sleep_pose}")
|