""" app.py — Gradio interface for the Multi-Agent RAG Research System. Agent 1 (Research Summarizer) — runs automatically when a paper is submitted. Agent 2 (Code Explainer) — runs on-demand via a button click. Set OPENAI_API_KEY in Space → Settings → Variables and secrets. Optionally set GITHUB_TOKEN for higher GitHub API rate limits. """ from __future__ import annotations import os import tempfile import gradio as gr from dotenv import load_dotenv from langchain_core.messages import AIMessage, HumanMessage load_dotenv() _MISSING_KEY_MSG = ( "**GROQ_API_KEY is not set.**\n\n" "Go to **Settings → Variables and secrets** and add your key." ) def _key_is_set() -> bool: return bool(os.getenv("GROQ_API_KEY")) _SUMMARY_COMPLETE = False _SUMMARY_RUNNING = False def reset_summary_state(): global _SUMMARY_COMPLETE, _SUMMARY_RUNNING _SUMMARY_COMPLETE = False _SUMMARY_RUNNING = False def _invoke_agent(agent, query: str) -> str: result = agent.invoke({"messages": [HumanMessage(content=query)]}) messages = result.get("messages", []) last = next( (m for m in reversed(messages) if isinstance(m, AIMessage)), AIMessage(content="(no response)"), ) return last.content # --------------------------------------------------------------------------- # Agent 1 — Research Summarizer # Returns: (summary_out update, explain_btn update) # --------------------------------------------------------------------------- def run_summarizer(paper_url: str, pdf_file): global _SUMMARY_COMPLETE, _SUMMARY_RUNNING if not _key_is_set(): return gr.update(value=_MISSING_KEY_MSG, visible=True), gr.update(visible=False) has_url = bool(paper_url and paper_url.strip()) has_pdf = pdf_file is not None if not has_url and not has_pdf: return ( gr.update(value="Please provide a paper URL / arXiv ID or upload a PDF.", visible=True), gr.update(visible=False), ) if _SUMMARY_RUNNING: return ( gr.update(value="⏳ Summary is already running.", visible=True), gr.update(visible=False), ) if _SUMMARY_COMPLETE: return ( gr.update(visible=True), gr.update(visible=True), ) _SUMMARY_RUNNING = True from vector_store import reset_store reset_store() try: from agents import build_paper_summariser if has_pdf: from tools import fetch_pdf_paper pdf_path = pdf_file if isinstance(pdf_file, str) else pdf_file.name fetch_pdf_paper.invoke({"pdf_path_or_url": pdf_path}) query = ( "The research paper has already been indexed from an uploaded PDF. " "Summarise it now — do NOT call fetch_arxiv_paper or fetch_pdf_paper again." ) else: query = f"Summarise this research paper: {paper_url.strip()}" agent = build_paper_summariser() summary = _invoke_agent(agent, query) _SUMMARY_COMPLETE = True _SUMMARY_RUNNING = False return gr.update(value=summary, visible=True), gr.update(visible=True) except Exception as exc: import traceback traceback.print_exc() return gr.update(value=f"Error: {exc}", visible=True), gr.update(visible=False) finally: _SUMMARY_RUNNING = False # --------------------------------------------------------------------------- # Agent 2 — Code Explainer # Returns: (code_out update, download_btn update) # --------------------------------------------------------------------------- def run_code_explainer(): if not _key_is_set(): return gr.update(value=_MISSING_KEY_MSG, visible=True), gr.update(visible=False) query = ( "Find all GitHub repository links mentioned in the indexed paper. " "Fetch and index each repository. " "Extract and explain every algorithm and pseudocode block from the paper. " "Then produce a complete how-to-run guide with exact terminal commands " "starting from git clone." ) try: from agents import build_code_explainer print("[DEBUG] Starting code explainer...") agent = build_code_explainer() explanation = _invoke_agent(agent, query) print("[DEBUG] Code explainer finished.") with tempfile.NamedTemporaryFile( mode="w", suffix=".md", delete=False, encoding="utf-8" ) as f: f.write(explanation) tmp_path = f.name return gr.update(value=explanation, visible=True), gr.update(value=tmp_path, visible=True) except Exception as exc: import traceback traceback.print_exc() return gr.update(value=f"Error: {exc}", visible=True), gr.update(visible=False) # --------------------------------------------------------------------------- # Gradio UI # --------------------------------------------------------------------------- def show_summarizer_working(): if _SUMMARY_COMPLETE: return gr.update(value="", visible=False) if _SUMMARY_RUNNING: return gr.update(value="⏳ Summary is already running.", visible=True) return gr.update(value="⏳ Summarizing paper... this may take a minute.", visible=True) def show_code_working(): return gr.update( value=( "⏳ Searching for repositories, indexing code evidence, and generating " "the explanation. This can take a few minutes." ), visible=True, ) TITLE_HTML = """