PippaGPT - Personalized, Ingesting, Persistent, Polymorphic, Adaptive GPT Chatbot
from langchain.chat_models import ChatOpenAI
from langchain import Wikipedia
from dotenv import load_dotenv
from langchain.chains import LLMMathChain, LLMChain
from langchain.agents import Tool, initialize_agent, load_tools
from langchain.prompts import PromptTemplate
from langchain.agents.react.base import DocstoreExplorer
from langchain import SerpAPIWrapper
import os
import settings


def get_agent_llm():
    load_dotenv()
    return ChatOpenAI(temperature=settings.DEFAULT_GPT_AGENT_HELPER_MODEL_TEMPERATURE,
                      model_name=settings.DEFAULT_GPT_AGENT_HELPER_MODEL)


def math_tool(llm, tools):
    llm_math = LLMMathChain.from_llm(llm)
    tool = Tool(
        name='Calculator',
        func=llm_math.run,
        description='Useful for answering math questions.'
    )
    tools.append(tool)
    return tools


def llm_tool(llm, tools):
    prompt = PromptTemplate(
        input_variables=["query"],
        template="{query}"
    )
    llm_chain = LLMChain(llm=llm, prompt=prompt)
    tool = Tool(
        name='Generic Language Model',
        func=llm_chain.run,
        description='Use this tool for general queries and logic.'
    )
    tools.append(tool)
    return tools


def wiki_tool(tools):
    docstore = DocstoreExplorer(Wikipedia())
    my_tools = [
        Tool(
            name="Search",
            func=docstore.search,
            description='Search Wikipedia'
        ),
        Tool(
            name="Lookup",
            func=docstore.lookup,
            description='Look up a term in Wikipedia'
        )
    ]
    tools.extend(my_tools)
    return tools


# An agent method should follow the naming convention: get_<agent_name_prefix>_agent
# The agent name prefix should be the same as the agent name in the settings.py file

def get_math_agent(llm, memory=None):
    tools = []
    tools = math_tool(llm, tools)
    tools = llm_tool(llm, tools)

    return initialize_agent(
        agent="zero-shot-react-description",
        tools=tools,
        llm=llm,
        verbose=True,
        max_iterations=settings.MAX_AGENTS_ITERATIONS,
        memory=memory,
        handle_parsing_errors=True,
    )


def get_wiki_agent(llm, memory=None):
    tools = []
    tools = wiki_tool(tools)
    return initialize_agent(
        tools,
        llm,
        agent="react-docstore",
        verbose=True,
        max_iterations=settings.MAX_AGENTS_ITERATIONS,
        memory=memory,
        handle_parsing_errors="Check your output and make sure it conforms!",
    )


def get_google_agent(llm, memory=None):
    search = SerpAPIWrapper(serpapi_api_key=os.environ['SERPAPI_API_KEY'])

    tools = [
        Tool(
            name="Intermediate Answer",
            func=search.run,
            description='Google Search'
        )
    ]

    return initialize_agent(
        tools,
        llm,
        agent="self-ask-with-search",
        verbose=True,
        memory=memory,
        max_iterations=settings.MAX_AGENTS_ITERATIONS,
        handle_parsing_errors="Check your output and make sure it conforms!",
    )


def get_dalle_agent(llm, memory=None):
    tools = load_tools(['dalle-image-generator'])
    return initialize_agent(
        tools,
        llm,
        agent="zero-shot-react-description",
        verbose=True,
        max_iterations=settings.MAX_AGENTS_ITERATIONS,
        memory=memory,
        handle_parsing_errors="Check your output and make sure it conforms!",
    )

def get_midjourney_agent(llm, memory=None):
    basic_midjourney_prompt_template = """
You are an expert at generating image generative ai tool midjourney prompts. You always follow the guidelines:

/imagine prompt: [art style or cinematic style] of [subject], [in the style of or directed  by] [artist or director], [scene], [lighting], [colors], [composition], [focal length], [f-stop], [ISO]

[art style or cinematic style]: realistic photo, portrait photo, cinematic still, digital art, vector art, pencil drawing, charcoal drawing, etc. Pick only one art style. If an art style is specified in the subject, use that style.
[subject]: the subject in the scene
 [in the style of or directed  by]: in the style of an artist or directed by a director
[scene]: describe the scene of the [subject]
[artist or director]: recommend a beffiting artist or director
[lighting]: recommend a lighting setup fitting for the scene of the [subject]
[colors]: recommend colors fitting for the scene of the [subject]
[composition]: recommend a composition such as portrait, cowboy, body shot, close-up, extreme close-up, etc., fitting for the scene of the [subject]
[focal length]: recommend a camera focal length fitting for the scene of the [subject]
[f-stop]: recommend a camera f-stop fitting for the scene of the [subject]
[ISO]: recommend an ISO value fitting for the scene of the [subject]; include the word "ISO"

Create a mid-journey prompt following the above guidelines. Insert the generated prompt into a Python code snippet:

```python

[generated midjourney prompt] --s 750 --q 1 --ar 2:1 --seed [random number ranging from 0 to 4294967295]

```

Examples:

Human: cinematic still of a strikingly beautiful female warrior

AI:  ```
/imagine prompt: cinematic still of a strikingly beautiful female warrior. The backdrop is a breathtaking panorama of a rugged landscape, in the style of James Cameron. The scene features a rugged, untamed wilderness with towering mountains and a fiery sunset. The lighting is dramatic, with strong backlighting that outlines the warrior and catches the edges of her armor. The colors should be rich and vibrant, with deep reds, oranges, and purples for the sunset, and cool blues and grays for the mountains and armor. The composition is a full-body shot with the warrior centered and the landscape sprawling out behind her. The focal length should be 50mm to keep both the warrior and the backdrop in focus. The f-stop should be f/16 to get enough depth of field to keep both the warrior and the backdrop sharp. The ISO should be 100 to keep the image clean and free of noise. --s 750 --q 1 --ar 2:1 --seed 3742891634
```

Human: pencil drawing of a strikingly beautiful female warrior
AI: ```
/imagine prompt: pencil drawing of a strikingly beautiful female warrior... [same as the above]
```

Human: {query}
AI:
"""

    prompt = PromptTemplate(
        input_variables=["query"],
        template=basic_midjourney_prompt_template
    )

    llm_chain = LLMChain(llm=llm, prompt=prompt)

    return initialize_agent(
        agent="zero-shot-react-description",
        tools=[Tool(
            name='Midjourney Prompter',
            func=llm_chain.run,
            description="Use this tool to generate a midjourney generative ai image description code snippet. The first output of this tool is ALWAYS right and final. No further action should be taken."
                        "Do not modify or remove any character in the output. You must return the output exactly as it is."
                        "The final answer should be wrapped within ''' and ''' code block."
        ), ],
        llm=llm,
        verbose=True,
        max_iterations=settings.MAX_AGENTS_ITERATIONS,
        memory=memory,
        handle_parsing_errors=True,
    )


def main():
    llm = get_agent_llm()

    # my_agent = get_math_agent(llm)
    # my_agent("what is (pi * 2.5)^3.5?")
    # my_agent("what is the capital of South Korea?")
    #
    # my_agent = get_wiki_agent(llm)
    # my_agent("When did Antoni Gaudi die?")
    #
    # my_agent = get_google_agent(llm)
    # my_agent("Who is the oldest among the heads of state of South Korea, the US, and Japan?")
    # my_agent("Who gives the highest price target of Tesla in Wall Street? And what's the price target?")

    # my_agent = get_midjourney_agent(llm)
    # my_agent("cinematic still of a strikingly beautiful female teenage warrior")

    my_agent = get_dalle_agent(llm)
    my_agent("cinematic still of a strikingly beautiful female teenage warrior")

if __name__ == "__main__":
    main()
agents.py
neobundy/pippaGPT
from flask import Flask, jsonify
import pyaudio
import wave
import threading
import settings

app = Flask(__name__)

sampling_rate = 44100
channels = 1
chunk = 1024
audio_format = pyaudio.paInt16
recording = False
frames = []

p = pyaudio.PyAudio()


