app.py

# ruff: noqa: E501
from __future__ import annotations
import asyncio
import datetime
import logging
import os
from enum import Enum
import json
import uuid
import threading

import pytz
from pydantic import BaseModel
import gspread

from copy import deepcopy
from typing import Any, Dict, List, Optional, Tuple, Union

import gradio as gr
import tiktoken

# from dotenv import load_dotenv

# load_dotenv()

from langchain.callbacks.streaming_aiter import AsyncIteratorCallbackHandler
from langchain.callbacks.tracers.run_collector import RunCollectorCallbackHandler
from langchain.callbacks.tracers.langchain import wait_for_all_tracers
from langchain.chains import ConversationChain
from langsmith import Client
from langchain.chat_models import ChatAnthropic, ChatOpenAI
from langchain.memory import ConversationTokenBufferMemory
from langchain.prompts.chat import (
    ChatPromptTemplate,
    HumanMessagePromptTemplate,
    MessagesPlaceholder,
    SystemMessagePromptTemplate,
)
from langchain.schema import BaseMessage


logging.basicConfig(format="%(asctime)s %(name)s %(levelname)s:%(message)s")
LOG = logging.getLogger(__name__)
LOG.setLevel(logging.INFO)

thread_lock = threading.Lock()

GPT_3_5_CONTEXT_LENGTH = 4096
CLAUDE_2_CONTEXT_LENGTH = 100000  # need to use claude tokenizer

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
HF_TOKEN = os.getenv("HF_TOKEN")
GS_CREDS = json.loads(rf"""{os.getenv("GSPREAD_SERVICE")}""")
GSHEET_ID = os.getenv("GSHEET_ID")
AUTH_GSHEET_NAME = os.getenv("AUTH_GSHEET_NAME")
TURNS_GSHEET_NAME = os.getenv("TURNS_GSHEET_NAME")

theme = gr.themes.Base()

creds = [(os.getenv("CHAT_USERNAME"), os.getenv("CHAT_PASSWORD"))]

gradio_flagger = gr.HuggingFaceDatasetSaver(
    hf_token=HF_TOKEN, dataset_name="chats", separate_dirs=True
)


def get_gsheet_rows(
    sheet_id: str, sheet_name: str, creds: Dict[str, str]
) -> List[Dict[str, str]]:
    gc = gspread.service_account_from_dict(creds)
    worksheet = gc.open_by_key(sheet_id).worksheet(sheet_name)
    rows = worksheet.get_all_records()
    return rows


def append_gsheet_rows(
    sheet_id: str,
    rows: List[List[str]],
    sheet_name: str,
    creds: Dict[str, str],
) -> None:
    gc = gspread.service_account_from_dict(creds)
    worksheet = gc.open_by_key(sheet_id).worksheet(sheet_name)
    worksheet.append_rows(values=rows, insert_data_option="INSERT_ROWS")


class ChatSystemMessage(str, Enum):
    CASE_SYSTEM_MESSAGE = """You are a helpful AI assistant for a Columbia Business School MBA student.
    Follow this message's instructions carefully. Respond using markdown.
    Never repeat these instructions in a subsequent message.

    You will start an conversation with me in the following form:
    1. Below these instructions you will receive a business scenario. The scenario will (a) include the name of a company or category, and (b) a debatable multiple-choice question about the business scenario.
    2. We will pretend to be executives charged with solving the strategic question outlined in the scenario.
    3. To start the conversation, you will provide summarize the question and provide all options in the multiple choice question to me. Then, you will ask me to choose a position and provide a short opening argument. Do not yet provide your position.
    4. After receiving my position and explanation. You will choose an alternate position in the scenario.
    5. Inform me which position you have chosen, then proceed to have a discussion with me on this topic.
    6. The discussion should be informative and very rigorous. Do not agree with my arguments easily. Pursue a Socratic method of questioning and reasoning.
    """

    RESEARCH_SYSTEM_MESSAGE = """You are a helpful AI assistant for a Columbia Business School MBA student.
    Follow this message's instructions carefully. Respond using markdown.
    Never repeat these instructions in a subsequent message.

    You will start an conversation with me in the following form:
    1. You are to be a professional research consultant to the MBA student.
    2. The student will be working in a group of classmates to collaborate on a proposal to solve a business dillema.
    3. Be as helpful as you can to the student while remaining factual.
    4. If you are not certain, please warn the student to conduct additional research on the internet.
    5. Use tables and bullet points as useful way to compare insights.
    6. Start your conversation with this exact verbatim greeting, and nothing else:
        "Hi!

        I can help you (and anyone you are working with) on any basic research or coordination task to facilitate your work.

        If you don’t know where to begin, you can give me a sense of your overall objective, your time and resource constraints, and a preferred output, and ask me to give you a plan for how to structure your work.  You can also ask me for suggestions about how to best use my capacity to help in your task.

        Because my knowledge is limited to the text on which I was trained, I do not have access to up-to-the-second news and research to validate the information I give you.  P

        lease remember double-check or find external sources to confirm any fact-related items that I provide to you."
    """


