ssube
/
taleweave-ai
1
0
Fork 0
Code Actions

split up simulation into planning and action systems

This commit is contained in:
Sean Sube 2024-06-12 23:02:25 -05:00
parent d0ddeca211
commit 3d833c683f
Signed by: ssube
GPG Key ID: 3EED7B957D362AF1
15 changed files with 589 additions and 466 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
Makefile
View File

@ -36,16 +36,16 @@ lint-check:
black --check tests/
flake8 taleweave
flake8 tests
isort --check-only --skip __init__.py --filter-files taleweave
isort --check-only --skip __init__.py --filter-files tests
isort --check-only --filter-files taleweave
isort --check-only --filter-files tests
lint-fix:
black taleweave/
black tests/
flake8 taleweave
flake8 tests
isort --skip __init__.py --filter-files taleweave
isort --skip __init__.py --filter-files tests
isort --filter-files taleweave
isort --filter-files tests
style: lint-fix

20
prompts/llama-digest.yml
View File

@ -1,13 +1,5 @@
prompts:
# digest system
digest_action_move_other_enter: |
{{event.character | name}} entered the room through the {{source_portal | name}}.
digest_action_move_other_exit: |
{{event.character | name}} left the room, heading through the {{destination_portal | name}}.
digest_action_move_self_enter: |
You entered the room through the {{source_portal | name}}.
digest_action_move_self_exit: |
You left the room, heading through the {{destination_portal | name}}.
# action digest
digest_action_move: |
{{event.character | name}} entered the room.
digest_action_take: |
@ -22,3 +14,13 @@ prompts:
{{event.character | name}} told {{event.parameters[character]}} about something.
digest_action_examine: |
{{event.character | name}} examined the {{event.parameters[target]}}.
# movement digest
digest_move_other_enter: |
{{event.character | name}} entered the room through the {{source_portal | name}}.
digest_move_other_exit: |
{{event.character | name}} left the room, heading through the {{destination_portal | name}}.
digest_move_self_enter: |
You entered the room through the {{source_portal | name}}.
digest_move_self_exit: |
You left the room, heading through the {{destination_portal | name}}.

2
taleweave/actions/optional.py
View File

@ -252,7 +252,7 @@ def action_use(item: str, target: str) -> str:
return outcome
def init() -> List[Callable]:
def init_optional() -> List[Callable]:
"""
Initialize the custom actions.
"""

10
taleweave/context.py
View File

@ -37,6 +37,7 @@ dungeon_master: Agent | None = None
# TODO: wrap this into a class that can be passed around
character_agents: Dict[str, Tuple[Character, Agent]] = {}
event_emitter = EventEmitter()
extra_actions: List[Callable[..., str]] = []
game_config: Config = DEFAULT_CONFIG
game_systems: List[GameSystem] = []
prompt_library: PromptLibrary = PromptLibrary(prompts={})
@ -185,6 +186,10 @@ def get_system_data(system: str) -> Any | None:
return system_data.get(system)
def get_extra_actions() -> List[Callable[..., str]]:
return extra_actions
# endregion
@ -237,6 +242,11 @@ def set_system_data(system: str, data: Any):
system_data[system] = data
def set_extra_actions(actions: List[Callable[..., str]]):
global extra_actions
extra_actions = actions
# endregion

63
taleweave/editor.py
View File

