Spaces:
Build error
Build error
File size: 3,034 Bytes
a173b58 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 |
"""
Monitoring Dashboard Component
Shows LLM usage statistics and costs in Streamlit
"""
import streamlit as st
import os
from typing import Dict, Any, List
from datetime import datetime, timedelta
from .langfuse_integration import langfuse_manager
def show_monitoring_dashboard():
"""Display monitoring dashboard in Streamlit sidebar"""
try:
if not langfuse_manager.is_enabled():
return
with st.sidebar:
st.subheader("📊 LLM Monitoring")
# Dashboard link
dashboard_url = langfuse_manager.get_dashboard_url()
st.markdown(f"[📈 Open Langfuse Dashboard]({dashboard_url})")
# Quick stats (if available)
try:
# Note: This would require additional Langfuse API calls
# For now, we'll show basic info
st.metric("Status", "🟢 Active")
st.metric("Session", st.session_state.get("session_id", "N/A"))
except Exception:
# Silently fail if stats can't be loaded
pass
except Exception:
# Dashboard should never break the main app
pass
def create_session_id() -> str:
"""Create a unique session ID for tracking"""
if "session_id" not in st.session_state:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
st.session_state.session_id = f"session_{timestamp}"
return st.session_state.session_id
def log_user_interaction(usecase: str, user_message: str, response: str):
"""Log user interaction for monitoring - fails silently if monitoring unavailable"""
try:
if not langfuse_manager.is_enabled():
return
session_id = create_session_id()
# Create trace for this interaction
trace = langfuse_manager.create_trace(
name=f"chat_interaction_{usecase}",
session_id=session_id,
user_id="streamlit_user",
metadata={
"usecase": usecase,
"timestamp": datetime.now().isoformat()
}
)
if trace:
# Log the generation
langfuse_manager.log_generation(
trace_id=trace.id,
model="groq_llm",
input_text=user_message,
output_text=response,
metadata={
"usecase": usecase,
"session_id": session_id
}
)
except Exception:
# Silently fail - logging should never break the app
pass
def show_cost_tracking():
"""Show cost tracking information"""
try:
if not langfuse_manager.is_enabled():
return
st.info("💰 Cost tracking is enabled via Langfuse. View detailed costs in the dashboard.")
except Exception:
# Silently fail if cost tracking info can't be shown
pass |