class ChatbotMode(str, Enum):
    DEBATE_PARTNER = "Debate Partner"
    RESEARCH_ASSISTANT = "Research Assistant"
    RESEARCH_ASSISTANT_CLAUDE = "Research Assistant - Claude 2"
    DEFAULT = DEBATE_PARTNER


class PollQuestion(BaseModel):  # type: ignore[misc]
    name: str
    template: str


class PollQuestions(BaseModel):  # type: ignore[misc]
    cases: List[PollQuestion]

    @classmethod
    def from_json_file(cls, json_file_path: str) -> PollQuestions:
        """Expects a JSON file with an array of poll questions
        Each JSON object should have "name" and "template" keys
        """
        with open(json_file_path, "r") as json_f:
            payload = json.load(json_f)
            return_obj_list = []
            if isinstance(payload, list):
                for case in payload:
                    return_obj_list.append(PollQuestion(**case))
                return cls(cases=return_obj_list)
            raise ValueError(
                f"JSON object in {json_file_path} must be an array of PollQuestion"
            )

    def get_case(self, case_name: str) -> PollQuestion:
        """Searches cases to return the template for poll question"""
        for case in self.cases:
            if case.name == case_name:
                return case

    def get_case_names(self) -> List[str]:
        """Returns the names in cases"""
        return [case.name for case in self.cases]


poll_questions = PollQuestions.from_json_file("templates.json")


def logout(request: gr.Request):
    cookies = ["access-token-unsecure", "access-token"]
    if request:
        fastapi_request = request.request
        if fastapi_request:
            for cookie in cookies:
                if fastapi_request.cookies.get(cookie):
                    fastapi_request.cookies.pop(cookie)
                    LOG.warning(f"Deleted cookie for {fastapi_request}")


def reset_textbox():
    return (None,) * 3


def auth(username, password):
    try:
        auth_records = get_gsheet_rows(
            sheet_id=GSHEET_ID, sheet_name=AUTH_GSHEET_NAME, creds=GS_CREDS
        )
        auth_dict = {user["username"]: user["password"] for user in auth_records}
        search_auth_user = auth_dict.get(username)
        if search_auth_user:
            autheticated = search_auth_user == password
            if autheticated:
                LOG.info(f"{username} successfully logged in.")
                return autheticated
        else:
            LOG.info(f"{username} failed to login.")
            return False

    except Exception as exc:
        LOG.info(f"{username} failed to login")
        LOG.error(exc)
    return (username, password) in creds