@ -9,6 +9,7 @@ from taleweave.generate import (
generate_item,
generate_portals,
generate_room,
link_rooms,
)
from taleweave.main import load_or_initialize_system_data
from taleweave.models.base import dump_model
@ -27,7 +28,6 @@ from taleweave.utils.search import (
)
from taleweave.utils.world import describe_entity
ENTITY_TYPES = ["room", "portal", "item", "character"]
@ -72,6 +72,9 @@ def parse_args():
"prompt", type=str, help="Prompt to generate the entity"
)
generate_parser.add_argument("--room", type=str, help="Room the entity is in")
generate_parser.add_argument(
"--dest-room", type=str, help="Destination room for portals"
)
# Set up the 'delete' command
delete_parser = subparsers.add_parser("delete", help="Delete an entity")
@ -91,6 +94,15 @@ def parse_args():
"--description", type=str, help="Description of the entity"
)
# Set up the 'link' command
link_parser = subparsers.add_parser("link", help="Link rooms")
link_parser.add_argument(
"rooms",
type=str,
nargs="*",
help="Rooms to link. Leave blank to link all rooms.",
)
return parser.parse_args()
@ -212,24 +224,47 @@ def command_generate(args):
dungeon_master = get_dungeon_master()
systems = get_game_systems()
# TODO: Generate the entity
if args.type == "room":
room = generate_room(dungeon_master, world, systems)
world.rooms.append(room)
if args.type == "portal":
portal = generate_portals(dungeon_master, world, "TODO", "TODO", systems)
# TODO: Add portal to room and generate reverse portal from destination room
source_room = find_room(world, args.room)
if not source_room:
print(f"Room {args.room} not found")
return
destination_room = find_room(world, args.dest_room)
if not destination_room:
print(f"Room {args.dest_room} not found")
return
outgoing_portal, incoming_portal = generate_portals(
dungeon_master, world, source_room, destination_room, systems
)
source_room.portals.append(outgoing_portal)
destination_room.portals.append(incoming_portal)
if args.type == "item":
# TODO: add item to character or container inventory
room = find_room(world, args.room)
if not room:
print(f"Room {args.room} not found")
return
item = generate_item(dungeon_master, world, systems)
# TODO: Add item to room or character inventory
room.items.append(item)
if args.type == "character":
room = find_room(world, args.room)
if not room:
print(f"Room {args.room} not found")
return
character = generate_character(
dungeon_master, world, systems, "TODO", args.prompt
dungeon_master, world, systems, room, args.prompt
)
# TODO: Add character to room
room.characters.append(character)
save_world(args.state, args.world, world, state)
@ -291,6 +326,19 @@ def command_update(args):
save_world(args.state, args.world, world, state)
def command_link(args):
print(f"Linking rooms {args.rooms}")
world, state = load_world(args.state, args.world)
print(world.name)
dungeon_master = get_dungeon_master()
systems = get_game_systems()
link_rooms(dungeon_master, world, systems)
save_world(args.state, args.world, world, state)
COMMAND_TABLE = {
"list": command_list,
"describe": command_describe,
@ -298,6 +346,7 @@ COMMAND_TABLE = {
"generate": command_generate,
"delete": command_delete,
"update": command_update,
"link": command_link,
}

73
taleweave/main.py
View File

@ -1,6 +1,8 @@
import argparse
import atexit
from functools import partial
from glob import glob
from itertools import count
from logging.config import dictConfig
from os import environ, path
from typing import List
@ -42,11 +44,15 @@ if environ.get("DEBUG", "false").lower() == "true":
if True:
from taleweave.context import (
get_current_turn,
get_prompt_library,
get_system_data,
set_current_turn,
set_current_world,
set_dungeon_master,
set_extra_actions,
set_game_config,
set_game_systems,
set_system_data,
subscribe,
)
@ -58,8 +64,9 @@ if True:
from taleweave.models.files import TemplateFile, WorldPrompt
from taleweave.models.prompt import PromptLibrary
from taleweave.plugins import load_plugin
from taleweave.simulate import simulate_world
from taleweave.state import create_agents, save_world, save_world_state
from taleweave.systems.action import init_action
from taleweave.systems.planning import init_planning
from taleweave.utils.template import format_prompt
@ -72,8 +79,6 @@ def int_or_inf(value: str) -> float | int:
# main
def parse_args():
import argparse
parser = argparse.ArgumentParser(
description="Generate and simulate a text adventure world"
)
@ -364,9 +369,9 @@ def main():
extra_actions = []
if args.optional_actions:
logger.info("loading optional actions")
from taleweave.actions.optional import init as init_optional_actions
from taleweave.actions.optional import init_optional
optional_actions = init_optional_actions()
optional_actions = init_optional()
logger.info(
f"loaded optional actions: {[action.__name__ for action in optional_actions]}"
)
@ -381,63 +386,61 @@ def main():
)
extra_actions.extend(module_actions)
set_extra_actions(extra_actions)
# set up the game systems
systems: List[GameSystem] = []
systems.extend(init_planning())
systems.extend(init_action())
# load extra systems from plugins
extra_systems: List[GameSystem] = []
for system_name in args.systems or []:
logger.info(f"loading extra systems from {system_name}")
module_systems = load_plugin(system_name)
logger.info(f"loaded extra systems: {module_systems}")
extra_systems.extend(module_systems)
systems.extend(module_systems)
# make sure the server system runs after any updates
if args.server:
from taleweave.server.websocket import server_system
extra_systems.append(GameSystem(name="server", simulate=server_system))
systems.append(GameSystem(name="server", simulate=server_system))
set_game_systems(systems)
# load or generate the world
world_prompt = get_world_prompt(args)
world, world_state_file, world_turn = load_or_generate_world(
args, config, players, extra_systems, world_prompt=world_prompt
args, config, players, systems, world_prompt=world_prompt
)
set_current_world(world)
# make sure the snapshot system runs last
def snapshot_system(world: World, turn: int, data: None = None) -> None:
logger.info("taking snapshot of world state")
save_world_state(world, turn, world_state_file)
extra_systems.append(GameSystem(name="snapshot", simulate=snapshot_system))
systems.append(GameSystem(name="snapshot", simulate=snapshot_system))
# hack: send a snapshot to the websocket server
if args.server:
server_system(world, world_turn)
# create the DM
llm = agent_easy_connect()
memory_factory = partial(
make_limited_memory, limit=config.world.character.memory_limit
)
world_builder = Agent(
"dungeon master",
format_prompt(
"world_generate_dungeon_master",
flavor=world_prompt.flavor,
theme=world_prompt.theme,
),
{},
llm,
memory_factory=memory_factory,
)
set_dungeon_master(world_builder)
# run game systems for each turn
logger.info(f"simulating the world for {args.turns} turns using systems: {systems}")
for i in count():
current_turn = get_current_turn()
logger.info(f"simulating turn {i} of {args.turns} (world turn {current_turn})")
# start the sim
logger.debug("simulating world: %s", world.name)
simulate_world(
world,
turns=args.turns,
actions=extra_actions,
systems=extra_systems,
)
for system in systems:
if system.simulate:
logger.info(f"running system {system.name}")
system.simulate(world, current_turn)
set_current_turn(current_turn + 1)
if i >= args.turns:
logger.info("reached turn limit at world turn %s", current_turn + 1)
break
if __name__ == "__main__":

