Spaces:
Paused
Paused
File size: 10,128 Bytes
6a42990 | 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 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 | from typing import Union, List
from tinytroupe.extraction import logger
from tinytroupe.utils import JsonSerializableRegistry
from tinytroupe.experimentation import Proposition
from tinytroupe.environment import TinyWorld
from tinytroupe.agent import TinyPerson
import tinytroupe.utils as utils
DEFAULT_FIRST_N = 10
DEFAULT_LAST_N = 100
class InterventionBatch:
"""
A wrapper around multiple Intervention instances that allows chaining set_* methods.
"""
def __init__(self, interventions):
self.interventions = interventions
def __iter__(self):
"""Makes the batch iterable and compatible with list()"""
return iter(self.interventions)
def set_textual_precondition(self, text):
for intervention in self.interventions:
intervention.set_textual_precondition(text)
return self
def set_functional_precondition(self, func):
for intervention in self.interventions:
intervention.set_functional_precondition(func)
return self
def set_effect(self, effect_func):
for intervention in self.interventions:
intervention.set_effect(effect_func)
return self
def set_propositional_precondition(self, proposition, threshold=None):
for intervention in self.interventions:
intervention.set_propositional_precondition(proposition, threshold)
return self
def as_list(self):
"""Return the list of individual interventions."""
return self.interventions
class Intervention:
def __init__(self, targets: Union[TinyPerson, TinyWorld, List[TinyPerson], List[TinyWorld]],
first_n:int=DEFAULT_FIRST_N, last_n:int=DEFAULT_LAST_N,
name: str = None):
"""
Initialize the intervention.
Args:
target (Union[TinyPerson, TinyWorld, List[TinyPerson], List[TinyWorld]]): the target to intervene on
first_n (int): the number of first interactions to consider in the context
last_n (int): the number of last interactions (most recent) to consider in the context
name (str): the name of the intervention
"""
self.targets = targets
# initialize the possible preconditions
self.text_precondition = None
self.precondition_func = None
# effects
self.effect_func = None
# which events to pay attention to?
self.first_n = first_n
self.last_n = last_n
# name
if name is None:
self.name = self.name = f"Intervention {utils.fresh_id(self.__class__.__name__)}"
else:
self.name = name
# the most recent precondition proposition used to check the precondition
self._last_text_precondition_proposition = None
self._last_functional_precondition_check = None
# propositional precondition (optional)
self.propositional_precondition = None
self.propositional_precondition_threshold = None
self._last_propositional_precondition_check = None
################################################################################################
# Intervention flow
################################################################################################
@classmethod
def create_for_each(cls, targets, first_n=DEFAULT_FIRST_N, last_n=DEFAULT_LAST_N, name=None):
"""
Create separate interventions for each target in the list.
Args:
targets (list): List of targets (TinyPerson or TinyWorld instances)
first_n (int): the number of first interactions to consider in the context
last_n (int): the number of last interactions (most recent) to consider in the context
name (str): the name of the intervention
Returns:
InterventionBatch: A wrapper that allows chaining set_* methods that will apply to all interventions
"""
if not isinstance(targets, list):
targets = [targets]
interventions = [cls(target, first_n=first_n, last_n=last_n,
name=f"{name}_{i}" if name else None)
for i, target in enumerate(targets)]
return InterventionBatch(interventions)
def __call__(self):
"""
Execute the intervention.
Returns:
bool: whether the intervention effect was applied.
"""
return self.execute()
def execute(self):
"""
Execute the intervention. It first checks the precondition, and if it is met, applies the effect.
This is the simplest method to run the intervention.
Returns:
bool: whether the intervention effect was applied.
"""
logger.debug(f"Executing intervention: {self}")
if self.check_precondition():
self.apply_effect()
logger.debug(f"Precondition was true, intervention effect was applied.")
return True
logger.debug(f"Precondition was false, intervention effect was not applied.")
return False
def check_precondition(self):
"""
Check if the precondition for the intervention is met.
"""
#
# Textual precondition
#
if self.text_precondition is not None:
self._last_text_precondition_proposition = Proposition(claim=self.text_precondition, target=self.targets, first_n=self.first_n, last_n=self.last_n)
llm_precondition_check = self._last_text_precondition_proposition.check()
else:
llm_precondition_check = True
#
# Functional precondition
#
if self.precondition_func is not None:
self._last_functional_precondition_check = self.precondition_func(self.targets)
else:
self._last_functional_precondition_check = True # default to True if no functional precondition is set
#
# Propositional precondition
#
self._last_propositional_precondition_check = True
if self.propositional_precondition is not None:
if self.propositional_precondition_threshold is not None:
score = self.propositional_precondition.score(target=self.targets)
if score >= self.propositional_precondition_threshold:
self._last_propositional_precondition_check = False
else:
if not self.propositional_precondition.check(target=self.targets):
self._last_propositional_precondition_check = False
return llm_precondition_check and self._last_functional_precondition_check and self._last_propositional_precondition_check
def apply_effect(self):
"""
Apply the intervention's effects. This won't check the precondition,
so it should be called after check_precondition.
"""
self.effect_func(self.targets)
################################################################################################
# Pre and post conditions
################################################################################################
def set_textual_precondition(self, text):
"""
Set a precondition as text, to be interpreted by a language model.
Args:
text (str): the text of the precondition
"""
self.text_precondition = text
return self # for chaining
def set_functional_precondition(self, func):
"""
Set a precondition as a function, to be evaluated by the code.
Args:
func (function): the function of the precondition.
Must have the a single argument, targets (either a TinyWorld or TinyPerson, or a list). Must return a boolean.
"""
self.precondition_func = func
return self # for chaining
def set_effect(self, effect_func):
"""
Set the effect of the intervention.
Args:
effect (str): the effect function of the intervention
"""
self.effect_func = effect_func
return self # for chaining
def set_propositional_precondition(self, proposition:Proposition, threshold:int=None):
"""
Set a propositional precondition using the Proposition class,
optionally with a score threshold.
"""
self.propositional_precondition = proposition
self.propositional_precondition_threshold = threshold
return self
################################################################################################
# Inspection
################################################################################################
def precondition_justification(self):
"""
Get the justification for the precondition.
"""
justification = ""
# text precondition justification
if self._last_text_precondition_proposition is not None:
justification += f"{self._last_text_precondition_proposition.justification} (confidence = {self._last_text_precondition_proposition.confidence})\n\n"
# functional precondition justification
if self.precondition_func is not None:
if self._last_functional_precondition_check == True:
justification += f"Functional precondition was met.\n\n"
else:
justification += "Preconditions do not appear to be met.\n\n"
# propositional precondition justification
if self.propositional_precondition is not None:
if self._last_propositional_precondition_check == True:
justification += f"Propositional precondition was met.\n\n"
else:
justification += "Propositional precondition was not met.\n\n"
return justification
return justification
|