From 5117db71504239a7d967494ca60656a5a10e7f72 Mon Sep 17 00:00:00 2001 From: Sean Sube Date: Tue, 7 May 2024 20:40:53 -0500 Subject: [PATCH] set up a proper dungeon master --- adventure/logic.py | 30 ++++++++++--------- adventure/main.py | 42 +++++++++++++++++++++------ adventure/optional_actions.py | 54 +++++++++++++++++++++++------------ 3 files changed, 86 insertions(+), 40 deletions(-) diff --git a/adventure/logic.py b/adventure/logic.py index 1b8c927..d4d6710 100644 --- a/adventure/logic.py +++ b/adventure/logic.py @@ -1,4 +1,4 @@ -from functools import partial +from functools import partial, wraps from logging import getLogger from random import random from typing import Callable, Dict, List, Optional @@ -45,7 +45,7 @@ class LogicTable: LogicTrigger = Callable[[Room | Actor | Item, Attributes], Attributes] -TriggerTable = Dict[LogicRule, List[LogicTrigger]] +TriggerTable = Dict[str, LogicTrigger] def update_attributes( @@ -99,9 +99,10 @@ def update_attributes( attributes.update(rule.set) logger.info("logic set state: %s", rule.set) - if rule in triggers: - for trigger in triggers[rule]: - attributes = trigger(entity, attributes) + if rule.trigger: + for trigger in rule.trigger: + if trigger in triggers: + attributes = triggers[trigger](entity, attributes) return attributes @@ -143,7 +144,7 @@ def format_logic(attributes: Attributes, rules: LogicTable, self=True) -> str: logger.debug("label has no relevant description: %s", label) if len(labels) > 0: - logger.info("adding labels: %s", labels) + logger.info("adding attribute labels: %s", labels) return " ".join(labels) @@ -152,14 +153,17 @@ def init_from_file(filename: str): logger.info("loading logic from file: %s", filename) with open(filename) as file: logic_rules = LogicTable(**load(file, Loader=Loader)) - logic_triggers = { - rule: [get_plugin_function(trigger) for trigger in rule.trigger] - for rule in logic_rules.rules - if rule.trigger - } + logic_triggers = {} + + for rule in logic_rules.rules: + if rule.trigger: + for trigger in rule.trigger: + logic_triggers[trigger] = get_plugin_function(trigger) logger.info("initialized logic system") return ( - partial(update_logic, rules=logic_rules, triggers=logic_triggers), - partial(format_logic, rules=logic_rules), + wraps(update_logic)( + partial(update_logic, rules=logic_rules, triggers=logic_triggers) + ), + wraps(format_logic)(partial(format_logic, rules=logic_rules)), ) diff --git a/adventure/main.py b/adventure/main.py index 3e7c067..250cbdc 100644 --- a/adventure/main.py +++ b/adventure/main.py @@ -1,7 +1,7 @@ from json import load from logging.config import dictConfig from os import environ, path -from typing import Callable, Dict, Sequence, Tuple +from typing import Callable, Sequence, Tuple from dotenv import load_dotenv from packit.agent import Agent, agent_easy_connect @@ -10,7 +10,8 @@ from packit.results import multi_function_or_str_result from packit.toolbox import Toolbox from packit.utils import logger_with_colors -from adventure.context import set_current_broadcast +from adventure.context import set_current_broadcast, set_dungeon_master +from adventure.models import Attributes from adventure.plugins import load_plugin # Configure logging @@ -50,7 +51,7 @@ if True: from adventure.models import Actor, Room, World, WorldState from adventure.state import create_agents, save_world, save_world_state -logger = logger_with_colors(__name__, level="INFO") +logger = logger_with_colors(__name__, level="DEBUG") load_dotenv(environ.get("ADVENTURE_ENV", ".env"), override=True) @@ -81,7 +82,7 @@ def simulate_world( steps: int = 10, actions: Sequence[Callable[..., str]] = [], systems: Sequence[ - Tuple[Callable[[World, int], None], Callable[[Dict[str, str]], str] | None] + Tuple[Callable[[World, int], None], Callable[[Attributes], str] | None] ] = [], event_callbacks: Sequence[Callable[[str], None]] = [], input_callbacks: Sequence[Callable[[Room, Actor, str], None]] = [], @@ -140,6 +141,9 @@ def simulate_world( def result_parser(value, agent, **kwargs): for callback in input_callbacks: + logger.info( + f"calling input callback for {actor_name}: {callback.__name__}" + ) callback(room, actor, value) return world_result_parser(value, agent, **kwargs) @@ -198,7 +202,13 @@ def parse_args(): help="Extra actions to include in the simulation", ) parser.add_argument( - "--flavor", type=str, help="Some additional flavor text for the generated world" + "--discord", type=bool, help="Whether to run the simulation in a Discord bot" + ) + parser.add_argument( + "--flavor", + type=str, + default="", + help="Some additional flavor text for the generated world", ) parser.add_argument( "--player", type=str, help="The name of the character to play as" @@ -289,7 +299,7 @@ def main(): else: logger.info(f"Generating a new {args.theme} world") llm = agent_easy_connect() - agent = Agent( + world_builder = Agent( "World Builder", f"You are an experienced game master creating a visually detailed {args.theme} world for a new adventure. {args.flavor}", {}, @@ -306,7 +316,7 @@ def main(): server_system(world, 0) world = generate_world( - agent, + world_builder, args.world, args.theme, room_count=args.rooms, @@ -349,14 +359,28 @@ def main(): logger.info(f"Loading extra systems from {system_name}") module_systems = load_plugin(system_name) logger.info( - f"Loaded extra systems: {[system.__name__ for system in module_systems]}" + f"Loaded extra systems: {[component.__name__ for system in module_systems for component in system]}" ) - extra_systems.append(module_systems) + extra_systems.extend(module_systems) # make sure the server system runs after any updates if args.server: extra_systems.append((server_system, None)) + # create the DM + llm = agent_easy_connect() + world_builder = Agent( + "dungeon master", + ( + f"You are the dungeon master in charge of a {world.theme} world. Be creative and original, and come up with " + f"interesting events that will keep players interested. {args.flavor}" + "Do not to repeat yourself unless you are given the same prompt with the same characters and actions." + ), + {}, + llm, + ) + set_dungeon_master(world_builder) + # start the sim logger.debug("Simulating world: %s", world) simulate_world( diff --git a/adventure/optional_actions.py b/adventure/optional_actions.py index 5befc9c..ca65505 100644 --- a/adventure/optional_actions.py +++ b/adventure/optional_actions.py @@ -3,21 +3,29 @@ from typing import Callable, List from packit.agent import Agent, agent_easy_connect -from adventure.context import broadcast, get_agent_for_actor, get_current_context +from adventure.context import ( + broadcast, + get_agent_for_actor, + get_current_context, + get_dungeon_master, + has_dungeon_master, + set_dungeon_master, +) from adventure.generate import OPPOSITE_DIRECTIONS, generate_item, generate_room logger = getLogger(__name__) - -llm = agent_easy_connect() - -# TODO: provide dungeon master with the world theme -dungeon_master = Agent( - "dungeon master", - "You are the dungeon master in charge of a fantasy world.", - {}, - llm, -) +# this is the fallback dungeon master if none is set +if not has_dungeon_master(): + llm = agent_easy_connect() + set_dungeon_master( + Agent( + "dungeon master", + "You are the dungeon master in charge of a fantasy world.", + {}, + llm, + ) + ) def action_explore(direction: str) -> str: @@ -29,6 +37,7 @@ def action_explore(direction: str) -> str: """ current_world, current_room, current_actor = get_current_context() + dungeon_master = get_dungeon_master() if not current_world: raise ValueError("No world found") @@ -53,12 +62,13 @@ def action_explore(direction: str) -> str: return f"You explore {direction} and find a new room: {new_room.name}" -def action_search() -> str: +def action_search(unused: bool) -> str: """ Search the room for hidden items. """ action_world, action_room, action_actor = get_current_context() + dungeon_master = get_dungeon_master() if len(action_room.items) > 2: return "You find nothing hidden in the room." @@ -88,12 +98,17 @@ def action_use(item: str, target: str) -> str: target: The name of the character to use the item on, or "self" to use the item on yourself. """ _, action_room, action_actor = get_current_context() + dungeon_master = get_dungeon_master() - available_items = [item.name for item in action_actor.items] + [ - item.name for item in action_room.items - ] - - if item not in available_items: + action_item = next( + ( + search_item + for search_item in (action_actor.items + action_room.items) + if search_item.name == item + ), + None, + ) + if not action_item: return f"The {item} item is not available to use." if target == "self": @@ -108,7 +123,10 @@ def action_use(item: str, target: str) -> str: broadcast(f"{action_actor.name} uses {item} on {target}") outcome = dungeon_master( - f"{action_actor.name} uses {item} on {target}. {action_actor.description}. {target_actor.description}. What happens? How does {target} react? " + f"{action_actor.name} uses {item} on {target}. " + f"{action_actor.description}. {target_actor.description}. {action_item.description}. " + f"What happens? How does {target} react? Be creative with the results. The outcome can be good, bad, or neutral." + "Decide based on the characters involved and the item being used." "Specify the outcome of the action. Do not include the question or any JSON. Only include the outcome of the action." ) broadcast(f"The action resulted in: {outcome}")