393
taleweave/simulate.py
View File

@ -1,393 +0,0 @@
from functools import partial
from itertools import count
from json import loads
from logging import getLogger
from math import inf
from typing import Callable, Sequence
from packit.agent import Agent
from packit.conditions import condition_or, condition_threshold
from packit.errors import ToolError
from packit.loops import loop_retry
from packit.results import function_result
from packit.toolbox import Toolbox
from taleweave.actions.base import (
action_ask,
action_examine,
action_give,
action_move,
action_take,
action_tell,
)
from taleweave.actions.planning import (
check_calendar,
edit_note,
erase_notes,
get_recent_notes,
read_notes,
schedule_event,
summarize_notes,
take_note,
)
from taleweave.context import (
broadcast,
get_character_agent_for_name,
get_character_for_agent,
get_current_turn,
get_current_world,
get_game_config,
get_prompt,
set_current_character,
set_current_room,
set_current_turn,
set_current_world,
set_game_systems,
)
from taleweave.errors import ActionError
from taleweave.game_system import GameSystem
from taleweave.models.entity import Character, Room, World
from taleweave.models.event import ActionEvent, ResultEvent
from taleweave.utils.conversation import make_keyword_condition, summarize_room
from taleweave.utils.effect import expire_effects
from taleweave.utils.planning import expire_events, get_upcoming_events
from taleweave.utils.search import find_containing_room
from taleweave.utils.template import format_prompt
from taleweave.utils.world import format_attributes
logger = getLogger(__name__)
def world_result_parser(value, agent, **kwargs):
current_world = get_current_world()
if not current_world:
raise ValueError(
"The current world must be set before calling world_result_parser"
)
logger.debug(f"parsing action for {agent.name}: {value}")
current_character = get_character_for_agent(agent)
current_room = next(
(room for room in current_world.rooms if current_character in room.characters),
None,
)
set_current_room(current_room)
set_current_character(current_character)
return function_result(value, agent=agent, **kwargs)
def prompt_character_action(
room, character, agent, action_toolbox, current_turn
) -> str:
action_names = action_toolbox.list_tools()
# collect data for the prompt
notes_prompt, events_prompt = get_notes_events(character, current_turn)
room_characters = [character.name for character in room.characters]
room_items = [item.name for item in room.items]
room_directions = [portal.name for portal in room.portals]
character_attributes = format_attributes(character)
# character_effects = [effect.name for effect in character.active_effects]
character_items = [item.name for item in character.items]
# set up a result parser for the agent
def result_parser(value, **kwargs):
if not room or not character:
raise ValueError("Room and character must be set before parsing results")
# trim suffixes that are used elsewhere
value = value.removesuffix("END").strip()
# fix the "action_ move" whitespace issue
if '"action_ ' in value:
value = value.replace('"action_ ', '"action_')
# fix unbalanced curly braces
if value.startswith("{") and not value.endswith("}"):
open_count = value.count("{")
close_count = value.count("}")
if open_count > close_count:
fixed_value = value + ("}" * (open_count - close_count))
try:
loads(fixed_value)
value = fixed_value
except Exception:
pass
try:
result = world_result_parser(value, **kwargs)
# TODO: try to avoid parsing the JSON twice
event = ActionEvent.from_json(value, room, character)
broadcast(event)
return result
except ToolError as e:
e_str = str(e)
if e_str and "Error running tool" in e_str:
# extract the tool name and rest of the message from the error
# the format is: "Error running tool: <action_name>: <message>"
action_name, message = e_str.split(":", 1)
action_name = action_name.removeprefix("Error running tool").strip()
message = message.strip()
raise ActionError(
format_prompt(
"world_simulate_character_action_error_action",
action=action_name,
message=message,
)
)
elif e_str and "Unknown tool" in e_str:
raise ActionError(
format_prompt(
"world_simulate_character_action_error_unknown_tool",
actions=action_names,
)
)
else:
raise ActionError(
format_prompt(
"world_simulate_character_action_error_json",
actions=action_names,
)
)
# prompt and act
logger.info("starting turn for character: %s", character.name)
result = loop_retry(
agent,
format_prompt(
"world_simulate_character_action",
actions=action_names,
character_items=character_items,
attributes=character_attributes,
directions=room_directions,
room=room,
visible_characters=room_characters,
visible_items=room_items,
notes_prompt=notes_prompt,
events_prompt=events_prompt,
),
result_parser=result_parser,
toolbox=action_toolbox,
)
logger.debug(f"{character.name} action result: {result}")
if agent.memory:
agent.memory.append(result)
return result
def get_notes_events(character: Character, current_turn: int):
recent_notes = get_recent_notes(character)
upcoming_events = get_upcoming_events(character, current_turn)
if len(recent_notes) > 0:
notes = "\n".join(recent_notes)
notes_prompt = format_prompt(
"world_simulate_character_planning_notes_some", notes=notes
)
else:
notes_prompt = format_prompt("world_simulate_character_planning_notes_none")
if len(upcoming_events) > 0:
current_turn = get_current_turn()
events = [
format_prompt(
"world_simulate_character_planning_events_item",
event=event,
turns=event.turn - current_turn,
)
for event in upcoming_events
]
events = "\n".join(events)
events_prompt = format_prompt(
"world_simulate_character_planning_events_some", events=events
)
else:
events_prompt = format_prompt("world_simulate_character_planning_events_none")
return notes_prompt, events_prompt
def prompt_character_planning(
room: Room,
character: Character,
agent: Agent,
planner_toolbox: Toolbox,
current_turn: int,
max_steps: int | None = None,
) -> str:
config = get_game_config()
max_steps = max_steps or config.world.turn.planning_steps
notes_prompt, events_prompt = get_notes_events(character, current_turn)
event_count = len(character.planner.calendar.events)
note_count = len(character.planner.notes)
def result_parser(value, **kwargs):
try:
return function_result(value, **kwargs)
except ToolError as e:
e_str = str(e)
if e_str and "Error running tool" in e_str:
# extract the tool name and rest of the message from the error
# the format is: "Error running tool: <action_name>: <message>"
action_name, message = e_str.split(":", 2)
action_name = action_name.removeprefix("Error running tool").strip()
message = message.strip()
raise ActionError(
format_prompt(
"world_simulate_character_planning_error_action",
action=action_name,
message=message,
)
)
elif e_str and "Unknown tool" in e_str:
raise ActionError(
format_prompt(
"world_simulate_character_planning_error_unknown_tool",
actions=planner_toolbox.list_tools(),
)
)
else:
raise ActionError(
format_prompt(
"world_simulate_character_planning_error_json",
actions=planner_toolbox.list_tools(),
)
)
logger.info("starting planning for character: %s", character.name)
_, condition_end, result_parser = make_keyword_condition(
get_prompt("world_simulate_character_planning_done"),
result_parser=result_parser,
)
stop_condition = condition_or(
condition_end, partial(condition_threshold, max=max_steps)
)
i = 0
while not stop_condition(current=i):
result = loop_retry(
agent,
format_prompt(
"world_simulate_character_planning",
event_count=event_count,
events_prompt=events_prompt,
note_count=note_count,
notes_prompt=notes_prompt,
room_summary=summarize_room(room, character),
),
result_parser=result_parser,
stop_condition=stop_condition,
toolbox=planner_toolbox,
)
if agent.memory:
agent.memory.append(result)
i += 1
return result
def simulate_world(
world: World,
turns: float | int = inf,
actions: Sequence[Callable[..., str]] = [],
systems: Sequence[GameSystem] = [],
):
logger.info("simulating the world")
set_current_world(world)
set_game_systems(systems)
# build a toolbox for the actions
action_tools = Toolbox(
[
action_ask,
action_give,
action_examine,
action_move,
action_take,
action_tell,
*actions,
]
)
# build a toolbox for the planners
planner_toolbox = Toolbox(
[
check_calendar,
erase_notes,
read_notes,
edit_note,
schedule_event,
summarize_notes,
take_note,
]
)
# simulate each character
for i in count():
current_turn = get_current_turn()
logger.info(f"simulating turn {i} of {turns} (world turn {current_turn})")
for character_name in world.order:
character, agent = get_character_agent_for_name(character_name)
if not agent or not character:
logger.error(f"agent or character not found for name {character_name}")
continue
room = find_containing_room(world, character)
if not room:
logger.error(f"character {character_name} is not in a room")
continue
# prep context
set_current_room(room)
set_current_character(character)
# decrement effects on the character and remove any that have expired
expire_effects(character)
expire_events(character, current_turn)
# give the character a chance to think and check their planner
if agent.memory and len(agent.memory) > 0:
try:
thoughts = prompt_character_planning(
room, character, agent, planner_toolbox, current_turn
)
logger.debug(f"{character.name} thinks: {thoughts}")
except Exception:
logger.exception(
f"error during planning for character {character.name}"
)
try:
result = prompt_character_action(
room, character, agent, action_tools, current_turn
)
result_event = ResultEvent(
result=result, room=room, character=character
)
broadcast(result_event)
except Exception:
logger.exception(f"error during action for character {character.name}")
for system in systems:
if system.simulate:
system.simulate(world, current_turn)
set_current_turn(current_turn + 1)
if i >= turns:
logger.info("reached turn limit at world turn %s", current_turn + 1)
break

