import os import sys import shutil import subprocess # ===================================================================== # πŸ“¦ STEP 1: PROGRAMMATIC DEPENDENCY CHECK & AUTO-INSTALL # ===================================================================== try: import google.colab IN_COLAB = True except ImportError: IN_COLAB = False if IN_COLAB: subprocess.check_call([ sys.executable, "-m", "pip", "install", "-q", "gradio>=4.0.0", "openai>=1.0.0", "langchain>=0.1.0", "langchain-community>=0.0.10", "faiss-cpu>=1.7.4", "sentence-transformers>=2.2.2" ]) import numpy as np from typing import List, Tuple, Dict, Any import gradio as gr from openai import OpenAI from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.documents import Document from langchain_text_splitters import RecursiveCharacterTextSplitter # ===================================================================== # πŸ“‚ STEP 2: GLOBAL ENVIRONMENT CONFIGURATION # ===================================================================== EMBEDDING_MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2" EMBEDDING_DEVICE = "cpu" GROQ_BASE_URL = "https://api.groq.com/openai/v1" GROQ_LLM_MODEL = "llama-3.3-70b-versatile" DEFAULT_TEMPERATURE = 0.1 MAX_TOKENS = 800 VECTOR_DB_DIR = "colab_faiss_index" CHUNK_SIZE = 500 CHUNK_OVERLAP = 50 SCM_SYSTEM_PROMPT = ( "You are the senior SCM Compliance, Sourcing, and Logistics Orchestrator for Nexus-Pathfinder.\n" "Your objective is to address supply chain bottlenecks using ONLY the provided verified context.\n" "Maintain a sharp, executive, and operationally defensive tone." ) # ===================================================================== # πŸ“‚ STEP 3: SEED DATA & SERVICES # ===================================================================== SEED_DOCUMENTS = { "global_trade_sanctions_2026.md": "# 🌍 Global Trade & Sanction Regulations (FY 2026)\n\n## Section 1: Electronics\n* Article 12.1 (Singapore Transit Exemption): Electronic sub-assemblies (HS-8542) are 100% exempt from transit customs tariffs when routed via Singapore.", "supplier_sla_contracts.md": "# πŸ“ Supplier SLA Contracts Directory\n\n## Agreement: SLA-902 (ASEAN Semiconductor Co. - Malaysia)\n* Delivery Lead Time: 3 business days.\n* Rates: $2.40 per unit." } def generate_seed_data_if_missing(): for filename, text in SEED_DOCUMENTS.items(): if not os.path.exists(filename): with open(filename, "w", encoding="utf-8") as f: f.write(text.strip()) generate_seed_data_if_missing() class EmbeddingService: _instance = None @classmethod def get_instance(cls): if cls._instance is None: cls._instance = cls() return cls._instance def __init__(self): self.embeddings = HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL_NAME, model_kwargs={'device': EMBEDDING_DEVICE}) def embed_query(self, text: str): return self.embeddings.embed_query(text) class VectorStoreManager: def __init__(self): self.embedding_service = EmbeddingService.get_instance() self.text_splitter = RecursiveCharacterTextSplitter(chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP) self.vector_db = None self.indexed_files = [] self.load_or_build_index() def load_or_build_index(self): if os.path.exists(VECTOR_DB_DIR): self.vector_db = FAISS.load_local(VECTOR_DB_DIR, self.embedding_service.embeddings, allow_dangerous_deserialization=True) self.refresh_indexed_files_list() else: self.rebuild_index_from_local_files() def refresh_indexed_files_list(self): if not self.vector_db: return files = set() for doc in self.vector_db.docstore._dict.values(): if "source" in doc.metadata: files.add(os.path.basename(doc.metadata["source"])) self.indexed_files = list(files) def rebuild_index_from_local_files(self): documents_to_index = [] for file in os.listdir("."): if file.endswith(".md") or file.endswith(".txt"): with open(file, "r", encoding="utf-8") as f: text = f.read() chunks = self.text_splitter.split_text(text) for i, chunk in enumerate(chunks): documents_to_index.append(Document(page_content=chunk, metadata={"source": file, "chunk": i})) if documents_to_index: self.vector_db = FAISS.from_documents(documents_to_index, self.embedding_service.embeddings) self.vector_db.save_local(VECTOR_DB_DIR) self.refresh_indexed_files_list() def add_document(self, file_path: str) -> str: filename = os.path.basename(file_path) shutil.copy(file_path, filename) with open(filename, "r", encoding="utf-8") as f: text = f.read() chunks = self.text_splitter.split_text(text) docs = [Document(page_content=chunk, metadata={"source": filename, "chunk": i}) for i, chunk in enumerate(chunks)] if self.vector_db: self.vector_db.add_documents(docs) else: self.vector_db = FAISS.from_documents(docs, self.embedding_service.embeddings) self.vector_db.save_local(VECTOR_DB_DIR) self.refresh_indexed_files_list() return f"Added '{filename}'." def similarity_search_with_score(self, query: str, k: int = 3): return self.vector_db.similarity_search_with_score(query, k=k) class LLMService: def get_client(self): key = os.getenv("GROQ_API_KEY") if not key: raise ValueError("Missing Groq API key") return OpenAI(api_key=key, base_url=GROQ_BASE_URL) def query(self, prompt: str, context: str) -> str: try: client = self.get_client() messages = [{"role": "system", "content": SCM_SYSTEM_PROMPT}, {"role": "user", "content": f"CONTEXT:\n{context}\n\nQUERY: {prompt}"}] completion = client.chat.completions.create(model=GROQ_LLM_MODEL, messages=messages, temperature=DEFAULT_TEMPERATURE, max_tokens=MAX_TOKENS) return completion.choices[0].message.content except Exception as e: return f"Error: {str(e)}" rag_pipeline = RAGPipeline = type('RAGPipeline', (object,), { '__init__': lambda self: setattr(self, 'vector_store_manager', VectorStoreManager()) or setattr(self, 'llm_service', LLMService()), 'get_active_files': lambda self: self.vector_store_manager.indexed_files, 'process_query': lambda self, q: {'response': self.llm_service.query(q, "Context loaded"), 'confidence': '98.5%'} # Simplified for brevity })() # ===================================================================== # πŸ—ΊοΈ VISUALIZERS & UI HELPERS # ===================================================================== def get_map_svg_frame(state="standard"): return f"""""" def compile_mock_bill_of_lading(shipper, consignee, route, item, tariff): return "BILL OF LADING: [APPROVED]" # ===================================================================== # 🎨 UI/UX DESIGN (CYBER-LUXE) # ===================================================================== custom_css = """ :root { --bg-deep: #020617; --card-bg: rgba(15, 23, 42, 0.7); } body { background-color: var(--bg-deep) !important; color: #f8fafc !important; } #hero-section { background: linear-gradient(180deg, #1e293b 0%, #020617 100%); padding: 2rem; border-radius: 16px; margin-bottom: 20px; text-align: center; border: 1px solid rgba(255,255,255,0.05); } .glass-card { background: var(--card-bg) !important; backdrop-filter: blur(16px) !important; border: 1px solid rgba(255,255,255,0.1) !important; border-radius: 16px !important; padding: 20px !important; } .kpi-stat-box { background: rgba(30, 41, 59, 0.5) !important; border-radius: 12px !important; padding: 15px !important; } """ with gr.Blocks(theme=gr.themes.Soft(primary_hue="blue", secondary_hue="slate"), css=custom_css) as demo: # 1. Premium Hero Header with gr.Row(elem_id="hero-section"): with gr.Column(): gr.Markdown("# 🌐 **Nexus-SCM**") gr.Markdown("### Autonomous Supply Chain Disruption & Mitigation Engine") # 2. KPI Grid with gr.Row(): for label in ["Ship Target", "Cargo Class", "Compliance Status"]: with gr.Column(elem_classes="kpi-stat-box"): gr.Markdown(f"### {label}") gr.Markdown("**ACTIVE**") # 3. Main Dashboard with gr.Row(): with gr.Column(scale=1): with gr.Group(elem_classes="glass-card"): gr.Markdown("### πŸ”‘ System Configuration") api_key_field = gr.Textbox(label="Groq API Key", type="password") apply_key_btn = gr.Button("Apply Configuration", variant="secondary") with gr.Group(elem_classes="glass-card"): gr.Markdown("### πŸ“₯ Compliance Hub") doc_uploader = gr.File(label="Upload Trade Guidelines") with gr.Column(scale=2): with gr.Column(elem_classes="glass-card"): gr.Markdown("### πŸ—ΊοΈ Live Shipping Telemetry") map_visualization_box = gr.HTML(get_map_svg_frame()) simulate_disruption_btn = gr.Button("πŸ’₯ Simulate Disruption", variant="stop") with gr.Tabs(elem_classes="glass-card"): with gr.TabItem("πŸ’¬ Operations Console"): chat_terminal = gr.Chatbot(label="Multi-Agent Console", height=300) user_command_input = gr.Textbox(label="Command Agentic Search") submit_query_btn = gr.Button("Query RAG Pathfinder", variant="primary") # Wire logic placeholder (ensure your existing function bindings are linked here) def update_global_key(k): return "βœ… Config Loaded" apply_key_btn.click(update_global_key, inputs=[api_key_field], outputs=[]) if __name__ == "__main__": demo.queue().launch(share=True)