# --- PYTHON 3.13 COMPATIBILITY MONKEY-PATCH --- import sys import types if 'audioop' not in sys.modules: mock_audioop = types.ModuleType('audioop') mock_audioop.error = Exception mock_audioop.getsample = lambda data, width, index: 0 sys.modules['audioop'] = mock_audioop # ---------------------------------------------- import os import sqlite3 import pandas as pd import json import gradio as gr from groq import Groq import requests from datetime import datetime # --- Free Tier Local Path Mapping --- DB_NAME = "strides_architecture.db" def init_db(): conn = sqlite3.connect(DB_NAME) cursor = conn.cursor() # 1. Core Requirements Records Table cursor.execute(''' CREATE TABLE IF NOT EXISTS project_tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, phase TEXT, task_name TEXT, owner TEXT, timeline TEXT, priority TEXT ) ''') # 2. Session Metastore Table cursor.execute(''' CREATE TABLE IF NOT EXISTS chat_sessions ( session_id TEXT PRIMARY KEY, title TEXT, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ''') # 3. Chat History Content Store Table cursor.execute(''' CREATE TABLE IF NOT EXISTS chat_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, role TEXT, content TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY(session_id) REFERENCES chat_sessions(session_id) ON DELETE CASCADE ) ''') cursor.execute("SELECT COUNT(*) FROM project_tasks") if cursor.fetchone()[0] == 0: dummy_data = [ ('BRD-Scope', 'Automated Audit Trail Log Consolidation', 'A. Sharma', 'Q3-2026', 'Critical'), ('GxP-Compliance', '21 CFR Part 11 Electronic Signature Verification', 'R. Verma', 'Q3-2026', 'Critical') ] cursor.executemany('INSERT INTO project_tasks (phase, task_name, owner, timeline, priority) VALUES (?,?,?,?,?)', dummy_data) conn.commit() conn.close() init_db() # --- Model Selection Profiles --- GROQ_MODELS = ["llama-3.1-8b-instant", "llama-3.1-70b-versatile", "whisper-large-v3-turbo"] OPENROUTER_MODELS = ["nvidia/nemotron-3-ultra-550b-a55b:free", "nvidia/nemotron-3.5-content-safety:free"] # --- Persistent Storage Controller Core Functions --- def get_all_sessions(): conn = sqlite3.connect(DB_NAME) cursor = conn.cursor() cursor.execute("SELECT session_id, title FROM chat_sessions ORDER BY updated_at DESC") sessions = cursor.fetchall() conn.close() if not sessions: return [("default_session", "New Compliance Session 1")] return sessions def load_session_history(session_id): conn = sqlite3.connect(DB_NAME) cursor = conn.cursor() cursor.execute("SELECT role, content FROM chat_history WHERE session_id = ? ORDER BY timestamp ASC", (session_id,)) turns = cursor.fetchall() conn.close() # Filter out internal tracking 'system' triggers to protect chat UI cleanliness return [{"role": t[0], "content": t[1]} for t in turns if t[0] in ["user", "assistant"]] def save_chat_turn(session_id, role, content, prompt_fallback_title=None): conn = sqlite3.connect(DB_NAME) cursor = conn.cursor() cursor.execute("SELECT COUNT(*) FROM chat_sessions WHERE session_id = ?", (session_id,)) if cursor.fetchone()[0] == 0: title = prompt_fallback_title if prompt_fallback_title else f"Chat Session ({datetime.now().strftime('%M:%S')})" cursor.execute("INSERT INTO chat_sessions (session_id, title) VALUES (?, ?)", (session_id, title)) else: if prompt_fallback_title: cursor.execute("UPDATE chat_sessions SET title = ?, updated_at = CURRENT_TIMESTAMP WHERE session_id = ?", (prompt_fallback_title, session_id)) else: cursor.execute("UPDATE chat_sessions SET updated_at = CURRENT_TIMESTAMP WHERE session_id = ?", (session_id,)) cursor.execute("INSERT INTO chat_history (session_id, role, content) VALUES (?, ?, ?)", (session_id, role, content)) conn.commit() conn.close() # --- Dropdown / State Modifiers --- def change_provider_options(provider_choice): choices = GROQ_MODELS if provider_choice == "Groq" else OPENROUTER_MODELS return gr.Dropdown(choices=choices, value=choices[0], interactive=True) def handle_session_switch(session_id): history = load_session_history(session_id) return history, f"Active Session: {session_id}" def create_brand_new_session(): new_id = f"session_{int(datetime.now().timestamp())}" save_chat_turn(new_id, "system", "Session initialized.", prompt_fallback_title="New Chat Session") updated_sessions = get_all_sessions() session_choices = [(title, sid) for sid, title in updated_sessions] return gr.Dropdown(choices=session_choices, value=new_id), [], f"Active Session: {new_id}" def wipe_entire_history(): conn = sqlite3.connect(DB_NAME) cursor = conn.cursor() cursor.execute("DELETE FROM chat_history") cursor.execute("DELETE FROM chat_sessions") conn.commit() conn.close() return create_brand_new_session() # --- Common Core Orchestrator Engine --- def call_llm(provider, api_key, model_choice, system_prompt, history_messages): # CRITICAL RE-STRUCTURE: Ensure the primary system card is at index 0 clean_messages = [{"role": "system", "content": system_prompt}] # Strip any potential hidden formatting metadata from past list items for item in history_messages: if item.get("role") in ["user", "assistant"]: clean_messages.append({ "role": item["role"], "content": str(item["content"]) }) if provider == "Groq": client = Groq(api_key=api_key) response = client.chat.completions.create(model=model_choice, messages=clean_messages, temperature=0.0) return response.choices[0].message.content.strip() elif provider == "OpenRouter": headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} payload = {"model": model_choice, "messages": clean_messages, "temperature": 0.0} response = requests.post("https://openrouter.ai/api/v1/chat/completions", headers=headers, json=payload) if response.status_code == 200: return response.json()['choices'][0]['message']['content'].strip() # Parse detailed error fields if provided out of target pipeline try: err_details = response.json() msg = err_details.get("error", {}).get("message", response.text) except Exception: msg = response.text raise Exception(f"OpenRouter Gateway Alert ({response.status_code}): {msg}") # --- Tab 1: Natural-Language-to-SQL Desk --- def execute_ai_query(provider, api_key, model_choice, user_question): if not api_key or not api_key.strip(): return None, "⚠️ Authentication missing. Input API Key on the Global Controller panel." system_prompt = ( "You are an expert SQL assistant. The database table is called 'project_tasks' with columns: " "[id, timestamp, phase, task_name, owner, timeline, priority]. Return ONLY a valid executable raw SQLite query." ) try: query_as_history = [{"role": "user", "content": user_question}] sql_query = call_llm(provider, api_key, model_choice, system_prompt, query_as_history) sql_query = sql_query.replace("```sql", "").replace("```", "").replace("`", "").strip() conn = sqlite3.connect(DB_NAME) df = pd.read_sql_query(sql_query, conn) conn.close() return df, f"⚡ **Executed Query:** `{sql_query}`" except Exception as e: return None, f"❌ Execution Error: {str(e)}" # --- Tab 2: ChatGPT-Like Multi-Step Chatbot Logic --- def append_user_message(chat_history, user_msg, current_session_id): if not user_msg or not user_msg.strip(): return chat_history, "", gr.Dropdown() # Dynamically update the thread title using the first chat interaction text slice is_first_msg = (len([m for m in chat_history if m['role'] in ['user', 'assistant']]) == 0) fallback_title = user_msg[:22] + "..." if is_first_msg else None save_chat_turn(current_session_id, "user", user_msg, prompt_fallback_title=fallback_title) chat_history.append({"role": "user", "content": user_msg}) updated_sessions = get_all_sessions() session_choices = [(title, sid) for sid, title in updated_sessions] return chat_history, "", gr.Dropdown(choices=session_choices, value=current_session_id) def generate_agent_response(chat_history, provider, api_key, model_choice, current_session_id): if not chat_history: return chat_history if not api_key or not api_key.strip(): err_msg = "⚠️ Authentication missing. Input API Key on the global controls tower." chat_history.append({"role": "assistant", "content": err_msg}) return chat_history system_prompt = ( "You are an expert Pharma Business Analyst for Strides Pharmaceuticals.\n" "Your objective is to ingest audit data, regulatory compliance case studies, and process gaps to generate formal BRD/URD clauses.\n" "If you have details to log a requirement entry, you MUST append a raw JSON block at the very end of your response exactly formatted as:\n" "||JSON_DATA: {\"phase\": \"...\", \"task_name\": \"...\", \"owner\": \"...\", \"timeline\": \"...\", \"priority\": \"...\"}||" ) try: raw_response = call_llm(provider, api_key, model_choice, system_prompt, chat_history) cleaned_response = raw_response database_committed_alert = "" if "||JSON_DATA:" in raw_response: try: parts = raw_response.split("||JSON_DATA:") cleaned_response = parts[0].strip() json_string = parts[1].replace("||", "").strip() data_payload = json.loads(json_string) conn = sqlite3.connect(DB_NAME) cursor = conn.cursor() cursor.execute( "INSERT INTO project_tasks (phase, task_name, owner, timeline, priority) VALUES (?, ?, ?, ?, ?)", (data_payload.get('phase'), data_payload.get('task_name'), data_payload.get('owner'), data_payload.get('timeline'), data_payload.get('priority')) ) conn.commit() conn.close() database_committed_alert = f"\n\n⚙️ **[SYSTEM UPDATE]:** Logged requirement '{data_payload.get('task_name')}' into records database." except Exception as inner_err: database_committed_alert = f"\n\n⚠️ **[SYSTEM NOTICE]:** Auto-logging skipped: {str(inner_err)}" final_display_text = cleaned_response + database_committed_alert save_chat_turn(current_session_id, "assistant", final_display_text) chat_history.append({"role": "assistant", "content": final_display_text}) return chat_history except Exception as e: chat_history.append({"role": "assistant", "content": f"❌ API Connection Failure: {str(e)}"}) return chat_history # --- Build Gradio 6.0 Workspace Layout --- with gr.Blocks() as demo: gr.Markdown("# 💊 Strides Pharma AI: Enterprise Requirement Hub") session_list_data = get_all_sessions() initial_session_id = session_list_data[0][0] initial_history = load_session_history(initial_session_id) current_session_state = gr.State(value=initial_session_id) with gr.Row(): # --- LEFT SIDEBAR PANEL --- with gr.Column(scale=1): gr.Markdown("### 🕒 Session Explorer") session_dropdown = gr.Dropdown( choices=[(title, sid) for sid, title in session_list_data], value=initial_session_id, label="Select Conversation Thread", interactive=True ) new_chat_btn = gr.Button("➕ New Chat Thread", variant="primary") clear_all_btn = gr.Button("🗑️ Clear Entire History Index", variant="stop") gr.Markdown("---") gr.Markdown("### 🔑 Gateway Settings") provider_select = gr.Dropdown(choices=["Groq", "OpenRouter"], value="Groq", label="API Gateway") token_input = gr.Textbox(label="User API Secret Key", type="password") model_select = gr.Dropdown(choices=GROQ_MODELS, value=GROQ_MODELS[0], label="Target Engine") status_session_lbl = gr.Markdown(f"Active Session: {initial_session_id}") # --- RIGHT MAIN APP FRAME WORKSPACE --- with gr.Column(scale=2): with gr.Tabs(): # Tab 1: ChatGPT Workspace Environment with gr.TabItem("🤖 BRD / URD Requirement Workspace"): chatbot_viewport = gr.Chatbot(value=initial_history, label="Interactive Audit Log Narrative") chat_input = gr.Textbox(placeholder="Type messages here...", label="Your Input") send_btn = gr.Button("Submit Message", variant="primary") send_btn.click( fn=append_user_message, inputs=[chatbot_viewport, chat_input, current_session_state], outputs=[chatbot_viewport, chat_input, session_dropdown], api_name=False ).then( fn=generate_agent_response, inputs=[chatbot_viewport, provider_select, token_input, model_select, current_session_state], outputs=[chatbot_viewport], api_name=False ) chat_input.submit( fn=append_user_message, inputs=[chatbot_viewport, chat_input, current_session_state], outputs=[chatbot_viewport, chat_input, session_dropdown], api_name=False ).then( fn=generate_agent_response, inputs=[chatbot_viewport, provider_select, token_input, model_select, current_session_state], outputs=[chatbot_viewport], api_name=False ) # Tab 2: Natural-Language-to-SQL Inventory Subsystem Node with gr.TabItem("🔎 Requirement Inventory Explorer"): query_input = gr.Textbox(label="Query inventory using plain text:") query_btn = gr.Button("Evaluate Subsystem Node", variant="secondary") sql_status_display = gr.Markdown() output_data_table = gr.DataFrame(label="Database Table View") query_btn.click( fn=execute_ai_query, inputs=[provider_select, token_input, model_select, query_input], outputs=[output_data_table, sql_status_display], api_name=False ) # --- STATE EVENT LISTENERS --- session_dropdown.change( fn=handle_session_switch, inputs=[session_dropdown], outputs=[chatbot_viewport, status_session_lbl], api_name=False ).then( fn=lambda val: val, inputs=[session_dropdown], outputs=[current_session_state], api_name=False ) new_chat_btn.click( fn=create_brand_new_session, inputs=[], outputs=[session_dropdown, chatbot_viewport, status_session_lbl], api_name=False ) clear_all_btn.click( fn=wipe_entire_history, inputs=[], outputs=[session_dropdown, chatbot_viewport, status_session_lbl], api_name=False ) provider_select.change( fn=change_provider_options, inputs=[provider_select], outputs=[model_select], api_name=False ) if __name__ == "__main__": demo.launch(theme=gr.themes.Soft())