218
taleweave/systems/action.py Normal file
View File

@ -0,0 +1,218 @@
from json import loads
from logging import getLogger
from typing import Any
from packit.errors import ToolError
from packit.loops import loop_retry
from packit.results import function_result
from packit.toolbox import Toolbox
from taleweave.actions.base import (
action_ask,
action_examine,
action_give,
action_move,
action_take,
action_tell,
)
from taleweave.context import (
broadcast,
get_character_agent_for_name,
get_character_for_agent,
get_current_world,
get_extra_actions,
set_current_character,
set_current_room,
)
from taleweave.errors import ActionError
from taleweave.game_system import GameSystem
from taleweave.models.entity import World
from taleweave.models.event import ActionEvent, ResultEvent
from taleweave.utils.effect import expire_effects
from taleweave.utils.search import find_containing_room
from taleweave.utils.template import format_prompt
from taleweave.utils.world import format_attributes
from .planning import get_notes_events
logger = getLogger(__name__)
def world_result_parser(value, agent, **kwargs):
current_world = get_current_world()
if not current_world:
raise ValueError(
"The current world must be set before calling world_result_parser"
)
logger.debug(f"parsing action for {agent.name}: {value}")
current_character = get_character_for_agent(agent)
current_room = next(
(room for room in current_world.rooms if current_character in room.characters),
None,
)
set_current_room(current_room)
set_current_character(current_character)
return function_result(value, agent=agent, **kwargs)
def prompt_character_action(
room, character, agent, action_toolbox, current_turn
) -> str:
action_names = action_toolbox.list_tools()
# collect data for the prompt
notes_prompt, events_prompt = get_notes_events(character, current_turn)
room_characters = [character.name for character in room.characters]
room_items = [item.name for item in room.items]
room_directions = [portal.name for portal in room.portals]
character_attributes = format_attributes(character)
# character_effects = [effect.name for effect in character.active_effects]
character_items = [item.name for item in character.items]
# set up a result parser for the agent
def result_parser(value, **kwargs):
if not room or not character:
raise ValueError("Room and character must be set before parsing results")
# trim suffixes that are used elsewhere
value = value.removesuffix("END").strip()
# fix the "action_ move" whitespace issue
if '"action_ ' in value:
value = value.replace('"action_ ', '"action_')
# fix unbalanced curly braces
if value.startswith("{") and not value.endswith("}"):
open_count = value.count("{")
close_count = value.count("}")
if open_count > close_count:
fixed_value = value + ("}" * (open_count - close_count))
try:
loads(fixed_value)
value = fixed_value
except Exception:
pass
try:
result = world_result_parser(value, **kwargs)
# TODO: try to avoid parsing the JSON twice
event = ActionEvent.from_json(value, room, character)
broadcast(event)
return result
except ToolError as e:
e_str = str(e)
if e_str and "Error running tool" in e_str:
# extract the tool name and rest of the message from the error
# the format is: "Error running tool: <action_name>: <message>"
action_name, message = e_str.split(":", 1)
action_name = action_name.removeprefix("Error running tool").strip()
message = message.strip()
raise ActionError(
format_prompt(
"world_simulate_character_action_error_action",
action=action_name,
message=message,
)
)
elif e_str and "Unknown tool" in e_str:
raise ActionError(
format_prompt(
"world_simulate_character_action_error_unknown_tool",
actions=action_names,
)
)
else:
raise ActionError(
format_prompt(
"world_simulate_character_action_error_json",
actions=action_names,
)
)
# prompt and act
logger.info("starting turn for character: %s", character.name)
result = loop_retry(
agent,
format_prompt(
"world_simulate_character_action",
actions=action_names,
character_items=character_items,
attributes=character_attributes,
directions=room_directions,
room=room,
visible_characters=room_characters,
visible_items=room_items,
notes_prompt=notes_prompt,
events_prompt=events_prompt,
),
result_parser=result_parser,
toolbox=action_toolbox,
)
logger.debug(f"{character.name} action result: {result}")
if agent.memory:
agent.memory.append(result)
return result
action_tools: Toolbox | None = None
def initialize_action(world: World):
global action_tools
extra_actions = get_extra_actions()
action_tools = Toolbox(
[
action_ask,
action_give,
action_examine,
action_move,
action_take,
action_tell,
*extra_actions,
]
)
def simulate_action(world: World, turn: int, data: Any | None = None):
for character_name in world.order:
character, agent = get_character_agent_for_name(character_name)
if not agent or not character:
logger.error(f"agent or character not found for name {character_name}")
continue
room = find_containing_room(world, character)
if not room:
logger.error(f"character {character_name} is not in a room")
continue
# prep context
set_current_room(room)
set_current_character(character)
# decrement effects on the character and remove any that have expired
expire_effects(character)
try:
result = prompt_character_action(room, character, agent, action_tools, turn)
result_event = ResultEvent(result=result, room=room, character=character)
broadcast(result_event)
except Exception:
logger.exception(f"error during action for character {character.name}")
def init_action():
return [
GameSystem("action", initialize=initialize_action, simulate=simulate_action)
]

