Spaces:
Runtime error
Runtime error
File size: 2,465 Bytes
90a9891 |
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 |
from datetime import datetime
class ResponseStepObservations:
def __init__(self, episode, step):
self.timestamp = datetime.utcnow()
self.episode = episode
self.step = step
self.llm_preview = ''
self.llm_responses = []
self.tts_raw_chunk_ids = []
def __str__(self):
state = ', '.join(f'{k}={v}' for k, v in self.__dict__.items() if k not in {'episode', 'step', 'timestamp'})
return f'episode={self.episode}, step={self.step}, timestamp={self.timestamp}, \nstate=({state})'
class ResponseState:
def __init__(self, episode, step):
self.timestamp = datetime.utcnow()
self.episode = episode
self.step = step
self.current_responses = []
self.speech_chunks_per_response = []
self.is_speaking = False
def __str__(self):
state = ', '.join(f'{k}={v}' for k, v in self.__dict__.items() if k not in {'episode', 'step'})
return f'episode={self.episode}, step={self.step}, \nstate=({state})'
class ResponseStateManager:
def __init__(self):
self.episode = 0
self.step = 0
self.response_step_obs = None
self.response_state = None
self.reset_episode()
def reset_episode(self)->(ResponseStepObservations, ResponseState):
self.episode += 1
self.step = 0
self.response_state = ResponseState(self.episode, self.step)
self.response_step_obs = ResponseStepObservations(self.episode, self.step)
return self.response_step_obs, self.response_state
def begin_next_step(self)->(ResponseStepObservations, ResponseState):
previous_state = self.response_step_obs
self.step += 1
self.response_step_obs = ResponseStepObservations(self.episode, self.step)
return previous_state, self.response_state
def set_llm_preview(self, llm_preview):
self.response_step_obs.llm_preview = llm_preview
def add_llm_response_and_clear_llm_preview(self, llm_response):
self.response_state.current_responses.append(llm_response)
self.response_state.speech_chunks_per_response.append(0)
self.response_step_obs.llm_responses.append(llm_response)
self.response_step_obs.llm_preview = ''
def add_tts_raw_chunk_id(self, chunk_id, llm_sentence_id):
self.response_state.speech_chunks_per_response[llm_sentence_id] += 1
self.response_step_obs.tts_raw_chunk_ids.append(chunk_id) |