""" Chat components for rendering messages and handling user interactions. """ import streamlit as st import pandas as pd import base64 from typing import List, Dict, Any from config import OUTLINE_INDIGO_USER, DARK_MODE_SLATE_AI, RETRY_BUTTON_TEXT, DOWNLOAD_BUTTON_TEXT class ChatRenderer: """Handles rendering of chat messages and interactions.""" def __init__(self): self.user_avatar = OUTLINE_INDIGO_USER self.ai_avatar = DARK_MODE_SLATE_AI def render_chat_history(self, messages: List[Dict[str, Any]]) -> None: """Render the complete chat history.""" st.markdown("
", unsafe_allow_html=True) for i, msg in enumerate(messages): if msg["role"] == "user": self._render_user_message(msg, i, messages) elif msg["role"] == "assistant": # Skip error messages that are already shown after user messages if not self._is_error_shown_after_user(i, messages): self._render_assistant_message(msg) st.markdown("
", unsafe_allow_html=True) def _render_user_message(self, msg: Dict[str, Any], index: int, messages: List[Dict[str, Any]]) -> None: """Render a user message with avatar.""" st.markdown( f'''
User
{msg["content"]}
''', unsafe_allow_html=True ) # Check if next message is an error and render retry option self._render_error_retry_if_needed(index, messages) def _render_error_retry_if_needed(self, index: int, messages: List[Dict[str, Any]]) -> None: """Render error message and retry button if the next message is an error.""" if ( index + 1 < len(messages) and messages[index + 1]["role"] == "assistant" and messages[index + 1].get("is_error") ): error_msg = messages[index + 1]["content"] cols = st.columns([0.3, 0.7]) with cols[1]: col1, col2 = st.columns([0.8, 0.2]) with col1: st.markdown( f'
{error_msg}
', unsafe_allow_html=True ) with col2: if st.button(RETRY_BUTTON_TEXT, key=f"retry_{index}"): self._handle_retry(index) def _handle_retry(self, index: int) -> None: """Handle retry button click.""" messages = st.session_state["messages"] st.session_state["messages"] = messages[:index + 1] st.session_state["messages"].append({ "role": "assistant", "content": "🤔 Thinking...", "is_placeholder": True }) st.rerun() def _is_error_shown_after_user(self, index: int, messages: List[Dict[str, Any]]) -> bool: """Check if this error message is already shown after the previous user message.""" if not messages[index].get("is_error"): return False # Check if this is an error that follows a user message if index > 0 and messages[index - 1]["role"] == "user": return True return False def _render_assistant_message(self, msg: Dict[str, Any]) -> None: """Render an assistant message with avatar and optional data/charts.""" bubble_class = "error-bubble" if msg.get("is_error") else "ai-bubble" # Render message bubble if msg.get("is_placeholder"): self._render_thinking_message(bubble_class) else: self._render_regular_message(msg["content"], bubble_class) # Render data table if present if msg.get("data"): self._render_data_table(msg["data"], msg) # Render chart if present if msg.get("chart"): self._render_chart(msg["chart"]) def _render_thinking_message(self, bubble_class: str) -> None: """Render a thinking/loading message with spinner.""" st.markdown( f'''
AI
Thinking...
''', unsafe_allow_html=True ) def _render_regular_message(self, content: str, bubble_class: str) -> None: """Render a regular assistant message.""" st.markdown( f'''
AI
{content}
''', unsafe_allow_html=True ) def _render_data_table(self, data: List[Dict], msg: Dict[str, Any]) -> None: """Render data table with download option.""" df = pd.DataFrame(data) st.dataframe(df, use_container_width=True) csv = df.to_csv(index=False).encode("utf-8") st.download_button( DOWNLOAD_BUTTON_TEXT, csv, "results.csv", "text/csv", key=f"download_csv_{id(msg)}" ) def _render_chart(self, chart_data: str) -> None: """Render chart from base64 data.""" img_data = base64.b64decode(chart_data) st.image(img_data, use_column_width=True)