"""AgAdvisor — Gradio frontend for Hugging Face Spaces. HF retired the Streamlit SDK, and Gradio/Docker Spaces on cpu-basic now require PRO, so the free path is a Gradio Space on ZeroGPU hardware. This module is a view layer only: it drives the SAME pipeline as src/streamlit_app_conversational.py (parse_query -> ToolMatcher -> ToolExecutor) and the SAME AccountsService (bcrypt auth, per-user daily quota, history synced to a private HF Dataset), so abstention and page-level citations behave identically. The Streamlit app remains the local / EC2 entrypoint; keep the two in sync at the pipeline boundary, not the UI. """ # HF Spaces' free tier for Gradio is ZeroGPU. Its launcher aborts at startup ("No # @spaces.GPU function detected") unless (a) `spaces` is imported BEFORE gradio/torch so # it can install its hooks, and (b) the source contains a literal @spaces.GPU decorator. # This app is CPU-only (all inference is via the OpenAI API), so the probe below only # needs to EXIST — it is never called. `spaces` isn't a local dev dependency (HF injects # it in the ZeroGPU build), so synthesize a no-op stub when absent, keeping the literal # @spaces.GPU decorator so ZeroGPU's source scan still detects it on the Space. try: import spaces # provided by the HF ZeroGPU runtime; MUST precede `import gradio` except Exception: # local / non-ZeroGPU: synthesize a no-op `spaces` so import still works import sys as _sys import types as _types spaces = _types.ModuleType("spaces") spaces.GPU = lambda fn=None, **_kw: (fn if callable(fn) else (lambda f: f)) _sys.modules["spaces"] = spaces @spaces.GPU def _zero_gpu_probe(): """Exists solely so ZeroGPU detects a @spaces.GPU function at startup; unused.""" return None import logging import gradio as gr from src.accounts.service import AccountsService from src.cdms.product_catalog import get_catalog from src.parser import parse_query from src.tools.tool_executor import ToolExecutor from src.tools.tool_matcher import ToolMatcher logger = logging.getLogger(__name__) # Built once per process, not per session — these load models and are the cold-start cost. accounts = AccountsService() tool_matcher = ToolMatcher() tool_executor = ToolExecutor() try: PRODUCTS = sorted(get_catalog().available_products()) except Exception: logger.exception("Could not load the product catalog") PRODUCTS = [] CONTEXT_TURNS = 5 # matches the Streamlit app's last-5-messages window _TOOL_LABELS = { "cdms_label": "CDMS label", "cdms": "CDMS label", "pesticide_label": "CDMS label", "rag": "CDMS label", "documentation": "CDMS label", "weather": "weather", "soil": "soil", "agriculture_web": "agriculture web", "ag_web": "agriculture web", } def _confidence_badge(tool: str, confidence: float) -> str: """Answer confidence/star badge (ISA feedback #3, ported from the Streamlit UI). Plain-markdown so it renders cleanly on mobile — the Streamlit version's badge was CSS-clipped on phones. `confidence` is the tool-router's match score (0-1). """ stars = max(1, min(5, round((confidence or 0.0) * 5))) bar = "★" * stars + "☆" * (5 - stars) label = _TOOL_LABELS.get(tool, tool) return f"\n\n---\n`{bar}` · **{confidence:.0%} confidence** · answered via *{label}*" # --- Response cache (ISA scalability/cost: LLM credits as the user base grows) --- # Agronomists ask the same product questions repeatedly. A cache hit returns the # stored answer and skips the WHOLE pipeline — query embedding, retrieval, and the # OpenAI generation call — so repeats cost nothing. Keyed by the normalized # question; only first-turn questions are cached so follow-ups (which depend on # conversation context) are never wrongly reused. Process-wide + TTL + size cap. import hashlib as _hashlib import time as _time _RESP_CACHE: dict[str, tuple[str, float]] = {} _RESP_CACHE_TTL = 6 * 3600 # 6 hours _RESP_CACHE_MAX = 500 def _cache_key(question: str) -> str: norm = " ".join(question.lower().split()) return _hashlib.sha256(norm.encode()).hexdigest() def _cache_put(key: str, reply: str) -> None: if len(_RESP_CACHE) >= _RESP_CACHE_MAX: # evict oldest _RESP_CACHE.pop(min(_RESP_CACHE, key=lambda k: _RESP_CACHE[k][1]), None) _RESP_CACHE[key] = (reply, _time.time()) def answer(question: str, history: list[dict], on_step=None) -> tuple[str, dict]: """Run one question through the tool pipeline. Mirrors the Streamlit flow. Retrieval is auto-mode (decided in the CDMS tool): serve from the committed index when the label is present, else live-fetch + index + cache it. `on_step(str)` is called at each stage so the UI can show the real process (and make first-time label fetches legible as the slow step). Returns (reply, debug) where debug holds routing/source details for the optional Debug panel. """ def _step(msg: str) -> None: if on_step: try: on_step(msg) except Exception: pass # Cache only first-turn questions (no context to depend on). cacheable = not history if cacheable: _ck = _cache_key(question) _hit = _RESP_CACHE.get(_ck) if _hit and (_time.time() - _hit[1]) < _RESP_CACHE_TTL: _step("⚡ Served from a recent cached answer.") return _hit[0], {"cached": True} context = [ {"role": m["role"], "content": m["content"]} for m in (history or [])[-CONTEXT_TURNS * 2:] ] _step("Understanding your question…") try: keywords = parse_query(question).get("extracted_keywords", []) except Exception: keywords = [] confidence = 0.3 _step("Choosing the right tool…") try: match = tool_matcher.match_tool(keywords, question, conversation_context=context) tool = match["tool_name"] confidence = float(match.get("confidence", 0.3) or 0.3) except Exception: logger.exception("Tool matching failed; falling back to the label tool") tool = "cdms_label" debug = {"tool": tool, "confidence": confidence, "keywords": keywords} try: result = tool_executor.execute( tool_name=tool, user_question=question, conversation_context=context, on_step=on_step, ) except Exception: # Never surface tracebacks to end users; log server-side. logger.exception("Tool execution failed") return "I ran into an unexpected problem answering that. Please try rephrasing.", debug raw = result.get("raw_data", {}) if isinstance(result, dict) else {} debug.update({ "tool_used": result.get("tool_used", tool), "source": raw.get("source", "index"), "chunks": raw.get("total_chunks_found", raw.get("pdfs_indexed")), "success": result.get("success", False), }) if not result.get("success", False): return result.get( "llm_response", "I couldn't find that in my label set." ), debug reply = result.get("llm_response", "I couldn't process that request.") # Show a confidence/star rating with real answers. Not on an abstention — a # confidence score on "I don't have that label" would be misleading. if "don't have the label" not in reply: reply += _confidence_badge(tool, confidence) # Cache only real answers (never abstentions) so a later retry can still # succeed — and only first-turn questions (cacheable). if cacheable: _cache_put(_ck, reply) return reply, debug import queue as _queue import threading as _threading def _format_steps(steps: list[str]) -> str: """Render the live pipeline steps as a small checklist for the accordion.""" if not steps: return "_working…_" return "\n".join(f"- {s}" for s in steps) def _format_debug(debug: dict | None) -> str: """Render the last answer's routing details for the Debug panel.""" if not debug: return "_No debug info yet — ask a question._" if debug.get("cached"): return "**Debug:** served from the in-process response cache (no pipeline run)." rows = [ f"- **Tool:** `{debug.get('tool_used', debug.get('tool', '?'))}`", f"- **Confidence:** {float(debug.get('confidence', 0) or 0):.0%}", f"- **Retrieval source:** `{debug.get('source', '?')}` " f"({'live-fetched + cached' if debug.get('source') == 'live' else 'served from index'})", f"- **Chunks used:** {debug.get('chunks', '?')}", f"- **Keywords:** {', '.join(debug.get('keywords') or []) or '—'}", ] return "**Debug — last answer**\n" + "\n".join(rows) def on_submit(question: str, history: list[dict], user: dict | None, last_mid): """Quota-gate, persist, answer, persist — as a streaming generator. Yields progressive updates so the UI shows the ACTUAL pipeline steps live (retrieval / live-fetch / indexing / writing), then collapses them into the "Process steps" accordion once the answer is in. The synchronous pipeline runs in a worker thread that pushes step strings onto a queue; this generator drains the queue and re-yields. Outputs: (chatbot, question, quota, steps_md, steps_accordion, last_msg_id, debug_md) """ history = history or [] if not user: yield history, "", "Please sign in first.", "", gr.update(), last_mid, gr.update() return if not question or not question.strip(): yield history, "", "", "", gr.update(), last_mid, gr.update() return uid, chat_id = user["id"], user["chat_id"] if not accounts.check_quota(uid): history = history + [ {"role": "user", "content": question}, {"role": "assistant", "content": "You've reached today's question limit. Please come back tomorrow."}, ] yield history, "", _quota_label(uid), "", gr.update(open=False), last_mid, gr.update() return accounts.add_message(chat_id, uid, "user", question) accounts.record_query(uid) # Show the user's turn immediately; stream steps into the open accordion. working_history = history + [{"role": "user", "content": question}] steps: list[str] = [] step_q: _queue.Queue = _queue.Queue() box: dict = {} def _run(): try: box["reply"], box["debug"] = answer(question, history, on_step=lambda m: step_q.put(m)) except Exception as e: # logged in answer(); keep a generic user-facing reply logger.exception("answer() failed in worker thread") box["error"] = e finally: step_q.put(None) # sentinel: pipeline finished worker = _threading.Thread(target=_run, daemon=True) worker.start() yield working_history, "", _quota_label(uid), _format_steps(steps), gr.update(open=True), last_mid, gr.update() while True: item = step_q.get() if item is None: break steps.append(item) yield working_history, "", _quota_label(uid), _format_steps(steps), gr.update(open=True), last_mid, gr.update() worker.join() if box.get("error") is not None: reply = "I ran into an unexpected problem answering that. Please try rephrasing." else: reply = box.get("reply", "I couldn't process that request.") # Persist the assistant turn; keep its id so ratings/comments can reference it. mid = accounts.add_message(chat_id, uid, "assistant", reply, metadata={"tool": "cdms_label"}) final_history = working_history + [{"role": "assistant", "content": reply}] steps.append("Done.") yield (final_history, "", _quota_label(uid), _format_steps(steps), gr.update(open=False), mid, gr.update(value=_format_debug(box.get("debug")))) def _quota_label(uid: int) -> str: try: return f"{accounts.remaining_quota(uid)} questions left today" except Exception: return "" def _load_user(user_row: dict): """Attach a chat (resuming the most recent) and hydrate its history.""" uid = user_row["id"] chats = accounts.list_chats(uid) chat_id = chats[0]["id"] if chats else accounts.create_chat(uid, "Chat 1") messages = accounts.get_messages(chat_id, uid) history = [{"role": m["role"], "content": m["content"]} for m in messages] user = {"id": uid, "username": user_row["username"], "chat_id": chat_id} return user, history def do_login(username: str, password: str): try: row = accounts.login(username, password) except Exception as e: return None, [], gr.update(visible=True), gr.update(visible=False), str(e), "", "", None user, history = _load_user(row) return ( user, history, gr.update(visible=False), gr.update(visible=True), "", f"Signed in as **{user['username']}**", _quota_label(user["id"]), None, ) def do_signup(username: str, password: str): try: row = accounts.signup(username, password) except Exception as e: return None, [], gr.update(visible=True), gr.update(visible=False), str(e), "", "", None user, history = _load_user(row) return ( user, history, gr.update(visible=False), gr.update(visible=True), "", f"Signed in as **{user['username']}**", _quota_label(user["id"]), None, ) def do_logout(): return ( None, [], gr.update(visible=True), gr.update(visible=False), "", "", "", None, ) def do_new_chat(user: dict | None): if not user: return [], user, None n = len(accounts.list_chats(user["id"])) + 1 chat_id = accounts.create_chat(user["id"], f"Chat {n}") return [], {**user, "chat_id": chat_id}, None # --- Feedback: thumbs + optional comment on the latest assistant answer ------- def on_like(user: dict | None, last_mid, evt: gr.LikeData) -> str: """Record a thumbs up/down for the most recent assistant answer.""" if not user or last_mid is None: return "" rating = 1 if evt.liked else -1 try: ok = accounts.add_feedback(last_mid, user["id"], rating=rating) except Exception: logger.exception("add_feedback (rating) failed") return "Couldn't record that rating." if ok is None: return "" return "Thanks — 👍 noted." if rating > 0 else "Thanks — 👎 noted." def submit_comment(user: dict | None, last_mid, comment: str): """Persist an optional free-text comment against the latest answer.""" if not user or last_mid is None: return comment, "Ask a question first, then comment on its answer." if not comment or not comment.strip(): return comment, "" try: ok = accounts.add_feedback(last_mid, user["id"], comment=comment.strip()) except Exception: logger.exception("add_feedback (comment) failed") return comment, "Couldn't save your comment." if ok is None: return comment, "Couldn't save your comment." return "", "Thanks for the feedback! 🙏" # --- Chat-session switcher -------------------------------------------------- def _chat_choices(user: dict | None): """(radio choices, current value) for the chat-session switcher.""" if not user: return [], None try: chats = accounts.list_chats_with_counts(user["id"]) except Exception: logger.exception("list_chats_with_counts failed") return [], user.get("chat_id") choices = [(f"{c['name']} · {c['msg_count']} msgs", c["id"]) for c in chats] return choices, user.get("chat_id") def load_chat(user: dict | None, chat_id: str | None): """Switch the active chat: load its messages, repoint the user's chat_id.""" if not user or not chat_id: return gr.update(), user, None try: messages = accounts.get_messages(chat_id, user["id"]) except Exception: logger.exception("get_messages failed") return gr.update(), user, None history = [{"role": m["role"], "content": m["content"]} for m in messages] return history, {**user, "chat_id": chat_id}, None def delete_selected_chat(user: dict | None, chat_id: str | None): """Delete the selected chat, then fall back to the most recent (or a fresh one).""" if not user or not chat_id: choices, cur = _chat_choices(user) return gr.update(choices=choices, value=cur), gr.update(), user, None try: accounts.delete_chat(chat_id, user["id"]) except Exception: logger.exception("delete_chat failed") chats = accounts.list_chats(user["id"]) new_id = chats[0]["id"] if chats else accounts.create_chat(user["id"], "Chat 1") messages = accounts.get_messages(new_id, user["id"]) history = [{"role": m["role"], "content": m["content"]} for m in messages] user2 = {**user, "chat_id": new_id} choices, _ = _chat_choices(user2) return gr.update(choices=choices, value=new_id), history, user2, None # --- Data export (admin-gated) ---------------------------------------------- import csv as _csv import os as _os import tempfile as _tempfile def _is_admin(user: dict | None) -> bool: """True if the user's name is in AGADVISOR_ADMINS (comma-separated env).""" if not user: return False admins = {a.strip().lower() for a in _os.getenv("AGADVISOR_ADMINS", "").split(",") if a.strip()} return user.get("username", "").lower() in admins def _stats_md(user: dict | None) -> str: """Counts panel: all-users for admins, own activity otherwise.""" if not user: return "" try: if _is_admin(user): s = accounts.stats(None) return (f"**📊 All users:** {s['feedback']} feedback · {s['queries']} queries · " f"{s['users']} users \n_(admin view — the CSV export covers everyone)_") s = accounts.stats(user["id"]) return f"**📊 Your activity:** {s['feedback']} feedback · {s['queries']} queries" except Exception: logger.exception("stats failed") return "" def export_csv(user: dict | None): """Write a feedback CSV (prompt/response/rating/comment) and return its path. Admins export all users; everyone else exports only their own.""" if not user: return None scope_all = _is_admin(user) try: rows = accounts.export_feedback(None if scope_all else user["id"]) except Exception: logger.exception("export_feedback failed") return None path = _tempfile.mktemp(prefix="agadvisor_feedback_", suffix=".csv") with open(path, "w", newline="", encoding="utf-8") as f: w = _csv.writer(f) w.writerow(["username", "timestamp", "rating", "comment", "prompt", "response"]) for r in rows: ts = _time.strftime("%Y-%m-%d %H:%M:%S", _time.localtime(r.get("created_at") or 0)) w.writerow([ r.get("username", ""), ts, r.get("rating", ""), r.get("comment") or "", r.get("prompt") or "", r.get("response") or "", ]) return path def _refresh_sidebar(user: dict | None): """Refresh the chat switcher + stats panel (chained after auth/answer/new-chat).""" choices, cur = _chat_choices(user) return gr.update(choices=choices, value=cur), _stats_md(user) import base64 as _base64 from pathlib import Path as _Path def _logo_uri(name: str) -> str: """Inline a logo as a base64 data URI (robust on Spaces — no static-path serving).""" try: b = (_Path(__file__).parent / "assets" / "logos" / name).read_bytes() return "data:image/png;base64," + _base64.b64encode(b).decode() except Exception: logger.exception("Could not load logo %s", name) return "" # Simple logo strip shown at the very top: the two Iowa State research-center logos # (TrAC + AIIRA) on small white cards, centered, wrapping on mobile. No background, # no title (the "AgAdvisor" heading below already carries the name). _LOGO_CARD = ("flex:1 1 0;display:flex;justify-content:center;align-items:center;" "background:#fff;border-radius:12px;padding:12px 10px;box-shadow:0 1px 5px rgba(0,0,0,.14);") _LOGO_IMG = "max-height:82px;max-width:100%;height:auto;width:auto;display:block;" _LOGO_BANNER = f"""
Translational AI Center (TrAC)
COALESCE
AI Institute for Resilient Agriculture (AIIRA)
""".replace("__TRAC__", _logo_uri("trac.png")) \ .replace("__COALESCE__", _logo_uri("coalesce.png")) \ .replace("__AIIRA__", _logo_uri("aiira.png")) # Mobile-friendly layout: full-width chat that grows with the viewport, and # button rows that wrap with tap-friendly targets. The logo cards stay white in # both themes (the logos need a light backing) — in dark mode we add a soft # border so they read as intentional cards rather than glare. _MOBILE_CSS = """ .gradio-container {max-width: 940px !important; margin: 0 auto !important;} #agadvisor-chat {height: 460px !important;} .theme-btn {min-width: 120px;} .dark .agadvisor-logos > div {box-shadow: 0 0 0 1px rgba(255,255,255,.12) !important;} @media (max-width: 700px) { .gradio-container {padding: 8px !important;} #agadvisor-chat {height: 60vh !important; min-height: 300px !important;} .agadvisor-btns {flex-wrap: wrap !important; gap: 8px !important;} .agadvisor-btns button {flex: 1 1 46% !important; min-height: 44px !important;} } """ with gr.Blocks(title="AgAdvisor", theme=gr.themes.Soft(primary_hue="green"), css=_MOBILE_CSS) as demo: user_state = gr.State(None) last_msg_state = gr.State(None) # id of the latest assistant message (for feedback) gr.HTML(_LOGO_BANNER) # two research-center logos at the very top gr.Markdown( "# 🌿 AgAdvisor\n" "A CDMS pesticide-label assistant with weather, soil, and agronomic tools. " "Answers include page-level citations.\n\n" "> Pesticide labels are legally binding. This is a research prototype and is " "**not** a substitute for reading the label. Always verify against the label of record." ) # --- Auth gate ----------------------------------------------------------- with gr.Column(visible=True) as login_view: gr.Markdown("### Sign in or create an account") username = gr.Textbox(label="Username", autofocus=True) password = gr.Textbox(label="Password", type="password") with gr.Row(): login_btn = gr.Button("Sign in", variant="primary") signup_btn = gr.Button("Create account") auth_error = gr.Markdown("") # --- Chat ---------------------------------------------------------------- with gr.Column(visible=False) as chat_view: with gr.Row(): who = gr.Markdown("") quota = gr.Markdown("") # Retrieval is automatic now (index-first, live-fetch on a miss), so # there's no mode toggle — just a light/dark switch for outdoor use. theme_btn = gr.Button( "🌓 Light / Dark", scale=0, min_width=120, elem_classes=["theme-btn"] ) # Switch between your saved chat sessions (name + message count). with gr.Accordion("💬 Chat sessions", open=False): chat_selector = gr.Radio(choices=[], label="Your chats", value=None) delete_chat_btn = gr.Button("🗑️ Delete selected chat", size="sm") # Thumbs up/down appear on each assistant message (Chatbot.like); they # rate the most recent answer. chatbot = gr.Chatbot(type="messages", height=460, label="AgAdvisor", elem_id="agadvisor-chat") # The real pipeline steps stream here live, then collapse (open=False) # once the answer is in — so first-time label fetches are legible. with gr.Accordion("🔎 Process steps", open=False) as steps_accordion: steps_md = gr.Markdown("") question = gr.Textbox( placeholder="e.g. What is the application rate for Roundup on soybeans?", label="Your question", autofocus=True, ) with gr.Row(elem_classes=["agadvisor-btns"]): send_btn = gr.Button("Ask", variant="primary") new_chat_btn = gr.Button("New chat") logout_btn = gr.Button("Log out") # Optional free-text feedback on the latest answer (thumbs are on the # message itself). Both persist to the accounts DB and sync to HF. with gr.Row(): comment_box = gr.Textbox( placeholder="Optional: tell us what was wrong or missing in the last answer", label="Feedback comment", scale=4, ) feedback_btn = gr.Button("Submit feedback", scale=1) feedback_status = gr.Markdown("") # Data export: your own activity, or (for admins in AGADVISOR_ADMINS) all # users. The CSV pairs each rating/comment with its prompt + response. with gr.Accordion("📊 Data export", open=False): export_stats = gr.Markdown("") with gr.Row(): refresh_stats_btn = gr.Button("Refresh stats", size="sm") download_btn = gr.Button("Download feedback CSV", size="sm") export_file = gr.File(label="Feedback export (CSV)") # Optional debug view of the last answer's routing (tool/source/chunks). with gr.Accordion("⚙️ Debug", open=False): debug_toggle = gr.Checkbox(label="Show debug info for the last answer", value=False) debug_md = gr.Markdown("", visible=False) if PRODUCTS: gr.Examples( examples=[ [f"What is the application rate for {p}?"] for p in PRODUCTS[:6] ], inputs=question, label=f"Labels in the index ({len(PRODUCTS)} products)", ) login_btn.click( do_login, [username, password], [user_state, chatbot, login_view, chat_view, auth_error, who, quota, last_msg_state], ).then(_refresh_sidebar, user_state, [chat_selector, export_stats]) signup_btn.click( do_signup, [username, password], [user_state, chatbot, login_view, chat_view, auth_error, who, quota, last_msg_state], ).then(_refresh_sidebar, user_state, [chat_selector, export_stats]) logout_btn.click( do_logout, None, [user_state, chatbot, login_view, chat_view, auth_error, who, quota, last_msg_state], ).then(_refresh_sidebar, user_state, [chat_selector, export_stats]) new_chat_btn.click( do_new_chat, user_state, [chatbot, user_state, last_msg_state] ).then(_refresh_sidebar, user_state, [chat_selector, export_stats]) # Light/dark toggle: flip Gradio's `dark` class on the document body. theme_btn.click(None, None, None, js="() => { document.body.classList.toggle('dark'); }") # Chat-session switcher. chat_selector.change(load_chat, [user_state, chat_selector], [chatbot, user_state, last_msg_state]) delete_chat_btn.click( delete_selected_chat, [user_state, chat_selector], [chat_selector, chatbot, user_state, last_msg_state], ) # Data export + debug. refresh_stats_btn.click(_stats_md, user_state, export_stats) download_btn.click(export_csv, user_state, export_file) debug_toggle.change(lambda on: gr.update(visible=on), debug_toggle, debug_md) # Feedback wiring. chatbot.like(on_like, [user_state, last_msg_state], feedback_status) feedback_btn.click( submit_comment, [user_state, last_msg_state, comment_box], [comment_box, feedback_status], ) for trigger in (send_btn.click, question.submit): trigger( on_submit, [question, chatbot, user_state, last_msg_state], [chatbot, question, quota, steps_md, steps_accordion, last_msg_state, debug_md], ).then(_refresh_sidebar, user_state, [chat_selector, export_stats]) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)