| 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): |
| |
| 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): |
| |
| pattern = re.compile(r'(\$\$.*?\$\$|\$.*?\$)') |
| latex_matches = pattern.findall(text) |
| |
| |
| placeholders = {} |
| for i, match in enumerate(latex_matches): |
| placeholder = f'__LATEX_PLACEHOLDER_{i}__' |
| placeholders[placeholder] = match |
| text = text.replace(match, placeholder) |
| |
| |
| text = text.replace('$', '\\$') |
| |
| |
| for placeholder, original in placeholders.items(): |
| text = text.replace(placeholder, original) |
| return text |