#!/usr/bin/env python """ Unified Web UI for Hybrid Software Bug Assistant - Simple queries → Local Gemma-2B (fast, free) - Complex queries → Cloud Gemini + Tools (powerful) - Single interface for both models """ import os import re import sys import json import queue import base64 from datetime import datetime from typing import Optional # Load .env FIRST before anything else from dotenv import load_dotenv load_dotenv(override=True) # Set defaults if not in .env os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", "FALSE") os.environ.setdefault("SKIP_AGENT_IMPORT", "TRUE") # Add project to path sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from flask import Flask, render_template_string, request, jsonify, stream_with_context, Response, make_response from software_bug_assistant.tools.local_tools import ( backfill_missing_embeddings, get_ticket_creator, query_stackexchange, ) from software_bug_assistant.hybrid_agent import HybridAgent, QueryClassifier, classify_query import threading from pathlib import Path # Application and Agent globals app = Flask(__name__) hybrid_agent = None adk_agent = None local_llm_with_tools = None _model_downloading = False conversation_state = {} embedding_backfill_attempted = False _model_init_lock = threading.Lock() def _load_local_instances(): """Load local LLM with tools and hybrid agent sharing a single model in memory""" global hybrid_agent, local_llm_with_tools if local_llm_with_tools is None: try: from software_bug_assistant.tools.local_tools import get_local_llm_with_tools local_llm_with_tools = get_local_llm_with_tools() if local_llm_with_tools and local_llm_with_tools.is_available(): print("✓ Local LLM with Tool Calling: READY") else: local_llm_with_tools = None except Exception as e: print(f"⚠ Local LLM with tools not available: {e}") local_llm_with_tools = None try: from software_bug_assistant.hybrid_agent import HybridAgent, LocalGemmaModel shared_model = local_llm_with_tools.model if local_llm_with_tools else None if hybrid_agent is None: local_gemma = LocalGemmaModel(model=shared_model) hybrid_agent = HybridAgent(enable_local=True, local_model=local_gemma) elif hybrid_agent.local_model is None or not hybrid_agent.local_model.is_available(): hybrid_agent.local_model = LocalGemmaModel(model=shared_model) if hybrid_agent and hybrid_agent.local_model and hybrid_agent.local_model.is_available(): print("✓ Local Gemma-2B: READY") except Exception as e: print(f"⚠ Local Gemma model init error: {e}") def init_local_models(): """Attempt to load Local LLM and Local LLM with tools, sharing model instance and downloading in background if needed""" global hybrid_agent, local_llm_with_tools, _model_downloading # Fast path if already loaded if local_llm_with_tools is not None and hybrid_agent is not None and hybrid_agent.local_model is not None and hybrid_agent.local_model.is_available(): return with _model_init_lock: model_path = Path("models") / "gemma-2-2b-it-Q4_K_M.gguf" if not model_path.exists(): if _model_downloading: return _model_downloading = True def _download_task(): global _model_downloading, hybrid_agent, local_llm_with_tools print("⏳ Gemma model not found. Starting background download (~1.5GB)...") try: from download_gemma import download_gemma_model download_gemma_model() except Exception as dl_err: print(f"⚠ Background Gemma download failed: {dl_err}") _model_downloading = False return _model_downloading = False if model_path.exists(): _load_local_instances() threading.Thread(target=_download_task, daemon=True).start() return # Model file already exists on disk (pre-downloaded in Docker image) _load_local_instances() # Kick off model initialization init_local_models() # Initialize direct database access functions (replaces MCP Toolbox) from software_bug_assistant.tools.db_tools import ( get_tickets_by_status as db_get_tickets_by_status, get_tickets_by_priority as db_get_tickets_by_priority, ) # Startup DB connection test try: _db_test = db_get_tickets_by_status("Open") if _db_test.get('success'): print(f"✓ Database connection: OK ({_db_test.get('count', 0)} open tickets)") else: print(f"⚠ Database connection FAILED: {_db_test.get('error', 'unknown')}") except Exception as _e: print(f"⚠ Database connection test error: {_e}") def query_database_directly(query: str) -> Optional[str]: """ Directly query the database using psycopg when cloud is unavailable. Returns formatted results or None if can't handle. """ query_lower = query.lower() try: result = None # Detect query type and call appropriate function if "open" in query_lower and "ticket" in query_lower: result = db_get_tickets_by_status("Open") elif "closed" in query_lower and "ticket" in query_lower: result = db_get_tickets_by_status("Closed") elif "in progress" in query_lower or "in-progress" in query_lower: result = db_get_tickets_by_status("In Progress") elif "all ticket" in query_lower or "show ticket" in query_lower or "list ticket" in query_lower: result = db_get_tickets_by_status("Open") elif "high" in query_lower and ("priority" in query_lower or "ticket" in query_lower): result = db_get_tickets_by_priority("P1 - High") elif "critical" in query_lower: result = db_get_tickets_by_priority("P0 - Critical") elif "medium" in query_lower and "priority" in query_lower: result = db_get_tickets_by_priority("P2 - Medium") elif "low" in query_lower and "priority" in query_lower: result = db_get_tickets_by_priority("P3 - Low") else: return None if result is None or result.get('error'): error_msg = result.get('error', 'Unknown error') if result else 'No result' return f"⚠️ Database error: {error_msg}\n\nPlease check the database connection and try again." tickets = result.get('tickets', []) if isinstance(tickets, list) and len(tickets) > 0: # Select key columns for display display_cols = ['ticket_id', 'title', 'status', 'priority', 'assignee'] # Build markdown table md_table = "| " + " | ".join(display_cols) + " |\n" md_table += "| " + " | ".join(["---"] * len(display_cols)) + " |\n" for row in tickets: if isinstance(row, dict): values = [] for col in display_cols: val = str(row.get(col, "")) if "@" in val: val = val.split("@")[0] if len(val) > 40: val = val[:37] + "..." values.append(val) md_table += "| " + " | ".join(values) + " |\n" output = f"📋 **Database Results:**\n\n{md_table}\n\n_Found {len(tickets)} ticket(s)._" if result.get('is_fallback'): output += ( "\n\n---\n> 💡 **Notice:** Remote Supabase database is unreachable (`FATAL: tenant/user not found` — project may be paused).\n" "> Loaded tickets from local database fallback. To connect live Supabase, resume your project in your [Supabase Dashboard](https://supabase.com/dashboard)." ) return output elif isinstance(tickets, list) and len(tickets) == 0: return "No tickets found matching your query." else: return f"📋 **Database Results:**\n\n{result}" except Exception as e: print(f"Direct database query failed: {e}") import traceback traceback.print_exc() return None def _get_client_state(client_id: str) -> dict: if client_id not in conversation_state: conversation_state[client_id] = { "stage": "idle", "last_issue": "", } return conversation_state[client_id] def _is_affirmative(text: str) -> bool: return bool(re.search(r"\b(yes|yep|yeah|fixed|solved|done|see|ok|okay)\b", text)) def _is_negative(text: str) -> bool: return bool(re.search(r"\b(no|not yet|still|didn't|did not|failed|nope)\b", text)) def _is_ticket_query(text: str) -> bool: has_ticket = bool(re.search(r"\b(ticket|tickets|bug|bugs|issue|issues)\b", text)) has_action = bool(re.search(r"\b(show|list|get|find|search|open|closed|status|priority|create|update)\b", text)) return has_ticket and has_action def _needs_problem_details(text: str) -> bool: is_help_request = bool(re.search(r"\b(suggestion|suggest|solution|fix|help|troubleshoot)\b", text)) has_issue_signal = bool(re.search(r"\b(error|issue|problem|slow|crash|freeze|not working|fail|broken|bug)\b", text)) return is_help_request and not has_issue_signal def _is_troubleshoot_request(text: str) -> bool: return bool(re.search(r"\b(error|issue|problem|slow|crash|freeze|not working|fail|broken|bug|fix|help|troubleshoot)\b", text)) def _looks_like_new_query(text: str) -> bool: """Detect whether user typed a new request instead of yes/no confirmation.""" if _is_ticket_query(text): return True if _needs_problem_details(text): return True if _is_troubleshoot_request(text): return True # Longer free-text messages are likely new queries, not confirmations. if len(text.split()) >= 3 and not _is_affirmative(text) and not _is_negative(text): return True return False def _is_generic_issue_text(text: str) -> bool: cleaned = (text or "").strip().lower() if not cleaned: return True return bool(re.match(r"^(how to fix (this|the) issue\??|fix (this|the) issue\??|this issue\??|issue\??|problem\??|help\??)$", cleaned)) def _extract_issue_from_assistant_text(text: str) -> str: if not text: return "" cleaned = text cleaned = re.split(r"\n\s*---", cleaned, maxsplit=1)[0] cleaned = re.split(r"Did that fix it\?", cleaned, flags=re.IGNORECASE, maxsplit=1)[0] cleaned = re.split(r"StackOverflow Related Issues", cleaned, flags=re.IGNORECASE, maxsplit=1)[0] cleaned = re.sub(r"\[(.*?)\]\((.*?)\)", r"\1", cleaned) lines = [] for line in cleaned.splitlines(): line = re.sub(r"^\s*[-*#>\d.\)]+\s*", "", line).strip() if not line: continue lines.append(line) if not lines: return "" summary = " ".join(lines) summary = re.sub(r"\s+", " ", summary).strip() return summary[:260] def _derive_issue_text_for_ticket(last_issue: str, history: list) -> str: candidate = (last_issue or "").strip() if candidate and not _is_generic_issue_text(candidate): return candidate for msg in reversed(history or []): if msg.get("role") != "assistant": continue extracted = _extract_issue_from_assistant_text(str(msg.get("content") or "")) if extracted and not _is_generic_issue_text(extracted): return extracted return candidate def _format_search_section(title: str, items: list, max_items: int = 5) -> str: if not items: return "" lines = [f"## {title}"] for i, item in enumerate(items[:max_items], 1): name = item.get("title", "") link = item.get("link", "") snippet = item.get("snippet") or item.get("score") or "" suffix = f" - {snippet}" if snippet else "" lines.append(f"{i}. [{name}]({link}){suffix}") return "\n".join(lines) # HTML Template HELP_TEMPLATE = """ Tandion Bug Assistant | Help & Examples
Help & Examples
Complete guide to using Local (Offline) & Online (Cloud) AI models, plus ticket management commands.
← Back to Chat
🤖 AI Models: How to Use Local vs Online
🏠 Local • Offline • Gemma-2B

