""" Agentic AI Deal Hunter & Price Estimator — Gradio app for Hugging Face Spaces. Wraps the existing agents/ package (Specialist -> Modal, Frontier -> Gemini+Chroma RAG, NeuralNetworkAgent -> local residual MLP, ScannerAgent -> RSS deals) behind an interactive UI, with every external dependency (Modal, Gemini, Chroma, HF weights) wrapped so a missing/misconfigured piece degrades gracefully instead of crashing the whole Space. """ import os import shutil import logging import traceback # --------------------------------------------------------------------------- # 0. Environment setup — MUST happen before any `agents.*` module is imported, # since agents/preprocessor.py reads its default model from the env at # import time. On a Space there's no local Ollama server, so we point the # preprocessor at Gemini (same key already used by the other agents). # --------------------------------------------------------------------------- os.environ.setdefault("PRICER_PREPROCESSOR_MODEL", "gemini/gemini-3.5-flash") logging.basicConfig(level=logging.INFO) log = logging.getLogger("space") import gradio as gr import pandas as pd # Optional ZeroGPU support: if the Space has ZeroGPU hardware, wrap the neural # network call so it gets a GPU slice; otherwise fall back to a no-op decorator. try: import spaces GPU_DECORATOR = spaces.GPU except Exception: # not running on a ZeroGPU Space, or package unavailable def GPU_DECORATOR(fn=None, **kwargs): if fn is None: return lambda f: f return fn # --------------------------------------------------------------------------- # 1. Fetch the neural-network weights from the Hub if not already present. # Your DeepNeuralNetworkInference.load() expects a relative path # "deep_neural_network.pth" in the working directory. # --------------------------------------------------------------------------- NN_REPO_ID = os.environ.get("NN_REPO_ID", "Md-Asif/Price-N_N") NN_LOCAL_PATH = "deep_neural_network.pth" def ensure_nn_weights(): if os.path.exists(NN_LOCAL_PATH): return NN_LOCAL_PATH from huggingface_hub import hf_hub_download # Try a couple of likely filenames on the repo since we can't be 100% sure # of the exact filename used when it was pushed. candidates = ["deep_neural_network.pth", "pytorch_model.pth", "model.pth"] last_err = None for fname in candidates: try: path = hf_hub_download(repo_id=NN_REPO_ID, filename=fname) shutil.copy(path, NN_LOCAL_PATH) return NN_LOCAL_PATH except Exception as e: # noqa: BLE001 last_err = e continue raise RuntimeError(f"Could not download NN weights from {NN_REPO_ID}: {last_err}") # --------------------------------------------------------------------------- # 2. Chroma vectorstore for the Frontier agent's RAG context. If you upload # your persisted Chroma directory alongside this app (e.g. as # `products_vectorstore/`), it will be picked up automatically. If it's # missing, the Frontier agent still works — it just runs without # retrieved comparables (empty context), which is a reasonable fallback. # --------------------------------------------------------------------------- CHROMA_PATH = os.environ.get("CHROMA_PATH", "products_vectorstore") CHROMA_COLLECTION_NAME = os.environ.get("CHROMA_COLLECTION", "products") def get_chroma_collection(): import chromadb client = chromadb.PersistentClient(path=CHROMA_PATH) try: return client.get_collection(CHROMA_COLLECTION_NAME) except Exception: return client.get_or_create_collection(CHROMA_COLLECTION_NAME) # --------------------------------------------------------------------------- # 3. Lazily build each agent, catching errors independently so one broken # dependency (no Modal deploy, no API key, no vectorstore) doesn't take # down the others. # --------------------------------------------------------------------------- STATE = { "preprocessor": None, "specialist": None, "frontier": None, "neural_network": None, "scanner": None, "messenger": None, "collection": None, "status": {}, } # Only notify Pushover if the estimated discount clears this bar — same idea # as PlanningAgent.DEAL_THRESHOLD in your original code. DEAL_THRESHOLD = float(os.environ.get("DEAL_THRESHOLD", "50")) def init_agents(): status = {} # Preprocessor (Gemini rewrite step) try: from agents.preprocessor import Preprocessor STATE["preprocessor"] = Preprocessor() status["Preprocessor (Gemini)"] = "✅ ready" except Exception as e: # noqa: BLE001 status["Preprocessor (Gemini)"] = f"⚠️ unavailable ({e})" # Specialist (Modal fine-tuned model) try: from agents.specialist_agent import SpecialistAgent STATE["specialist"] = SpecialistAgent() status["Specialist (Modal)"] = "✅ ready" except Exception as e: # noqa: BLE001 status["Specialist (Modal)"] = f"⚠️ unavailable ({e})" # Chroma collection, needed by Frontier try: STATE["collection"] = get_chroma_collection() count = STATE["collection"].count() status["Chroma vectorstore"] = f"✅ loaded ({count} items)" except Exception as e: # noqa: BLE001 status["Chroma vectorstore"] = f"⚠️ unavailable ({e})" # Frontier (Gemini + RAG) try: from agents.frontier_agent import FrontierAgent STATE["frontier"] = FrontierAgent(STATE["collection"]) status["Frontier (Gemini + RAG)"] = "✅ ready" except Exception as e: # noqa: BLE001 status["Frontier (Gemini + RAG)"] = f"⚠️ unavailable ({e})" # Neural network (local residual MLP) try: ensure_nn_weights() from agents.neural_network_agent import NeuralNetworkAgent STATE["neural_network"] = NeuralNetworkAgent() status["Neural Network"] = "✅ ready" except Exception as e: # noqa: BLE001 status["Neural Network"] = f"⚠️ unavailable ({e})" # Scanner (RSS deal feed) try: from agents.scanner_agent import ScannerAgent STATE["scanner"] = ScannerAgent() status["Scanner (RSS + Gemini)"] = "✅ ready" except Exception as e: # noqa: BLE001 status["Scanner (RSS + Gemini)"] = f"⚠️ unavailable ({e})" # Messenger (Pushover) try: from agents.messaging_agent import MessagingAgent STATE["messenger"] = MessagingAgent() status["Messenger (Pushover)"] = "✅ ready" except Exception as e: # noqa: BLE001 status["Messenger (Pushover)"] = f"⚠️ unavailable ({e})" STATE["status"] = status return status def status_markdown(): lines = ["### System status"] for name, val in STATE["status"].items(): lines.append(f"- **{name}**: {val}") if not any("✅" in v for v in STATE["status"].values()): lines.append( "\n⚠️ No agents initialized. Check your Space secrets " "(`GOOGLE_API_KEY`, `MODAL_TOKEN_ID`, `MODAL_TOKEN_SECRET`, " "`PUSHOVER_USER`, `PUSHOVER_TOKEN`)." ) return "\n".join(lines) # --------------------------------------------------------------------------- # 4. Weighted ensemble that degrades gracefully — same 0.8/0.1/0.1 weighting # as your EnsembleAgent, but renormalized over whichever sub-agents # actually returned a result. # --------------------------------------------------------------------------- BASE_WEIGHTS = {"frontier": 0.8, "specialist": 0.1, "neural_network": 0.1} @GPU_DECORATOR def run_neural_network(text): return STATE["neural_network"].price(text) def _price_with_combined(description: str): """Core pricing logic shared by the Price Estimator tab and the Deal Scanner. Returns (summary_markdown, breakdown_df, combined_price_or_None, rewritten_description).""" if not description or not description.strip(): return "Please enter a product description.", None, None, description rewrite = description if STATE["preprocessor"] is not None: try: rewrite = STATE["preprocessor"].preprocess(description) except Exception as e: # noqa: BLE001 log.warning("Preprocessor failed, using raw text: %s", e) results = {} errors = {} if STATE["specialist"] is not None: try: results["specialist"] = STATE["specialist"].price(rewrite) except Exception as e: # noqa: BLE001 errors["specialist"] = str(e) if STATE["frontier"] is not None: try: results["frontier"] = STATE["frontier"].price(rewrite) except Exception as e: # noqa: BLE001 errors["frontier"] = str(e) if STATE["neural_network"] is not None: try: results["neural_network"] = run_neural_network(rewrite) except Exception as e: # noqa: BLE001 errors["neural_network"] = str(e) if not results: detail = "\n".join(f"- {k}: {v}" for k, v in errors.items()) or "No agents available." return f"❌ All pricing agents failed.\n{detail}", None, None, rewrite weight_sum = sum(BASE_WEIGHTS[k] for k in results) combined = sum(results[k] * BASE_WEIGHTS[k] for k in results) / weight_sum rows = [] label_map = { "specialist": "Specialist (fine-tuned LLM, Modal)", "frontier": "Frontier (Gemini + RAG)", "neural_network": "Neural Network (residual MLP)", } for key, label in label_map.items(): if key in results: rows.append([label, f"${results[key]:,.2f}", f"{BASE_WEIGHTS[key]:.0%}"]) elif key in errors: rows.append([label, "unavailable", "-"]) df = pd.DataFrame(rows, columns=["Agent", "Estimate", "Weight"]) summary = f"## 💰 Estimated price: **${combined:,.2f}**\n\n**Rewritten description used:**\n> {rewrite}" return summary, df, combined, rewrite def estimate_price(description: str): summary, df, _combined, _rewrite = _price_with_combined(description) return summary, df # --------------------------------------------------------------------------- # 5. Deal scanner — pulls live deals from RSS, prices each with the same # graceful ensemble logic, and surfaces the best discounts. # --------------------------------------------------------------------------- def scan_deals(notify_enabled: bool, threshold: float, progress=gr.Progress()): if STATE["scanner"] is None: return "⚠️ Scanner agent is unavailable (check GOOGLE_API_KEY).", None progress(0, desc="Scanning RSS feeds...") try: selection = STATE["scanner"].scan(memory=[]) except Exception as e: # noqa: BLE001 return f"❌ Scan failed: {e}", None if not selection or not selection.deals: return "No qualifying deals found right now — try again shortly.", None rows = [] priced_deals = [] # (deal, estimate, discount) for notification pass n = len(selection.deals[:5]) for i, deal in enumerate(selection.deals[:5]): progress((i + 1) / n, desc=f"Pricing deal {i + 1}/{n}...") _, _df, combined, _rewrite = _price_with_combined(deal.product_description) discount = (combined - deal.price) if combined is not None else None if combined is not None: priced_deals.append((deal, combined, discount)) rows.append( [ deal.product_description[:90] + "...", f"${deal.price:,.2f}", f"${combined:,.2f}" if combined is not None else "n/a", f"${discount:,.2f}" if discount is not None else "n/a", deal.url, ] ) df_out = pd.DataFrame(rows, columns=["Description", "Deal price", "Est. value", "Discount", "URL"]) df_out = df_out.sort_values( by="Discount", key=lambda s: s.str.replace("$", "").replace("n/a", "-999999", regex=False).astype(float), ascending=False, ) status_line = f"Found {len(selection.deals)} candidate deals, priced top {n}." if notify_enabled and priced_deals: if STATE["messenger"] is None: status_line += "\n\n⚠️ Notification skipped — Messenger agent is unavailable (check PUSHOVER_USER / PUSHOVER_TOKEN secrets)." else: best_deal, best_estimate, best_discount = max(priced_deals, key=lambda t: t[2]) if best_discount > threshold: try: STATE["messenger"].notify( best_deal.product_description, best_deal.price, best_estimate, best_deal.url ) status_line += f"\n\n📲 Pushover notification sent for best deal (discount ${best_discount:,.2f})." except Exception as e: # noqa: BLE001 status_line += f"\n\n⚠️ Notification failed: {e}" else: status_line += f"\n\nNo deal cleared the ${threshold:,.2f} discount threshold — no notification sent." return status_line, df_out # --------------------------------------------------------------------------- # 6. Build the UI # --------------------------------------------------------------------------- with gr.Blocks(title="Agentic Price Estimator") as demo: gr.Markdown("# 🤖 Agentic AI Price Estimator & Deal Scanner") gr.Markdown( "An ensemble of a fine-tuned LLM (via Modal), a RAG-grounded frontier model " "(Gemini), and a locally-run residual neural network, coordinated by a " "preprocessing agent. Enter a product description to get a price estimate, " "or scan live deal feeds for bargains." ) status_box = gr.Markdown("Initializing agents...") with gr.Tab("💰 Price Estimator"): desc_input = gr.Textbox( label="Product description", placeholder="e.g. Dell XPS 13 laptop, 16GB RAM, 512GB SSD, Intel i7, 13.4-inch FHD+ display", lines=4, ) estimate_btn = gr.Button("Estimate price", variant="primary") estimate_output = gr.Markdown() breakdown_output = gr.Dataframe(label="Per-agent breakdown", wrap=True) estimate_btn.click( fn=estimate_price, inputs=desc_input, outputs=[estimate_output, breakdown_output] ) with gr.Tab("🔍 Live Deal Scanner"): gr.Markdown( "Scrapes current listings from DealNews RSS feeds, has Gemini select the " "5 with the clearest descriptions and prices, then prices each one with " "the ensemble to surface the best discounts." ) with gr.Row(): notify_toggle = gr.Checkbox( label="Send Pushover notification for the best deal", value=False ) threshold_slider = gr.Slider( label="Minimum discount to notify on ($)", minimum=0, maximum=500, step=10, value=DEAL_THRESHOLD, ) scan_btn = gr.Button("Scan for deals", variant="primary") scan_status = gr.Markdown() scan_output = gr.Dataframe(label="Candidate deals", wrap=True) scan_btn.click( fn=scan_deals, inputs=[notify_toggle, threshold_slider], outputs=[scan_status, scan_output], ) with gr.Tab("⚙️ System status"): refresh_btn = gr.Button("Re-check agent status") detailed_status = gr.Markdown() refresh_btn.click(fn=lambda: status_markdown(), outputs=detailed_status) demo.load(fn=lambda: status_markdown(), outputs=status_box) demo.load(fn=lambda: status_markdown(), outputs=detailed_status) # Initialize agents once at import/startup time (before the server starts # accepting requests), so the status panel is accurate from the first load. try: init_agents() except Exception: # noqa: BLE001 log.error("Agent initialization failed:\n%s", traceback.format_exc()) STATE["status"] = {"Startup": f"❌ fatal error during init — see logs"} if __name__ == "__main__": demo.queue().launch()