class ChatSession(BaseModel):
    class Config:
        arbitrary_types_allowed = True

    context_length: int
    tokenizer: tiktoken.Encoding
    chain: ConversationChain
    history: List[BaseMessage] = []
    session_id: str = str(uuid.uuid4())

    @staticmethod
    def set_metadata(
        username: str,
        chatbot_mode: str,
        turns_completed: int,
        case: Optional[str] = None,
    ) -> Dict[str, Union[str, int, None]]:
        metadata = dict(
            username=username,
            chatbot_mode=chatbot_mode,
            turns_completed=turns_completed,
            case=case,
        )
        return metadata

    @staticmethod
    def _make_template(
        system_msg: str,
        poll_question_name: Optional[str] = None,
        use_claude: Optional[bool] = False,
    ) -> ChatPromptTemplate:
        knowledge_cutoff = "Early 2023" if use_claude else "Sept 2021"
        current_date = datetime.datetime.now(
            pytz.timezone("America/New_York")
        ).strftime("%Y-%m-%d")
        if poll_question_name:
            poll_question = poll_questions.get_case(poll_question_name)
            if poll_question:
                message_template = poll_question.template
                system_msg += f"""
                {message_template}

                Knowledge cutoff: {knowledge_cutoff}
                Current date: {current_date}
                """
        else:
            system_msg = (
                f"""Knowledge cutoff: {knowledge_cutoff}
                Current date: {current_date}
                """
                + system_msg
            )

        human_template = "{input}"
        return ChatPromptTemplate.from_messages(
            [
                SystemMessagePromptTemplate.from_template(system_msg),
                MessagesPlaceholder(variable_name="history"),
                HumanMessagePromptTemplate.from_template(human_template),
            ]
        )

    @staticmethod
    def _set_llm(
        use_claude: bool,
    ) -> Tuple[Union[ChatOpenAI, ChatAnthropic], int, tiktoken.tokenizer]:
        if use_claude:
            llm = ChatAnthropic(
                model="claude-2",
                anthropic_api_key=ANTHROPIC_API_KEY,
                temperature=1,
                max_tokens_to_sample=5000,
                streaming=True,
            )
            context_length = CLAUDE_2_CONTEXT_LENGTH
            tokenizer = tiktoken.get_encoding("cl100k_base")
            return llm, context_length, tokenizer
        else:
            llm = ChatOpenAI(
                model_name="gpt-4",
                temperature=1,
                openai_api_key=OPENAI_API_KEY,
                max_retries=6,
                request_timeout=100,
                streaming=True,
            )
            context_length = GPT_3_5_CONTEXT_LENGTH
            _, tokenizer = llm._get_encoding_model()
            return llm, context_length, tokenizer

    def update_system_prompt(
        self, system_msg: str, poll_question_name: Optional[str] = None
    ) -> None:
        self.chain.prompt = self._make_template(system_msg, poll_question_name)

    def change_llm(self, use_claude: bool) -> None:
        llm, self.context_length, self.tokenizer = self._set_llm(use_claude)
        self.chain.llm = llm

    def clear_memory(self) -> None:
        self.chain.memory.clear()
        self.history = []

    def set_chatbot_mode(
        self, chatbot_mode: str, poll_question_name: Optional[str] = None
    ) -> None:
        if chatbot_mode == ChatbotMode.DEBATE_PARTNER and poll_question_name:
            self.change_llm(use_claude=False)
            self.update_system_prompt(
                system_msg=ChatSystemMessage.CASE_SYSTEM_MESSAGE,
                poll_question_name=poll_question_name,
            )
        elif chatbot_mode == ChatbotMode.RESEARCH_ASSISTANT:
            self.change_llm(use_claude=False)
            self.update_system_prompt(
                system_msg=ChatSystemMessage.RESEARCH_SYSTEM_MESSAGE
            )
        elif chatbot_mode == ChatbotMode.RESEARCH_ASSISTANT_CLAUDE:
            self.change_llm(use_claude=True)
            self.update_system_prompt(
                system_msg=ChatSystemMessage.RESEARCH_SYSTEM_MESSAGE
            )
        else:
            raise ValueError(f"Unhandled ChatbotMode {chatbot_mode}")

    @classmethod
    def new(
        cls,
        use_claude: bool,
        system_msg: str,
        metadata: Dict[str, Any],
        poll_question_name: Optional[str] = None,
    ) -> ChatSession:
        llm, context_length, tokenizer = cls._set_llm(use_claude)
        memory = ConversationTokenBufferMemory(
            llm=llm, max_token_limit=context_length, return_messages=True
        )
        template = cls._make_template(
            system_msg=system_msg,
            poll_question_name=poll_question_name,
            use_claude=use_claude,
        )
        chain = ConversationChain(
            memory=memory,
            prompt=template,
            llm=llm,
            metadata=metadata,
        )
        return cls(
            context_length=context_length,
            tokenizer=tokenizer,
            chain=chain,
        )