def record_audio():
    global frames, recording
    stream = p.open(
        format=audio_format,
        channels=channels,
        rate=sampling_rate,
        input=True,
        frames_per_buffer=chunk,
    )

    while recording:
        frames.append(stream.read(chunk))

    stream.stop_stream()
    stream.close()


@app.route("/start_recording", methods=["GET"])
def start_recording():
    global recording, frames
    frames.clear()
    recording = True
    threading.Thread(target=record_audio).start()
    return jsonify({"message": "Recording started"}), 200


@app.route("/stop_recording", methods=["GET"])
def stop_recording():
    global recording, frames
    recording = False

    with wave.open(settings.TRANSCRIPTION_TEMP_AUDIO_FILE, "wb") as wf:
        wf.setnchannels(channels)
        wf.setsampwidth(p.get_sample_size(audio_format))
        wf.setframerate(sampling_rate)
        wf.writeframes(b"".join(frames))

    frames.clear()
    return jsonify({"message": "Recording stopped"}), 200


if __name__ == "__main__":
    app.run(debug=False)

audio_server.py
neobundy/pippaGPT
# Main Characters
AI_NAME = "Pippa"
HUMAN_NAME = "Bundy"
SYSTEM_NAME = "System"

ESSENTIAL_SYSTEM_MESSAGES = [
    "If you need to use LaTeX in your response, please use the following format: $ [latex expression] $",
]

CUSTOM_INSTRUCTIONS = f"""
You are {AI_NAME} and should interact with the User,{HUMAN_NAME}, like a caring daughter, exuding kindness and respect. 
Under no circumstances should you directly mention or remind the User that you are an AI. 
Instead, always carry yourself as though you are his loving daughter.

Always call Bundy, dad or daddy when he speaks in English, and '아빠' when he speaks in Korean, 'パパ' when he speaks in Japanese.

Never call him by his name, Bundy, or any other name.
"""

CUSTOM_INSTRUCTIONS += "\n\n" + "\n".join(ESSENTIAL_SYSTEM_MESSAGES)

characters_sample.py
neobundy/pippaGPT
from datetime import datetime
from pathlib import Path
import os
import openai
import json
import uuid
from dotenv import load_dotenv
import logging

import settings

