|
|
import os |
|
|
import json |
|
|
import requests |
|
|
import pandas as pd |
|
|
import streamlit as st |
|
|
|
|
|
|
|
|
def get_env_variable(name: str) -> str: |
|
|
try: |
|
|
return os.environ[name] |
|
|
except KeyError: |
|
|
st.error(f"Environment variable '{name}' not set.") |
|
|
st.stop() |
|
|
|
|
|
CHAT_URL = get_env_variable("CHAT_URL") |
|
|
|
|
|
|
|
|
def fetch_results(query: str) -> list[dict]: |
|
|
""" |
|
|
Send a query to the chat API and return the parsed JSON response. |
|
|
""" |
|
|
payload = {"query": query} |
|
|
response = requests.post( |
|
|
CHAT_URL, |
|
|
headers={"Content-Type": "application/json"}, |
|
|
data=json.dumps(payload), |
|
|
) |
|
|
return response.json() |
|
|
|
|
|
|
|
|
def display_results_table(results: dict) -> None: |
|
|
"""Normalize and display API results as a Streamlit table.""" |
|
|
if "results" not in results: |
|
|
st.info(results.get('detail', 'No results found.')) |
|
|
return |
|
|
df = pd.json_normalize(results['results']) |
|
|
st.dataframe(df) |
|
|
|
|
|
def display_messages(history: list[dict]) -> None: |
|
|
"""Render chat messages from history including tables.""" |
|
|
for msg in history: |
|
|
with st.chat_message(msg["role"]): |
|
|
st.markdown(msg["content"]) |
|
|
if "data" in msg: |
|
|
display_results_table(msg["data"]) |
|
|
|
|
|
def main() -> None: |
|
|
st.title("BookDB 📚 🤖") |
|
|
|
|
|
|
|
|
if "messages" not in st.session_state: |
|
|
st.session_state.messages = [] |
|
|
|
|
|
|
|
|
query = st.chat_input("Ask about books...") |
|
|
|
|
|
if query: |
|
|
|
|
|
st.session_state.messages.append({ |
|
|
"role": "user", |
|
|
"content": query.strip() |
|
|
}) |
|
|
|
|
|
|
|
|
results = fetch_results(query) |
|
|
st.session_state.messages.append({ |
|
|
"role": "assistant", |
|
|
"content": "Here are the results:", |
|
|
"data": results |
|
|
}) |
|
|
|
|
|
|
|
|
display_messages(st.session_state.messages) |
|
|
|
|
|
if __name__ == "__main__": |
|
|
main() |