HR-Assistant / app.py
HassanJalil's picture
Upload 13 files
0a9f9c2 verified
import streamlit as st
import os
from pathlib import Path
import time
from typing import List, Dict, Any
from datetime import datetime
import google.generativeai as genai
from vector_store import VectorStore
from admin import AdminPanel
from config import Config
from utils import validate_api_key, format_response, log_interaction
# Page configuration
st.set_page_config(
page_title="BLUESCARF AI - HR Assistant",
page_icon="πŸ”·",
layout="wide",
initial_sidebar_state="collapsed"
)
# Custom CSS for enhanced UX and professional styling
st.markdown("""
<style>
/* Modern Color Palette & Typography */
:root {
--primary-blue: #1e40af;
--light-blue: #3b82f6;
--accent-blue: #60a5fa;
--surface-light: #f8fafc;
--surface-white: #ffffff;
--text-primary: #1f2937;
--text-secondary: #6b7280;
--border-light: #e5e7eb;
--success-green: #10b981;
--warning-orange: #f59e0b;
--error-red: #ef4444;
--shadow-soft: 0 1px 3px rgba(0,0,0,0.1);
--shadow-medium: 0 4px 6px rgba(0,0,0,0.1);
--radius-md: 8px;
--radius-lg: 12px;
}
/* Remove Streamlit Default Padding */
.main .block-container {
padding-top: 2rem;
padding-bottom: 2rem;
max-width: 1200px;
}
/* Enhanced Header Design */
.main-header {
background: linear-gradient(135deg, var(--primary-blue) 0%, var(--light-blue) 100%);
padding: 2.5rem;
border-radius: var(--radius-lg);
margin-bottom: 2rem;
text-align: center;
box-shadow: var(--shadow-medium);
position: relative;
overflow: hidden;
}
.main-header::before {
content: '';
position: absolute;
top: 0;
left: 0;
right: 0;
bottom: 0;
background: url('data:image/svg+xml,<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 100 100"><defs><pattern id="grid" width="10" height="10" patternUnits="userSpaceOnUse"><path d="M 10 0 L 0 0 0 10" fill="none" stroke="rgba(255,255,255,0.1)" stroke-width="0.5"/></pattern></defs><rect width="100" height="100" fill="url(%23grid)"/></svg>');
opacity: 0.3;
}
.main-header h1, .main-header h3 {
position: relative;
z-index: 1;
margin: 0;
}
.main-header h1 {
color: white;
font-size: 2.5rem;
font-weight: 700;
letter-spacing: -0.02em;
}
.main-header h3 {
color: #bfdbfe;
font-size: 1.25rem;
font-weight: 400;
margin-top: 0.5rem;
}
/* Logo Styling */
.company-logo {
max-width: 120px;
margin: 1rem auto;
display: block;
border-radius: var(--radius-md);
box-shadow: var(--shadow-soft);
}
/* Chat Interface Enhancements */
.chat-main-container {
background: var(--surface-white);
border-radius: var(--radius-lg);
padding: 1.5rem;
margin: 1rem 0;
box-shadow: var(--shadow-medium);
border: 1px solid var(--border-light);
}
.chat-messages-container {
min-height: 300px;
max-height: 500px;
overflow-y: auto;
padding: 1rem;
background: var(--surface-light);
border-radius: var(--radius-md);
margin-bottom: 1.5rem;
border: 1px solid var(--border-light);
}
.chat-messages-container::-webkit-scrollbar {
width: 6px;
}
.chat-messages-container::-webkit-scrollbar-track {
background: #f1f5f9;
border-radius: 3px;
}
.chat-messages-container::-webkit-scrollbar-thumb {
background: #cbd5e1;
border-radius: 3px;
}
.chat-messages-container::-webkit-scrollbar-thumb:hover {
background: #94a3b8;
}
/* Enhanced Message Bubbles */
.user-message {
background: linear-gradient(135deg, var(--light-blue), var(--accent-blue));
color: white;
padding: 1rem 1.25rem;
border-radius: 1.5rem 1.5rem 0.5rem 1.5rem;
margin: 0.75rem 0 0.75rem auto;
max-width: 80%;
box-shadow: var(--shadow-soft);
animation: slideInRight 0.3s ease-out;
position: relative;
}
.assistant-message {
background: var(--surface-white);
color: var(--text-primary);
padding: 1rem 1.25rem;
border-radius: 1.5rem 1.5rem 1.5rem 0.5rem;
margin: 0.75rem auto 0.75rem 0;
max-width: 80%;
box-shadow: var(--shadow-soft);
border: 1px solid var(--border-light);
animation: slideInLeft 0.3s ease-out;
position: relative;
}
@keyframes slideInRight {
from { opacity: 0; transform: translateX(20px); }
to { opacity: 1; transform: translateX(0); }
}
@keyframes slideInLeft {
from { opacity: 0; transform: translateX(-20px); }
to { opacity: 1; transform: translateX(0); }
}
.message-meta {
font-size: 0.75rem;
opacity: 0.7;
margin-top: 0.5rem;
}
/* Perfect Chat Input Layout */
.chat-input-container {
display: flex;
gap: 0.75rem;
align-items: flex-end;
padding: 1rem;
background: var(--surface-light);
border-radius: var(--radius-md);
border: 2px solid transparent;
transition: border-color 0.2s ease;
}
.chat-input-container:focus-within {
border-color: var(--light-blue);
box-shadow: 0 0 0 3px rgba(59, 130, 246, 0.1);
}
.chat-input-field {
flex: 1;
min-height: 44px;
max-height: 120px;
padding: 0.75rem 1rem;
border: 1px solid var(--border-light);
border-radius: var(--radius-md);
font-size: 1rem;
resize: vertical;
transition: all 0.2s ease;
background: var(--surface-white);
}
.chat-input-field:focus {
outline: none;
border-color: var(--light-blue);
box-shadow: 0 0 0 3px rgba(59, 130, 246, 0.1);
}
.chat-send-button {
min-width: 44px;
height: 44px;
background: linear-gradient(135deg, var(--light-blue), var(--primary-blue));
color: white;
border: none;
border-radius: var(--radius-md);
cursor: pointer;
transition: all 0.2s ease;
display: flex;
align-items: center;
justify-content: center;
font-weight: 600;
box-shadow: var(--shadow-soft);
}
.chat-send-button:hover:not(:disabled) {
transform: translateY(-1px);
box-shadow: 0 4px 12px rgba(59, 130, 246, 0.3);
}
.chat-send-button:disabled {
opacity: 0.6;
cursor: not-allowed;
transform: none;
}
/* Enhanced Button Styles */
.stButton > button {
background: linear-gradient(135deg, var(--light-blue), var(--primary-blue));
color: white;
border: none;
border-radius: var(--radius-md);
padding: 0.6rem 1.2rem;
font-weight: 600;
transition: all 0.2s ease;
box-shadow: var(--shadow-soft);
}
.stButton > button:hover {
transform: translateY(-1px);
box-shadow: 0 4px 12px rgba(59, 130, 246, 0.3);
}
/* Loading States */
.loading-indicator {
display: flex;
align-items: center;
gap: 0.5rem;
padding: 1rem;
background: var(--surface-light);
border-radius: var(--radius-md);
margin: 0.5rem 0;
}
.loading-dots {
display: flex;
gap: 0.25rem;
}
.loading-dot {
width: 6px;
height: 6px;
background: var(--light-blue);
border-radius: 50%;
animation: loadingPulse 1.4s infinite ease-in-out;
}
.loading-dot:nth-child(1) { animation-delay: -0.32s; }
.loading-dot:nth-child(2) { animation-delay: -0.16s; }
@keyframes loadingPulse {
0%, 80%, 100% { transform: scale(0.8); opacity: 0.5; }
40% { transform: scale(1); opacity: 1; }
}
/* Admin Section Enhancements */
.admin-section {
background: linear-gradient(135deg, #fef2f2, #fdf2f8);
border: 1px solid #fecaca;
border-radius: var(--radius-lg);
padding: 1.5rem;
margin-top: 2rem;
position: relative;
overflow: hidden;
}
.admin-section::before {
content: 'πŸ”';
position: absolute;
top: 1rem;
right: 1rem;
font-size: 1.5rem;
opacity: 0.3;
}
/* Status Indicators */
.status-indicator {
display: inline-flex;
align-items: center;
gap: 0.5rem;
padding: 0.375rem 0.75rem;
border-radius: 9999px;
font-size: 0.875rem;
font-weight: 500;
}
.status-success {
background: #dcfce7;
color: #166534;
border: 1px solid #bbf7d0;
}
.status-warning {
background: #fef3c7;
color: #92400e;
border: 1px solid #fde68a;
}
.status-error {
background: #fee2e2;
color: #991b1b;
border: 1px solid #fecaca;
}
/* Enhanced Metrics */
.metric-card {
background: var(--surface-white);
padding: 1.5rem;
border-radius: var(--radius-md);
box-shadow: var(--shadow-soft);
border: 1px solid var(--border-light);
text-align: center;
transition: transform 0.2s ease;
}
.metric-card:hover {
transform: translateY(-2px);
box-shadow: var(--shadow-medium);
}
.metric-value {
font-size: 2rem;
font-weight: 700;
color: var(--primary-blue);
margin-bottom: 0.5rem;
}
.metric-label {
font-size: 0.875rem;
color: var(--text-secondary);
font-weight: 500;
}
/* Footer Enhancement */
.footer {
text-align: center;
padding: 2rem;
color: var(--text-secondary);
border-top: 1px solid var(--border-light);
margin-top: 3rem;
background: var(--surface-light);
border-radius: var(--radius-md);
}
/* Mobile Responsiveness */
@media (max-width: 768px) {
.main-header {
padding: 1.5rem;
}
.main-header h1 {
font-size: 1.875rem;
}
.chat-input-container {
flex-direction: column;
gap: 0.75rem;
}
.chat-send-button {
width: 100%;
height: 48px;
}
.user-message, .assistant-message {
max-width: 95%;
}
}
/* Performance Optimization - Reduce Repaints */
.main .block-container {
will-change: transform;
}
/* Accessibility Enhancements */
.chat-input-field:focus,
.stButton > button:focus {
outline: 2px solid var(--light-blue);
outline-offset: 2px;
}
/* High Contrast Mode Support */
@media (prefers-contrast: high) {
:root {
--primary-blue: #0056b3;
--light-blue: #0066cc;
--border-light: #666666;
}
}
/* Reduced Motion Support */
@media (prefers-reduced-motion: reduce) {
* {
animation-duration: 0.01ms !important;
animation-iteration-count: 1 !important;
transition-duration: 0.01ms !important;
}
}
</style>
""", unsafe_allow_html=True)
class HRAssistant:
def __init__(self):
self.config = Config()
self.vector_store = VectorStore()
self.admin_panel = AdminPanel()
def initialize_session_state(self):
"""Initialize session state variables"""
if 'messages' not in st.session_state:
st.session_state.messages = []
if 'api_key_validated' not in st.session_state:
st.session_state.api_key_validated = False
if 'show_admin' not in st.session_state:
st.session_state.show_admin = False
if 'admin_authenticated' not in st.session_state:
st.session_state.admin_authenticated = False
def render_header(self):
"""Render application header with logo"""
st.markdown("""
<div class="main-header">
<h1 style="color: white; margin: 0;">BLUESCARF ARTIFICIAL INTELLIGENCE</h1>
<h3 style="color: #bfdbfe; margin: 0.5rem 0 0 0;">HR Assistant</h3>
</div>
""", unsafe_allow_html=True)
# Logo placeholder - replace logo.png with actual company logo
logo_path = Path("logo.png")
if logo_path.exists():
st.image("logo.png", width=200)
else:
st.info("πŸ“‹ Replace 'logo.png' with your company logo")
def setup_gemini_api(self, api_key: str) -> bool:
"""Configure Gemini API with provided key"""
try:
if not validate_api_key(api_key):
return False
genai.configure(api_key=api_key)
# Test API connection
model = genai.GenerativeModel('gemini-1.5-flash')
test_response = model.generate_content("Hello")
st.session_state.api_key_validated = True
st.session_state.model = model
return True
except Exception as e:
st.error(f"API Configuration Error: {str(e)}")
return False
def get_relevant_context(self, query: str) -> List[Dict[str, Any]]:
"""Retrieve relevant context from vector store"""
return self._retrieve_relevant_context(query)
def generate_response(self, query: str, context: List[Dict[str, Any]]) -> str:
"""Generate response using Gemini API with retrieved context"""
return self._generate_contextual_response(query, context)
def is_hr_related_query(self, query: str) -> bool:
"""Check if query is HR-related using enhanced classification"""
return self._is_hr_related_query(query)
# Log interaction
log_interaction(query, response)
def render_chat_interface(self):
"""Render the main chat interface with robust state management"""
st.markdown("### πŸ’¬ Chat with HR Assistant")
# Initialize input state management
if 'input_processed' not in st.session_state:
st.session_state.input_processed = False
if 'last_input' not in st.session_state:
st.session_state.last_input = ""
# Chat message container
self._render_chat_messages()
# Input interface with intelligent state handling
self._render_chat_input()
# Chat controls
self._render_chat_controls()
def _render_chat_messages(self):
"""Render chat message history with optimized layout"""
if not st.session_state.messages:
st.info("πŸ‘‹ Welcome! Ask me anything about BLUESCARF AI HR policies and procedures.")
return
# Create scrollable chat container
chat_container = st.container()
with chat_container:
for idx, message in enumerate(st.session_state.messages):
message_key = f"msg_{idx}_{message.get('timestamp', time.time())}"
if message["role"] == "user":
st.markdown(f"""
<div class="user-message" id="{message_key}">
<strong>You:</strong> {message["content"]}
</div>
""", unsafe_allow_html=True)
else:
st.markdown(f"""
<div class="assistant-message" id="{message_key}">
<strong>HR Assistant:</strong> {message["content"]}
</div>
""", unsafe_allow_html=True)
def _render_chat_input(self):
"""Render chat input with intelligent state management to prevent loops"""
col1, col2 = st.columns([5, 1])
with col1:
# Dynamic input key to prevent state persistence issues
input_key = f"chat_input_{len(st.session_state.messages)}"
user_input = st.text_input(
"Ask me about company policies, benefits, procedures...",
key=input_key,
placeholder="Type your HR question here...",
value="" # Always start with empty value
)
with col2:
send_button = st.button("Send", type="primary", key=f"send_{len(st.session_state.messages)}")
# Process input with anti-loop protection
if send_button and user_input and user_input.strip():
# Prevent duplicate processing
if user_input != st.session_state.last_input or not st.session_state.input_processed:
self._process_user_query(user_input.strip())
st.session_state.last_input = user_input.strip()
st.session_state.input_processed = True
# Trigger rerun to update UI with new messages
st.rerun()
else:
st.warning("⚠️ Query already processed. Please ask a new question.")
# Reset processing flag when input changes
if user_input != st.session_state.last_input:
st.session_state.input_processed = False
def _render_chat_controls(self):
"""Render chat control buttons with proper state management"""
if not st.session_state.messages:
return
col1, col2, col3 = st.columns([2, 2, 2])
with col1:
if st.button("πŸ—‘οΈ Clear Chat", key="clear_chat_btn"):
self._clear_chat_session()
with col2:
if st.button("πŸ“₯ Export Chat", key="export_chat_btn"):
self._export_chat_history()
with col3:
st.caption(f"πŸ’¬ {len(st.session_state.messages)} messages")
def _process_user_query(self, query: str):
"""Process user query with enhanced error handling and state management"""
if not query or len(query.strip()) < 3:
st.warning("⚠️ Please enter a meaningful question.")
return
# Add user message to chat history
user_message = {
"role": "user",
"content": query,
"timestamp": time.time(),
"message_id": self._generate_message_id()
}
st.session_state.messages.append(user_message)
# Process query and generate response
try:
with st.spinner("πŸ€” Thinking..."):
response = self._generate_intelligent_response(query)
# Add assistant response to chat history
assistant_message = {
"role": "assistant",
"content": response,
"timestamp": time.time(),
"message_id": self._generate_message_id(),
"query_processed": query
}
st.session_state.messages.append(assistant_message)
# Log successful interaction
self._log_successful_interaction(query, response)
except Exception as e:
error_response = f"I apologize, but I encountered an error processing your request: {str(e)}. Please try rephrasing your question."
assistant_message = {
"role": "assistant",
"content": error_response,
"timestamp": time.time(),
"message_id": self._generate_message_id(),
"error": True
}
st.session_state.messages.append(assistant_message)
# Log error for debugging
self._log_error_interaction(query, str(e))
def _generate_intelligent_response(self, query: str) -> str:
"""Generate contextually aware response using RAG pipeline"""
# Validate query scope
if not self._is_hr_related_query(query):
return self._get_scope_redirect_message()
# Retrieve relevant context
context_chunks = self._retrieve_relevant_context(query)
if not context_chunks:
return self._get_no_context_message()
# Generate response using Gemini API
return self._generate_contextual_response(query, context_chunks)
def _retrieve_relevant_context(self, query: str) -> List[Dict[str, Any]]:
"""Retrieve relevant context with enhanced error handling"""
try:
return self.vector_store.similarity_search(
query,
k=self.config.MAX_CONTEXT_CHUNKS
)
except Exception as e:
st.error(f"Context retrieval error: {str(e)}")
return []
def _generate_contextual_response(self, query: str, context: List[Dict[str, Any]]) -> str:
"""Generate response using Gemini API with retrieved context"""
try:
# Prepare context for prompt engineering
context_text = self._format_context_for_prompt(context)
# Construct optimized prompt
prompt = self._build_contextual_prompt(query, context_text)
# Generate response with error handling
response = st.session_state.model.generate_content(prompt)
return self._format_and_validate_response(response.text)
except Exception as e:
return f"I apologize, but I encountered an error generating a response: {str(e)}. Please try rephrasing your question."
def _format_context_for_prompt(self, context: List[Dict[str, Any]]) -> str:
"""Format context chunks for optimal prompt engineering"""
formatted_sections = []
for idx, chunk in enumerate(context, 1):
source = chunk['metadata'].get('source', 'Company Document')
content = chunk['content']
formatted_sections.append(
f"[Document {idx}: {source}]\n{content}\n"
)
return "\n".join(formatted_sections)
def _build_contextual_prompt(self, query: str, context_text: str) -> str:
"""Build optimized prompt for Gemini API"""
system_context = self.config.get_hr_context_prompt()
return f"""{system_context}
COMPANY DOCUMENT CONTEXT:
{context_text}
USER QUESTION: {query}
RESPONSE GUIDELINES:
- Answer based ONLY on the provided company documents
- Be specific and reference relevant policies
- If information is incomplete, state what's available and suggest contacting HR
- Maintain professional, helpful tone
- Provide actionable guidance when possible
RESPONSE:"""
def _format_and_validate_response(self, response_text: str) -> str:
"""Format and validate AI response for optimal user experience"""
if not response_text or len(response_text.strip()) < 10:
return "I apologize, but I couldn't generate a meaningful response. Please try rephrasing your question."
# Enhanced text formatting
formatted_response = self._enhance_response_formatting(response_text.strip())
# Add contextual footer if response is substantial
if len(formatted_response) > 150:
formatted_response += "\n\n*For additional assistance, please contact the HR department.*"
return formatted_response
def _enhance_response_formatting(self, text: str) -> str:
"""Apply intelligent formatting enhancements"""
# Remove AI response artifacts
cleaned = text.replace("Based on the provided documents,", "")
cleaned = cleaned.replace("According to the company policies,", "")
# Ensure proper sentence spacing
sentences = cleaned.split('. ')
properly_spaced = '. '.join(sentence.strip() for sentence in sentences if sentence.strip())
return properly_spaced
def _is_hr_related_query(self, query: str) -> bool:
"""Enhanced HR query classification with fuzzy matching"""
hr_indicators = [
'policy', 'leave', 'vacation', 'sick', 'holiday', 'benefit', 'insurance',
'salary', 'compensation', 'promotion', 'performance', 'review', 'training',
'onboarding', 'handbook', 'procedure', 'guideline', 'hr', 'human resources',
'employee', 'staff', 'team', 'department', 'work', 'job', 'role',
'resignation', 'termination', 'disciplinary', 'conduct', 'harassment'
]
query_lower = query.lower()
return any(indicator in query_lower for indicator in hr_indicators)
def _get_scope_redirect_message(self) -> str:
"""Get polite redirect message for non-HR queries"""
return ("I'm specifically designed to assist with BLUESCARF AI HR-related questions "
"using our company policies and documents. Please ask me about company "
"policies, benefits, leave procedures, or other HR matters.")
def _get_no_context_message(self) -> str:
"""Get message when no relevant context is found"""
return ("I couldn't find relevant information in our company documents for your "
"question. Please contact HR directly for assistance, or try rephrasing "
"your question using different terms.")
def _clear_chat_session(self):
"""Clear chat session with proper state reset"""
st.session_state.messages = []
st.session_state.input_processed = False
st.session_state.last_input = ""
st.success("πŸ—‘οΈ Chat history cleared!")
st.rerun()
def _export_chat_history(self):
"""Export chat history for user reference"""
if not st.session_state.messages:
st.warning("No chat history to export.")
return
# Create exportable format
export_content = "BLUESCARF AI HR Assistant - Chat Export\n"
export_content += f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"
for message in st.session_state.messages:
role = "You" if message["role"] == "user" else "HR Assistant"
timestamp = datetime.fromtimestamp(message["timestamp"]).strftime('%H:%M:%S')
export_content += f"[{timestamp}] {role}: {message['content']}\n\n"
st.download_button(
label="πŸ“₯ Download Chat History",
data=export_content,
file_name=f"hr_chat_export_{int(time.time())}.txt",
mime="text/plain"
)
def _generate_message_id(self) -> str:
"""Generate unique message identifier"""
return f"msg_{int(time.time() * 1000)}_{len(st.session_state.messages)}"
def _log_successful_interaction(self, query: str, response: str):
"""Log successful interaction for analytics"""
try:
log_interaction(query, response, {
'success': True,
'response_length': len(response),
'session_messages': len(st.session_state.messages)
})
except Exception:
pass # Silent fail for logging
def _log_error_interaction(self, query: str, error: str):
"""Log error interaction for debugging"""
try:
log_interaction(query, f"ERROR: {error}", {
'success': False,
'error_type': 'processing_error',
'session_messages': len(st.session_state.messages)
})
except Exception:
pass # Silent fail for logging
def render_admin_section(self):
"""Render admin panel section"""
st.markdown("---")
col1, col2 = st.columns([3, 1])
with col1:
st.markdown("### πŸ”§ Administrator Panel")
st.markdown("*Manage knowledge base and update company documents*")
with col2:
if st.button("Admin Access"):
st.session_state.show_admin = not st.session_state.show_admin
if st.session_state.show_admin:
self.admin_panel.render()
def render_footer(self):
"""Render application footer"""
st.markdown("""
<div class="footer">
<p><strong>BLUESCARF ARTIFICIAL INTELLIGENCE</strong> | HR Assistant v1.0</p>
<p>Powered by Google Gemini AI | Built with Streamlit</p>
</div>
""", unsafe_allow_html=True)
def run(self):
"""Main application entry point"""
self.initialize_session_state()
self.render_header()
# API Key input
if not st.session_state.api_key_validated:
st.markdown("### πŸ”‘ API Configuration")
with st.form("api_key_form"):
api_key = st.text_input(
"Enter your Google Gemini API Key:",
type="password",
help="Get your API key from https://makersuite.google.com/app/apikey"
)
submitted = st.form_submit_button("Connect", type="primary")
if submitted and api_key:
with st.spinner("Validating API key..."):
if self.setup_gemini_api(api_key):
st.success("βœ… API key validated successfully!")
st.rerun()
else:
st.error("❌ Invalid API key. Please check and try again.")
# Show knowledge base status
doc_count = self.vector_store.get_document_count()
if doc_count > 0:
st.info(f"πŸ“š Knowledge base contains {doc_count} processed documents")
else:
st.warning("⚠️ No documents in knowledge base. Please use admin panel to add company documents.")
else:
# Main application interface
self.render_chat_interface()
self.render_admin_section()
self.render_footer()
def main():
"""Application entry point"""
app = HRAssistant()
app.run()
if __name__ == "__main__":
main()