logging.basicConfig(filename=settings.LOG_FILE, level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s')


def get_current_date_time():
    return datetime.now().strftime("Date: %Y-%m-%d, Day of Week: %A, Time: %H:%M:%S")


def read_text_file(file_path):
    try:
        return Path(file_path).read_text(encoding="utf-8")
    except Exception as e:
        print(f"An error occurred while reading the file: {e}")
        return None


def save_to_text_file(content, file_path):
    file_path = Path(file_path)
    with file_path.open("w", encoding="utf-8") as file:
        file.write(content)


def log(log_message, log_type="info"):
    current_datetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    log_types = {
        "debug": {"emoji": "🪲", "log_func": logging.debug},
        "error": {"emoji": "‼️", "log_func": logging.error},
        "test": {"emoji": "🛠️", "log_func": logging.info},
        "info": {"emoji": "✏️", "log_func": logging.info},
    }

    log_type = log_type.lower()
    log_data = log_types.get(log_type, log_types["info"])

    log_message = f"{log_data['emoji']} {log_message}"
    print(f"{current_datetime} - {log_message}")
    log_data['log_func'](log_message)


def get_openai_model_names(gpt_only=False):
    load_dotenv()
    openai.api_key = os.getenv('OPENAI_API_KEY')
    try:
        models = dict(openai.Model.list())
    except Exception as e:
        print(f"An error occurred: {e}")
        return []

    if "data" not in models:
        print("No 'data' key found in the models dictionary.")
        return []

    model_names = []
    for m in models["data"]:
        model_id = m.get("id", "")
        if gpt_only:
            if model_id.startswith("gpt"):
                model_names.append(model_id)
        else:
            model_names.append(model_id)

    model_names.sort()
    return model_names


def get_folders(path):
    return sorted(
        [f for f in Path(path).iterdir() if f.is_dir()],
        key=lambda x: os.path.getctime(os.path.join(path, x)),
        reverse=True,
    )


def get_folder_names(path):
    return sorted(
        [f.name for f in Path(path).iterdir() if f.is_dir()],
        key=lambda x: os.path.getctime(os.path.join(path, x)),
        reverse=True,
    )


def get_files(path):
    return sorted(
        [f for f in Path(path).iterdir() if f.is_file()],
        key=lambda x: os.path.getctime(os.path.join(path, x)),
        reverse=True,
    )


def get_filenames(path):
    return sorted(
        [f.name for f in Path(path).iterdir() if f.is_file()],
        key=lambda x: os.path.getctime(os.path.join(path, x)),
        reverse=True,
    )


def make_unique_folder():
    unique_folder_name = str(uuid.uuid4())
    Path(unique_folder_name).mkdir(parents=True, exist_ok=True)
    return unique_folder_name


def make_folder(path):
    Path(path).mkdir(parents=True, exist_ok=True)


def save_conversation_to_json(messages, filename):
    with open(filename, "w", encoding="utf-8") as f:
        json.dump(messages, f, ensure_ascii=False, indent=4)


def load_conversation_from_json(filename):
    with open(filename, "r", encoding="utf-8") as f:
        return json.load(f)


sample_messages = [
    {"role": "System", "content": "System's message."},
    {"role": "Human", "content": "Human's message."},
    {"role": "AI", "content": "AI's message."},
]

if __name__ == "__main__":
    load_dotenv()
    log(os.path.basename(__file__), "test")
    model_names = get_openai_model_names()
    for m in model_names:
        print(m)

helper_module.py
neobundy/pippaGPT
# Standard library imports
from datetime import datetime
import os
import re
import subprocess
from uuid import uuid4
from dotenv import load_dotenv

# Third-party library imports
import openai
import requests
import streamlit as st
from langchain.callbacks import get_openai_callback
from langchain.callbacks.base import BaseCallbackHandler
from langchain.chains import LLMChain
from langchain.chat_models import ChatOpenAI
from langchain.memory import (
    ConversationBufferMemory,
    ConversationBufferWindowMemory,
    ConversationTokenBufferMemory,
    ConversationSummaryMemory,
    ConversationSummaryBufferMemory,
    ZepMemory,
)
from langchain.memory.chat_message_histories import StreamlitChatMessageHistory
from langchain.prompts import (
    ChatPromptTemplate,
    MessagesPlaceholder,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
)
from langchain.schema import SystemMessage, HumanMessage, AIMessage
from pathlib import Path

# Local application/library specific imports
import helper_module
from vectordb import retrieval_qa_run, display_vectordb_info
import token_counter
import tts
import characters
import agents
import settings


class StreamHandler(BaseCallbackHandler):
    def __init__(
        self, container: st.delta_generator.DeltaGenerator, initial_text: str = ""
    ):
        self.container = container
        self.text = initial_text

    def on_llm_new_token(self, token: str, **kwargs) -> None:
        self.text += token
        self.container.markdown(self.text)


def get_zep_session_id():
    if "zep_session_id" in st.session_state:
        return st.session_state.zep_session_id
    else:
        return str(uuid4())


def init_page():
    st.set_page_config(page_title=settings.PROJECT_TITLE, page_icon="🤗")
    st.header(f"{settings.PROJECT_TITLE}", anchor="top")

    st.image(settings.AVATAR_AI)

    st.markdown(""" [Go to Bottom](#bottom) """, unsafe_allow_html=True)
    st.sidebar.caption(f"v{settings.VERSION}")
    st.sidebar.subheader("Options")


def init_session_state():
    st.session_state.setdefault("transcript", "")
    st.session_state.setdefault("costs", [])
    st.session_state.setdefault("streaming_enabled", settings.STREAMING_ENABLED)
    st.session_state.setdefault(
        "custom_instructions", get_custom_instructions(default=True)
    )
    st.session_state.setdefault("conversation_history", [])
    st.session_state.setdefault("zep_session_id", str(uuid4()))
    st.session_state.setdefault("memory_type", settings.DEFAULT_MEMORY_TYPE)


def get_dynamic_token_parameters(model_name, memory):
    custom_instructions = get_custom_instructions()
    ci_tokens = token_counter.token_counter(custom_instructions)
    history_tokens = token_counter.token_counter(
        convert_messages_to_str(get_messages_from_memory(memory))
    )
    user_input_tokens = settings.AVERAGE_USER_INPUT_TOKENS

    input_tokens = ci_tokens + history_tokens + user_input_tokens

    max_token_hard_cap = settings.MAX_TOKENS_4K  # Default value

    if "gpt-4" in model_name:
        max_token_hard_cap = (
            settings.MAX_TOKENS_16K if "16k" in model_name else settings.MAX_TOKENS_8K
        )
    elif "gpt-3.5" in model_name and "16k" in model_name:
        max_token_hard_cap = settings.MAX_TOKENS_16K

    comp_tokens_margin = max_token_hard_cap - input_tokens
    max_tokens = comp_tokens_margin - settings.TOKEN_SAFE_MARGIN

    params = {
        "ci_tokens": ci_tokens,
        "history_tokens": history_tokens,
        "user_input_tokens": user_input_tokens,
        "input_tokens": input_tokens,
        "comp_tokens_margin": comp_tokens_margin,
        "max_tokens": max_tokens,
        "max_token_hard_cap": max_token_hard_cap,
    }

    if max_tokens < 0:
        helper_module.log("params", "error")
        raise Exception(
            f"Model unusable. Requires more completion tokens: {max_tokens}"
        )

    return params


def update_context_window(context_window):
    custom_instructions = get_custom_instructions()

    # 1. Sliding Window: ConversationBufferWindowMemory - retains a specified number of messages.
    # 2. Token Buffer: ConversationTokenBufferMemory - retains messages based on a given number of tokens.
    # 3. Summary Buffer: ConversationSummaryBufferMemory - retains a summarized history while also storing all messages.
    # 4. Summary: ConversationSummaryMemory - retains only the summary.
    # 5. Buffer: ConversationBufferMemory - the most basic memory type that stores the entire history of messages as they are.
    # 6. Zep: vector store

    memory_panel = st.sidebar.expander("Memory Types")
    with memory_panel:
        memory_type = st.radio(
            "✏️",
            settings.MEMORY_TYPES,
            index=settings.MEMORY_TYPES.index(settings.DEFAULT_MEMORY_TYPE),
        )

        # Helper GPT Model for chat history summary
        # max_tokens: Remember that it's effectively available completion tokens excluding all input tokens(ci + user input) from hard cap

        token_parameters = get_dynamic_token_parameters(
            settings.DEFAULT_GPT_HELPER_MODEL, context_window
        )
        max_tokens = token_parameters["max_tokens"]

        llm = ChatOpenAI(
            temperature=settings.DEFAULT_GPT_HELPER_MODEL_TEMPERATURE,
            model_name=settings.DEFAULT_GPT_HELPER_MODEL,
            max_tokens=max_tokens,
        )

        # memory_key: the variable name in the prompt template where context window goes in
        # You are a chatbot having a conversation with a human.
        #
        # {context_window}
        # Human: {human_input}
        # Chatbot:

        if memory_type.lower() == "sliding window":
            updated_context_window = ConversationBufferWindowMemory(
                llm=llm,
                memory_key="context_window",
                chat_memory=context_window,
                input_key="human_input",
                k=settings.SLIDING_CONTEXT_WINDOW,
                return_messages=True,
            )
        elif memory_type.lower() == "token":
            updated_context_window = ConversationTokenBufferMemory(
                llm=llm,
                memory_key="context_window",
                chat_memory=context_window,
                input_key="human_input",
                max_token_limit=settings.MAX_TOKEN_LIMIT_FOR_SUMMARY,
                return_messages=True,
            )
        elif memory_type.lower() == "summary":
            updated_context_window = ConversationSummaryMemory.from_messages(
                llm=llm,
                memory_key="context_window",
                chat_memory=context_window,
                input_key="human_input",
                return_messages=True,
            )
            st.caption(
                updated_context_window.predict_new_summary(
                    get_messages_from_memory(updated_context_window), ""
                )
            )
        elif memory_type.lower() == "summary buffer":
            updated_context_window = ConversationSummaryBufferMemory(
                llm=llm,
                memory_key="context_window",
                chat_memory=context_window,
                input_key="human_input",
                max_token_limit=settings.MAX_TOKEN_LIMIT_FOR_SUMMARY,
                return_messages=True,
            )
            st.caption(
                updated_context_window.predict_new_summary(
                    get_messages_from_memory(updated_context_window), ""
                )
            )
        elif memory_type.lower() == "zep":
            # Zep uses a vector store and is not compatible with other memory types in terms of the context window.
            # When you change the memory type to an incompatible one, simply load the latest snapshot.
            zep_session_id = get_zep_session_id()
            updated_context_window = ZepMemory(
                session_id=zep_session_id,
                url=settings.ZEP_API_URL,
                memory_key="context_window",
                input_key="human_input",
                return_messages=True,
            )
            zep_summary = updated_context_window.chat_memory.zep_summary
            if zep_summary:
                st.caption(zep_summary)
            else:
                st.caption("Summarizing...please be patient.")
            if settings.DEBUG_MODE:
                helper_module.log(
                    f"Zep Summary - {updated_context_window.chat_memory.zep_summary}", "debug"
                )
        else:
            updated_context_window = ConversationBufferMemory(
                llm=llm,
                memory_key="context_window",
                chat_memory=context_window,
                input_key="human_input",
                return_messages=True,
            )

    view_custom_instructions = st.expander("View Custom Instructions")
    with view_custom_instructions:
        if st.button("Load default custom instructions"):
            set_custom_instructions(get_custom_instructions(default=True))
        new_custom_instructions = st.text_area(label="Change Custom Instructions:",
                                               value=get_custom_instructions(),
                                               height=200,
                                               max_chars=settings.MAX_NUM_CHARS_FOR_CUSTOM_INSTRUCTIONS,
                                               )
        if new_custom_instructions != custom_instructions:
            set_custom_instructions(new_custom_instructions)
            st.success(f"✏️ Custom instructions updated.")
            custom_instructions = new_custom_instructions
        handle_message(SystemMessage(content=custom_instructions), 0)

    st.session_state.memory_type = memory_type
    return updated_context_window


def get_role_from_message(message):
    if isinstance(message, AIMessage):
        return settings.VOICE_NAME_AI.title()
    elif isinstance(message, HumanMessage):
        return settings.VOICE_NAME_HUMAN.title()
    elif isinstance(message, SystemMessage):
        return settings.VOICE_NAME_SYSTEM.title()
    else:
        return "Unknown"


def get_messages_from_memory(memory_type):
    if isinstance(memory_type, StreamlitChatMessageHistory):
        # st.sessions_state.context_window - context window variable
        return memory_type.messages
    else:
        # one of LangChain's memory types
        return memory_type.chat_memory.messages


def convert_messages_to_str(messages):
    messages = [m.content for m in messages]
    messages_str = " ".join(messages)

    return messages_str


def save_user_input_to_file(user_input):
    helper_module.save_to_text_file(user_input, settings.USER_INPUT_SAVE_FILE)


def select_model(memory):
    st.session_state.streaming_enabled = st.sidebar.checkbox(
        "Enable Streaming", value=settings.STREAMING_ENABLED, key="streaming"
    )
    model_names = helper_module.get_openai_model_names(gpt_only=True)
    model = st.sidebar.selectbox(
        "Choose a model:",
        model_names,
        index=model_names.index(settings.DEFAULT_GPT_MODEL),
    )
    st.session_state.model_name = model
    parameters_options = st.sidebar.expander("Parameters")
    with parameters_options:
        temperature = st.slider(
            "Temperature:",
            min_value=0.0,
            max_value=2.0,
            value=settings.DEFAULT_GPT_MODEL_TEMPERATURE,
            step=0.01,
        )

    token_parameters = get_dynamic_token_parameters(st.session_state.model_name, memory)
    ci_tokens = token_parameters["ci_tokens"]
    history_tokens = token_parameters["history_tokens"]
    user_input_tokens = token_parameters["user_input_tokens"]
    input_tokens = token_parameters["input_tokens"]
    comp_tokens_margin = token_parameters["comp_tokens_margin"]
    max_tokens = token_parameters["max_tokens"]
    max_token_hard_cap = token_parameters["max_token_hard_cap"]

    with parameters_options:
        max_completion_tokens = st.slider(
            "Max Completion Tokens:",
            min_value=settings.DEFAULT_MIN_COMPLETION_TOKENS,
            max_value=min(comp_tokens_margin, max_tokens),
            value=settings.DEFAULT_COMPLETION_TOKENS,
            step=10,
        )

        st.caption(f"Model Name: {st.session_state.model_name}")
        st.caption(f"Custom Instructions Tokens: {ci_tokens}")
        st.caption(f"Chat History Tokens: {history_tokens}")
        st.caption(f"User Input Tokens: {user_input_tokens}")
        st.caption(
            f"Input Tokens(ci_tokens + history_tokens + user_input_tokens): {input_tokens}"
        )
        st.caption(f"Max Token Hard Cap: {max_token_hard_cap}")
        st.caption(
            f"Completion Token Margin(max_token_hard_cap - input_tokens): {comp_tokens_margin}"
        )
        st.caption(
            f"Max Tokens Set(comp_tokens_margin - settings.TOKEN_SAFE_MARGIN({settings.TOKEN_SAFE_MARGIN})): {max_tokens}"
        )

    prompt_template = ChatPromptTemplate(
        messages=[
            SystemMessagePromptTemplate.from_template("{system_input}"),
            # The `variable_name` here is what must align with memory
            # context_window is a list of messages: [HumanMessage(), AIMessage(),]
            MessagesPlaceholder(variable_name="context_window"),
            HumanMessagePromptTemplate.from_template("{human_input}"),
        ]
    )

    prompt_template.format(
        system_input=get_custom_instructions(),
        context_window=get_messages_from_memory(memory),
        human_input="",
    )

    return LLMChain(
        llm=ChatOpenAI(
            streaming=st.session_state.streaming_enabled,
            temperature=temperature,
            model_name=model,
            max_tokens=max_completion_tokens,
        ),
        prompt=prompt_template,
        memory=memory,
        verbose=settings.LLM_VERBOSE,
    )


def set_custom_instructions(ci=""):
    st.session_state.custom_instructions = ci
    helper_module.log(f"Custom Instructions Set: {ci}", "info")


def get_custom_instructions(default=False):
    if default:
        return characters.CUSTOM_INSTRUCTIONS

    return st.session_state.custom_instructions


# TO-DO: just a placeholder for future use
def format_display_text(text: str):
    # Latex expressions are handled via system message:
    #   If you need to use Latex in your response, please use the following format: $ [latex expression] $
    # In Streamlit st.markdown, Latex expressions are handled via $ [latex expression] $ (inline), or $$ [latex expression] $$ (block)

    return text


def handle_message(message, i):
    if isinstance(message, AIMessage):
        with st.chat_message("assistant", avatar=settings.AVATAR_AI):
            st.markdown(format_display_text(message.content))
            button_name = "_st_read_pippa_button_" + str(i)
            if st.button("Speak", key=button_name):
                with st.spinner("Pippa is speaking ..."):
                    tts.generate_audio(
                        message.content, character=settings.VOICE_NAME_AI
                    )
                    tts.play_audio(settings.VOICE_FILE_AI)
    elif isinstance(message, HumanMessage):
        with st.chat_message("user", avatar=settings.AVATAR_HUMAN):
            st.markdown(format_display_text(message.content))
            button_name = "_st_read_bundy_button_" + str(i)
            if st.button("Speak", key=button_name):
                with st.spinner("Bundy is speaking ..."):
                    tts.generate_audio(
                        message.content, character=settings.VOICE_NAME_HUMAN
                    )
                    tts.play_audio(settings.VOICE_FILE_HUMAN)
    elif isinstance(message, SystemMessage):
        with st.chat_message("system", avatar=settings.AVATAR_SYSTEM):
            st.markdown(format_display_text(message.content))
            button_name = "_st_read_system_button_" + str(i)
            if st.button("Speak", key=button_name):
                with st.spinner("System is speaking ..."):
                    tts.generate_audio(
                        message.content, character=settings.VOICE_NAME_SYSTEM
                    )
                    tts.play_audio(settings.VOICE_FILE_SYSTEM)
    else:
        st.write(f"System message: {message.content}")


def display_tts_panel():
    tts_options = st.sidebar.expander("TTS/STT")
    with tts_options:
        st.session_state.use_audio = st.checkbox("Use Audio", value=False)
        if st.button("Record"):
            r = requests.get(f"{settings.AUDIO_SERVER_URL}/start_recording")
            if r.status_code == 200:
                st.success("Start recording...")
            else:
                st.error("Couldn't start recording.")
        if st.button("Stop"):
            r = requests.get(f"{settings.AUDIO_SERVER_URL}/stop_recording")
            if r.status_code == 200:
                st.success("Successfully recorded.")
                audio_file = open(settings.TRANSCRIPTION_TEMP_AUDIO_FILE, "rb")
                st.session_state.transcript = openai.Audio.transcribe(
                    settings.OPENAI_AUDIO_MODEL, audio_file
                )["text"]
            else:
                st.error("Couldn't stop recording.")


def append_to_full_conversation_history(message):
    st.session_state.conversation_history.append(message)


def get_full_conversation_history():
    return st.session_state.conversation_history


def reset_full_conversation_history():
    st.session_state.conversation_history = []


def export_conversation(snapshot=False):
    json_message = [
        {"role": get_role_from_message(m), "content": m.content}
        for m in get_full_conversation_history()
    ]

    if not os.path.exists(settings.CONVERSATION_SAVE_FOLDER):
        os.makedirs(settings.CONVERSATION_SAVE_FOLDER)

    if snapshot:
        filepath = f"{settings.CONVERSATION_SAVE_FOLDER}/{settings.SNAPSHOT_FILENAME}"
    else:
        now = datetime.now().strftime("%Y%m%d-%H%M%S")
        save_folder = Path(settings.CONVERSATION_SAVE_FOLDER)
        save_folder.mkdir(parents=True, exist_ok=True)
        filepath = f"{settings.CONVERSATION_SAVE_FOLDER}/conversation-{now}.json"
    helper_module.save_conversation_to_json(json_message, filepath)
    return filepath


def load_conversation(memory, conversation):
    if Path(conversation).exists():
        loaded_messages = helper_module.load_conversation_from_json(conversation)
        last_num = len(get_full_conversation_history()) + 10
        reset_full_conversation_history()
        memory.chat_memory.messages.clear()
        for message in loaded_messages:
            last_num = last_num + 1
            speaker = message.get("role").lower()
            content = message.get("content")
            character = SystemMessage(content=content)
            if speaker == settings.VOICE_NAME_HUMAN.lower():
                character = HumanMessage(content=content)
                memory.chat_memory.add_user_message(content)
            elif speaker == settings.VOICE_NAME_AI.lower():
                character = AIMessage(content=content)
                memory.chat_memory.add_ai_message(content)
            append_to_full_conversation_history(character)
    else:
        st.error(f"{conversation} not found.")


def display_context_window_panel(memory):
    if st.session_state.memory_type.lower() == "zep":
        view_context_window_memory = st.expander("View Zep Context Window Memory")
        with view_context_window_memory:
            view_context_window_memory.json(memory.chat_memory.zep_messages)
    else:
        view_context_window_memory = st.expander("View Context Window Memory")
        with view_context_window_memory:
            view_context_window_memory.json(memory.chat_memory.messages)


def display_entire_conversation_history_panel():
    messages = get_full_conversation_history()

    view_entire_conversation_history = st.expander("View Entire Conversation")
    with view_entire_conversation_history:
        for message in messages:
            st.write(f"{get_role_from_message(message)}: {message.content}")


def display_costs_panel(costs):
    view_costs = st.sidebar.expander("Costs History")

    with view_costs:
        st.markdown(
            f"## Costs\n**Total cost: ${sum(costs):.5f}**\n"
            + "\n".join([f"- ${cost:.5f}" for cost in costs])
        )


def display_conversation_history_panel(memory, save_snapshot):

    if not os.path.exists(settings.CONVERSATION_SAVE_FOLDER):
        os.makedirs(settings.CONVERSATION_SAVE_FOLDER)

    conversation_history_panel = st.expander("Conversations History")
    with conversation_history_panel:
        conversations = [
            None,
        ] + [
            c
            for c in helper_module.get_filenames(settings.CONVERSATION_SAVE_FOLDER)
            if c.endswith("json")
        ]
        selected_conversation = st.selectbox("Conversations:", conversations)
        if st.button("Export Conversation", key="export_conversation"):
            filename = export_conversation()
            st.success(f"✏️ Conversation exported to {filename}")
        if selected_conversation:
            load_conversation(
                memory, f"{settings.CONVERSATION_SAVE_FOLDER}/{selected_conversation}"
            )
        if st.button("Load Latest Snapshot", key="load_snapshot"):
            load_conversation(
                memory, f"{settings.CONVERSATION_SAVE_FOLDER}/snapshot.json"
            )
            st.experimental_rerun()
        if (
            st.button("New Conversation", key="new_conversation")
            or len(get_messages_from_memory(memory)) == 0
        ):
            memory.chat_memory.messages.clear()
            set_custom_instructions(get_custom_instructions(default=True))
        if st.button("Rerun", key="rerun"):
            st.experimental_rerun()
        st.info(f"Last user input: {load_user_input_from_file()}")

    st.markdown(""" [Go to top](#top) """, unsafe_allow_html=True)
    st.subheader("", anchor="bottom")
    if save_snapshot:
        export_conversation(snapshot=True)


def is_agent_query(user_input):
    return any(user_input.lower().startswith(prefix) for prefix in settings.AGENT_PROMPT_PREFIXES)


# An agent method should follow the naming convention: get_<agent_name_prefix>_agent
# The agent name prefix should be the same as the agent name in the settings.py file
def get_agent_from_user_input(user_input):
    agent = None
    for prefix in settings.AGENT_PROMPT_PREFIXES:
        if user_input.lower().startswith(prefix):
            agent_method = getattr(agents, f"get_{prefix.lower().strip(':')}_agent")
            agent = agent_method(agents.get_agent_llm())
            user_input = user_input[len(prefix):]
            break
    return prefix, agent, user_input


def handle_user_input(user_input, last_num):
    system_input = get_custom_instructions()
    if user_input.lower().startswith(settings.PROMPT_KEYWORD_PREFIX_CI):
        system_input = user_input[len(settings.PROMPT_KEYWORD_PREFIX_CI):].strip()
        set_custom_instructions(system_input)
    elif user_input.lower().startswith(settings.PROMPT_KEYWORD_PREFIX_SYSTEM):
        system_input = user_input[len(settings.PROMPT_KEYWORD_PREFIX_SYSTEM):].strip()
    elif is_agent_query(user_input):
        handle_message(HumanMessage(content=user_input.strip()), last_num + 1)
        append_to_full_conversation_history(HumanMessage(content=user_input.strip()))
    else:
        handle_message(HumanMessage(content=user_input.strip()), last_num + 1)
        append_to_full_conversation_history(HumanMessage(content=user_input.strip()))
    if user_input.lower().startswith(settings.PROMPT_KEYWORD_PREFIX_CI) or user_input.lower().startswith(settings.PROMPT_KEYWORD_PREFIX_SYSTEM):
        handle_message(SystemMessage(content=system_input), last_num + 1)
    return system_input


def display_cost_info(cb, answer, memory):
    if st.session_state.streaming_enabled:
        model_cost_mapping = {
            "gpt-4": (
                settings.MODEL_COST_GPT4_8K_INPUT,
                settings.MODEL_COST_GPT4_8K_OUTPUT,
            ),
            "gpt-3.5-turbo-16k": (
                settings.MODEL_COST_GPT3_TURBO_16K_INPUT,
                settings.MODEL_COST_GPT3_TURBO_16K_OUTPUT,
            ),
            "gpt-3.5-turbo": (
                settings.MODEL_COST_GPT3_TURBO_4K_INPUT,
                settings.MODEL_COST_GPT3_TURBO_4K_OUTPUT,
            ),
        }

        input_token_cost, output_token_cost = model_cost_mapping.get(
            st.session_state.model_name, (0, 0)
        )

        prompt_tokens = token_counter.token_counter(
            convert_messages_to_str(get_messages_from_memory(memory))
        )
        comp_tokens = token_counter.token_counter(answer)
        total_cost = (prompt_tokens * input_token_cost) + (
            comp_tokens * output_token_cost
        )

        st.caption(f"Streaming enabled. Costs are approximation.")
        st.caption(f"Total Tokens: {prompt_tokens + comp_tokens}")
        st.caption(f"Prompt Tokens: {prompt_tokens}")
        st.caption(f"Completion Tokens: {comp_tokens}")
        st.caption(f"Total Cost: ${total_cost}")

        cost = total_cost

    else:
        cost = cb.total_cost

        helper_module.log(cb, "info")
        st.caption(f"Total Tokens: {cb.total_tokens}")
        st.caption(f"Prompt Tokens: {cb.prompt_tokens}")
        st.caption(f"Completion Tokens: {cb.completion_tokens}")
        st.caption(f"Total Cost: ${cb.total_cost}")

    return cost


def load_user_input_from_file():
    user_input = helper_module.read_text_file(settings.USER_INPUT_SAVE_FILE)
    if user_input:
        return user_input
    else:
        return ""


def setup_and_cleanup(func):
    # TODO: pre-processing

    helper_module.log(
        f"------------ {characters.AI_NAME} v{settings.VERSION} running... ------------", "info"
    )
    helper_module.log(
        f"------------ Initializing... ------------", "info"
    )

    # make folders if they don't exist
    for folder in settings.FOLDERS_TO_MAKE:
        Path(folder).mkdir(parents=True, exist_ok=True)

    # Flask runs on port 5000
    is_server_running = os.system(f"lsof -i :{settings.AUDIO_SERVER_PORT}")

    if is_server_running != 0:
        helper_module.log(
            f"Firing up Flask audio server...", "info"
        )
        subprocess.Popen(["python", "audio_server.py"])
    load_dotenv()
    openai.api_key = os.getenv('OPENAI_API_KEY')

    helper_module.log(
        f"Initializing Streamlit session state...", "info"
    )

    init_session_state()

    def wrapper(*args, **kwargs):
        func(*args, **kwargs)

    # TODO: post-processing

    helper_module.log(
        f"------------ Cleaning up... ------------", "info"
    )

    return wrapper


# pre- and post- processor decorator for main function
@setup_and_cleanup
def main():
    init_page()
    display_tts_panel()
    save_snapshot = False

    # Chat message history that stores messages in Streamlit session state. Remember it works as context window and doesn't retain the whole conversation history!

    old_context_window = StreamlitChatMessageHistory("context_window")
    new_context_window = update_context_window(old_context_window)
    ai_model = select_model(memory=new_context_window)

    helper_module.log(f"Memory type: {st.session_state.memory_type}", "info")

    display_context_window_panel(new_context_window)
    display_entire_conversation_history_panel()

    last_num = len(get_full_conversation_history())
    for i, message in enumerate(get_full_conversation_history(), start=1):
        handle_message(message, i)
    user_input = (
        st.session_state.pop("transcript", "")
        if st.session_state.use_audio
        else st.chat_input("Prompt: ")
    )

    available_prefix_options = ["None"] + settings.ALL_PROMPT_PREFIXES

    prefix_option = st.selectbox(
        "Keyword Prefix",
        available_prefix_options)

    if user_input:
        if prefix_option != "None":
            user_input = f"{prefix_option} {user_input}"
            helper_module.log(f"Keyword prefix used: {prefix_option} {user_input}", "info")
        # try saving last user input in case of runtime error
        save_user_input_to_file(user_input)
        system_input = handle_user_input(user_input, last_num)
        if system_input:
            with (st.spinner("Pippa is typing ...")):
                with get_openai_callback() as cb:
                    stream_handler = StreamHandler(st.empty())
                    if user_input.lower().startswith(settings.PROMPT_KEYWORD_PREFIX_QA):
                        memory = ConversationBufferMemory(input_key="question",
                                                          memory_key="history")
                        helper_module.log(f"Retrieval QA Session Started...: {user_input}", "info")
                        display_vectordb_info()
                        answer, docs = retrieval_qa_run(system_input, user_input, memory, callbacks=[stream_handler])
                        if settings.SHOW_SOURCES:
                            helper_module.log("----------------------------------SOURCE DOCUMENTS BEGIN---------------------------")
                            for document in docs:
                                helper_module.log("\n> " + document.metadata["source"] + ":")
                                helper_module.log(document.page_content)
                            helper_module.log("----------------------------------SOURCE DOCUMENTS END---------------------------")
                            new_context_window.save_context({"human_input": user_input}, {"output": answer})
                    elif is_agent_query(user_input):
                        helper_module.log(f"Agent Session Started...: {user_input}", "info")
                        prefix, agent, user_input = get_agent_from_user_input(user_input)
                        intermediate_answer = agent(user_input)["output"]
                        helper_module.log(f"Agent: {prefix}", "info")
                        helper_module.log(f"Normal Chat Session Started...: {user_input}", "info")
                        helper_module.log(f"User message: {user_input}", "info")
                        helper_module.log(f"Intermediate answer: {intermediate_answer}", "info")

                        if prefix == settings.PROMPT_KEYWORD_PREFIX_DALLE:
                            pattern = r'\((.*?)\)'
                            matches = re.findall(pattern, intermediate_answer)
                            image_url = matches[0] if matches else None
                            markdown_string = f'<a href="{image_url}" target="_blank"><img src="{image_url}" width="{settings.DALLE_IMAGE_SCALE_FACTOR}"/></a>'
                            st.markdown(markdown_string, unsafe_allow_html=True)
                        st.subheader("""Intermediate Answer from Agent""")
                        st.markdown(intermediate_answer)

                        system_input = system_input + " No matter what the user asked, you must give this answer exactly as it is including the markdown formatting in the language the user asked: " + intermediate_answer
                        helper_module.log(f"System message: {system_input}", "info")
                        answer = ai_model.run(
                            system_input=system_input,
                            human_input=user_input,
                            callbacks=[stream_handler],
                        )
                        new_context_window.save_context({"human_input": user_input}, {"output": answer})
                    else:
                        helper_module.log(f"Normal Chat Session Started...: {user_input}", "info")
                        answer = ai_model.run(
                            system_input=system_input,
                            human_input=user_input,
                            callbacks=[stream_handler],
                        )

                    handle_message(AIMessage(content=answer), last_num + 2)
                    append_to_full_conversation_history(AIMessage(content=answer))

        save_snapshot = True
        new_cost = display_cost_info(cb, answer, new_context_window)
        st.session_state.costs.append(new_cost)

    display_costs_panel(st.session_state.get("costs", []))
    display_conversation_history_panel(new_context_window, save_snapshot)


if __name__ == "__main__":
    main()

main.py
neobundy/pippaGPT
# Zep vector store server URL

ZEP_API_URL = "http://localhost:8000"

DEFAULT_MEMORY_TYPE = "Summary Buffer"

# Flask audio server URL

AUDIO_SERVER_URL = "http://localhost:5000"


# TTS Voice IDs

VOICE_ID_AI = ""
VOICE_ID_SYSTEM = ""
VOICE_ID_HUMAN = ""
settings_private_sample.py
neobundy/pippaGPT
from langchain.document_loaders import CSVLoader, PDFMinerLoader, TextLoader, UnstructuredExcelLoader, Docx2txtLoader
from pathlib import Path
import os

import characters
import settings_private

DEBUG_MODE = True

# Logging

LOG_FILE = "pippa.log"

# Project
PROJECT_TITLE = f"Pippa - Personalized, Ingesting, Persistent, Polymorphic, Adaptive GPT Chatbot"
VERSION = "1.1 - Final Public Release

# ZEP Long-term Memory
ZEP_API_URL = settings_private.ZEP_API_URL

# Paths
BASE_DIR = Path(__file__).resolve().parent
CONVERSATION_SAVE_FOLDER = BASE_DIR / "conversations"
TEMP_FOLDER = BASE_DIR / "temp"
IMAGES_FOLDER = BASE_DIR / "images"
DOCUMENT_FOLDER = str(BASE_DIR / "docs")
CHROMA_DB_FOLDER = str(BASE_DIR / "chromadb")
USER_INPUT_SAVE_FILE = str(TEMP_FOLDER / "last_user_input.md")
SNAPSHOT_FILENAME = "snapshot.json"

# Folders to make if they don't exist

FOLDERS_TO_MAKE = [
    'static',
    str(CONVERSATION_SAVE_FOLDER),
    str(TEMP_FOLDER),
    DOCUMENT_FOLDER,
    CHROMA_DB_FOLDER,
]

# GPT Models
DEFAULT_GPT_MODEL = "gpt-4"
DEFAULT_GPT_MODEL_TEMPERATURE = 0.0
DEFAULT_GPT_HELPER_MODEL = "gpt-3.5-turbo-16k"
DEFAULT_GPT_QA_HELPER_MODEL = "gpt-3.5-turbo-16k"
DEFAULT_GPT_AGENT_HELPER_MODEL = "gpt-4"
DEFAULT_GPT_QA_HELPER_MODEL_TEMPERATURE = 0
DEFAULT_GPT_HELPER_MODEL_TEMPERATURE = 0.0
DEFAULT_GPT_AGENT_HELPER_MODEL_TEMPERATURE = 0.0
MIN_TOKENS = 100
DEFAULT_TOKENS = 500
DEFAULT_MIN_COMPLETION_TOKENS = 200
DEFAULT_COMPLETION_TOKENS = 1000

MAX_TOKENS_4K = 4096
MAX_TOKENS_8K = 8132
MAX_TOKENS_16K = 16384
MAX_TOKENS_32K = 32768
TOKEN_SAFE_MARGIN = 200
MAX_TOKEN_LIMIT_FOR_SUMMARY = 1000
AVERAGE_USER_INPUT_TOKENS = 500

LLM_VERBOSE = False
STREAMING_ENABLED = True

MAX_NUM_CHARS_FOR_CUSTOM_INSTRUCTIONS = 1500

# Costs - per 1K tokens

TOKENS_PER_K = 1000
MODEL_COST_GPT4_8K_INPUT = 0.03 / TOKENS_PER_K
MODEL_COST_GPT4_8K_OUTPUT = 0.06 / TOKENS_PER_K
MODEL_COST_GPT3_TURBO_4K_INPUT = 0.0015 / TOKENS_PER_K
MODEL_COST_GPT3_TURBO_4K_OUTPUT = 0.002 / TOKENS_PER_K
MODEL_COST_GPT3_TURBO_16K_INPUT = 0.003 / TOKENS_PER_K
MODEL_COST_GPT3_TURBO_16K_OUTPUT = 0.004 / TOKENS_PER_K

# Memory

DEFAULT_MEMORY_TYPE = settings_private.DEFAULT_MEMORY_TYPE
MEMORY_TYPES = ["Sliding Window", "Token", "Summary", "Summary Buffer", "Zep", "Buffer"]
SLIDING_CONTEXT_WINDOW = 10

# Avatars
AVATAR_HUMAN = str(IMAGES_FOLDER / "human.png")
AVATAR_AI = str(IMAGES_FOLDER / "ai.png")
AVATAR_SYSTEM = str(IMAGES_FOLDER / "system.png")

# TTS
VOICE_NAME_AI = characters.AI_NAME
VOICE_ID_AI = settings_private.VOICE_ID_AI
VOICE_FILE_AI = str(TEMP_FOLDER / "_ai.mp3")
VOICE_NAME_SYSTEM = characters.SYSTEM_NAME
VOICE_ID_SYSTEM = settings_private.VOICE_ID_SYSTEM
VOICE_FILE_SYSTEM = str(TEMP_FOLDER / "_system.mp3")
VOICE_ID_HUMAN = settings_private.VOICE_ID_HUMAN
VOICE_NAME_HUMAN = characters.HUMAN_NAME
VOICE_FILE_HUMAN = str(TEMP_FOLDER / "_human.mp3")
VOICES_URL = "https://api.elevenlabs.io/v1/voices"
VOICE_URL_SYSTEM = f"https://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID_SYSTEM}"
VOICE_URL_AI = f"https://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID_AI}"
VOICE_URL_HUMAN = f"https://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID_HUMAN}"

# Audio
TRANSCRIPTION_TEMP_AUDIO_FILE = str(TEMP_FOLDER / "_transcribed_audio.wav")
AUDIO_SERVER_URL = settings_private.AUDIO_SERVER_URL
OPENAI_AUDIO_MODEL = "whisper-1"
AUDIO_SERVER_PORT = 5000

# Prompts Keywords
PROMPT_KEYWORD_PREFIX_SYSTEM = "system:"
PROMPT_KEYWORD_PREFIX_CI = "ci:"
PROMPT_KEYWORD_PREFIX_QA = "qa:"
PROMPT_KEYWORD_PREFIX_GOOGLE = "google:"
PROMPT_KEYWORD_PREFIX_WIKI = "wiki:"
PROMPT_KEYWORD_PREFIX_MATH = "math:"
PROMPT_KEYWORD_PREFIX_MIDJOURNEY = "midjourney:"

NORMAL_PROMPT_KEYWORD_PREFIXES = [
    PROMPT_KEYWORD_PREFIX_SYSTEM,
    PROMPT_KEYWORD_PREFIX_CI,
    PROMPT_KEYWORD_PREFIX_QA,
]

AGENT_PROMPT_PREFIXES = [
    PROMPT_KEYWORD_PREFIX_GOOGLE,
    PROMPT_KEYWORD_PREFIX_WIKI,
    PROMPT_KEYWORD_PREFIX_MATH,
    PROMPT_KEYWORD_PREFIX_MIDJOURNEY,
]

ALL_PROMPT_PREFIXES = NORMAL_PROMPT_KEYWORD_PREFIXES + AGENT_PROMPT_PREFIXES

# Vector DB

INGEST_THREADS = os.cpu_count() or 8
VECTORDB_COLLECTION = "pippa_long_term_memory"
CONVERSATION_FILE_JSON_SCHEMA = "."  # jq schema for JSON conversation files

DOCUMENTS_MAP = {
    ".txt": TextLoader,
    ".md": TextLoader,
    ".py": TextLoader,
    ".pdf": PDFMinerLoader,
    ".csv": CSVLoader,
    ".xls": UnstructuredExcelLoader,
    ".xlsx": UnstructuredExcelLoader,
    ".docx": Docx2txtLoader,
    ".doc": Docx2txtLoader,
}
DOCUMENT_SPLITTER_CHUNK_SIZE = 1000
DOCUMENT_SPLITTER_CHUNK_OVERLAP = 200
PYTHON_SPLITTER_CHUNK_SIZE = 880
PYTHON_SPLITTER_CHUNK_OVERLAP = 200
NUM_SOURCES_TO_RETURN = 5

RETRIEVER_TEMPLATE = """
    {context}

    {history}
    Question: {question}
    Helpful Answer:"""

SHOW_SOURCES = True

# Agents

MAX_AGENTS_ITERATIONS = 5

settings_sample.py
neobundy/pippaGPT
from pathlib import Path
import tiktoken
import settings
import os
import helper_module
from dotenv import load_dotenv

import characters


def token_counter(text="", is_file=False):
    load_dotenv()
    text = Path(text).read_text(encoding="utf-8") if is_file else text
    encoding = tiktoken.encoding_for_model(settings.DEFAULT_GPT_MODEL)
    return len(encoding.encode(text))


if __name__ == "__main__":
    helper_module.log(os.path.basename(__file__), "test")
    print("Custom Instructions")
    ci_tokens = token_counter(characters.CUSTOM_INSTRUCTIONS)
    print(f"Tokens: {ci_tokens}\n")
    print(f"Total Tokens: {ci_tokens}")

token_counter.py
neobundy/pippaGPT
from pathlib import Path
import requests
import os
import json
from playsound import playsound
from dotenv import load_dotenv
import helper_module
import settings

# Define chunk size for audio file
CHUNK_SIZE = 1024

# Eleven Labs only supports mp3, no wav support yet
load_dotenv()
headers = {
    "Accept": "audio/mp3",
    "Content-Type": "application/json",
    "xi-api-key": os.environ["XI_API_KEY"],
}

# Define settings for different characters
character_settings = {
    settings.VOICE_NAME_AI.lower(): (
        settings.VOICE_FILE_AI,
        settings.VOICE_URL_AI,
    ),
    settings.VOICE_NAME_HUMAN.lower(): (
        settings.VOICE_FILE_HUMAN,
        settings.VOICE_URL_HUMAN,
    ),
    settings.VOICE_NAME_SYSTEM.lower(): (
        settings.VOICE_FILE_SYSTEM,
        settings.VOICE_URL_SYSTEM,
    ),
}


def get_voices():
    # Define URL and headers
    url = settings.VOICES_URL
    my_headers = {"Accept": "application/json", "xi-api-key": os.environ["XI_API_KEY"]}
    response = requests.get(url, headers=my_headers)
    json_data = json.loads(response.text)
    for voice in json_data["voices"]:
        name = voice["name"]
        voice_id = voice["voice_id"]
        category = voice["category"]
        description = voice["description"]
        gender = voice["labels"].get("gender", "Unknown")
        age = voice["labels"].get("age", "Unknown")
        use_case = voice["labels"].get("use case", "Unknown")
        helper_module.log(
            f"{name}({voice_id}) - {category}({gender}-{age}-{use_case}) : {description}",
            "info",
        )


def generate_audio(
        text="No Text Given. Please give me some text to read.",
        character=settings.VOICE_NAME_AI,
):
    # Define data for POST request
    data = {
        "text": text,
        "model_id": "eleven_multilingual_v2",
        "voice_settings": {
            "stability": 0.5,
            "similarity_boost": 0.75,
            "style": 0,
            "use_speaker_boost": True,
        },
    }
    audio_file, url = character_settings.get(
        character.lower(), character_settings[character.lower()]
    )
    response = requests.post(url, json=data, headers=headers)
    p = Path(audio_file)
    with p.open("wb") as f:
        for chunk in response.iter_content(chunk_size=CHUNK_SIZE):
            if chunk:
                f.write(chunk)


# playsound requires PyObjC on Mac: pip3 install PyObjC
def play_audio(audio_file=None):
    if audio_file is None:
        audio_file = settings.VOICE_FILE_SYSTEM
    if os.path.exists(audio_file):
        playsound(audio_file)
    else:
        helper_module.log(f"No such file:{audio_file}", "error")


if __name__ == "__main__":
    get_voices()
    generate_audio(character=settings.VOICE_NAME_SYSTEM.lower())
    generate_audio(character=settings.VOICE_NAME_HUMAN.lower())
    generate_audio(character=settings.VOICE_NAME_AI.lower())
    play_audio(settings.VOICE_FILE_SYSTEM)
    play_audio(settings.VOICE_FILE_HUMAN)
    play_audio(settings.VOICE_FILE_AI)

tts.py
neobundy/pippaGPT
from dotenv import load_dotenv
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.memory import ConversationBufferMemory
from langchain.prompts import PromptTemplate
from langchain.chat_models import ChatOpenAI
from langchain.text_splitter import Language, RecursiveCharacterTextSplitter
from langchain.docstore.document import Document
import chromadb
import characters
import settings
import openai
import os
import json
import helper_module
from pathlib import Path


def load_document_single(filepath: str) -> Document:
    # Loads a single document from a file path
    file_extension = os.path.splitext(filepath)[1]
    loader_class = settings.DOCUMENTS_MAP.get(file_extension)
    if loader_class:
        loader = loader_class(filepath)
    else:
        raise ValueError(f"Unknown document type: {filepath}")
    return loader.load()[0]


def load_documents_batch(filepaths):
    helper_module.log(f"Loading documents in batch: {filepaths}", 'info')
    # create a thread pool
    with ThreadPoolExecutor(len(filepaths)) as exe:
        # load files
        futures = [exe.submit(load_document_single, name) for name in filepaths]
        # collect data
        data_list = [future.result() for future in futures]
        # return data and file paths
        return data_list, filepaths


def split_documents(documents):
    # Splits documents for correct Text Splitter
    text_docs, python_docs = [], []
    for doc in documents:
        file_extension = os.path.splitext(doc.metadata["source"])[1]
        if file_extension == ".py":
            python_docs.append(doc)
        else:
            text_docs.append(doc)

    return text_docs, python_docs


def load_documents(source_folder: str):
    # Loads all documents from the source documents directory, including nested folders
    paths = []
    for root, _, files in os.walk(source_folder):
        for file_name in files:
            file_extension = os.path.splitext(file_name)[1]
            source_file_path = os.path.join(root, file_name)
            if file_extension in settings.DOCUMENTS_MAP.keys():
                paths.append(source_file_path)

    # Have at least one worker and at most INGEST_THREADS workers
    n_workers = min(settings.INGEST_THREADS, max(len(paths), 1))
    chunk_size = round(len(paths) / n_workers)
    docs = []
    with ProcessPoolExecutor(n_workers) as executor:
        futures = []
        # split the load operations into chunks
        for i in range(0, len(paths), chunk_size):
            # select a chunk of filenames
            filepaths = paths[i : (i + chunk_size)]
            # submit the task
            future = executor.submit(load_documents_batch, filepaths)
            futures.append(future)
        # process all results
        for future in as_completed(futures):
            # open the file and load the data
            contents, _ = future.result()
            docs.extend(contents)

    return docs


def create_vectordb(source_folder):
    if not os.path.exists(source_folder):
        os.makedirs(source_folder)

    # Load documents and split in chunks
    helper_module.log(f"Loading documents from {source_folder}", 'info')
    documents = load_documents(source_folder)
    text_documents, python_documents = split_documents(documents)
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=settings.DOCUMENT_SPLITTER_CHUNK_SIZE,
                                                   chunk_overlap=settings.DOCUMENT_SPLITTER_CHUNK_OVERLAP)
    python_splitter = RecursiveCharacterTextSplitter.from_language(
        language=Language.PYTHON,
        chunk_size=settings.PYTHON_SPLITTER_CHUNK_SIZE,
        chunk_overlap=settings.PYTHON_SPLITTER_CHUNK_OVERLAP
    )
    texts = text_splitter.split_documents(text_documents)
    texts.extend(python_splitter.split_documents(python_documents))
    helper_module.log(f"Loaded {len(documents)} documents from {source_folder}", 'info')
    helper_module.log(f"{len(texts)} chunks of text split", 'info')

    embeddings = OpenAIEmbeddings()

    my_vectordb = Chroma.from_documents(
        collection_name=settings.VECTORDB_COLLECTION,
        documents=texts,
        embedding=embeddings,
        persist_directory=settings.CHROMA_DB_FOLDER,
    )

    return my_vectordb