Local / Offline Model

Runs 100% locally inside the container via llama.cpp CPU inference. Zero external API keys required, private, fast, and operates completely offline.

Force Local Prefix: @local <query>
  • @local how to debug memory leaks in Python
  • @local explain deadlock vs race condition
  • @local what causes segmentation faults in C++?
  • @local show all open tickets
☁️ Cloud • Online • Gemini

Online / Cloud Model

Powered by Google Gemini with live tool execution, Google Search, StackOverflow issue cross-referencing, multi-modal screenshot analysis, and deep reasoning.

Force Cloud Prefix: @cloud <query>
  • @cloud investigate CVE-2024-3094 vulnerability
  • @cloud analyze this stack trace and suggest root causes
  • @cloud find solutions for AWS RDS pool exhaustion
  • Attach error screenshots using the 📷 camera button
⚡

Smart Automatic Routing (Default)

If you don't prefix with @local or @cloud, Tandion automatically routes simple questions, greetings, and ticket queries to the fast Local model, and complex bug troubleshooting to Cloud Gemini. If external cloud services are ever unavailable, the system automatically falls back to your local model!

📋 Ticket Management Commands

Create Ticket

  • Create a ticket: login fails after update
  • Report a bug: dashboard loads forever
  • Open a ticket for VPN not connecting

