# app.py (Gradio) — Delight AI Agent Prototype + RFQ Simulation + Business Impact Dashboard # ✅ Stable on Hugging Face Gradio Spaces # ✅ Dummy supplier DB (~50) stored in /tmp # ✅ Inquiry → extract → shortlist → RFQ simulation → pick best → quote draft → download # ✅ Business Impact dashboard uses fixed assumptions (25–30 inquiries/day, 10 suppliers, 10 mins each) import os import re import random import uuid from datetime import datetime, timedelta import pandas as pd import gradio as gr # ---------------------------- # Config # ---------------------------- random.seed(42) DATA_PATH = os.environ.get("SUPPLIER_DB_PATH", "/tmp/supplier_db.csv") PRODUCT_CATALOG = [ ("LED Panel", ["panel", "led panel", "ceiling panel"], (14, 28)), ("Downlight", ["downlight", "spot", "spotlight"], (6, 18)), ("Flood Light", ["flood", "floodlight"], (18, 55)), ("High Bay", ["high bay", "warehouse bay"], (35, 120)), ("Tube Light", ["tube", "t8", "batten"], (4, 12)), ("Track Light", ["track", "rail light"], (10, 30)), ("Street Light", ["street", "road light"], (40, 160)), ("LED Strip", ["strip", "led strip", "tape"], (3, 15)), ] BRANDS = ["Philips", "Osram", "Panasonic", "Schneider", "Opple", "NVC", "Crompton", "Wipro", "Havells", "Generic"] REGIONS = ["SG Central", "SG East", "SG West", "SG North", "Johor", "KL", "Batam"] # ---------------------------- # Business assumptions (fixed for demo) # ---------------------------- INQUIRIES_PER_DAY_LOW = 25 INQUIRIES_PER_DAY_HIGH = 30 SUPPLIERS_PER_INQUIRY = 10 MINUTES_PER_SUPPLIER = 10 # write email + follow up + read/understand response MANUAL_MINUTES_PER_INQUIRY = SUPPLIERS_PER_INQUIRY * MINUTES_PER_SUPPLIER # 100 mins AI_MINUTES_PER_INQUIRY = 5 # conservative: review + clicks + draft quote WORKING_DAYS_PER_MONTH = 22 # ---------------------------- # Dummy Supplier DB # ---------------------------- def make_supplier_name(i: int) -> str: prefixes = ["Bright", "Nova", "Apex", "Luma", "Spark", "Prime", "Zen", "Vertex", "Delta", "Orion"] suffixes = ["Lighting", "Electrics", "Solutions", "Supply", "Traders", "Distributors", "Imports", "Wholesale", "Mart", "Hub"] return f"{random.choice(prefixes)} {random.choice(suffixes)} Pte Ltd #{i:02d}" def generate_supplier_db(n_suppliers: int = 50) -> pd.DataFrame: rows = [] for i in range(1, n_suppliers + 1): supplier = make_supplier_name(i) region = random.choice(REGIONS) supported_categories = random.sample([c[0] for c in PRODUCT_CATALOG], k=random.randint(2, 4)) reliability = round(random.uniform(0.60, 0.98), 2) lead_days = random.randint(2, 21) moq = random.choice([1, 5, 10, 20, 30, 50]) competitiveness = round(random.uniform(0.85, 1.20), 2) brands_supported = random.sample(BRANDS, k=random.randint(2, 5)) rows.append({ "supplier_id": f"SUP-{1000+i}", "supplier_name": supplier, "region": region, "supported_categories": "|".join(supported_categories), "brands_supported": "|".join(brands_supported), "reliability_score": reliability, "lead_time_days": lead_days, "moq": moq, "price_competitiveness_factor": competitiveness, "contact_email": f"sales{i:02d}@example-supplier.com", "last_updated": (datetime.today() - timedelta(days=random.randint(0, 60))).strftime("%Y-%m-%d"), }) return pd.DataFrame(rows) def load_or_create_db() -> pd.DataFrame: if os.path.exists(DATA_PATH): try: return pd.read_csv(DATA_PATH) except Exception: pass df = generate_supplier_db(50) df.to_csv(DATA_PATH, index=False) return df SUPPLIERS_DF = load_or_create_db() def regenerate_db(): global SUPPLIERS_DF SUPPLIERS_DF = generate_supplier_db(50) SUPPLIERS_DF.to_csv(DATA_PATH, index=False) return SUPPLIERS_DF.head(10), DATA_PATH # ---------------------------- # Parsing # ---------------------------- def normalize_text(t: str) -> str: return re.sub(r"\s+", " ", (t or "").strip().lower()) def detect_quantity(text: str): patterns = [ r"\bqty[:\s]*([0-9]{1,5})\b", r"\bquantity[:\s]*([0-9]{1,5})\b", r"\b([0-9]{1,5})\s*(pcs|pc|pieces|nos|units)\b", ] for p in patterns: m = re.search(p, text, flags=re.IGNORECASE) if m: return int(m.group(1)) return None def detect_wattage(text: str): m = re.search(r"\b([0-9]{1,4})\s*(w|watt|watts)\b", text, flags=re.IGNORECASE) return int(m.group(1)) if m else None def detect_brand(text: str): t = (text or "").lower() for b in BRANDS: if b.lower() in t: return b return None def detect_category(text: str): t = normalize_text(text) for category, keywords, _rng in PRODUCT_CATALOG: for kw in keywords: if kw in t: return category return None def detect_location(text: str): t = normalize_text(text) loc_map = { "singapore": "SG", "sg": "SG", "jurong": "SG West", "tampines": "SG East", "woodlands": "SG North", "batam": "Batam", "johor": "Johor", "kuala lumpur": "KL", "kl": "KL", } for k, v in loc_map.items(): if k in t: return v return None def parse_inquiry(text: str) -> dict: return { "quantity": detect_quantity(text) or 10, "wattage": detect_wattage(text), "brand": detect_brand(text), "category": detect_category(text), "location": detect_location(text), } def estimate_market_range(category: str | None, wattage: int | None): if not category: return (10.0, 40.0) base = None for c, _kw, rng in PRODUCT_CATALOG: if c == category: base = rng break if not base: return (10.0, 40.0) lo, hi = base if wattage: scale = min(2.0, max(0.7, wattage / 18.0)) lo = lo * (0.85 + 0.15 * scale) hi = hi * (0.85 + 0.20 * scale) return (round(lo, 2), round(hi, 2)) def pick_margin(pricing_mode: str, base_margin: float): if pricing_mode == "Competitive": return max(5, base_margin - 6) if pricing_mode == "High Margin": return min(40, base_margin + 8) return base_margin # ---------------------------- # Offers + RFQ Simulation # ---------------------------- def compute_offers(req: dict, suppliers: pd.DataFrame, margin_pct: float): category = req.get("category") brand = req.get("brand") qty = int(req.get("quantity") or 10) candidates = suppliers.copy() if category: candidates = candidates[candidates["supported_categories"].astype(str).str.contains(category, na=False)] if brand: bm = candidates["brands_supported"].astype(str).str.contains(brand, na=False) if bm.sum() > 0: candidates = candidates[bm] candidates = candidates[candidates["moq"].fillna(1).astype(int) <= qty] if candidates.empty: return pd.DataFrame(), None market_lo, market_hi = estimate_market_range(category, req.get("wattage")) market_mid = (market_lo + market_hi) / 2 rows = [] for _, s in candidates.iterrows(): factor = float(s["price_competitiveness_factor"]) supplier_cost_est = market_mid * factor * random.uniform(0.92, 1.06) region = str(s["region"]) if region in ["Johor", "KL", "Batam"]: supplier_cost_est *= 1.05 supplier_cost_est = round(supplier_cost_est, 2) sell_price = round(supplier_cost_est / (1 - margin_pct / 100.0), 2) reliability = float(s["reliability_score"]) lead = int(s["lead_time_days"]) score = (1 / max(sell_price, 0.01)) * 100 + reliability * 10 + (1 / max(lead, 1)) * 5 rows.append({ "supplier_id": s["supplier_id"], "supplier_name": s["supplier_name"], "region": s["region"], "reliability_score": reliability, "lead_time_days": lead, "moq": int(s["moq"]), "est_supplier_cost_sgd": supplier_cost_est, "recommended_sell_price_sgd": sell_price, "score": round(score, 4), "contact_email": s["contact_email"], }) offers = pd.DataFrame(rows).sort_values("score", ascending=False).head(10).reset_index(drop=True) return offers, (market_lo, market_hi) def simulate_rfq(offers_df: pd.DataFrame, qty: int, n: int = 5) -> pd.DataFrame: if offers_df.empty: return pd.DataFrame() top = offers_df.head(n).copy() rfq_rows = [] for _, row in top.iterrows(): reliability = float(row["reliability_score"]) base_cost = float(row["est_supplier_cost_sgd"]) base_lead = int(row["lead_time_days"]) price_variance = (1.5 - reliability) * 0.08 quoted_unit = base_cost * random.uniform(1 - price_variance, 1 + price_variance) quoted_unit = round(quoted_unit, 2) if qty >= 100: quoted_unit = round(quoted_unit * 0.97, 2) elif qty >= 50: quoted_unit = round(quoted_unit * 0.985, 2) lead = base_lead + random.choice([-2, -1, 0, 1, 2]) lead = max(2, lead) response_hours = int(max(1, (1.2 - reliability) * random.uniform(3, 12))) rfq_rows.append({ "rfq_id": f"RFQ-{uuid.uuid4().hex[:6].upper()}", "supplier_name": row["supplier_name"], "supplier_id": row["supplier_id"], "region": row["region"], "reliability_score": reliability, "quoted_unit_price_sgd": quoted_unit, "quoted_total_sgd": round(quoted_unit * qty, 2), "quoted_lead_days": lead, "response_time_hours": response_hours, }) rfq_df = pd.DataFrame(rfq_rows) rfq_df = rfq_df.sort_values( ["quoted_unit_price_sgd", "quoted_lead_days", "reliability_score"], ascending=[True, True, False] ).reset_index(drop=True) rfq_df.insert(0, "rank", range(1, len(rfq_df) + 1)) return rfq_df def build_quote_text(req, unit_price_sgd, lead_days, pricing_mode, margin_used, company_name, customer_name, valid_days): qty = int(req.get("quantity") or 10) category = req.get("category") or "Lighting Product" brand = req.get("brand") or "Brand-agnostic" wattage = f"{req.get('wattage')}W" if req.get("wattage") else "" total = round(unit_price_sgd * qty, 2) valid_until = (datetime.today() + timedelta(days=int(valid_days))).strftime("%Y-%m-%d") return f"""Subject: Quotation - {brand} {wattage} {category} (Qty: {qty}) Hi {customer_name}, Thanks for your inquiry. Please find our quotation below: Item: {brand} {wattage} {category} Quantity: {qty} Unit Price: SGD {unit_price_sgd:.2f} Total: SGD {total:.2f} Estimated Lead Time: {int(lead_days)} days Validity: Until {valid_until} Pricing Mode: {pricing_mode} (Margin applied: {margin_used:.0f}%) Regards, Sales Team {company_name} """ # ---------------------------- # Business Impact Dashboard (fixed assumptions) # ---------------------------- def _hours(mins: float) -> float: return round(mins / 60.0, 2) def compute_business_impact(qty: int, internal_best_unit_cost: float | None, rfq_best_unit_cost: float | None): # Time (manual vs AI) per inquiry manual_min_inq = MANUAL_MINUTES_PER_INQUIRY ai_min_inq = AI_MINUTES_PER_INQUIRY saved_min_inq = max(0, manual_min_inq - ai_min_inq) # Daily ranges (25–30 inquiries) daily_manual_low = _hours(manual_min_inq * INQUIRIES_PER_DAY_LOW) daily_manual_high = _hours(manual_min_inq * INQUIRIES_PER_DAY_HIGH) daily_ai_low = _hours(ai_min_inq * INQUIRIES_PER_DAY_LOW) daily_ai_high = _hours(ai_min_inq * INQUIRIES_PER_DAY_HIGH) daily_saved_low = _hours(saved_min_inq * INQUIRIES_PER_DAY_LOW) daily_saved_high = _hours(saved_min_inq * INQUIRIES_PER_DAY_HIGH) # Monthly ranges (22 working days) monthly_saved_low = round(daily_saved_low * WORKING_DAYS_PER_MONTH, 2) monthly_saved_high = round(daily_saved_high * WORKING_DAYS_PER_MONTH, 2) # Effort: supplier interactions avoided (emails/threads processed) supplier_interactions_daily_low = INQUIRIES_PER_DAY_LOW * SUPPLIERS_PER_INQUIRY supplier_interactions_daily_high = INQUIRIES_PER_DAY_HIGH * SUPPLIERS_PER_INQUIRY # Cost savings (only after RFQ simulation) cost_savings_per_unit = None order_savings = None if internal_best_unit_cost is not None and rfq_best_unit_cost is not None: cost_savings_per_unit = round(internal_best_unit_cost - rfq_best_unit_cost, 2) order_savings = round(cost_savings_per_unit * qty, 2) return { "assumptions": { "inquiries_per_day": f"{INQUIRIES_PER_DAY_LOW}–{INQUIRIES_PER_DAY_HIGH}", "suppliers_checked_per_inquiry": SUPPLIERS_PER_INQUIRY, "minutes_per_supplier_email_cycle": MINUTES_PER_SUPPLIER, "manual_minutes_per_inquiry": manual_min_inq, "ai_minutes_per_inquiry": ai_min_inq, "working_days_per_month": WORKING_DAYS_PER_MONTH, }, "time_impact": { "daily_manual_hours": f"{daily_manual_low}–{daily_manual_high}", "daily_ai_hours": f"{daily_ai_low}–{daily_ai_high}", "daily_hours_saved": f"{daily_saved_low}–{daily_saved_high}", "monthly_hours_saved": f"{monthly_saved_low}–{monthly_saved_high}", }, "effort_impact": { "supplier_interactions_avoided_per_day": f"{supplier_interactions_daily_low}–{supplier_interactions_daily_high}", "supplier_interactions_avoided_per_month": f"{supplier_interactions_daily_low*WORKING_DAYS_PER_MONTH}–{supplier_interactions_daily_high*WORKING_DAYS_PER_MONTH}", "note": "Interaction = one supplier email cycle (write + follow-up + read/understand reply).", }, "cost_impact": { "cost_savings_per_unit_sgd": cost_savings_per_unit, "order_savings_sgd": order_savings, "note": "Cost impact shown only after RFQ simulation compares internal estimate vs RFQ best quote.", } } # ---------------------------- # Agent runners # ---------------------------- def run_agent(inquiry_text, base_margin, pricing_mode, top_n, company_name, customer_name, valid_days): req = parse_inquiry(inquiry_text) margin_used = pick_margin(pricing_mode, float(base_margin)) offers_df, market_rng = compute_offers(req, SUPPLIERS_DF, margin_used) if market_rng: market_text = f"Estimated market range: SGD {market_rng[0]:.2f} – {market_rng[1]:.2f} per unit" else: lo, hi = estimate_market_range(req.get("category"), req.get("wattage")) market_text = f"Estimated market range: SGD {lo:.2f} – {hi:.2f} per unit" steps = [] steps.append(f"Step 1 — Extracted requirement: {req}") steps.append(f"Step 2 — Market intelligence: {market_text}") steps.append(f"Step 3 — Pricing mode: {pricing_mode} | Margin applied: {margin_used:.0f}%") # Default empty RFQ table empty_rfq = pd.DataFrame(columns=[ "rank","rfq_id","supplier_name","supplier_id","region","reliability_score", "quoted_unit_price_sgd","quoted_total_sgd","quoted_lead_days","response_time_hours" ]) if offers_df.empty: steps.append("Step 4 — No internal matches found → New Product Mode.") quote_text = ( "No matching suppliers found.\n\n" "New Product Mode:\n" "1) Research market range online\n" "2) Identify supplier categories\n" "3) Send RFQs to shortlisted suppliers\n" "4) Update internal catalog once confirmed\n" ) empty_offers = pd.DataFrame(columns=[ "supplier_id","supplier_name","region","reliability_score","lead_time_days","moq", "est_supplier_cost_sgd","recommended_sell_price_sgd","score","contact_email" ]) impact = compute_business_impact(qty=int(req.get("quantity") or 10), internal_best_unit_cost=None, rfq_best_unit_cost=None) quote_path = "/tmp/quote_draft.txt" with open(quote_path, "w", encoding="utf-8") as f: f.write(quote_text) return req, "\n".join(steps), market_text, empty_offers, empty_rfq, "", impact, quote_text, quote_path, DATA_PATH steps.append(f"Step 4 — Shortlisted top {int(top_n)} suppliers from internal DB.") offers_view = offers_df.head(int(top_n)).copy() # Internal estimate recommendation best = offers_view.iloc[0] internal_best_unit_cost = float(best["est_supplier_cost_sgd"]) unit_price = round(internal_best_unit_cost / (1 - margin_used / 100.0), 2) lead_days = int(best["lead_time_days"]) quote_text = build_quote_text(req, unit_price, lead_days, pricing_mode, margin_used, company_name, customer_name, valid_days) quote_path = "/tmp/quote_draft.txt" with open(quote_path, "w", encoding="utf-8") as f: f.write(quote_text) summary = ( f"Current recommendation (internal estimate): {best['supplier_name']} | " f"Est Cost SGD {internal_best_unit_cost:.2f}/unit → Sell SGD {unit_price:.2f}/unit | Lead {lead_days} days" ) # Dashboard (cost impact not shown yet until RFQ run) impact = compute_business_impact( qty=int(req.get("quantity") or 10), internal_best_unit_cost=internal_best_unit_cost, rfq_best_unit_cost=None ) return req, "\n".join(steps), market_text, offers_view, empty_rfq, summary, impact, quote_text, quote_path, DATA_PATH def run_rfq_simulation(req, offers_df, base_margin, pricing_mode, company_name, customer_name, valid_days): if offers_df is None or len(offers_df) == 0: empty = pd.DataFrame() return empty, "No offers available for RFQ simulation.", "", {}, "", None qty = int(req.get("quantity") or 10) margin_used = pick_margin(pricing_mode, float(base_margin)) # offers_df might come back as list-of-lists depending on Gradio; # ensure it's a DataFrame. if not isinstance(offers_df, pd.DataFrame): try: offers_df = pd.DataFrame(offers_df) except Exception: offers_df = pd.DataFrame() if offers_df.empty: empty = pd.DataFrame() return empty, "No offers available for RFQ simulation.", "", {}, "", None rfq_df = simulate_rfq(offers_df, qty=qty, n=min(5, len(offers_df))) best = rfq_df.iloc[0] rfq_best_unit_cost = float(best["quoted_unit_price_sgd"]) rfq_best_lead = int(best["quoted_lead_days"]) # Internal best cost (from internal estimate row 0) internal_best_unit_cost = None if "est_supplier_cost_sgd" in offers_df.columns: internal_best_unit_cost = float(offers_df.iloc[0]["est_supplier_cost_sgd"]) # Update selling price based on RFQ best unit cost sell_unit = round(rfq_best_unit_cost / (1 - margin_used / 100.0), 2) summary = ( f"RFQ best supplier: {best['supplier_name']} | " f"Quoted Cost SGD {rfq_best_unit_cost:.2f}/unit → Sell SGD {sell_unit:.2f}/unit | " f"Lead {rfq_best_lead} days | Response ~{int(best['response_time_hours'])}h" ) quote_text = build_quote_text(req, sell_unit, rfq_best_lead, pricing_mode, margin_used, company_name, customer_name, valid_days) quote_path = "/tmp/quote_draft.txt" with open(quote_path, "w", encoding="utf-8") as f: f.write(quote_text) impact = compute_business_impact( qty=qty, internal_best_unit_cost=internal_best_unit_cost, rfq_best_unit_cost=rfq_best_unit_cost ) return rfq_df, "✅ RFQ simulation completed. Best quote selected.", summary, impact, quote_text, quote_path # ---------------------------- # UI (Gradio) # ---------------------------- with gr.Blocks(title="Delight AI Agent (Prototype)") as demo: gr.Markdown( """ # 💡 Delight AI Agent (Prototype) Paste a customer inquiry → agent extracts requirement → ranks suppliers → (simulated RFQ) → recommends pricing → generates quotation draft. **Demo assumptions (fixed):** - 25–30 inquiries/day - 10 suppliers checked per inquiry - ~10 mins per supplier email cycle - Manual effort ≈ 100 mins per inquiry """ ) # Inputs inquiry = gr.Textbox( label="Customer Inquiry", lines=6, value="Hi, please quote best price for 50 pcs Philips 18W LED panel light. Delivery to Singapore in 2 weeks." ) with gr.Row(): base_margin = gr.Slider(5, 40, value=20, step=1, label="Base Margin (%)") pricing_mode = gr.Radio(["Balanced", "Competitive", "High Margin"], value="Balanced", label="Pricing Mode") top_n = gr.Slider(3, 10, value=5, step=1, label="Top offers to show") with gr.Row(): company_name = gr.Textbox(label="Your company name", value="Delight Lighting (Demo)") customer_name = gr.Textbox(label="Customer name", value="Customer") valid_days = gr.Slider(1, 30, value=7, step=1, label="Quote validity (days)") # Buttons run_btn = gr.Button("🚀 Run Agent") # Outputs with gr.Row(): parsed_req = gr.JSON(label="Extracted Requirement") agent_steps = gr.Textbox(label="Agent Steps", lines=10) market_info = gr.Textbox(label="Market Intelligence", lines=2) offers_table = gr.Dataframe( label="Supplier Shortlist (Top N)", interactive=False, wrap=True ) gr.Markdown("## 📡 RFQ Simulation (Agentic Step)") rfq_btn = gr.Button("📨 Simulate RFQ to Top Suppliers") rfq_status = gr.Textbox(label="RFQ Status", lines=2) rfq_table = gr.Dataframe(label="RFQ Responses (Simulated)", interactive=False, wrap=True) recommendation = gr.Textbox(label="Recommendation Summary", lines=2) gr.Markdown("## 📊 Business Impact Dashboard (Auto)") impact_output = gr.JSON(label="Business Impact Metrics") quote_text = gr.Textbox(label="Generated Quote Draft", lines=14) with gr.Row(): quote_file = gr.File(label="Download Quote Draft (.txt)") supplier_csv = gr.File(label="Download Supplier DB (.csv)") with gr.Accordion("⚙️ Admin: Regenerate Dummy Supplier DB", open=False): regen_btn = gr.Button("Regenerate DB (50 suppliers)") db_preview = gr.Dataframe(label="DB Preview (Top 10)", interactive=False) db_file = gr.File(label="Download Fresh DB (.csv)") # State state_req = gr.State({}) state_offers = gr.State(pd.DataFrame()) # Wire actions run_btn.click( fn=run_agent, inputs=[inquiry, base_margin, pricing_mode, top_n, company_name, customer_name, valid_days], outputs=[parsed_req, agent_steps, market_info, offers_table, rfq_table, recommendation, impact_output, quote_text, quote_file, supplier_csv], ).then( fn=lambda req, offers: (req, offers), inputs=[parsed_req, offers_table], outputs=[state_req, state_offers], ) rfq_btn.click( fn=run_rfq_simulation, inputs=[state_req, state_offers, base_margin, pricing_mode, company_name, customer_name, valid_days], outputs=[rfq_table, rfq_status, recommendation, impact_output, quote_text, quote_file], ) regen_btn.click( fn=regenerate_db, inputs=[], outputs=[db_preview, db_file], ) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)