app_utils.py

import datetime as dt
import streamlit as st
from streamlit.logger import get_logger
import langchain
from langchain.memory import ConversationBufferMemory

from app_config import ENVIRON
from models.openai.finetuned_models import finetuned_models, get_finetuned_chain
from models.openai.role_models import get_role_chain, role_templates
from mongo_utils import new_convo

langchain.verbose = ENVIRON=="dev"
logger = get_logger(__name__)

def add_initial_message(model_name, memory):
    if "Spanish" in model_name:
        memory.chat_memory.add_ai_message("Hola necesito ayuda") 
    else:
        memory.chat_memory.add_ai_message("Hi I need help")


def push_convo2db(memories, username, language):
    if len(memories) == 1:
        issue =  memories['memory']['issue']
        model_one = memories['memory']['source']
        new_convo(st.session_state['db_client'], issue, language, username, False, model_one)
    else:
        issue =  memories['commonMemory']['issue']
        model_one = memories['memoryA']['source']
        model_two = memories['memoryB']['source']
        new_convo(st.session_state['db_client'], issue, language, username, True, model_one, model_two)

def change_memories(memories, username, language, changed_source=False):
    for memory, params in memories.items():
        if (memory not in st.session_state) or changed_source:
            source = params['source']
            logger.info(f"Source for memory {memory} is {source}")
            if source in ('OA_rolemodel','OA_finetuned'):
                st.session_state[memory] = ConversationBufferMemory(ai_prefix='texter', human_prefix='helper')
    
    if ("convo_id" in st.session_state) and changed_source:
        del st.session_state['convo_id']
    

def clear_memory(memories, username, language):
    for memory, _ in memories.items():
        st.session_state[memory].clear()

    if "convo_id" in st.session_state:
        del st.session_state['convo_id']
    

def create_memory_add_initial_message(memories, username, language, changed_source=False):
    change_memories(memories, username, language, changed_source=changed_source)
    for memory, _ in memories.items():
        if len(st.session_state[memory].buffer_as_messages) < 1:
            add_initial_message(language, st.session_state[memory])


def get_chain(issue, language, source, memory, temperature):
    if source in ("OA_finetuned"):
        OA_engine = finetuned_models[f"{issue}-{language}"]
        return get_finetuned_chain(OA_engine, memory, temperature)
    elif source in ('OA_rolemodel'):
        template = role_templates[f"{issue}-{language}"]
        return get_role_chain(template, memory, temperature)