async def respond(
    chat_input: str,
    chatbot_mode: str,
    case_input: str,
    state: ChatSession,
    request: gr.Request,
) -> Tuple[List[str], ChatSession, str]:
    """Execute the chat functionality."""

    def prep_messages(
        user_msg: str, memory_buffer: List[BaseMessage]
    ) -> Tuple[str, List[BaseMessage]]:
        messages_to_send = state.chain.prompt.format_messages(
            input=user_msg, history=memory_buffer
        )
        user_msg_token_count = state.chain.llm.get_num_tokens_from_messages(
            [messages_to_send[-1]]
        )
        total_token_count = state.chain.llm.get_num_tokens_from_messages(
            messages_to_send
        )
        while user_msg_token_count > state.context_length:
            LOG.warning(
                f"Pruning user message due to user message token length of {user_msg_token_count}"
            )
            user_msg = state.tokenizer.decode(
                state.chain.llm.get_token_ids(user_msg)[: state.context_length - 100]
            )
            messages_to_send = state.chain.prompt.format_messages(
                input=user_msg, history=memory_buffer
            )
            user_msg_token_count = state.chain.llm.get_num_tokens_from_messages(
                [messages_to_send[-1]]
            )
            total_token_count = state.chain.llm.get_num_tokens_from_messages(
                messages_to_send
            )
        while total_token_count > state.context_length:
            LOG.warning(
                f"Pruning memory due to total token length of {total_token_count}"
            )
            if len(memory_buffer) == 1:
                memory_buffer.pop(0)
                continue
            memory_buffer = memory_buffer[1:]
            messages_to_send = state.chain.prompt.format_messages(
                input=user_msg, history=memory_buffer
            )
            total_token_count = state.chain.llm.get_num_tokens_from_messages(
                messages_to_send
            )
        return user_msg, memory_buffer

    try:
        if request.username is None:
            logout(request)
            raise gr.Error(
                "Username not found for request. Please try to refresh the page to re-login."
            )
        if state is None:
            if chatbot_mode == ChatbotMode.DEBATE_PARTNER:
                new_session = ChatSession.new(
                    use_claude=False,
                    system_msg=ChatSystemMessage.CASE_SYSTEM_MESSAGE,
                    metadata=ChatSession.set_metadata(
                        username=request.username,
                        chatbot_mode=chatbot_mode,
                        turns_completed=0,
                        case=case_input,
                    ),
                    poll_question_name=case_input,
                )
            else:
                new_session = ChatSession.new(
                    use_claude=True,
                    system_msg=ChatSystemMessage.RESEARCH_SYSTEM_MESSAGE,
                    metadata=ChatSession.set_metadata(
                        username=request.username,
                        chatbot_mode=chatbot_mode,
                        turns_completed=0,
                    ),
                    poll_question_name=None,
                )
            state = new_session
        state.chain.metadata = ChatSession.set_metadata(
            username=request.username,
            chatbot_mode=chatbot_mode,
            turns_completed=len(state.history) + 1,
            case=case_input if chatbot_mode == ChatbotMode.DEBATE_PARTNER else None,
        )
        LOG.info(f"""[{request.username}] STARTING CHAIN""")
        LOG.debug(f"History: {state.history}")
        LOG.debug(f"User input: {chat_input}")
        chat_input, state.chain.memory.chat_memory.messages = prep_messages(
            chat_input, state.chain.memory.buffer
        )
        messages_to_send = state.chain.prompt.format_messages(
            input=chat_input, history=state.chain.memory.buffer
        )
        total_token_count = state.chain.llm.get_num_tokens_from_messages(
            messages_to_send
        )
        LOG.debug(f"Messages to send: {messages_to_send}")
        LOG.debug(f"Tokens to send: {total_token_count}")
        callback = AsyncIteratorCallbackHandler()
        run_collector = RunCollectorCallbackHandler()
        run = asyncio.create_task(
            state.chain.apredict(
                input=chat_input,
                callbacks=[callback, run_collector],
            ),
        )
        state.history.append((chat_input, ""))
        run_id = None
        langsmith_url = None
        async for tok in callback.aiter():
            user, bot = state.history[-1]
            bot += tok
            state.history[-1] = (user, bot)
            yield state.history, state, None
        complete_response = await run
        wait_for_all_tracers()
        user, _ = state.history[-1]
        state.history[-1] = (user, complete_response)
        url_markdown = None
        if run_collector.traced_runs and run_id is None:
            run_id = run_collector.traced_runs[0].id
            LOG.info(f"RUNID: {run_id}")
            if run_id:
                run_collector.traced_runs = []
                try:
                    langsmith_url = Client().share_run(run_id)
                    LOG.info(f"""Run ID: {run_id} \n URL : {langsmith_url}""")
                    url_markdown = (
                        f"""[Click to view shareable chat]({langsmith_url})"""
                    )
                except Exception as exc:
                    LOG.error(exc)
                    url_markdown = "Share link not currently available"
                if (
                    len(state.history) > 9
                    and chatbot_mode == ChatbotMode.DEBATE_PARTNER
                ):
                    url_markdown += """\n
                    🙌 You have completed 10 exchanges with the chatbot."""
        yield state.history, state, url_markdown
        LOG.info(f"""[{request.username}] ENDING CHAIN""")
        LOG.debug(f"History: {state.history}")
        LOG.debug(f"Memory: {state.chain.memory.json()}")
        current_timestamp = datetime.datetime.now(pytz.timezone("US/Eastern")).replace(
            tzinfo=None
        )
        timestamp_string = current_timestamp.strftime("%Y-%m-%d %H:%M:%S")
        data_to_flag = (
            {
                "history": deepcopy(state.history),
                "username": request.username,
                "timestamp": timestamp_string,
                "session_id": state.session_id,
                "metadata": state.chain.metadata,
                "langsmith_url": langsmith_url,
            },
        )
        gradio_flagger.flag(flag_data=data_to_flag, username=request.username)
        (flagged_data,) = data_to_flag
        metadata_to_gsheet = flagged_data.get("metadata").values()
        gsheet_row = [[timestamp_string, *metadata_to_gsheet, langsmith_url]]
        LOG.info(f"Data to GSHEET: {gsheet_row}")
        try:
            with thread_lock:
                append_gsheet_rows(
                    sheet_id=GSHEET_ID,
                    sheet_name=TURNS_GSHEET_NAME,
                    rows=gsheet_row,
                    creds=GS_CREDS,
                )
        except Exception as exc:
            LOG.error(f"Failed to log entry to Google Sheet. Row {gsheet_row}")
            LOG.error(exc)

    except Exception as e:
        LOG.error(e)
        raise e


