""" Market Performance Sentinel โ€” Hugging Face Space demo. Standalone Streamlit chatbot that showcases a LangGraph multi-agent pharma analytics workflow on 100% synthetic data. The fictional company in the demo is NovaPharma. """ from __future__ import annotations import os import sys from pathlib import Path HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import streamlit as st from chatbot_agents import build_langgraph_chatbot, invoke_langgraph_chatbot from db import DB_PATH, create_all_tables, db_is_seeded, get_metrics_distinct_values, get_metrics_from_db_filtered from llm_client import DEFAULT_MODEL, MissingHFTokenError, create_chat_client, describe_client, get_hf_token # --------------------------------------------------------------------------- # Page config and one-time setup # --------------------------------------------------------------------------- st.set_page_config( page_title="Market Performance Sentinel (Demo)", page_icon="๐Ÿงฌ", layout="wide", initial_sidebar_state="expanded", ) @st.cache_resource(show_spinner="Initialising the demo database โ€ฆ") def _bootstrap_database() -> str: """Create the SQLite schema and seed it if empty. Returns the DB path.""" create_all_tables() if not db_is_seeded(): from seed_data import main as seed_main seed_main() return str(DB_PATH) @st.cache_resource(show_spinner="Connecting to Hugging Face Inference Providers โ€ฆ") def _bootstrap_chat_client(): return create_chat_client() @st.cache_data(ttl=600) def _metrics_metadata() -> dict: return get_metrics_distinct_values() @st.cache_resource def _build_app(_client_marker: str, _db_marker: str): """Compile the LangGraph once and reuse across reruns. The marker arguments aren't read โ€” they just invalidate the cache when the client or DB changes. """ column_values = _metrics_metadata() client = _bootstrap_chat_client() return build_langgraph_chatbot( chat_client=client, db_query_fn=get_metrics_from_db_filtered, column_values=column_values, ) # --------------------------------------------------------------------------- # Sidebar โ€” architecture overview + config # --------------------------------------------------------------------------- with st.sidebar: st.markdown("## ๐Ÿงฌ Sentinel Demo") st.caption( "A LangGraph multi-agent chatbot for pharma market analytics, running on " "100% synthetic data for the fictional company **NovaPharma**." ) st.divider() st.markdown("### Agent graph") st.code( """query_analyzer โ”œโ†’ data_extraction โ”€โ” โ”‚ โ””โ†’ leading_country โ”€โ”€โ” โ”œโ†’ data_knowledge โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค โ””โ†’ (out_of_scope) โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค โ–ผ generate_response""", language="text", ) st.caption( "**query_analyzer** classifies intent ยท **data_extraction** parses filters " "& issues a targeted SQL query ยท **data_knowledge** looks up parameter " "tables ยท **leading_country** computes deterministic deltas ยท " "**generate_response** synthesises the final reply." ) st.divider() st.markdown("### Configuration") if not get_hf_token(): st.error( "๐Ÿ”‘ No `HF_TOKEN` detected. Add it under " "**Space โ†’ Settings โ†’ Variables and secrets**, then restart this Space." ) else: try: cfg = describe_client(_bootstrap_chat_client()) st.success(f"LLM: `{cfg['model']}`") st.caption(f"Provider: `{cfg['provider']}`") except MissingHFTokenError as exc: st.error(str(exc)) st.caption(f"Default model: `{DEFAULT_MODEL}` (override with `HF_MODEL` env var).") st.divider() st.markdown("### Source code") st.markdown( "This Space is the public, anonymized demo of a larger internal project. " "See the README for architecture details." ) # --------------------------------------------------------------------------- # Header & demo disclaimer # --------------------------------------------------------------------------- col_title, col_badge = st.columns([4, 1]) with col_title: st.title("Market Performance Sentinel ๐Ÿ’ฌ") st.caption( "Ask questions about pharma market performance or about **parameters** " "(cluster mapping, market summary, country-region). Powered by a LangGraph " "supervisor with specialised agents." ) with col_badge: st.markdown( "
" "" "100% Synthetic Demo Data
", unsafe_allow_html=True, ) st.info( "**This demo uses fictional data.** Company name (NovaPharma), product names " "(NOVACOR, NOVAGLU, NOVAFERT-A, โ€ฆ), and every numeric value are synthetic โ€” " "generated procedurally for demonstration purposes only.", icon="โ„น๏ธ", ) # --------------------------------------------------------------------------- # Bootstrap (DB + LLM client + graph) # --------------------------------------------------------------------------- try: db_path = _bootstrap_database() except Exception as exc: st.error("Could not initialise the demo database.") st.exception(exc) st.stop() column_values = _metrics_metadata() if not column_values or not column_values.get("Period"): st.warning( "The demo database is empty. Run `python seed_data.py` to populate it, " "then refresh the page." ) st.stop() try: chat_client = _bootstrap_chat_client() except MissingHFTokenError as exc: st.error(str(exc)) st.stop() except Exception as exc: st.error("Could not connect to Hugging Face Inference Providers.") st.exception(exc) st.stop() app = _build_app( _client_marker=getattr(chat_client, "_default_deployment", "default"), _db_marker=db_path, ) # --------------------------------------------------------------------------- # Session state # --------------------------------------------------------------------------- if "chat_messages" not in st.session_state: st.session_state.chat_messages = [] if "last_result_state" not in st.session_state: st.session_state.last_result_state = None if "last_user_query" not in st.session_state: st.session_state.last_user_query = None # --------------------------------------------------------------------------- # Available filters popover # --------------------------------------------------------------------------- _FILTER_DISPLAY_LABELS: dict[str, str] = { "Region": "Regions", "Period": "Periods", "Cluster": "Countries / Subregions / Clusters", "Calculation_Type": "Timeframes", "Product": "Products", "TA Market": "TA Markets", "Class": "Classes", } with st.popover("โ„น๏ธ Available filters"): st.markdown("**Filters you can use in your questions:**") st.caption("Region and Period are mandatory; everything else is optional.") st.divider() for col, label in _FILTER_DISPLAY_LABELS.items(): vals = column_values.get(col, []) if not vals: continue st.markdown(f"**{label}**") display = vals[:50] if col == "Product" else vals if len(display) <= 15: st.caption(", ".join(display)) else: st.caption(", ".join(display[:15]) + " โ€ฆ") st.write("") # --------------------------------------------------------------------------- # Example questions # --------------------------------------------------------------------------- EXAMPLE_QUESTIONS = [ "What's the market share of the top 3 products in France for the Growth Hormone Market?", "In LATAM, which country has the biggest total value QTR-QoQ in the Hypothyroid Market in 25Q3?", "What are the names of the clusters available for the Injectable Platform Market in APAC?", "Who is leading the total market changes in APAC for the Anti-EGFR Market?", ] st.caption("**Try an example:**") ex_col1, ex_col2 = st.columns(2) for i, question in enumerate(EXAMPLE_QUESTIONS): target_col = ex_col1 if i % 2 == 0 else ex_col2 with target_col: if st.button(question, key=f"ex_{i}", use_container_width=True): st.session_state.pending_query = question st.rerun() # --------------------------------------------------------------------------- # Chat history # --------------------------------------------------------------------------- for msg in st.session_state.chat_messages: with st.chat_message(msg["role"]): st.markdown(msg["content"]) prompt = st.session_state.pop("pending_query", None) chat_prompt = st.chat_input("Ask about market performance or parameters โ€ฆ") if prompt is None: prompt = chat_prompt if prompt: st.session_state.chat_messages.append({"role": "user", "content": prompt}) with st.chat_message("user"): st.markdown(prompt) with st.chat_message("assistant"): with st.spinner("Thinking โ€ฆ"): try: recent_history = st.session_state.chat_messages[-8:] prior_filters = {} if isinstance(st.session_state.last_result_state, dict): previous = st.session_state.last_result_state.get("filters", {}) if isinstance(previous, dict): prior_filters = previous result_state = invoke_langgraph_chatbot( app=app, user_query=prompt, conversation_history=recent_history, prior_filters=prior_filters, ) st.session_state.last_result_state = result_state st.session_state.last_user_query = prompt reply_text = (result_state.get("final_response", "") or "").strip() or "No reply generated." st.markdown(reply_text) st.session_state.chat_messages.append({"role": "assistant", "content": reply_text}) except Exception as exc: st.error(f"Error running the agent: {exc}") if st.session_state.chat_messages and st.session_state.chat_messages[-1]["role"] == "user": st.session_state.chat_messages.pop() if st.session_state.chat_messages: if st.button("Clear chat", type="secondary"): st.session_state.chat_messages = [] st.session_state.last_result_state = None st.session_state.last_user_query = None st.rerun() # --------------------------------------------------------------------------- # Data scope expander (latest filters + row counts) # --------------------------------------------------------------------------- with st.expander("Data scope (metrics used for answers)"): st.caption("Rows are fetched on demand via targeted SQL queries โ€” no full table scan.") st.write("**Periods in data:**", ", ".join(column_values.get("Period", []))) st.write("**Regions in data:**", ", ".join(column_values.get("Region", []))) st.write( "**Calculation types:**", ", ".join(column_values.get("Calculation_Type", [])), ) st.divider() st.write("**Latest query context:**") last_state = st.session_state.get("last_result_state") or {} last_query = st.session_state.get("last_user_query") if last_query: st.caption(f"Query: {last_query}") latest_filters = last_state.get("filters", {}) if isinstance(last_state, dict) else {} if latest_filters: st.json(latest_filters) else: st.caption("No filters captured yet. Ask a query to see extracted filters.") extracted = (last_state.get("extracted_data") or []) if isinstance(last_state, dict) else [] parameter = (last_state.get("parameter_data") or []) if isinstance(last_state, dict) else [] leading = (last_state.get("leading_country_result") or []) if isinstance(last_state, dict) else [] st.write(f"**Extracted rows:** {len(extracted):,}") if parameter: st.write(f"**Parameter rows:** {len(parameter):,}") if leading: st.write(f"**Leading-country rows:** {len(leading):,}")