Cortex_LLM/Streamlit_Cortex_LLM.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Simple Streamlit App for Cortex LLM\n",
    "\n",
    "Simple app showing off ability to do completion, sentiment, summarization, question/ansswe, and translation with Cortex LLM."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import python packages\n",
    "import streamlit as st\n",
    "from streamlit import cache_resource, cache_data\n",
    "from snowflake.snowpark.session import Session\n",
    "from snowflake.snowpark.context import get_active_session\n",
    "import re\n",
    "import json\n",
    "\n",
    "session = get_active_session()\n",
    "\n",
    "@cache_data(show_spinner='CALLING LLM')\n",
    "def complete(model, prompt):\n",
    "    global session\n",
    "    prompt = re.sub(\"'\", \"''\", prompt)\n",
    "    sql_code = f\"SELECT SNOWFLAKE.CORTEX.COMPLETE('{model}', '{prompt}') as answer\"\n",
    "    df = session.sql(sql_code).to_pandas()\n",
    "    return df['ANSWER'][0], sql_code\n",
    "\n",
    "@cache_data(show_spinner='CALLING LLM')\n",
    "def get_sentiment(text):\n",
    "    global session\n",
    "    text = re.sub(\"'\", \"''\", text)\n",
    "    sql_code = f\"SELECT SNOWFLAKE.CORTEX.SENTIMENT('{text}') as answer\"\n",
    "    df = session.sql(sql_code).to_pandas()\n",
    "    return df['ANSWER'][0], sql_code\n",
    "\n",
    "@cache_data(show_spinner='CALLING LLM')\n",
    "def summarize(text):\n",
    "    global session\n",
    "    text = re.sub(\"'\", \"''\", text)\n",
    "    sql_code = f\"SELECT SNOWFLAKE.CORTEX.SUMMARIZE('{text}') as answer\"\n",
    "    df = session.sql(sql_code).to_pandas()\n",
    "    return df['ANSWER'][0], sql_code\n",
    "\n",
    "\n",
    "@cache_data(show_spinner='CALLING LLM')\n",
    "def extract_answer(from_text, question):\n",
    "    global session\n",
    "    from_text = re.sub(\"'\", \"''\", from_text)\n",
    "    question =  re.sub(\"'\", \"''\", question)\n",
    "    sql_code = f\"SELECT SNOWFLAKE.CORTEX.EXTRACT_ANSWER('{from_text}','{question}') as answer\"\n",
    "    #st.write (sql_code)\n",
    "    \n",
    "    df = session.sql(sql_code).to_pandas()\n",
    "    return df['ANSWER'][0], sql_code\n",
    "\n",
    "\n",
    "@cache_data(show_spinner='CALLING LLM')\n",
    "def translate(from_lang, to_lang, text):\n",
    "    global session\n",
    "    text = re.sub(\"'\", \"''\", text)\n",
    "    sql_code = f\"SELECT SNOWFLAKE.CORTEX.TRANSLATE('{text}', '{from_lang}', '{to_lang}') as answer\"\n",
    "    df = session.sql(sql_code).to_pandas()\n",
    "    return df['ANSWER'][0], sql_code\n",
    "\n",
    "\n",
    "tab1, tab2,tab3,tab4, tab5 = st.tabs(['Completion', 'Sentiment','Summarize', 'Translate', 'Extract Answer'])\n",
    "\n",
    "\n",
    "with tab1:\n",
    "\n",
    "    st.header(\"Completion\")\n",
    "    st.subheader(\"Ask the LLM anything\")\n",
    "    completion_input = st.text_area(\"Prompt\",height=400)\n",
    "    completion_model = st.selectbox(\"Choose Model\", options = ['llama2-70b-chat','mistral-7b'])\n",
    "    if st.button(\"Complete\"):\n",
    "        if completion_input is not None and completion_model is not None:\n",
    "            result, sql_code = complete(completion_model,completion_input)\n",
    "            st.success(result)\n",
    "            with st.expander(\"See SQL\"):\n",
    "                st.code(sql_code,language='SQL')\n",
    "                \n",
    "           \n",
    "with tab2:\n",
    "\n",
    "    st.header(\"Sentiment Analysis\")\n",
    "    st.subheader(\"Get the Sentiment of a Text\")\n",
    "    sentiment_input = st.text_area(\"Text for Sentiment Analysis\",height=400)\n",
    "    \n",
    "    if st.button(\"Get Sentiment\"):\n",
    "        if sentiment_input is not None: \n",
    "            result, sql_code = get_sentiment(sentiment_input)\n",
    "            if result > 0:\n",
    "                st.success(f\" Positive. The Sentiment Value is {result}\")\n",
    "            else:\n",
    "                \n",
    "                st.warning(f\" Negative. The Sentiment Value is {result}\")\n",
    "            with st.expander(\"See SQL\"):\n",
    "                st.code(sql_code,language='SQL')\n",
    "\n",
    "\n",
    "with tab3:\n",
    "\n",
    "    st.header(\"Summarize\")\n",
    "    st.subheader(\"Summarize a Text\")\n",
    "    sentiment_input = st.text_area(\"Text for Summarization\", height=400)\n",
    "    \n",
    "    if st.button(\"Summarize\"):\n",
    "        if sentiment_input is not None: \n",
    "            result, sql_code = summarize(sentiment_input)\n",
    "            st.success(result)\n",
    "            with st.expander(\"See SQL\"):\n",
    "                st.code(sql_code,language='SQL')\n",
    "\n",
    "with tab4:\n",
    "\n",
    "    st.header(\"Translate\")\n",
    "    st.subheader(\"Translate a Text\")\n",
    "    language_options = {\"German\":'de', \"French\":'fr', \"Turkish\":'tr', \"Korean\":'ko',\n",
    "                \"Potrtuguese\":'pt', \"English\":'en' , \"Italian\": 'it', \"Japanese\":'ja',\n",
    "                \"Spanish\":'es',\"Russian\": 'ru',\"Swedish\": 'sv', \"Polish\": 'pl' }\n",
    "    language_names = list(language_options.keys())\n",
    "    from_language = st.selectbox(\"From\", options=language_names )\n",
    "    if from_language is not None:\n",
    "        to_language_names =  language_names.copy()\n",
    "        \n",
    "        to_language_names.remove(from_language)\n",
    "        to_language =   st.selectbox(\"To\", options =  to_language_names)\n",
    "        translate_input = st.text_area(\"Text for Translation\", height=400)\n",
    "        from_language_code = language_options[from_language]\n",
    "        \n",
    "    if to_language is not None:\n",
    "        to_language_code = language_options[to_language]\n",
    "    \n",
    "    if st.button(\"Translate\"):\n",
    "        if to_language_code is not None: \n",
    "            result, sql_code = translate(from_language_code,to_language_code,translate_input)\n",
    "            st.success(result)\n",
    "            with st.expander(\"See SQL\"):\n",
    "                st.code(sql_code,language='SQL')\n",
    "\n",
    "with tab5:\n",
    "    st.header(\"Extract Answer\")\n",
    "    st.subheader(\"Extract answer from a Piece of Text\")\n",
    "    input_text = st.text_area(\"Source Text\", height=400)\n",
    "    question = st.text_input(\"Question\")\n",
    "    \n",
    "    if st.button(\"Get Answer\"):\n",
    "        if input_text is not None and question is not None: \n",
    "            result, sql_code = extract_answer(input_text,question)\n",
    "            st.success(json.loads(result)[0]['answer'])\n",
    "            with st.expander(\"See SQL\"):\n",
    "                st.code(sql_code,language='SQL')\n"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}