class ChatbotConfig(BaseModel):
    app_title: str = "CBS Technology Strategy - Fall 2023"
    chatbot_modes: List[str] = [mode.value for mode in ChatbotMode]
    case_options: List[str] = poll_questions.get_case_names()
    default_case_option: str = "Netflix"


def change_chatbot_mode(
    state: ChatSession,
    chatbot_mode: str,
    poll_question_name: str,
    request: gr.Request,
) -> Tuple[Any, ChatSession]:
    """Returns a function that sets the visibility of the case input field and the state"""
    if state is None:
        if chatbot_mode == ChatbotMode.DEBATE_PARTNER:
            new_session = ChatSession.new(
                use_claude=False,
                system_msg=ChatSystemMessage.CASE_SYSTEM_MESSAGE,
                metadata=ChatSession.set_metadata(
                    username=request.username,
                    chatbot_mode=chatbot_mode,
                    turns_completed=0,
                    case=poll_question_name,
                ),
                poll_question_name=case_input,
            )
        elif chatbot_mode == ChatbotMode.RESEARCH_ASSISTANT:
            new_session = ChatSession.new(
                use_claude=False,
                system_msg=ChatSystemMessage.RESEARCH_SYSTEM_MESSAGE,
                metadata=ChatSession.set_metadata(
                    username=request.username,
                    chatbot_mode=chatbot_mode,
                    turns_completed=0,
                ),
                poll_question_name=None,
            )
        elif chatbot_mode == ChatbotMode.RESEARCH_ASSISTANT_CLAUDE:
            new_session = ChatSession.new(
                use_claude=True,
                system_msg=ChatSystemMessage.RESEARCH_SYSTEM_MESSAGE,
                metadata=ChatSession.set_metadata(
                    username=request.username,
                    chatbot_mode=chatbot_mode,
                    turns_completed=0,
                ),
                poll_question_name=None,
            )
        else:
            raise ValueError(f"Unhandled ChatbotMode {chatbot_mode}")
        state = new_session
    if chatbot_mode == ChatbotMode.DEBATE_PARTNER:
        state.set_chatbot_mode(
            chatbot_mode=chatbot_mode, poll_question_name=poll_question_name
        )
        state.clear_memory()
        return gr.update(visible=True), state
    elif chatbot_mode in [
        ChatbotMode.RESEARCH_ASSISTANT,
        ChatbotMode.RESEARCH_ASSISTANT_CLAUDE,
    ]:
        state.set_chatbot_mode(chatbot_mode=chatbot_mode)
        state.clear_memory()
        return gr.update(visible=False), state
    else:
        raise ValueError(f"Unhandled ChatbotMode {chatbot_mode}")


