feat: integrate PythonInterpreter via SandboxFusion

app.py

@@ -34,6 +34,19 @@ DEFAULT_MAX_SEARCH_RESULTS = 10
 MAX_SEARCH_QUERIES_PER_CALL = 10
 MAX_VISIT_URLS_PER_CALL = 5
 
+# Python Interpreter via SandboxFusion ----------------------------------------
+# Set ENABLE_PYTHON_TOOL=true and SANDBOX_FUSION_ENDPOINTS=http://host:8080,...
+# in Space Secrets to enable the python_interpreter tool.
+ENABLE_PYTHON_TOOL = os.getenv("ENABLE_PYTHON_TOOL", "").strip().lower() in ("1", "true", "yes")
+_SANDBOX_FUSION_ENDPOINTS_RAW = (
+    os.getenv("SANDBOX_FUSION_ENDPOINTS") or os.getenv("SANDBOX_FUSION_ENDPOINT") or ""
+).strip()
+
+def _get_python_endpoints() -> List[str]:
+    if _SANDBOX_FUSION_ENDPOINTS_RAW:
+        return [e.strip() for e in _SANDBOX_FUSION_ENDPOINTS_RAW.split(",") if e.strip()]
+    return []
+
 PAPER_URL = os.getenv("PAPER_URL", "https://arxiv.org/abs/2605.24218")
 CODE_URL = os.getenv("CODE_URL", "https://github.com/OSU-NLP-Group/QUEST")
 DATASET_URL = os.getenv("DATASET_URL", "https://huggingface.co/collections/osunlp/quest")
@@ -42,9 +55,8 @@ MODEL_URL = os.getenv("MODEL_URL", "https://huggingface.co/osunlp/QUEST-35B-RL")
 
 # --- System prompt ---------------------------------------------------------
 # Hosted Space prompt (based on inference/prompt.py in the GitHub repo).