8
taleweave/systems/digest.py
View File

@ -6,7 +6,7 @@ from taleweave.game_system import FormatPerspective, GameSystem
from taleweave.models.entity import Character, Room, World, WorldEntity
from taleweave.models.event import ActionEvent, GameEvent
from taleweave.utils.search import find_containing_room, find_portal, find_room
from taleweave.utils.template import format_str
from taleweave.utils.template import format_prompt, format_str
logger = getLogger(__name__)
@ -44,11 +44,12 @@ def create_move_digest(
character_mode = "self" if (event.character == active_character) else "other"
direction_mode = "enter" if (destination_room == active_room) else "exit"
message = format_str(
message = format_prompt(
f"digest_move_{character_mode}_{direction_mode}",
destination_portal=destination_portal,
destination_room=destination_room,
direction=direction,
event=event,
source_portal=source_portal,
source_room=source_room,
)
@ -118,6 +119,9 @@ def format_digest(
if not isinstance(entity, Character):
return ""
if perspective != FormatPerspective.SECOND_PERSON:
return ""
buffer = character_buffers[entity.name]
world = get_current_world()

202
taleweave/systems/planning.py Normal file
View File

@ -0,0 +1,202 @@
from functools import partial
from logging import getLogger
from typing import Any
from packit.agent import Agent
from packit.conditions import condition_or, condition_threshold
from packit.errors import ToolError
from packit.loops import loop_retry
from packit.results import function_result
from packit.toolbox import Toolbox
from taleweave.actions.planning import (
check_calendar,
edit_note,
erase_notes,
get_recent_notes,
read_notes,
schedule_event,
summarize_notes,
take_note,
)
from taleweave.context import (
get_character_agent_for_name,
get_current_turn,
get_game_config,
get_prompt,
set_current_character,
set_current_room,
)
from taleweave.errors import ActionError
from taleweave.game_system import GameSystem
from taleweave.models.entity import Character, Room, World
from taleweave.utils.conversation import make_keyword_condition, summarize_room
from taleweave.utils.planning import expire_events, get_upcoming_events
from taleweave.utils.search import find_containing_room
from taleweave.utils.template import format_prompt
logger = getLogger(__name__)
# build a toolbox for the planners
planner_toolbox = Toolbox(
[
check_calendar,
erase_notes,
read_notes,
edit_note,
schedule_event,
summarize_notes,
take_note,
]
)
def get_notes_events(character: Character, current_turn: int):
recent_notes = get_recent_notes(character)
upcoming_events = get_upcoming_events(character, current_turn)
if len(recent_notes) > 0:
notes = "\n".join(recent_notes)
notes_prompt = format_prompt(
"world_simulate_character_planning_notes_some", notes=notes
)
else:
notes_prompt = format_prompt("world_simulate_character_planning_notes_none")
if len(upcoming_events) > 0:
current_turn = get_current_turn()
events = [
format_prompt(
"world_simulate_character_planning_events_item",
event=event,
turns=event.turn - current_turn,
)
for event in upcoming_events
]
events = "\n".join(events)
events_prompt = format_prompt(
"world_simulate_character_planning_events_some", events=events
)
else:
events_prompt = format_prompt("world_simulate_character_planning_events_none")
return notes_prompt, events_prompt
def prompt_character_planning(
room: Room,
character: Character,
agent: Agent,
planner_toolbox: Toolbox,
current_turn: int,
max_steps: int | None = None,
) -> str:
config = get_game_config()
max_steps = max_steps or config.world.turn.planning_steps
notes_prompt, events_prompt = get_notes_events(character, current_turn)
event_count = len(character.planner.calendar.events)
note_count = len(character.planner.notes)
def result_parser(value, **kwargs):
try:
return function_result(value, **kwargs)
except ToolError as e:
e_str = str(e)
if e_str and "Error running tool" in e_str:
# extract the tool name and rest of the message from the error
# the format is: "Error running tool: <action_name>: <message>"
action_name, message = e_str.split(":", 2)
action_name = action_name.removeprefix("Error running tool").strip()
message = message.strip()
raise ActionError(
format_prompt(
"world_simulate_character_planning_error_action",
action=action_name,
message=message,
)
)
elif e_str and "Unknown tool" in e_str:
raise ActionError(
format_prompt(
"world_simulate_character_planning_error_unknown_tool",
actions=planner_toolbox.list_tools(),
)
)
else:
raise ActionError(
format_prompt(
"world_simulate_character_planning_error_json",
actions=planner_toolbox.list_tools(),
)
)
logger.info("starting planning for character: %s", character.name)
_, condition_end, result_parser = make_keyword_condition(
get_prompt("world_simulate_character_planning_done"),
result_parser=result_parser,
)
stop_condition = condition_or(
condition_end, partial(condition_threshold, max=max_steps)
)
i = 0
while not stop_condition(current=i):
result = loop_retry(
agent,
format_prompt(
"world_simulate_character_planning",
event_count=event_count,
events_prompt=events_prompt,
note_count=note_count,
notes_prompt=notes_prompt,
room_summary=summarize_room(room, character),
),
result_parser=result_parser,
stop_condition=stop_condition,
toolbox=planner_toolbox,
)
if agent.memory:
agent.memory.append(result)
i += 1
return result
def simulate_planning(world: World, turn: int, data: Any | None = None):
for character_name in world.order:
character, agent = get_character_agent_for_name(character_name)
if not agent or not character:
logger.error(f"agent or character not found for name {character_name}")
continue
room = find_containing_room(world, character)
if not room:
logger.error(f"character {character_name} is not in a room")
continue
# prep context
set_current_room(room)
set_current_character(character)
# decrement effects on the character and remove any that have expired
expire_events(character, turn) # TODO: move to planning
# give the character a chance to think and check their planner
if agent.memory and len(agent.memory) > 0:
try:
thoughts = prompt_character_planning(
room, character, agent, planner_toolbox, turn
)
logger.debug(f"{character.name} thinks: {thoughts}")
except Exception:
logger.exception(
f"error during planning for character {character.name}"
)
def init_planning():
return [GameSystem("planning", simulate=simulate_planning)]

4
taleweave/systems/sim/__init__.py
View File

@ -1,9 +1,9 @@
from taleweave.systems.logic import load_logic
from .hunger_actions import action_cook, action_eat
from .hygiene_actions import action_wash
from .sleeping_actions import action_sleep
from taleweave.systems.logic import load_logic
LOGIC_FILES = [
"./taleweave/systems/sim/environment_logic.yaml",
"./taleweave/systems/sim/hunger_logic.yaml",

18
taleweave/systems/snapshot.py Normal file
View File

@ -0,0 +1,18 @@
from logging import getLogger
from typing import Any
from taleweave.game_system import GameSystem
from taleweave.models.entity import World
from taleweave.state import save_world_state
logger = getLogger(__name__)
def simulate_snapshot(world: World, turn: int, data: Any | None = None):
logger.info("taking snapshot of world state")
world_state_file = "TODO" # TODO: get world state file from somewhere
save_world_state(world, turn, world_state_file)
def init():
return [GameSystem("snapshot", simulate=simulate_snapshot)]

23
taleweave/systems/weather/__init__.py
View File

@ -1,16 +1,17 @@
from functools import partial
from typing import List
from taleweave.context import get_dungeon_master
from taleweave.models.base import dataclass
from taleweave.models.entity import World
from taleweave.systems.logic import load_logic
from taleweave.game_system import GameSystem
from packit.agent import Agent
from taleweave.models.entity import Room, WorldEntity
from taleweave.utils.string import or_list
from packit.results import enum_result
from packit.loops import loop_retry
from logging import getLogger
from typing import List
from packit.agent import Agent
from packit.loops import loop_retry
from packit.results import enum_result
from taleweave.context import get_dungeon_master
from taleweave.game_system import GameSystem
from taleweave.models.base import dataclass
from taleweave.models.entity import Room, World, WorldEntity
from taleweave.systems.logic import load_logic
from taleweave.utils.string import or_list
logger = getLogger(__name__)

5
taleweave/utils/world.py
View File

@ -58,6 +58,11 @@ def format_attributes(
for system in systems
if system.format
]
attribute_descriptions = [
description
for description in attribute_descriptions
if len(description.strip()) > 0
]
return f"{'. '.join(attribute_descriptions)}"

8
worlds.yml
View File

@ -28,8 +28,12 @@ templates:
theme: post-apocalyptic world where the only survivors are sentient robots
flavor: create a world where the only survivors of a nuclear apocalypse are sentient robots, who must now rebuild society from scratch
- name: haunted-house
theme: haunted house in the middle of nowhere
flavor: create a spooky and suspenseful world where a group of people are trapped in a haunted house in the middle of nowhere
theme: group of unexpecting friends find a haunted house in the middle of nowhere
flavor: |
create a spooky and suspenseful world where a group of innocent, unexpecting people find and become trapped in a
haunted house in the middle of nowhere. make sure you create some rooms for the road up to the house as well as
the interior of the house itself. include a variety of characters and make sure they will fully utilize all of the
actions available to them in this world, exploring and interacting with each other and the environment.
- name: magical-kingdom
theme: dangerous magical fantasy world
flavor: make a strange and dangerous world where magic winds its way through everything and incredibly powerful beings drink, fight, and wander the halls