import json import os import time from typing import List, Tuple import gradio as gr import requests from duckduckgo_search import DDGS DEFAULT_MODEL = "gemma-4-31b-it" GOOGLE_API_BASE = "https://generativelanguage.googleapis.com/v1beta/models" DEBUG_LOG_PATH = "debug-1caba1.log" DEBUG_SESSION_ID = "1caba1" def debug_log(run_id: str, hypothesis_id: str, location: str, message: str, data: dict) -> None: payload = { "sessionId": DEBUG_SESSION_ID, "runId": run_id, "hypothesisId": hypothesis_id, "location": location, "message": message, "data": data, "timestamp": int(time.time() * 1000), } # region agent log with open(DEBUG_LOG_PATH, "a", encoding="utf-8") as f: f.write(json.dumps(payload, ensure_ascii=True) + "\n") # endregion def normalize_model_id(model_input: str) -> str: cleaned = (model_input or "").strip().lower().replace(" ", "-") aliases = { "gemma-4-31b-it": "gemma-4-31b-it", "gemma4-31b-it": "gemma-4-31b-it", "gemma-4-31b-instruct": "gemma-4-31b-it", "gemma-4-31b": "gemma-4-31b-it", } return aliases.get(cleaned, cleaned) def parse_model_response(text: str, option_a: str, option_b: str) -> Tuple[str, str, str]: fallback_choice = option_a or option_b or "Expand to Singapore Market" fallback_reason = "High expected growth with favorable risk profile." fallback_confidence = "76%" if not text: return fallback_choice, fallback_reason, fallback_confidence lines = [line.strip() for line in text.split("\n") if line.strip()] choice = "" reason = "" confidence = "" for line in lines: lower = line.lower() if not choice and lower.startswith("choice:"): choice = line[7:].strip() elif not reason and lower.startswith("reason:"): reason = line[7:].strip() elif not confidence and lower.startswith("confidence:"): confidence = line[11:].strip() if not reason and lines: reason = " ".join(lines[:2]) return ( choice or fallback_choice, reason or fallback_reason, confidence or fallback_confidence, ) def build_web_context(dilemma: str, option_a: str, option_b: str, max_results: int) -> str: query = " ".join( [ dilemma or "", option_a or "", option_b or "", "market trends risk analysis recent data", ] ).strip() if not query: return "No web context available." rows: List[str] = [] # region agent log debug_log( "pre-fix", "H1", "app.py:build_web_context", "DuckDuckGo query prepared", {"query_len": len(query), "max_results": int(max_results)}, ) # endregion with DDGS() as ddgs: results = ddgs.text(query, max_results=max_results) for idx, item in enumerate(results, start=1): title = (item.get("title") or "").strip() href = (item.get("href") or "").strip() body = (item.get("body") or "").strip() if not (title or href or body): continue rows.append(f"{idx}. {title}\nURL: {href}\nSnippet: {body}") # region agent log debug_log( "pre-fix", "H1", "app.py:build_web_context", "DuckDuckGo query completed", {"result_count": len(rows)}, ) # endregion return "\n\n".join(rows) if rows else "No web results returned by DuckDuckGo." def call_google_model( api_key: str, model: str, dilemma: str, option_a: str, option_b: str, web_context: str, ) -> str: endpoint = f"{GOOGLE_API_BASE}/{model}:generateContent?key={api_key}" # region agent log debug_log( "pre-fix", "H3", "app.py:call_google_model", "Prepared model endpoint", {"model": model, "endpoint_without_key": f"{GOOGLE_API_BASE}/{model}:generateContent"}, ) # endregion prompt = "\n".join( [ "You are a strategic decision assistant.", "Given a dilemma and two options, pick the best option.", "Return exactly 3 lines in this strict format:", "CHOICE: ", "REASON: ", "CONFIDENCE: <0-100%>", "", "Use the web context below as additional evidence when relevant.", f"WEB_CONTEXT:\n{web_context or 'No web context provided.'}", "", f"DILEMMA: {dilemma or 'N/A'}", f"OPTION_ALPHA: {option_a or 'N/A'}", f"OPTION_BETA: {option_b or 'N/A'}", ] ) response = requests.post( endpoint, headers={"Content-Type": "application/json"}, data=json.dumps({"contents": [{"parts": [{"text": prompt}]}]}), timeout=60, ) data = response.json() # region agent log debug_log( "pre-fix", "H4", "app.py:call_google_model", "Model response received", { "status_code": int(response.status_code), "ok": bool(response.ok), "has_candidates": isinstance(data.get("candidates", []), list), "error_message": data.get("error", {}).get("message", ""), }, ) # endregion if not response.ok: message = data.get("error", {}).get("message", "Google AI Studio request failed.") raise RuntimeError(message) candidates = data.get("candidates", []) if not candidates: return "" parts = candidates[0].get("content", {}).get("parts", []) return "\n".join(part.get("text", "") for part in parts).strip() def analyze_decision( dilemma: str, option_a: str, option_b: str, model_input: str, api_key_input: str, use_web_search: bool, max_search_results: int, ) -> Tuple[str, str, str, str, str]: api_key = (api_key_input or "").strip() or os.getenv("GOOGLE_API_KEY", "").strip() model = normalize_model_id(model_input) or DEFAULT_MODEL web_context = "Web search disabled." # region agent log debug_log( "pre-fix", "H2", "app.py:analyze_decision", "Decision analysis started", { "has_api_key": bool(api_key), "model": model, "use_web_search": bool(use_web_search), "max_search_results": int(max_search_results), "has_dilemma": bool((dilemma or "").strip()), "has_option_a": bool((option_a or "").strip()), "has_option_b": bool((option_b or "").strip()), }, ) # endregion if not api_key: return ( option_a or option_b or "Fallback Recommendation", "Missing API key. Add `GOOGLE_API_KEY` in Hugging Face Space secrets or input it in the field.", "N/A", model, web_context, ) try: if use_web_search: safe_results = max(1, min(int(max_search_results), 8)) web_context = build_web_context(dilemma, option_a, option_b, safe_results) raw = call_google_model(api_key, model, dilemma, option_a, option_b, web_context) choice, reason, confidence = parse_model_response(raw, option_a, option_b) # region agent log debug_log( "pre-fix", "H5", "app.py:analyze_decision", "Parsed model output", { "raw_len": len(raw or ""), "choice_len": len(choice or ""), "reason_len": len(reason or ""), "confidence": confidence, }, ) # endregion return choice, reason, confidence, model, web_context except Exception as exc: # region agent log debug_log( "pre-fix", "H4", "app.py:analyze_decision", "Analyze decision failed", {"error": str(exc)}, ) # endregion return ( option_a or option_b or "Fallback Recommendation", f"Model call failed: {exc}", "N/A", model, web_context, ) with gr.Blocks(theme=gr.themes.Soft(), title="The Oracle | Decision Engine") as demo: gr.Markdown("## The Oracle - Hugging Face Spaces App") gr.Markdown( "Enter your dilemma and options, then run analysis with Google AI Studio (Gemma). " "Default model is `gemma-4-31b-it`." ) with gr.Row(): dilemma = gr.Textbox( label="Core dilemma", placeholder="Should we pivot strategy toward AI-integrated logistics by Q3?", lines=4, ) with gr.Row(): option_a = gr.Textbox(label="Option Alpha", placeholder="Immediate full pivot") option_b = gr.Textbox(label="Option Beta", placeholder="Incremental integration") with gr.Row(): model_input = gr.Textbox(label="Model ID", value=DEFAULT_MODEL) api_key_input = gr.Textbox( label="Google API key (optional if set in Space secrets as GOOGLE_API_KEY)", type="password", ) with gr.Row(): use_web_search = gr.Checkbox(label="Enable DuckDuckGo web search context", value=True) max_search_results = gr.Slider( label="DuckDuckGo results to use", minimum=1, maximum=8, step=1, value=4, ) run_btn = gr.Button("Generate Analysis", variant="primary") with gr.Row(): out_choice = gr.Textbox(label="Recommended Choice") out_reason = gr.Textbox(label="Reason") with gr.Row(): out_confidence = gr.Textbox(label="Confidence") out_model = gr.Textbox(label="Model Used") out_web_context = gr.Textbox(label="DuckDuckGo Research Context", lines=12) run_btn.click( fn=analyze_decision, inputs=[ dilemma, option_a, option_b, model_input, api_key_input, use_web_search, max_search_results, ], outputs=[out_choice, out_reason, out_confidence, out_model, out_web_context], ) if __name__ == "__main__": demo.launch()