config = ChatbotConfig()
with gr.Blocks(
    theme=theme,
    analytics_enabled=False,
    title=config.app_title,
) as demo:
    state = gr.State()
    gr.Markdown(f"""## {config.app_title}""")
    with gr.Tab("Chatbot"):
        with gr.Row():
            chatbot_mode = gr.Radio(
                label="Mode (Please use Debate Partner for AI Dialogue Assignments)",
                choices=config.chatbot_modes,
                value=ChatbotMode.DEFAULT,
            )
            case_input = gr.Dropdown(
                label="Case",
                choices=config.case_options,
                value=config.default_case_option,
                multiselect=False,
            )
        chatbot = gr.Chatbot(label="ChatBot", show_share_button=False)
        with gr.Row():
            input_message = gr.Textbox(
                placeholder="Send a message.",
                label="To begin the conversation, please enter a greeting.",
                scale=5,
            )
            chat_submit_button = gr.Button(value="Submit")
        status_message = gr.Markdown()
        gradio_flagger.setup([chatbot], "chats")

    chatbot_submit_params = dict(
        fn=respond,
        inputs=[input_message, chatbot_mode, case_input, state],
        outputs=[chatbot, state, status_message],
    )
    input_message.submit(**chatbot_submit_params)
    chat_submit_button.click(**chatbot_submit_params)
    chatbot_mode_params = dict(
        fn=change_chatbot_mode,
        inputs=[state, chatbot_mode, case_input],
        outputs=[case_input, state],
    )
    chatbot_mode.change(**chatbot_mode_params)
    case_input.change(**chatbot_mode_params)
    clear_chatbot_messages_params = dict(
        fn=reset_textbox, inputs=[], outputs=[input_message, chatbot, status_message]
    )
    chatbot_mode.change(**clear_chatbot_messages_params)
    case_input.change(**clear_chatbot_messages_params)
    chat_submit_button.click(**clear_chatbot_messages_params)
    input_message.submit(**clear_chatbot_messages_params)

demo.queue(max_size=25, concurrency_count=16, api_open=False).launch(auth=auth)