from omegaconf import OmegaConf from query import VectaraQuery import os import streamlit as st from PIL import Image def isTrue(x) -> bool: if isinstance(x, bool): return x return x.strip().lower() == 'true' def launch_bot(): def generate_response(question): response = vq.submit_query(question) return response def generate_streaming_response(question): response = vq.submit_query_streaming(question) return response if 'cfg' not in st.session_state: corpus_ids = str(os.environ['corpus_ids']).split(',') cfg = OmegaConf.create({ 'customer_id': str(os.environ['customer_id']), 'corpus_ids': corpus_ids, 'api_key': str(os.environ['api_key']), 'title': os.environ['title'], 'description': os.environ['description'], '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', '') }) st.session_state.cfg = cfg st.session_state.ex_prompt = None st.session_state.vq = VectaraQuery(cfg.api_key, cfg.customer_id, cfg.corpus_ids, 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.markdown(f"## Welcome to {cfg.title}\n\n" f"This demo uses Retrieval Augmented Generation to ask questions about {cfg.source_data_desc}\n\n") 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.image(image, width=250) st.markdown(f"

Vectara chat demo: {cfg.title}

", unsafe_allow_html=True) st.markdown(f"

{cfg.description}

", unsafe_allow_html=True) if "messages" not in st.session_state.keys(): st.session_state.messages = [{"role": "assistant", "content": "How may I help you?"}] max_examples = 4 example_messages = [example.strip() for example in cfg.examples.split(",")] example_messages = [em for em in example_messages if len(em)>0][:max_examples] if len(example_messages) > 0: st.markdown("
Queries To Try:
", unsafe_allow_html=True) ex_cols = st.columns(max_examples) # Display chat messages for message in st.session_state.messages: with st.chat_message(message["role"]): st.write(message["content"]) # User-provided prompt if st.session_state.ex_prompt: prompt = st.session_state.ex_prompt st.session_state.ex_prompt = None else: prompt = st.chat_input() if prompt: st.session_state.messages.append({"role": "user", "content": prompt}) with st.chat_message("user"): st.write(prompt) # Example prompt for i, example in enumerate(example_messages): button_pressed = False with ex_cols[i]: if st.button(example): st.session_state.ex_prompt = example if button_pressed: st.session_state.messages.append({"role": "user", "content": prompt}) with st.chat_message("user"): st.write(prompt) # 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) if __name__ == "__main__": launch_bot()