import gradio as gr import time import threading import logging import spaces from main import run_repository_ranking # Import the repository ranking function # --------------------------- # Global Logging Buffer Setup # --------------------------- LOG_BUFFER = [] LOG_BUFFER_LOCK = threading.Lock() class BufferLogHandler(logging.Handler): def emit(self, record): log_entry = self.format(record) with LOG_BUFFER_LOCK: LOG_BUFFER.append(log_entry) root_logger = logging.getLogger() if not any(isinstance(h, BufferLogHandler) for h in root_logger.handlers): handler = BufferLogHandler() formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") handler.setFormatter(formatter) root_logger.addHandler(handler) def filter_logs(logs): filtered = [] last_was_fetching = False for log in logs: if "HTTP Request:" in log: if not last_was_fetching: filtered.append("Fetching repositories...") last_was_fetching = True else: filtered.append(log) last_was_fetching = False return filtered # --------------------------- # Minimal Title, Favicon & Description # --------------------------- favicon_html = """ DeepGit Lite Research Agent """ title = """

DeepGit Lite

⚙️ A lightweight GitHub research agent for deep semantic search and ranking.

""" description = """

DeepGit Lite is a streamlined tool for semantic search on GitHub repositories. It retrieves repositories using dense retrieval, ranks them by similarity, and then presents the top results.

""" footer = """
Made with ❤️ by Zamal
""" def parse_result_to_html(raw_result: str) -> str: entries = raw_result.strip().split("Final Rank:") html = """ """ for entry in entries[1:]: lines = entry.strip().split("\n") data = {} data["Final Rank"] = lines[0].strip() for line in lines[1:]: if ": " in line: key, val = line.split(": ", 1) data[key.strip()] = val.strip() html += f""" """ html += "
Rank Title Link Combined Score
{data.get('Final Rank', '')} {data.get('Title', '')} GitHub {data.get('Combined Score', '')}
" return html # --------------------------- # GPU-enabled Wrapper for Repository Ranking # --------------------------- @spaces.GPU(duration=180) def gpu_run_repo(topic: str): return run_repository_ranking(topic) def run_lite_workflow(topic, result_container): result = gpu_run_repo(topic) result_container["raw_result"] = result def stream_lite_workflow(topic): with LOG_BUFFER_LOCK: LOG_BUFFER.clear() result_container = {} workflow_thread = threading.Thread(target=run_lite_workflow, args=(topic, result_container)) workflow_thread.start() last_index = 0 while workflow_thread.is_alive() or (last_index < len(LOG_BUFFER)): with LOG_BUFFER_LOCK: new_logs = LOG_BUFFER[last_index:] last_index = len(LOG_BUFFER) if new_logs: filtered_logs = filter_logs(new_logs) status_msg = filtered_logs[-1] detail_msg = "
".join(filtered_logs) yield status_msg, detail_msg time.sleep(0.5) workflow_thread.join() with LOG_BUFFER_LOCK: final_logs = LOG_BUFFER[:] raw_result = result_container.get("raw_result", "No results returned.") html_result = parse_result_to_html(raw_result) yield "", html_result # --------------------------- # App UI Setup Using Default Gradio Theme # --------------------------- with gr.Blocks() as demo: gr.HTML(favicon_html) gr.HTML(title) gr.HTML(description) with gr.Column(): research_input = gr.Textbox( label="Research Topic", placeholder="Enter your research topic here, e.g., 'Fine tuning Instruction tuned LLama models...'", lines=3 ) run_button = gr.Button("Run DeepGit Lite", variant="primary") status_display = gr.Markdown("") detail_display = gr.HTML("") output_html = gr.HTML("") state = gr.State([]) # Yield an initial status message for immediate feedback def lite_runner(topic): yield "Workflow started", "

Processing your request. Please wait...

" for status, details in stream_lite_workflow(topic): yield status, details run_button.click( fn=lite_runner, inputs=[research_input], outputs=[status_display, detail_display], api_name="deepgit_lite", show_progress=True ) research_input.submit( fn=lite_runner, inputs=[research_input], outputs=[status_display, detail_display], api_name="deepgit_lite_submit", show_progress=True ) gr.HTML(footer) demo.queue(max_size=10).launch()