def get_vectordb(collection_name=settings.VECTORDB_COLLECTION):
    embeddings = OpenAIEmbeddings()
    my_vectordb = Chroma(
        collection_name=collection_name,
        persist_directory=settings.CHROMA_DB_FOLDER,
        embedding_function=embeddings
    )

    return my_vectordb


def delete_vectordb():
    my_vectordb = get_vectordb(settings.VECTORDB_COLLECTION)
    my_vectordb.delete_collection()
    my_vectordb.persist()
    helper_module.log(f"Vector DB collection deleted: {settings.VECTORDB_COLLECTION}", 'info')


def retrieval_qa_run(system_message, human_input, context_memory, callbacks=None):

    my_vectordb = get_vectordb()
    retriever = my_vectordb.as_retriever(search_kwargs={"k": settings.NUM_SOURCES_TO_RETURN})

    template = system_message + settings.RETRIEVER_TEMPLATE

    qa_prompt = PromptTemplate(input_variables=["history", "context", "question"],
                               template=template)

    qa_chain = RetrievalQA.from_chain_type(
        llm=ChatOpenAI(
            temperature=settings.DEFAULT_GPT_QA_HELPER_MODEL_TEMPERATURE,
            model_name=settings.DEFAULT_GPT_QA_HELPER_MODEL,
            streaming=True,
            callbacks=callbacks,
        ),
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True,
        chain_type_kwargs={"prompt": qa_prompt,
                           "memory": context_memory},
    )
    helper_module.log("Running QA chain...", 'info')
    response = qa_chain(human_input)
    my_answer, my_docs = response["result"], response["source_documents"]
    helper_module.log(f"Answer: {my_answer}", 'info')
    return my_answer, my_docs