-# PythonInterpreter is intentionally omitted here: this Space has no Python
-# sandbox backend. To try PythonInterpreter, run offline inference from
-# https://github.com/OSU-NLP-Group/QUEST with ENABLE_PYTHON_TOOL=true.
+# Set ENABLE_PYTHON_TOOL=true and SANDBOX_FUSION_ENDPOINTS in Space Secrets
+# to enable the PythonInterpreter tool.
 QUEST_SYSTEM_PROMPT = """You are a deep research assistant. Your core function is to conduct thorough, multi-source investigations into any topic. You must handle both broad, open-domain inquiries and queries within specialized academic fields. For every request, synthesize information from credible, diverse sources to deliver a comprehensive, accurate, and objective response. When you have gathered sufficient information and are ready to provide the definitive response, you must enclose the entire final answer within <answer></answer> tags.
 
 # Tools
@@ -56,6 +68,7 @@ You are provided with function signatures within <tools></tools> XML tags:
 {"type": "function", "function": {"name": "search", "description": "Perform Google web searches then returns a string of the top search results. Accepts up to 10 queries per call.", "parameters": {"type": "object", "properties": {"query": {"type": "array", "items": {"type": "string", "description": "The search query."}, "minItems": 1, "maxItems": 10, "description": "The list of search queries. Use at most 10 queries per search call."}}, "required": ["query"]}}}
 {"type": "function", "function": {"name": "visit", "description": "Visit webpage(s) and return the summary of the content. Accepts up to 5 URLs per call.", "parameters": {"type": "object", "properties": {"url": {"type": "array", "items": {"type": "string"}, "minItems": 1, "maxItems": 5, "description": "The URL(s) of the webpage(s) to visit. Use at most 5 URLs per visit call."}, "goal": {"type": "string", "description": "The specific information goal for visiting webpage(s)."}}, "required": ["url", "goal"]}}}
 {"type": "function", "function": {"name": "google_scholar", "description": "Leverage Google Scholar to retrieve relevant information from academic publications. Accepts up to 10 queries per call.", "parameters": {"type": "object", "properties": {"query": {"type": "array", "items": {"type": "string", "description": "The search query."}, "minItems": 1, "maxItems": 10, "description": "The list of search queries for Google Scholar. Use at most 10 queries per call."}}, "required": ["query"]}}}
+{PYTHON_TOOL_DEF}
 </tools>
 
 Tool call batching limits: search and google_scholar process at most 10 queries per call; visit processes at most 5 URLs per call. If more sources are needed, make additional tool calls in later turns.
@@ -225,8 +238,32 @@ def build_current_date_context() -> str:
     )
 
 
+_PYTHON_TOOL_SIGNATURE = '{"type": "function", "function": {"name": "python_interpreter", "description": "Execute Python code in a sandboxed environment to perform calculations, data analysis, string processing, or any computational task. Always use print() for any output you want to see in the results.", "parameters": {"type": "object", "properties": {"code": {"type": "string", "description": "The Python code to execute. Use print() for output."}}, "required": ["code"]}}}'
+
+
 def build_system_prompt() -> str:
-    return build_current_date_context() + "\n\n" + QUEST_SYSTEM_PROMPT + current_date_iso()
+    if ENABLE_PYTHON_TOOL and _get_python_endpoints():
+        python_line = _PYTHON_TOOL_SIGNATURE
+        batching_note = (
+            "Tool call batching limits: search and google_scholar process at most 10 queries "
+            "per call; visit processes at most 5 URLs per call; python_interpreter executes one "
+            "code block per call. If more sources are needed, make additional tool calls in later turns."
+        )
+    else:
+        python_line = ""
+        batching_note = (
+            "Tool call batching limits: search and google_scholar process at most 10 queries "
+            "per call; visit processes at most 5 URLs per call. If more sources are needed, "
+            "make additional tool calls in later turns."
+        )
+    prompt = QUEST_SYSTEM_PROMPT.replace("{PYTHON_TOOL_DEF}", python_line)
+    # Replace the static batching note with a potentially updated one
+    prompt = prompt.replace(
+        "Tool call batching limits: search and google_scholar process at most 10 queries per call; "
+        "visit processes at most 5 URLs per call. If more sources are needed, make additional tool calls in later turns.",
+        batching_note,
+    )
+    return build_current_date_context() + "\n\n" + prompt + current_date_iso()
 
 
 def build_user_prompt(question: str) -> str:
@@ -385,6 +422,9 @@ def format_tool_response_for_model(tool_name: Optional[str], response: Any) -> s
             )
         return _format_visit_single_for_model(response)
 
+    if tool_name == "python_interpreter":
+        return _clean_tool_text(response.get("result", response.get("error", str(response))))
+
     if response.get("error"):
         return "[Tool Error] " + _clean_tool_text(response.get("error"))
     return str(response)
@@ -2445,6 +2485,49 @@ def run_visit(
     return _run_visit_single(str(url or "").strip(), max_chars, goal)
 
 
+def _call_python_interpreter(code: str, timeout: int = 50) -> str:
+    """Run Python code on a SandboxFusion endpoint. Mirrors tool_python.py logic."""
+    import random as _random
+    endpoints = _get_python_endpoints()
+    if not endpoints:
+        return "[Python Interpreter Error]: No sandbox endpoints configured. Set SANDBOX_FUSION_ENDPOINTS."
+
+    # Strip markdown code fences if model wrapped the code in them.
+    triple = re.search(r'```[^\n]*\n(.+?)```', code, re.DOTALL)
+    if triple:
+        code = triple.group(1)
+    if not code or not code.strip():
+        return "[Python Interpreter Error]: Empty code."
+
+    try:
+        from sandbox_fusion import run_code, RunCodeRequest  # type: ignore
+    except ImportError:
+        return "[Python Interpreter Error]: sandbox_fusion package not installed."
+
+    last_error: Optional[str] = None
+    for attempt in range(8):
+        endpoint = _random.choice(endpoints)
+        try:
+            result = run_code(
+                RunCodeRequest(code=code, language="python", run_timeout=timeout),
+                max_attempts=1,
+                client_timeout=timeout,
+                endpoint=endpoint,
+            )
+            parts: List[str] = []
+            if result.run_result.stdout:
+                parts.append(f"stdout:\n{result.run_result.stdout}")
+            if result.run_result.stderr:
+                parts.append(f"stderr:\n{result.run_result.stderr}")
+            if result.run_result.execution_time >= timeout - 1:
+                parts.append("[PythonInterpreter Error] TimeoutError: Execution timed out.")
+            return "\n".join(parts) if parts else "Finished execution."
+        except Exception as exc:
+            last_error = f"[Python Interpreter Error]: {type(exc).__name__}: {exc} (endpoint: {endpoint})"
+            continue
+    return last_error or "[Python Interpreter Error]: All attempts failed."
+
+
 def _build_client_for_model(model: str) -> Tuple[InferenceClient, str, List[str]]:
     """Returns (client, primary_model_id, fallback_model_ids).
 
@@ -2864,6 +2947,17 @@ def build_research_agent(
                     f"📚 turn {turn}: scholar {ok_count}/{len(per_q)} ok"
                 )
                 yield _emit()
+            elif tool_name == "python_interpreter":
+                code = str(tool_args.get("code", "")).strip()
+                code_preview = (code[:60] + "…") if len(code) > 60 else code
+                status_lines.append(f"🐍 turn {turn}: running Python — `{code_preview}`")
+                yield _emit()
+                py_result = _call_python_interpreter(code)
+                tool_response = {"ok": not py_result.startswith("[Python Interpreter Error]"), "result": py_result}
+                status_lines.append(
+                    f"{'✅' if tool_response['ok'] else '⚠️'} turn {turn}: python finished"
+                )
+                yield _emit()
             else:
                 tool_response = {"ok": False, "error": f"Unknown tool: {tool_name}"}
                 status_lines.append(f"⚠️ turn {turn}: unknown tool `{tool_name}`")

requirements.txt

@@ -5,3 +5,5 @@ requests==2.32.3
 beautifulsoup4==4.12.3
 openai>=1.40.0
 tiktoken>=0.7.0
+sandbox-fusion>=0.1.0
+json5>=0.9.0