diff --git a/agentops/time_travel.py b/agentops/time_travel.py index 207e8888..35c73eed 100644 --- a/agentops/time_travel.py +++ b/agentops/time_travel.py @@ -4,16 +4,12 @@ from .http_client import HttpClient from .exceptions import ApiServerException from .singleton import singleton -import threading - -_time_travel_lock = threading.Lock() @singleton class TimeTravel: def __init__(self): self._completion_overrides = {} - self._initialized = False script_dir = os.path.dirname(os.path.abspath(__file__)) parent_dir = os.path.dirname(script_dir) @@ -51,9 +47,9 @@ def fetch_time_travel_id(ttd_id): set_time_travel_active_state(True) except ApiServerException as e: - manage_time_travel_state(activated=False, error=e) + print(f"🖇️ Agentops: ⏰ Time Travel Error - {e}") except Exception as e: - manage_time_travel_state(activated=False, error=e) + print(f"🖇️ Agentops: ⏰ Time Travel Error - {e}") def fetch_completion_override_from_time_travel_cache(kwargs): @@ -68,14 +64,14 @@ def fetch_completion_override_from_time_travel_cache(kwargs): def find_cache_hit(prompt_messages, completion_overrides): if not isinstance(prompt_messages, (list, tuple)): print( - "🖇⏰ AgentOps: Time Travel Error - unexpected type for prompt_messages. Expected 'list' or 'tuple'. Got ", + "🖇️ Agentops: ⏰ Time Travel Error - unexpected type for prompt_messages. Expected 'list' or 'tuple'. Got ", type(prompt_messages), ) return None if not isinstance(completion_overrides, dict): print( - "🖇⏰ AgentOps: Time Travel Error - unexpected type for completion_overrides. Expected 'dict'. Got ", + "🖇️ Agentops: ⏰ Time Travel Error - unexpected type for completion_overrides. Expected 'dict'. Got ", type(completion_overrides), ) return None @@ -84,7 +80,7 @@ def find_cache_hit(prompt_messages, completion_overrides): completion_override_dict = eval(key) if not isinstance(completion_override_dict, dict): print( - "🖇⏰ AgentOps: Time Travel Error - unexpected type for completion_override_dict. Expected 'dict'. Got ", + "🖇️ Agentops: ⏰ Time Travel Error - unexpected type for completion_override_dict. Expected 'dict'. Got ", type(completion_override_dict), ) continue @@ -92,7 +88,7 @@ def find_cache_hit(prompt_messages, completion_overrides): cached_messages = completion_override_dict.get("messages") if not isinstance(cached_messages, list): print( - "🖇⏰ AgentOps: Time Travel Error - unexpected type for cached_messages. Expected 'list'. Got ", + "🖇️ Agentops: ⏰ Time Travel Error - unexpected type for cached_messages. Expected 'list'. Got ", type(cached_messages), ) continue @@ -109,11 +105,11 @@ def find_cache_hit(prompt_messages, completion_overrides): return value except (SyntaxError, ValueError, TypeError) as e: print( - f"🖇⏰ AgentOps: Time Travel Error - Error processing completion_overrides item: {e}" + f"🖇️ Agentops: ⏰ Time Travel Error - Error processing completion_overrides item: {e}" ) except Exception as e: print( - f"🖇⏰ AgentOps: Time Travel Error - Unexpected error in find_cache_hit: {e}" + f"🖇️ Agentops: ⏰ Time Travel Error - Unexpected error in find_cache_hit: {e}" ) return None @@ -126,14 +122,10 @@ def check_time_travel_active(): try: with open(config_file_path, "r") as config_file: config = yaml.safe_load(config_file) - if config.get("Time_Travel_Debugging_Active") is True: - manage_time_travel_state(activated=True) - return True + return config.get("Time_Travel_Debugging_Active", False) except FileNotFoundError: return False - return False - def set_time_travel_active_state(is_active: bool): config_path = ".agentops_time_travel.yaml" @@ -150,35 +142,11 @@ def set_time_travel_active_state(is_active: bool): yaml.dump(config, config_file) except: print( - f"🖇⏰ AgentOps: Unable to write to {config_path}. Time Travel not activated" + f"🖇️ Agentops: ⏰ Time Travel Error - Unable to write to {config_path}. Time Travel not activated" ) return - if is_active: - manage_time_travel_state(activated=True) - print("🖇⏰ AgentOps: Time Travel Activated") - else: - manage_time_travel_state(activated=False) - print("🖇⏰ AgentOps: Time Travel Deactivated") - - -def add_time_travel_terminal_indicator(): - with _time_travel_lock: - if not TimeTravel()._initialized: - print(f"🖇⏰ AgentOps: Time Travel Activated") - - -def reset_terminal(): - print("\033[0m", end="") - - -def manage_time_travel_state(activated=False, error=None): - if activated: - add_time_travel_terminal_indicator() - TimeTravel()._initialized = True + if is_active: + print("🖇️ Agentops: ⏰ Time Travel Activated") else: - reset_terminal() - print("🖇⏰ AgentOps: Deactivating Time Travel. Error with configuration") - if error: - print(error) - TimeTravel()._initialized = False + print("🖇️ Agentops: ⏰ Time Travel Deactivated")