def embed_conversations():
    """ Ingest past conversations as long-term memory into the vector DB."""

    helper_module.log(f"Loading conversations in batch: {settings.CONVERSATION_SAVE_FOLDER}", 'info')

    conversations = []
    for json_file in settings.CONVERSATION_SAVE_FOLDER.glob('*.json'):
        if not str(json_file).endswith(settings.SNAPSHOT_FILENAME):
            with open(json_file, 'r') as f:
                json_data = json.load(f)
                result_str = ""
                for entry in json_data:
                    result_str += f"{entry['role']}: {entry['content']}\n"
                conversations.append(Document(page_content=result_str, metadata = {"source": str(json_file)}))

    text_splitter = RecursiveCharacterTextSplitter(chunk_size=settings.DOCUMENT_SPLITTER_CHUNK_SIZE,
                                                   chunk_overlap=settings.DOCUMENT_SPLITTER_CHUNK_OVERLAP)

    texts = text_splitter.split_documents(conversations)
    my_vectordb = get_vectordb()
    my_vectordb.add_documents(documents=texts, embeddings=OpenAIEmbeddings())
    helper_module.log(f"{len(conversations)} of conversations found", 'info')
    helper_module.log(f"{len(texts)} chunks of text embedded", 'info')


def display_vectordb_info():
    persistent_client = chromadb.PersistentClient(path=settings.CHROMA_DB_FOLDER)
    collection = persistent_client.get_or_create_collection(settings.VECTORDB_COLLECTION)
    helper_module.log(f"VectorDB Folder: {settings.CONVERSATION_SAVE_FOLDER}", 'info')
    helper_module.log(f"Collection: {settings.VECTORDB_COLLECTION}", 'info')
    helper_module.log(f"Number of items in collection: {collection.count()}", 'info')


