syedkhizarrayaz/BM-AI-Analysis-And-Alert-Prioritization-Agent

1# imports2from typing import override3from openai import AssistantEventHandler4 5# Define the event handler to handle the streaming responses6class EventHandler(AssistantEventHandler):7    def __init__(self):8        super().__init__()9        # self.response_data = []  # List to accumulate response text10        self.response_text = ""11    # text creation12    @override13    def on_text_created(self, text) -> None:14        # print(f"\nassistant > ", end="", flush=True)15        # self.response_data.append(str(text))  # Convert to string and store initial text16        # if text.value:  17        #     self.response_data.append(str(text.value))18        pass19 20    @override21    def on_text_delta(self, delta, snapshot):22        print(delta.value, end="", flush=True)23        # self.response_data.append(str(delta.value))  # Convert delta to string and append24        if delta.value:  25            # self.response_data.append(str(delta.value))26            self.response_text += str(delta.value)27 28    def on_tool_call_created(self, tool_call):29        pass30        # print(f"\nassistant > {tool_call.type}\n", flush=True)31 32    def on_tool_call_delta(self, delta, snapshot):33        if delta.type == 'code_interpreter':34            if delta.code_interpreter.input:35                print(delta.code_interpreter.input, end="", flush=True)36                # self.response_data.append(str(delta.code_interpreter.input))  # Ensure it's a string37                self.response_text += delta.code_interpreter.input38            if delta.code_interpreter.outputs:39                print(f"\n\noutput >", flush=True)40                for output in delta.code_interpreter.outputs:41                    if output.type == "logs":42                        # print(f"\n{output.logs}", flush=True)43                        # self.response_data.append(str(output.logs))  # Ensure logs are stored as strings44                        self.response_text += output.logs45    # return response46    def get_final_response(self) -> str:47        # Filter out unwanted text parts48        # filtered_response = [part for part in self.response_data if "Text(annotations=[], value='1')" not in part]49        50        # Combine all collected response parts into a single string51        # return ''.join(self.response_data).strip()52        return self.response_text.strip() if self.response_text else "No response generated."53 54 55 56