Spaces:
Runtime error
Runtime error
File size: 7,676 Bytes
82a7a28 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 |
import pytest
import os
import sys
sys.path.append('../../tinytroupe/')
sys.path.append('../../')
sys.path.append('..')
from tinytroupe.examples import create_oscar_the_architect, create_lisa_the_data_scientist
from tinytroupe.agent import TinyPerson, TinyToolUse
from tinytroupe.environment import TinyWorld
from tinytroupe.control import Simulation
import tinytroupe.control as control
from tinytroupe.factory import TinyPersonFactory
from tinytroupe.enrichment import TinyEnricher
from tinytroupe.extraction import ArtifactExporter
from tinytroupe.tools import TinyWordProcessor
import logging
logger = logging.getLogger("tinytroupe")
import importlib
from testing_utils import *
def test_begin_checkpoint_end_with_agent_only(setup):
# erase the file if it exists
remove_file_if_exists("control_test.cache.json")
control.reset()
assert control._current_simulations["default"] is None, "There should be no simulation running at this point."
# erase the file if it exists
remove_file_if_exists("control_test.cache.json")
control.begin("control_test.cache.json")
assert control._current_simulations["default"].status == Simulation.STATUS_STARTED, "The simulation should be started at this point."
exporter = ArtifactExporter(base_output_folder="./synthetic_data_exports_3/")
enricher = TinyEnricher()
tooluse_faculty = TinyToolUse(tools=[TinyWordProcessor(exporter=exporter, enricher=enricher)])
agent_1 = create_oscar_the_architect()
agent_1.add_mental_faculties([tooluse_faculty])
agent_1.define("age", 19)
agent_1.define("nationality", "Brazilian")
agent_2 = create_lisa_the_data_scientist()
agent_2.add_mental_faculties([tooluse_faculty])
agent_2.define("age", 80)
agent_2.define("nationality", "Argentinian")
assert control._current_simulations["default"].cached_trace is not None, "There should be a cached trace at this point."
assert control._current_simulations["default"].execution_trace is not None, "There should be an execution trace at this point."
control.checkpoint()
agent_1.listen_and_act("How are you doing?")
agent_2.listen_and_act("What's up?")
# check if the file was created
assert os.path.exists("control_test.cache.json"), "The checkpoint file should have been created."
control.end()
assert control._current_simulations["default"].status == Simulation.STATUS_STOPPED, "The simulation should be ended at this point."
def test_begin_checkpoint_end_with_world(setup):
# erase the file if it exists
remove_file_if_exists("control_test_world.cache.json")
control.reset()
assert control._current_simulations["default"] is None, "There should be no simulation running at this point."
control.begin("control_test_world.cache.json")
assert control._current_simulations["default"].status == Simulation.STATUS_STARTED, "The simulation should be started at this point."
world = TinyWorld("Test World", [create_oscar_the_architect(), create_lisa_the_data_scientist()])
world.make_everyone_accessible()
assert control._current_simulations["default"].cached_trace is not None, "There should be a cached trace at this point."
assert control._current_simulations["default"].execution_trace is not None, "There should be an execution trace at this point."
world.run(2)
control.checkpoint()
# check if the file was created
assert os.path.exists("control_test_world.cache.json"), "The checkpoint file should have been created."
control.end()
assert control._current_simulations["default"].status == Simulation.STATUS_STOPPED, "The simulation should be ended at this point."
def test_begin_checkpoint_end_with_factory(setup):
# erase the file if it exists
remove_file_if_exists("control_test_personfactory.cache.json")
control.reset()
def aux_simulation_to_repeat(iteration, verbose=False):
control.reset()
assert control._current_simulations["default"] is None, "There should be no simulation running at this point."
control.begin("control_test_personfactory.cache.json")
assert control._current_simulations["default"].status == Simulation.STATUS_STARTED, "The simulation should be started at this point."
factory = TinyPersonFactory("We are interested in experts in the production of the traditional Gazpacho soup.")
assert control._current_simulations["default"].cached_trace is not None, "There should be a cached trace at this point."
assert control._current_simulations["default"].execution_trace is not None, "There should be an execution trace at this point."
agent = factory.generate_person("A Brazilian tourist who learned about Gazpaccho in a trip to Spain.")
assert control._current_simulations["default"].cached_trace is not None, "There should be a cached trace at this point."
assert control._current_simulations["default"].execution_trace is not None, "There should be an execution trace at this point."
control.checkpoint()
# check if the file was created
assert os.path.exists("control_test_personfactory.cache.json"), "The checkpoint file should have been created."
control.end()
assert control._current_simulations["default"].status == Simulation.STATUS_STOPPED, "The simulation should be ended at this point."
if verbose:
logger.debug(f"###################################################################################### Sim Iteration:{iteration}")
logger.debug(f"###################################################################################### Agent persona configs:{agent._persona}")
return agent
assert control.cache_misses() == 0, "There should be no cache misses in this test."
assert control.cache_hits() == 0, "There should be no cache hits here"
# FIRST simulation ########################################################
agent_1 = aux_simulation_to_repeat(1, verbose=True)
age_1 = agent_1.get("age")
nationality_1 = agent_1.get("nationality")
minibio_1 = agent_1.minibio()
print("minibio_1 =", minibio_1)
# SECOND simulation ########################################################
logger.debug(">>>>>>>>>>>>>>>>>>>>>>>>>> Second simulation...")
agent_2 = aux_simulation_to_repeat(2, verbose=True)
age_2 = agent_2.get("age")
nationality_2 = agent_2.get("nationality")
minibio_2 = agent_2.minibio()
print("minibio_2 =", minibio_2)
assert control.cache_misses() == 0, "There should be no cache misses in this test."
assert control.cache_hits() > 0, "There should be cache hits here."
assert age_1 == age_2, "The age should be the same in both simulations."
assert nationality_1 == nationality_2, "The nationality should be the same in both simulations."
assert minibio_1 == minibio_2, "The minibio should be the same in both simulations."
#
# let's also check the contents of the cache file, as raw text, not dict
#
with open("control_test_personfactory.cache.json", "r") as f:
cache_contents = f.read()
assert "'_aux_model_call'" in cache_contents, "The cache file should contain the '_aux_model_call' call."
assert "'_setup_agent'" in cache_contents, "The cache file should contain the '_setup_agent' call."
assert "'define'" not in cache_contents, "The cache file should not contain the 'define' methods, as these are reentrant."
assert "'define_several'" not in cache_contents, "The cache file should not contain the 'define_several' methods, as these are reentrant."
|