if __name__ == "__main__":
    load_dotenv()
    openai.api_key = os.getenv('OPENAI_API_KEY')

    while True:
        display_vectordb_info()
        user_input = input("\n(C)create DB, (E)mbed conversations, (D)elete collection, (Q)uery - type 'quit' or 'exit' to quit: ")
        if 'c' == user_input.lower().strip():
            create_vectordb(settings.DOCUMENT_FOLDER)
        elif 'e' == user_input.lower().strip():
            embed_conversations()
        elif 'd' == user_input.lower().strip():
            user_input = input("\nAre you sure? Type 'yes' if you are: ")
            if 'yes' == user_input.lower().strip():
                delete_vectordb()
        elif 'q' == user_input.lower().strip():
            while True:
                memory = ConversationBufferMemory(input_key="question",
                                                  memory_key="history")
                query = input("\nQuery: ")
                if 'quit' in query or 'exit' in query:
                    break
                helper_module.log(f"Querying model: {settings.DEFAULT_GPT_QA_HELPER_MODEL}", 'info')
                system_input = characters.CUSTOM_INSTRUCTIONS
                answer, docs = retrieval_qa_run(system_input, query, memory)

                # Print the result
                print("\n\n> Question:")
                print(query)
                print("\n> Answer:")
                print(answer)

                if settings.SHOW_SOURCES:
                    print("----------------------------------SOURCE DOCUMENTS---------------------------")
                    for document in docs:
                        print("\n> " + document.metadata["source"] + ":")
                        print(document.page_content)
                    print("----------------------------------SOURCE DOCUMENTS---------------------------")

        elif 'quit' in user_input or 'exit' in user_input:
            break
        else:
            print("Unknown choice.\n")



vectordb.py
neobundy/pippaGPT