KaniumCogs/reginaldCog/llm_clients/llm_client.py

from abc import ABC, abstractmethod
from openai import OpenAI
from reginaldCog.messenger_clients.messenger_client import ClientMessage


class ILLMContent(ABC):
    pass


class OpenAIContent(ILLMContent):
    def __init__(self):
        self._content_items: list[dict[str, str]] = []

    @property
    def content_items(self) -> list[dict]:
        return self._content_items

    @content_items.setter
    def content_items(self, value: list[dict[str, str]]):
        self._content_items = value


class ILLMContentBuilder(ABC):
    pass


class OpenAIContentBuilder(ILLMContentBuilder):
    def __init__(self, content: OpenAIContent):
        self.content = content

    def add_output_text(self, text: str):
        item = {"type": "output_text", "text": text}
        self.content.content_items.append(item)
        return self

    def add_input_text(self, text: str):
        item = {"type": "input_text", "text": text}
        self.content.content_items.append(item)
        return self

    def add_input_image(self, image_url: str):
        item = {"type": "input_image", "image_url": image_url}
        self.content.content_items.append(item)
        return self

    def add_from_dict(self, item: dict):
        self.content.content_items.append(item)
        return self


class ILLMMessage(ABC):
    pass


class OpenAIMessage(ILLMMessage):
    def __init__(self):
        self.content = OpenAIContent()
        self.role = ""

    @property
    def to_dict(self):
        return {"role": self.role, "content": self.content.content_items}


class ILLMMessageBuilder(ABC):
    pass


class OpenAIMessageBuilder(ILLMMessageBuilder):
    def __init__(self, message: OpenAIMessage):
        self.message = message

    def set_role(self, role: str):
        self.message.role = role
        return self

    def set_content(self, content: OpenAIContent):
        self.message.content = content
        return self


class ILLMPrompt(ABC):
    pass


class OpenAIPrompt(ILLMPrompt):
    def __init__(self):
        self.messages = []

    @property
    def to_list(self):
        return [i_message.to_dict for i_message in self.messages]


class ILLMPromptBuilder(ABC):
    @abstractmethod
    def __init__(self, prompt: ILLMPrompt):
        pass

    @abstractmethod
    def add_message(self, message: ILLMMessage):
        pass


class OpenAIPromptBuilder(ILLMPromptBuilder):
    def __init__(self, prompt: OpenAIPrompt):
        self.prompt = prompt

    def add_message(self, message: OpenAIMessage):
        self.prompt.messages.append(message)


class ILLMClient(ABC):
    @abstractmethod
    def get_response(self, prompt: ILLMPrompt):
        pass


class OpenAIClient(ILLMClient):
    def __init__(self):
        self.model = 'gpt-4.1-mini'
        self.client = OpenAI()

    def get_response(self, prompt: OpenAIPrompt):
        response_input = {"model": self.model, "input": prompt.to_list}
        return self.client.responses.create(**response_input)


class IMessageAdapter(ABC):
    @abstractmethod
    def to_message(self) -> ILLMMessage:
        pass


class OpenAIResponseAdapter(IMessageAdapter):
    def __init__(self, response):
        self.response = response
        self.response_output = response.output[0]

    def to_message(self) -> OpenAIMessage:
        content = OpenAIContent()
        content_builder = OpenAIContentBuilder(content)
        message = OpenAIMessage()
        message_builder = OpenAIMessageBuilder(message)

        message_builder.set_role(self.response_output.role)\
                       .set_content(content)

        for i_content_item in self.response_output.content:
            item = i_content_item.to_dict()
            content_builder.add_from_dict(item)

        return message


class MessengerClientMessageAdapter(IMessageAdapter):
    def __init__(self, message: ClientMessage):  # Dependency Inversion violation, figure out later
        self.message = message

    def to_message(self) -> ILLMMessage:
        content = OpenAIContent()
        content_builder = OpenAIContentBuilder(content)
        message = OpenAIMessage()
        message_builder = OpenAIMessageBuilder(message)

        message_builder.set_role("user")\
                       .set_content(content)

        if self.message:
            content_builder.add_input_text(self.message.content)
        for i_image_url in self.message.image_urls:
            content_builder.add_input_image(i_image_url)

        return message


if __name__ == "__main__":
    dev_message_text = "You are a butler named Reginald. Give your answers in a refined manners of the English butler."
    user_message_text = "Tell me what's on this picture"
    pic_url = "https://media.discordapp.net/attachments/913835221814763550/1408534120421654669/image.png?ex=68ad62df&is=68ac115f&hm=bdfcf878f20e154a575a528b9cb274a8c2162f84ce12fc3876856e821ef9cc63&=&format=webp&quality=lossless&width=1342&height=755"

    # Create developer message content and message
    dev_content = OpenAIContent()
    dev_content_builder = OpenAIContentBuilder(dev_content).add_input_text(dev_message_text)
    dev_message = OpenAIMessage()
    OpenAIMessageBuilder(dev_message).set_role("developer").set_content(dev_content)

    # Create user message content and message
    user_content = OpenAIContent()
    OpenAIContentBuilder(user_content).add_input_text(user_message_text).add_input_image(pic_url)
    user_message = OpenAIMessage()
    OpenAIMessageBuilder(user_message).set_role("user").set_content(user_content)

    # Create prompt and add messages
    test_prompt = OpenAIPrompt()
    prompt_builder = OpenAIPromptBuilder(test_prompt)
    prompt_builder.add_message(dev_message)
    prompt_builder.add_message(user_message)

    # Create client and get response
    client = OpenAIClient()
    test_response = client.get_response(test_prompt)

    # Create response adapter and add response to prompt
    response_adapter = OpenAIResponseAdapter(test_response)
    response_message = response_adapter.to_message()
    prompt_builder.add_message(response_message)

    # Print messages
    print(test_prompt.to_list)