Add utils.py

app.py

@@ -1,15 +1,15 @@
 from omegaconf import OmegaConf
 from query import VectaraQuery
 import os
-import requests
-import json
+from PIL import Image
 import uuid
 
 import streamlit as st
 from streamlit_pills import pills
 from streamlit_feedback import streamlit_feedback
 
-from PIL import Image
+from utils import thumbs_feedback, send_amplitude_data
+
 
 max_examples = 6
 languages = {'English': 'eng', 'Spanish': 'spa', 'French': 'frs', 'Chinese': 'zho', 'German': 'deu', 'Hindi': 'hin', 'Arabic': 'ara',
@@ -21,36 +21,6 @@ languages = {'English': 'eng', 'Spanish': 'spa', 'French': 'frs', 'Chinese': 'zh
 if 'device_id' not in st.session_state:
     st.session_state.device_id = str(uuid.uuid4())
 
-headers = {
-    'Content-Type': 'application/json',
-    'Accept': '*/*'
-}
-amp_api_key = os.getenv('AMPLITUDE_TOKEN')
-
-def thumbs_feedback(feedback, **kwargs):
-    """
-    Sends feedback to Amplitude Analytics
-    """
-    data = {
-            "api_key": amp_api_key,
-            "events": [{
-                "device_id": st.session_state.device_id,
-                "event_type": "provided_feedback",
-                "event_properties": {
-                    "Space Name": kwargs.get("title", "Unknown Space Name"),
-                    "Demo Type": "Chat bot",
-                    "query": kwargs.get("prompt", "No user input"),
-                    "response": kwargs.get("response", "No chat response"),
-                    "feedback": feedback["score"],
-                    "Response Language": st.session_state.language
-                }
-            }]
-        }
-    response = requests.post('https://api2.amplitude.com/2/httpapi', headers=headers, data=json.dumps(data))
-    if response.status_code != 200:
-        print(f"Request failed with status code {response.status_code}. Response Text: {response.text}")
-    
-    st.session_state.feedback_key += 1
 
 if "feedback_key" not in st.session_state:
         st.session_state.feedback_key = 0
@@ -157,7 +127,7 @@ def launch_bot():
     if st.session_state.messages[-1]["role"] != "assistant":
         with st.chat_message("assistant"):
             if cfg.streaming:
-                stream = generate_streaming_response(prompt) 
+                stream = generate_streaming_response(prompt)
                 response = st.write_stream(stream) 
             else:
                 with st.spinner("Thinking..."):
@@ -167,178 +137,20 @@ def launch_bot():
             st.session_state.messages.append(message)
 
             # Send query and response to Amplitude Analytics
-            data = {
-                "api_key": amp_api_key,
-                "events": [{
-                    "device_id": st.session_state.device_id,
-                    "event_type": "submitted_query",
-                    "event_properties": {
-                        "Space Name": cfg["title"],
-                        "Demo Type": "Chat bot",
-                        "query": st.session_state.messages[-2]["content"],
-                        "response": st.session_state.messages[-1]["content"],
-                        "Response Language": st.session_state.language
-                    }
-                }]
-            }
-            response = requests.post('https://api2.amplitude.com/2/httpapi', headers=headers, data=json.dumps(data))
-            if response.status_code != 200:
-                print(f"Request failed with status code {response.status_code}. Response Text: {response.text}")
+            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 = {"prompt": st.session_state.messages[-2]["content"],
-                                                "response": st.session_state.messages[-1]["content"],
-                                                "title": cfg["title"]})
+                                      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__":
-    launch_bot()
-
-
-
-# from omegaconf import OmegaConf
-# from query import VectaraQuery
-# import os
-
-# import streamlit as st
-# from streamlit_pills import pills
-# from streamlit_feedback import streamlit_feedback
-
-# from PIL import Image
-
-# max_examples = 4
-# languages = {'English': 'eng', 'Spanish': 'spa', 'French': 'frs', 'Chinese': 'zho', 'German': 'deu', 'Hindi': 'hin', 'Arabic': 'ara',
-#              'Portuguese': 'por', 'Italian': 'ita', 'Japanese': 'jpn', 'Korean': 'kor', 'Russian': 'rus', 'Turkish': 'tur', 'Persian (Farsi)': 'fas',
-#              'Vietnamese': 'vie', 'Thai': 'tha', 'Hebrew': 'heb', 'Dutch': 'nld', 'Indonesian': 'ind', 'Polish': 'pol', 'Ukrainian': 'ukr',
-#              'Romanian': 'ron', 'Swedish': 'swe', 'Czech': 'ces', 'Greek': 'ell', 'Bengali': 'ben', 'Malay (or Malaysian)': 'msa', 'Urdu': 'urd'}
-
-# def isTrue(x) -> bool:
-#     if isinstance(x, bool):
-#         return x
-#     return x.strip().lower() == 'true'
-
-# def thumbs_feedback(feedback, **kwargs):
-#     print(f'Debug: Feedback Received {feedback["score"]} FROM user question {kwargs.get("prompt", "No user input")} AND chat response {kwargs.get("response", "No chat response")}. Detected response language {kwargs.get("language", "unknown")}')
-#     st.session_state.feedback_key += 1
-
-# if "feedback_key" not in st.session_state:
-#         st.session_state.feedback_key = 0
-
-# def launch_bot():
-#     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("Queries 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 Retrieval Augmented Generation 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
-#             print(f"DEBUG: Language changed to {st.session_state.language}")
-#             st.rerun()
-
-#         st.markdown("---")
-#         st.markdown(
-#             "## How this works?\n"
-#             "This app was built with [Vectara](https://vectara.com).\n"
-#             "Vectara's [Indexing API](https://docs.vectara.com/docs/api-reference/indexing-apis/indexing) was used to ingest the data into a Vectara corpus (or index).\n\n"
-#             "This app uses Vectara [Chat API](https://docs.vectara.com/docs/console-ui/vectara-chat-overview) to query the corpus and present the results to you, answering your question.\n\n"
-#         )
-#         st.markdown("---")
-        
-
-#     st.markdown(f"<center> <h2> Vectara AI Assistant: {cfg.title} </h2> </center>", unsafe_allow_html=True)
-
-#     if "messages" not in st.session_state.keys():
-#         st.session_state.messages = [{"role": "assistant", "content": "How may I help you?"}]
-
-    
-                
-#     # Display chat messages
-#     for message in st.session_state.messages:
-#         with st.chat_message(message["role"]):
-#             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})
-#         with st.chat_message("user"):
-#             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"):
-#             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)
-#             message = {"role": "assistant", "content": response}
-#             st.session_state.messages.append(message)
-#             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 = {"prompt": st.session_state.messages[-2]["content"],
-#                                                 "response": st.session_state.messages[-1]["content"],
-#                                                 "language": st.session_state.language})
-        
-# if __name__ == "__main__":
-#     launch_bot()
+    launch_bot()

query.py

@@ -125,138 +125,4 @@ class VectaraQuery():
                         chunks.append(chunk)
                         yield chunk
 
-        return ''.join(chunks)
-
-
-# import requests
-# import json
-
-
-# class VectaraQuery():
-#     def __init__(self, api_key: str, corpus_keys: list[str], prompt_name: str = None):
-#         self.corpus_keys = corpus_keys
-#         self.api_key = api_key
-#         self.prompt_name = prompt_name if prompt_name else "vectara-experimental-summary-ext-2023-12-11-sml"
-#         self.conv_id = None
-
-    
-#     def get_body(self, query_str: str, response_lang: str, stream: False):
-#         corpora_list = [{
-#                 'corpus_key': corpus_key, 'lexical_interpolation': 0.005
-#             } for corpus_key in self.corpus_keys
-#         ]
-
-#         return {
-#             'query': query_str,
-#             'search':
-#             {
-#                 'corpora': corpora_list,
-#                 'offset': 0,
-#                 'limit': 50,
-#                 'context_configuration':
-#                 {
-#                     'sentences_before': 2,
-#                     'sentences_after': 2,
-#                     'start_tag': "%START_SNIPPET%",
-#                     'end_tag': "%END_SNIPPET%",
-#                 },
-#                 'reranker':
-#                 {
-#                     'type': 'mmr'
-#                 },
-#             },
-#             'generation':
-#             {
-#                 'prompt_name': self.prompt_name,
-#                 'max_used_search_results': 10,
-#                 'response_language': response_lang,
-#                 'citations':
-#                 {
-#                     'style': 'none'
-#                 },
-#                 'enable_factual_consistency_score': False
-#             },
-#             'chat':
-#             {
-#                 'store': True
-#             },
-#             'stream_response': stream
-#         }
-    
-
-#     def get_headers(self):
-#         return {
-#             "Content-Type": "application/json",
-#             "Accept": "application/json",
-#             "x-api-key": self.api_key,
-#             "grpc-timeout": "60S"
-#         }
-    
-#     def get_stream_headers(self):
-#         return {
-#             "Content-Type": "application/json",
-#             "Accept": "text/event-stream",
-#             "x-api-key": self.api_key,
-#             "grpc-timeout": "60S"
-#         }
-
-#     def submit_query(self, query_str: str, language: str):
-
-#         if self.conv_id:
-#             endpoint = f"https://api.vectara.io/v2/chats/{self.conv_id}/turns"
-#         else:
-#             endpoint = "https://api.vectara.io/v2/chats"
-
-#         body = self.get_body(query_str, language, stream=False)
-
-#         response = requests.post(endpoint, data=json.dumps(body), verify=True, headers=self.get_headers())
-#         if response.status_code != 200:
-#             print(f"Query failed with code {response.status_code}, reason {response.reason}, text {response.text}")
-#             return "Sorry, something went wrong in my brain. Please try again later."
-
-#         res = response.json()
-
-#         if self.conv_id is None:
-#             self.conv_id = res['chat_id']
-
-#         summary = res['answer']
-
-#         # FIGURE OUT HOW TO IMPLEMENT THIS IN APIV2
-#         # if chat and chat['status'] is not None:
-#         #     st_code = chat['status']
-#         #     print(f"Chat query failed with code {st_code}")
-#         #     if st_code == 'RESOURCE_EXHAUSTED':
-#         #         self.conv_id = None
-#         #         return 'Sorry, Vectara chat turns exceeds plan limit.'
-#         #     return 'Sorry, something went wrong in my brain. Please try again later.'
-        
-#         return summary
-
-#     def submit_query_streaming(self, query_str: str, language: str):
-
-#         if self.conv_id:
-#             endpoint = f"https://api.vectara.io/v2/chats/{self.conv_id}/turns"
-#         else:
-#             endpoint = "https://api.vectara.io/v2/chats"
-
-#         body = self.get_body(query_str, language, stream=True)
-
-#         response = requests.post(endpoint, data=json.dumps(body), verify=True, headers=self.get_stream_headers(), stream=True) 
-        
-#         if response.status_code != 200:
-#             print(f"Query failed with code {response.status_code}, reason {response.reason}, text {response.text}")
-#             return "Sorry, something went wrong in my brain. Please try again later."
-
-#         chunks = []
-#         for line in response.iter_lines():
-#             line = line.decode('utf-8')
-#             if line:  # filter out keep-alive new lines
-#                 key, value = line.split(':', 1)
-#                 if key == 'data':
-#                     line = json.loads(value)
-#                     if line['type'] == 'generation_chunk':
-#                         chunk = line['generation_chunk']
-#                         chunks.append(chunk)
-#                         yield chunk
-
-#         return ''.join(chunks)
+        return ''.join(chunks)

utils.py

@@ -0,0 +1,69 @@
+import os
+import requests
+import json
+# import re
+
+import streamlit as st
+
+headers = {
+    'Content-Type': 'application/json',
+    'Accept': '*/*'
+}
+
+def thumbs_feedback(feedback, **kwargs):
+    """
+    Sends feedback to Amplitude Analytics
+    """
+    
+    send_amplitude_data(
+        user_query=kwargs.get("user_query", "No user input"),
+        chat_response=kwargs.get("chat_response", "No bot response"),
+        demo_name=kwargs.get("demo_name", "Unknown"),
+        language = kwargs.get("response_language", "Unknown"),
+        feedback=feedback["score"],
+    )    
+    st.session_state.feedback_key += 1
+
+def send_amplitude_data(user_query, chat_response, demo_name, language, feedback=None):
+    # Send query and response to Amplitude Analytics
+    data = {
+        "api_key": os.getenv('AMPLITUDE_TOKEN'),
+        "events": [{
+            "device_id": st.session_state.device_id,
+            "event_type": "submitted_query",
+            "event_properties": {
+                "Space Name": demo_name,
+                "Demo Type": "chatbot",
+                "query": user_query,
+                "response": chat_response,
+                "Response Language": language
+            }
+        }]
+    }
+    if feedback:
+        data["events"][0]["event_properties"]["feedback"] = feedback
+
+    response = requests.post('https://api2.amplitude.com/2/httpapi', headers=headers, data=json.dumps(data))
+    if response.status_code != 200:
+        print(f"Amplitude request failed with status code {response.status_code}. Response Text: {response.text}")
+
+# def escape_dollars_outside_latex(text):
+#     # Define a regex pattern to find LaTeX equations (either single $ or double $$)
+#     pattern = re.compile(r'(\$\$.*?\$\$|\$.*?\$)')
+#     latex_matches = pattern.findall(text)
+    
+#     # Placeholder to temporarily store LaTeX equations
+#     placeholders = {}
+#     for i, match in enumerate(latex_matches):
+#         placeholder = f'__LATEX_PLACEHOLDER_{i}__'
+#         placeholders[placeholder] = match
+#         text = text.replace(match, placeholder)
+    
+#     # Escape dollar signs in the rest of the text
+#     text = text.replace('$', '\\$')
+    
+#     # Replace placeholders with the original LaTeX equations
+#     for placeholder, original in placeholders.items():
+#         text = text.replace(placeholder, original)
+#     return text
+