Update Status

  • Close ticket 12
  • Resolve ticket #7
  • Mark ticket 5 as in progress

Update Priority

  • Set ticket 9 to critical
  • Change priority of ticket 3 to P1
  • Make ticket 14 low priority

View & Search

  • Show all open tickets
  • List critical tickets
  • Search tickets for "login error"
""" HTML_TEMPLATE = """ Tandion Bug Assistant | Next-Gen AI Issue Intelligence
💡 Guide
⚡

Tandion Bug Assistant

High-performance AI bug triage, ticket operations, and stack trace debugging. Choose your routing engine or select an interactive prompt below to begin.

🚨 Critical Bug Triage
Find blocking P0/Critical issues requiring immediate fix.
Show critical priority tickets
🏠 Local Fast Debug
Private offline CPU analysis for Python memory leak diagnosis.
@local debug memory leaks
☁️ Cloud CVE & Security
Online deep research with live web tools and vulnerability insights.
@cloud investigate CVE-2024-3094
📋 Ticket Dashboard
Query active tickets, assignee distribution, and priority overview.
Show all open tickets
Mode:
Attached Preview Screenshot attached
Press Enter ↵ to send • Shift+Enter for newline • Paste screenshots directly from clipboard
""" @app.route('/') def index(): """Serve the main UI""" resp = make_response(render_template_string(HTML_TEMPLATE)) resp.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate, max-age=0' resp.headers['Pragma'] = 'no-cache' resp.headers['Expires'] = '0' return resp @app.route('/help') def help_page(): """Serve the help/examples page""" resp = make_response(render_template_string(HELP_TEMPLATE)) resp.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate, max-age=0' resp.headers['Pragma'] = 'no-cache' resp.headers['Expires'] = '0' return resp def init_agents(): """Initialize both hybrid agent and ADK agent""" global hybrid_agent, adk_agent, embedding_backfill_attempted # Initialize hybrid agent (for local Gemma) if hybrid_agent is None: try: from software_bug_assistant.hybrid_agent import HybridAgent, LocalGemmaModel shared_model = local_llm_with_tools.model if local_llm_with_tools else None local_gemma = LocalGemmaModel(model=shared_model) hybrid_agent = HybridAgent(enable_local=True, local_model=local_gemma) except Exception as e: print(f"Warning: Hybrid agent init failed: {e}") if not embedding_backfill_attempted and os.getenv("AUTO_EMBED_ON_START", "").upper() == "TRUE": embedding_backfill_attempted = True backfill_missing_embeddings() # Initialize ADK agent (for cloud Gemini with tools) if adk_agent is None: try: # Configure google-genai client BEFORE importing agent # ADK uses google-genai (not google-generativeai) from google import genai api_key = os.environ.get("GOOGLE_API_KEY") if api_key: # Create a client to verify the API key works client = genai.Client(api_key=api_key) # Temporarily allow agent import original_skip = os.environ.get("SKIP_AGENT_IMPORT") os.environ["SKIP_AGENT_IMPORT"] = "FALSE" from software_bug_assistant.agent import root_agent adk_agent = root_agent # Restore setting if original_skip: os.environ["SKIP_AGENT_IMPORT"] = original_skip else: os.environ.pop("SKIP_AGENT_IMPORT", None) print("✓ ADK agent initialized for cloud Gemini") except Exception as e: print(f"Warning: ADK agent init failed: {e}") adk_agent = None @app.route('/api/status', methods=['GET']) def get_status(): """Get status of local and cloud models""" global hybrid_agent, adk_agent, local_llm_with_tools, _model_downloading init_agents() init_local_models() local_available = ( (hybrid_agent is not None and hybrid_agent.local_model is not None and hybrid_agent.local_model.is_available()) or (local_llm_with_tools is not None and local_llm_with_tools.is_available()) ) cloud_available = adk_agent is not None return jsonify({ "local_available": local_available, "local_downloading": _model_downloading, "cloud_available": cloud_available }) def sse_event(payload: dict) -> str: """Format SSE event string for client streaming.""" return f"data: {json.dumps(payload)}\n\n" def _get_fast_greeting(msg_lower: str) -> Optional[str]: """Return an instant, high-quality greeting/intro (< 20ms) for common pleasantries.""" cleaned = re.sub(r'[^\w\s]', '', msg_lower).strip() greetings = { "hi": "Hello! 👋 I'm your Tandion Software Bug Assistant. How can I assist you with your software, tickets, or troubleshooting today?", "hello": "Hello! 👋 I'm your Tandion Software Bug Assistant. How can I help you troubleshoot a bug or manage your tickets today?", "hey": "Hey there! 👋 What issue or ticket can I help you with today?", "good morning": "Good morning! ☀️ How can I assist you with software debugging or tickets today?", "good evening": "Good evening! 🌙 How can I help you today?", "who are you": "I am the **Hybrid AI Software Bug Assistant** for **Tandion IT Software**. I combine local Gemma-2B processing with Cloud Gemini to help you triage errors, search documentation, manage support tickets, and resolve bugs efficiently.", "what can you do": "Here is what I can do for you:\n\n- 🔍 **Triage & Debug Code**: Analyze stack traces, syntax errors, and system exceptions.\n- 🎫 **Manage Tickets**: View open/closed tickets, update status and priority, or create new bug tickets directly.\n- 📚 **Search Solutions**: Find relevant troubleshooting guides and StackOverflow discussions.\n- 🏠 **Hybrid Local & Cloud**: Run queries locally with `@local` (Gemma-2B) or use advanced reasoning with `@cloud` (Gemini).", "help": "How can I help you? You can:\n1. Describe any bug or error message (or paste a screenshot).\n2. Ask to view tickets (e.g. *\"Show all open tickets\"* or *\"Ticket #11\"*).\n3. Update tickets (e.g. *\"Close ticket 11\"* or *\"Change priority of ticket 11 to critical\"*).\n4. Use `@local` to run locally on Gemma-2B or `@cloud` for Cloud Gemini." } return greetings.get(cleaned) def stream_cloud_gemini(adk_agent_instance, issue_text: str, image_bytes=None, image_mime="image/png"): """Stream Gemini Cloud tokens through a thread-safe queue for real-time SSE delivery.""" q = queue.Queue() def worker(): try: import asyncio import uuid from google.adk.runners import InMemoryRunner from google.genai import types async def run_async_agent(include_image: bool = True): parts = [types.Part(text=issue_text)] if include_image and image_bytes is not None: parts.append(types.Part.from_bytes(data=image_bytes, mime_type=image_mime)) content = types.Content(role="user", parts=parts) import warnings with warnings.catch_warnings(): warnings.filterwarnings("ignore", message=".*mismatch.*") runner = InMemoryRunner(agent=adk_agent_instance, app_name="software_bug_assistant") user_id = "web_ui_user" session_id = f"session_{uuid.uuid4().hex[:8]}" await runner.session_service.create_session( app_name="software_bug_assistant", user_id=user_id, session_id=session_id, ) async for event in runner.run_async( user_id=user_id, session_id=session_id, new_message=content ): if hasattr(event, 'content') and event.content: if hasattr(event.content, 'parts') and event.content.parts: for response_part in event.content.parts: if hasattr(response_part, 'text') and response_part.text: q.put(("chunk", response_part.text)) elif hasattr(event.content, 'text') and event.content.text: q.put(("chunk", event.content.text)) elif hasattr(event, 'text') and event.text: q.put(("chunk", event.text)) elif isinstance(event, str) and event: q.put(("chunk", event)) try: asyncio.run(run_async_agent(include_image=True)) except Exception as image_run_err: if image_bytes is not None: print(f"Cloud triage image processing failed, retrying text-only: {image_run_err}") q.put(("chunk", "I couldn't fully process the image, so I used your text prompt instead.\n\n")) asyncio.run(run_async_agent(include_image=False)) else: raise image_run_err # Append StackExchange results non-blockingly at the end of the stream try: stack = query_stackexchange(issue_text) stack_section = _format_search_section("📚 StackOverflow Related Issues", stack.get("results", [])) if stack_section: q.put(("chunk", "\n\n---\n\n" + stack_section)) except Exception as search_err: print(f"StackExchange search skipped: {search_err}") q.put(("done", None)) except Exception as e: q.put(("error", e)) t = threading.Thread(target=worker, daemon=True) t.start() while True: try: kind, val = q.get(timeout=100) if kind == "chunk": yield val elif kind == "done": break elif kind == "error": raise val except queue.Empty: break def send_chat_response( content, model_used: str, classification: str, ask_fix_confirmation: bool = False, is_stream: bool = False, error: bool = False, state: Optional[dict] = None ): """Unified response builder supporting both real-time SSE streaming and standard JSON.""" if not is_stream: if isinstance(content, str): final_text = content else: final_text = "".join(list(content)) if ask_fix_confirmation: final_text += "\n\n---\n\nDid that fix it? Reply **yes** or **no**. If no, I will create a ticket." if state is not None: state["stage"] = "awaiting_fix_confirmation" resp_payload = { "response": final_text, "model_used": model_used, "classification": classification } if error: resp_payload["error"] = True return jsonify(resp_payload) # Streaming Response via Server-Sent Events (SSE) def event_stream(): yield sse_event({"type": "meta", "model_used": model_used, "classification": classification}) if isinstance(content, str): yield sse_event({"type": "chunk", "text": content}) else: try: for chunk in content: if chunk: yield sse_event({"type": "chunk", "text": chunk}) except Exception as stream_err: print(f"Streaming error: {stream_err}") yield sse_event({"type": "chunk", "text": f"\n\n*(Stream error: {stream_err})*"}) if ask_fix_confirmation: yield sse_event({"type": "chunk", "text": "\n\n---\n\nDid that fix it? Reply **yes** or **no**. If no, I will create a ticket."}) if state is not None: state["stage"] = "awaiting_fix_confirmation" yield sse_event({"type": "done"}) return Response(stream_with_context(event_stream()), mimetype='text/event-stream') @app.route('/api/chat', methods=['POST']) def chat(): """Handle chat messages and route to appropriate model with optional live streaming""" global hybrid_agent, adk_agent, local_llm_with_tools, _model_downloading # Initialize agents if needed init_agents() init_local_models() if hybrid_agent is None: return jsonify({ "error": "Failed to initialize hybrid agent" }), 500 data = request.json or {} message = data.get('message', '').strip() history = data.get('history', []) client_id = data.get('client_id') or request.remote_addr or "anonymous" image_data = data.get('image_data') image_mime = data.get('image_mime', 'image/png') is_stream = bool(data.get('stream', False)) state = _get_client_state(client_id) image_bytes = None if image_data: try: image_bytes = base64.b64decode(image_data) except Exception: return jsonify({"error": "Invalid image payload"}), 400 if not message and image_bytes is None: return jsonify({"error": "Message or image is required"}), 400 # Check for @cloud prefix to force Cloud Gemini force_cloud = False if message.lower().startswith('@cloud'): force_cloud = True message = re.sub(r'^@cloud\s*:?', '', message, flags=re.IGNORECASE).strip() if not message: return send_chat_response( "Please add a query after @cloud, for example: @cloud show open tickets", model_used="gemini-cloud", classification="FORCED_CLOUD", error=True, is_stream=is_stream, state=state ) print(f"🌩️ Force Cloud Gemini requested for: {message}") # Check for @local prefix to force Local Gemma-2B force_local = False if message.lower().startswith('@local'): force_local = True message = re.sub(r'^@local\s*:?', '', message, flags=re.IGNORECASE).strip() if not message: return send_chat_response( "Please add a query after @local, for example: @local how to troubleshoot slow queries", model_used="gemma-2b-local", classification="FORCED_LOCAL", error=True, is_stream=is_stream, state=state ) print(f"🏠 Force Local Gemma requested for: {message}") if image_bytes is not None: force_cloud = True if not message: message = "Analyze the uploaded image and provide troubleshooting steps." message_lower = message.lower() if force_cloud and state["stage"] == "awaiting_fix_confirmation": state["stage"] = "idle" state["last_issue"] = "" if state["stage"] == "awaiting_fix_confirmation": if _is_affirmative(message_lower): state["stage"] = "idle" state["last_issue"] = "" return send_chat_response( "✅ Glad it is fixed. I will not create a ticket. Anything else you need?", model_used="triage", classification="FIX_CONFIRMED", is_stream=is_stream, state=state ) if _is_negative(message_lower): issue_text = _derive_issue_text_for_ticket(state.get("last_issue", ""), history) state["stage"] = "idle" state["last_issue"] = "" if not issue_text: return send_chat_response( "I can create a ticket, but I need the issue details first. What problem should I file?", model_used="triage", classification="NEED_DETAILS", is_stream=is_stream, state=state ) ticket_creator = get_ticket_creator() ticket_response = ticket_creator.create_ticket_from_query(issue_text) return send_chat_response( ticket_response, model_used="database-direct", classification="TICKET_CREATED", is_stream=is_stream, state=state ) if _looks_like_new_query(message_lower): state["stage"] = "idle" state["last_issue"] = "" else: return send_chat_response( "Please confirm if the issue is fixed. Reply 'yes' or 'no'.", model_used="triage", classification="NEED_CONFIRMATION", is_stream=is_stream, state=state ) if state["stage"] == "awaiting_problem": state["stage"] = "idle" message_lower = message.lower() state["last_issue"] = message # Build context from conversation history context = { "current_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "conversation_history": [ f"{msg['role']}: {msg['content']}" for msg in history[-5:] ] if history else [] } if _needs_problem_details(message_lower): state["stage"] = "awaiting_problem" return send_chat_response( "What problem should I fix? Please describe the issue (symptoms, error text, when it happens).", model_used="triage", classification="NEED_DETAILS", is_stream=is_stream, state=state ) # 0. Fast-path for common pleasantries & greetings (< 20ms response time) if not force_cloud and not force_local: fast_greeting = _get_fast_greeting(message_lower) if fast_greeting: return send_chat_response( fast_greeting, model_used="gemma-2b-local", classification="FAST_GREETING", is_stream=is_stream, state=state ) # If user explicitly requested @local, handle exclusively with local model/tools if force_local: # Check direct database and tools for ticket-related queries if _is_ticket_query(message_lower): db_result = query_database_directly(message) if db_result: return send_chat_response(db_result, "gemma-2b-tools", "FORCED_LOCAL", is_stream=is_stream, state=state) if local_llm_with_tools and local_llm_with_tools.is_available(): try: local_tool_response = local_llm_with_tools.chat(message) if local_tool_response and local_tool_response.strip(): return send_chat_response(local_tool_response, "gemma-2b-tools", "FORCED_LOCAL", is_stream=is_stream, state=state) except Exception as e: print(f"Force local tools error: {e}") # For conversational / technical questions with @local, stream directly via Gemma-2B if hybrid_agent and hybrid_agent.local_model and hybrid_agent.local_model.is_available(): try: if is_stream: gen = hybrid_agent.local_model.generate_stream(message, max_tokens=150, context=context) return send_chat_response(gen, "gemma-2b-local", "FORCED_LOCAL", is_stream=True, state=state) else: local_response = hybrid_agent.local_model.generate(message, max_tokens=150, context=context) if local_response and local_response.strip(): return send_chat_response(local_response, "gemma-2b-local", "FORCED_LOCAL", is_stream=False, state=state) except Exception as e: print(f"Force local model error: {e}") # Fallback to local tools if direct generation was unavailable or empty if local_llm_with_tools and local_llm_with_tools.is_available(): try: local_tool_response = local_llm_with_tools.chat(message) if local_tool_response and local_tool_response.strip(): return send_chat_response(local_tool_response, "gemma-2b-tools", "FORCED_LOCAL", is_stream=is_stream, state=state) except Exception as e: print(f"Force local tools fallback error: {e}") if _model_downloading: return send_chat_response( "⏳ **Local Gemma-2B is currently downloading in the background (~1.5GB).**\n\nPlease wait a moment for the download to complete, or ask without `@local` to use Cloud Gemini in the meantime.", model_used="offline", classification="FORCED_LOCAL", error=True, is_stream=is_stream, state=state ) return send_chat_response( "⚠️ Local Gemma-2B model is not available right now. Please try again in a few moments, or ask without `@local` to use Cloud Gemini.", model_used="offline", classification="FORCED_LOCAL", error=True, is_stream=is_stream, state=state ) # 1. Simple queries (Greetings, general assistant questions, small talk) -> Local Gemma-2B if not force_cloud and QueryClassifier.is_simple(message): if hybrid_agent and hybrid_agent.local_model and hybrid_agent.local_model.is_available(): try: if is_stream: gen = hybrid_agent.local_model.generate_stream(message, max_tokens=100, context=context) return send_chat_response(gen, "gemma-2b-local", "SIMPLE", is_stream=True, state=state) else: local_response = hybrid_agent.local_model.generate(message, max_tokens=100, context=context) if local_response and local_response.strip(): return send_chat_response(local_response, "gemma-2b-local", "SIMPLE", is_stream=False, state=state) except Exception as e: print(f"Local model simple query error: {e}") # 2. Ticket queries & direct database lookups -> Local database / Local tools if not force_cloud and _is_ticket_query(message_lower): db_result = query_database_directly(message) if db_result: return send_chat_response(db_result, "gemma-2b-tools", "DATABASE_DIRECT", is_stream=is_stream, state=state) if local_llm_with_tools and local_llm_with_tools.is_available(): try: local_tool_response = local_llm_with_tools.chat(message) if local_tool_response and local_tool_response.strip(): return send_chat_response(local_tool_response, "gemma-2b-tools", "TICKET_TOOL", is_stream=is_stream, state=state) except Exception as e: print(f"Local tools error on ticket query: {e}") # 3. Technical issues, code debugging, and triage -> Cloud Gemini (with local fallback) issue_text = state.get("last_issue") or message state["last_issue"] = issue_text ask_fix_confirmation = _is_troubleshoot_request(message_lower) # If user explicitly requested @cloud, but Cloud Gemini is not configured/available if force_cloud and adk_agent is None: return send_chat_response( "@cloud was requested, but Cloud Gemini is not available right now. Check API key/network and try again.", model_used="gemini-cloud", classification="FORCED_CLOUD", error=True, is_stream=is_stream, state=state ) # Try Cloud Gemini first when available if adk_agent is not None: try: if is_stream: gen = stream_cloud_gemini(adk_agent, issue_text, image_bytes, image_mime) return send_chat_response( gen, model_used="gemini-cloud", classification="FORCED_CLOUD" if force_cloud else "TRIAGE", ask_fix_confirmation=ask_fix_confirmation, is_stream=True, state=state ) else: import asyncio import uuid from google.adk.runners import InMemoryRunner from google.genai import types async def run_agent_async(include_image: bool = True): parts = [types.Part(text=issue_text)] if include_image and image_bytes is not None: parts.append(types.Part.from_bytes(data=image_bytes, mime_type=image_mime)) content = types.Content(role="user", parts=parts) import warnings with warnings.catch_warnings(): warnings.filterwarnings("ignore", message=".*mismatch.*") runner = InMemoryRunner(agent=adk_agent, app_name="software_bug_assistant") user_id = "web_ui_user" session_id = f"session_{uuid.uuid4().hex[:8]}" await runner.session_service.create_session( app_name="software_bug_assistant", user_id=user_id, session_id=session_id, ) response_chunks = [] async for event in runner.run_async( user_id=user_id, session_id=session_id, new_message=content ): if hasattr(event, 'content') and event.content: if hasattr(event.content, 'parts') and event.content.parts: for response_part in event.content.parts: if hasattr(response_part, 'text') and response_part.text: response_chunks.append(response_part.text) elif hasattr(event.content, 'text'): response_chunks.append(event.content.text) elif hasattr(event, 'text'): response_chunks.append(event.text) elif isinstance(event, str): response_chunks.append(event) return ''.join(response_chunks) if response_chunks else None try: response_text = asyncio.run(run_agent_async(include_image=True)) except Exception as image_run_err: if image_bytes is not None: print(f"Cloud triage image processing failed, retrying text-only: {image_run_err}") response_text = asyncio.run(run_agent_async(include_image=False)) response_text = ( "I couldn't fully process the image this time, so I used your text prompt instead.\n\n" + (response_text or "") ) else: raise if response_text and response_text.strip(): response_text = response_text.strip() try: stack = query_stackexchange(issue_text) stack_section = _format_search_section("📚 StackOverflow Related Issues", stack.get("results", [])) if stack_section: response_text += "\n\n---\n\n" + stack_section except Exception as search_err: print(f"StackExchange search skipped: {search_err}") return send_chat_response( response_text, model_used="gemini-cloud", classification="FORCED_CLOUD" if force_cloud else "TRIAGE", ask_fix_confirmation=ask_fix_confirmation, is_stream=False, state=state ) except Exception as e: print(f"Cloud triage error: {e}") import traceback traceback.print_exc() if force_cloud: return send_chat_response( f"Cloud analysis failed: {e}. Please retry or ask without `@cloud` to use the local Gemma model.", model_used="gemini-cloud", classification="FORCED_CLOUD", error=True, is_stream=is_stream, state=state ) print("Falling back to local models after cloud failure...") # Fallback to local models (when cloud is unavailable or failed) if local_llm_with_tools and local_llm_with_tools.is_available(): try: local_tool_response = local_llm_with_tools.chat(issue_text) response_text = local_tool_response or "" response_text = response_text.strip() or "I can help, but I need more details about the issue." return send_chat_response( response_text, model_used="gemma-2b-tools", classification="TRIAGE", ask_fix_confirmation=ask_fix_confirmation, is_stream=is_stream, state=state ) except Exception as local_err: print(f"Local LLM with tools fallback failed: {local_err}") if hybrid_agent and hybrid_agent.local_model and hybrid_agent.local_model.is_available(): try: if is_stream: gen = hybrid_agent.local_model.generate_stream(issue_text, max_tokens=150, context=context) return send_chat_response( gen, model_used="gemma-2b-local", classification="TRIAGE", ask_fix_confirmation=ask_fix_confirmation, is_stream=True, state=state ) else: local_response = hybrid_agent.local_model.generate(issue_text, max_tokens=150, context=context) if local_response and local_response.strip(): return send_chat_response( local_response.strip(), model_used="gemma-2b-local", classification="TRIAGE", ask_fix_confirmation=ask_fix_confirmation, is_stream=False, state=state ) except Exception as local_err: print(f"Local model fallback failed: {local_err}") # Final offline fallback if neither cloud nor local model could respond response_text = ( "I'm currently unable to reach the AI models. Please check your connection or provide more details." ) try: stack = query_stackexchange(issue_text) stack_section = _format_search_section("📚 StackOverflow Related Issues", stack.get("results", [])) if stack_section: response_text += "\n\n---\n\n" + stack_section except Exception as search_err: print(f"StackExchange search skipped: {search_err}") return send_chat_response( response_text, model_used="offline", classification="TRIAGE", ask_fix_confirmation=ask_fix_confirmation, is_stream=is_stream, state=state ) if __name__ == '__main__': print("=" * 70) print("🚀 Starting Unified Hybrid AI Web UI") print("=" * 70) print() import threading import subprocess def background_download(): print("Starting background download of Gemma model...") try: subprocess.run(["uv", "run", "python", "download_gemma.py"], check=False) print("Gemma background download finished (or already exists).") except Exception as e: print(f"Background download error: {e}") threading.Thread(target=background_download, daemon=True).start() # Initialize agents at startup print("Initializing agents...") init_agents() if hybrid_agent and hybrid_agent.local_model and hybrid_agent.local_model.is_available(): print("✓ Local Gemma-2B (Simple): READY") else: print("⚠ Local Gemma-2B (Simple): NOT AVAILABLE") if local_llm_with_tools and local_llm_with_tools.is_available(): print("✓ Local Gemma-2B (Tool Calling): READY") print(f" └── {len(local_llm_with_tools.tool_registry.tools)} tools available") else: print("⚠ Local Gemma-2B (Tool Calling): NOT AVAILABLE") if adk_agent: print("✓ Cloud Gemini + Tools: READY") else: print("⚠ Cloud Gemini: NOT AVAILABLE (check GOOGLE_API_KEY and .env)") print() port = int(os.environ.get("PORT", 7860)) print(f"📍 Access the UI at: http://localhost:{port}") print() print("Features:") print(" • Simple queries → 🏠 Local Gemma-2B (fast, free)") print(" • Complex queries → ☁️ Cloud Gemini + Tools") print(" • Fallback → 🔧 Local Gemma-2B with Tool Calling") print(" • Single unified interface") print() print("Press Ctrl+C to stop") print("=" * 70) print() # Suppress Flask/click banner to avoid Windows console error 6 import click click.echo = lambda *a, **kw: None from werkzeug.serving import run_simple run_simple('0.0.0.0', port, app, use_reloader=False, use_debugger=False)