Upload app.py

app.py

@@ -0,0 +1,156 @@
+import streamlit as st
+import torch
+import uuid
+import nest_asyncio
+import asyncio
+
+torch.classes.__path__ = []
+
+# Setup for HTTP API Calls
+if 'device_id' not in st.session_state:
+    st.session_state.device_id = str(uuid.uuid3())
+
+if "feedback_key" not in st.session_state:
+    st.session_state.feedback_key = 0
+
+
+def launch_bot():
+    def reset():
+        st.session_state.messages = [{"role": "assistant", "content": "How may I help you?", "avatar": '🤖'}]
+        st.session_state.ex_prompt = None
+        st.session_state.first_turn = True
+
+
+    def generate_response(question):
+        response = vq.submit_query(question, languages[st.session_state.language])
+        return response
+    
+    def generate_streaming_response(question):
+        response = vq.submit_query_streaming(question, languages[st.session_state.language])
+        return response
+    
+    def show_example_questions():        
+        if len(st.session_state.example_messages) > 0 and st.session_state.first_turn:            
+            selected_example = pills("Questions to Try:", st.session_state.example_messages, index=None)
+            if selected_example:
+                st.session_state.ex_prompt = selected_example
+                st.session_state.first_turn = False
+                return True
+        return False
+
+    if 'cfg' not in st.session_state:
+        corpus_keys = str(os.environ['corpus_keys']).split(',')
+        cfg = OmegaConf.create({
+            'corpus_keys': corpus_keys,
+            'api_key': str(os.environ['api_key']),
+            'title': os.environ['title'],
+            'source_data_desc': os.environ['source_data_desc'],
+            'streaming': isTrue(os.environ.get('streaming', False)),
+            'prompt_name': os.environ.get('prompt_name', None),
+            'examples': os.environ.get('examples', None),
+            'language': 'English'
+        })
+        st.session_state.cfg = cfg
+        st.session_state.ex_prompt = None
+        st.session_state.first_turn = True
+        st.session_state.language = cfg.language
+        example_messages = [example.strip() for example in cfg.examples.split(",")]
+        st.session_state.example_messages = [em for em in example_messages if len(em)>0][:max_examples]
+        
+        st.session_state.vq = VectaraQuery(cfg.api_key, cfg.corpus_keys, cfg.prompt_name)
+
+    cfg = st.session_state.cfg
+    vq = st.session_state.vq
+    st.set_page_config(page_title=cfg.title, layout="wide")
+
+    # left side content
+    with st.sidebar:
+        #image = Image.open('Vectara-logo.png')
+        #st.image(image, width=175)
+        st.markdown(f"## About\n\n"
+                    f"This demo uses outside RAG to ask questions about {cfg.source_data_desc}\n")
+        
+        cfg.language = st.selectbox('Language:', languages.keys())
+        if st.session_state.language != cfg.language:
+            st.session_state.language = cfg.language
+            reset()
+            st.rerun()
+
+        st.markdown("\n")
+        bc1, _ = st.columns([1, 1])
+        with bc1:
+            if st.button('Start Over'):
+                reset()
+                st.rerun()
+
+        st.markdown("---")
+        st.markdown(
+            "## Temporary test demo only\n"
+        )       
+
+    st.markdown(f"<center> <h2> Header Demo Test: {cfg.title} </h2> </center>", unsafe_allow_html=True)
+
+    if "messages" not in st.session_state.keys():
+        reset()
+                
+    # Display chat messages
+    for message in st.session_state.messages:
+        with st.chat_message(message["role"], avatar=message["avatar"]):
+            st.write(message["content"])
+
+    example_container = st.empty()
+    with example_container:
+        if show_example_questions():
+            example_container.empty()
+            st.rerun()
+
+    # select prompt from example question or user provided input
+    if st.session_state.ex_prompt:
+        prompt = st.session_state.ex_prompt
+    else:
+        prompt = st.chat_input()
+    if prompt:
+        st.session_state.messages.append({"role": "user", "content": prompt, "avatar": '🧑‍💻'})
+        with st.chat_message("user", avatar="🧑‍💻"):
+            st.write(prompt)
+        st.session_state.ex_prompt = None
+        
+    # Generate a new response if last message is not from assistant
+    if st.session_state.messages[-1]["role"] != "assistant":
+        with st.chat_message("assistant", avatar="🤖"):
+            if cfg.streaming:
+                stream = generate_streaming_response(prompt)
+                response = st.write_stream(stream)
+            else:
+                with st.spinner("Thinking..."):
+                    response = generate_response(prompt)
+                    st.write(response)
+
+            response = escape_dollars_outside_latex(response)
+            message = {"role": "assistant", "content": response, "avatar": '🤖'}
+            st.session_state.messages.append(message)
+
+            # Send query and response to Amplitude Analytics
+            send_amplitude_data(
+                user_query=st.session_state.messages[-2]["content"],
+                chat_response=st.session_state.messages[-1]["content"],
+                demo_name=cfg["title"],
+                language=st.session_state.language
+            )
+            st.rerun()
+
+    if (st.session_state.messages[-1]["role"] == "assistant") & (st.session_state.messages[-1]["content"] != "How may I help you?"):
+        streamlit_feedback(feedback_type="thumbs", on_submit = thumbs_feedback, key = st.session_state.feedback_key,
+                                      kwargs = {"user_query": st.session_state.messages[-2]["content"],
+                                                "chat_response": st.session_state.messages[-1]["content"],
+                                                "demo_name": cfg["title"],
+                                                "response_language": st.session_state.language})
+    
+
+    
+     
+ if __name__ == "__main__":
+   st.set_page_config(page_title="Sticky toolbar test", layout="wide")
+   nest_asyncio.apply()
+   asyncio.run(launch_bot())
+