interstellar_ai/py/ai.py

from mistralai import Mistral
from openai import OpenAI
import google.generativeai as genai
import anthropic
import ollama


class AI:
    @staticmethod
    def process_local(model, messages, return_class, access_token):
        stream = ollama.chat(
            model=model,
            messages=messages,
            stream=True,
            options={"temperature": 0.5},
        )

        with return_class.ai_response_lock:
            return_class.ai_response[access_token] = ""

        for chunk in stream:
            with return_class.ai_response_lock:
                return_class.ai_response[access_token] += chunk["message"]["content"]

    @staticmethod
    def process_mistralai(model, messages, return_class, access_token, api_key):
        client = Mistral(api_key=api_key)

        stream_response = client.chat.stream(model=model, messages=messages)

        with return_class.ai_response_lock:
            return_class.ai_response[access_token] = ""

        for chunk in stream_response:
            with return_class.ai_response_lock:
                return_class.ai_response[access_token] += chunk.data.choices[
                    0
                ].delta.content

    @staticmethod
    def process_openai(model, messages, return_class, access_token, api_key):
        client = OpenAI(api_key=api_key)

        stream_response = client.chat.completions.create(
            model=model, messages=messages, stream=True
        )

        with return_class.ai_response_lock:
            return_class.ai_response[access_token] = ""

        for chunk in stream_response:
            with return_class.ai_response_lock:
                return_class.ai_response[access_token] += chunk.choices[0].delta.content

    @staticmethod
    def process_anthropic(model, messages, return_class, access_token, api_key):
        client = anthropic.Anthropic(api_key=api_key)

        with return_class.ai_response_lock:
            return_class.ai_response[access_token] = ""

        with client.messages.stream(
            max_tokens=1024,
            model=model,
            messages=messages,
        ) as stream:
            for text in stream.text_stream:
                with return_class.ai_response_lock:
                    return_class.ai_response[access_token] += text

    @staticmethod
    def process_google(model, messages, return_class, access_token, api_key):
        message = messages[-1]["content"]
        messages.pop()

        for msg in messages:
            msg["parts"] = msg.pop()["content"]

        for msg in messages:
            if msg["role"] == "assistant":
                msg["role"] = "model"

        genai.configure(api_key=api_key)

        model = genai.GenerativeModel(model)

        chat = model.start_chat(
            history=messages,
        )

        response = chat.send_message(message, stream=True)
        for chunk in response:
            return_class.ai_response[access_token] += chunk.text