Initial document-explorer product

Gradio web app for exploring document workspaces with an LLM agent.
Uses E2B sandboxes for code execution, depends on a-simple-llm-harness
for the agent loop.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

README.md

@@ -0,0 +1,34 @@
+---
+title: Document Explorer
+emoji: 📄
+colorFrom: blue
+colorTo: green
+sdk: gradio
+sdk_version: "6.9.0"
+app_file: app.py
+pinned: false
+---
+
+# Document Explorer
+
+Upload text or CSV files and ask questions. The model explores your documents
+by writing and running Python code in a sandboxed E2B environment.
+
+Built on [a-simple-llm-harness](https://github.com/chuckfinca/a-simple-llm-harness).
+
+## Setup
+
+Set environment variables:
+
+```
+LH_MODEL=openrouter/qwen/qwen3-coder
+LH_ACCESS_TOKEN=your-secret-token
+E2B_API_KEY=your-e2b-api-key
+```
+
+Run locally:
+
+```
+pip install -r requirements.txt
+python app.py
+```

app.py

@@ -0,0 +1,225 @@
+"""Web interface for exploring document workspaces with an LLM agent.
+
+Usage:
+    python app.py
+
+Requires LH_MODEL and LH_ACCESS_TOKEN in .env or environment.
+Uses E2B sandboxes for code execution (no Docker required).
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import time
+from dataclasses import asdict
+from pathlib import Path
+
+import gradio as gr
+import litellm
+from dotenv import load_dotenv
+
+from llm_harness.agent import run_agent_loop
+from llm_harness.prompt import build_system_prompt
+from llm_harness.tools import TOOL_DEFINITIONS
+from llm_harness.types import Message, ToolCallEvent, ToolResultEvent
+
+from sandbox_e2b import run_python as e2b_run_python
+
+load_dotenv()
+litellm.suppress_debug_info = True
+
+MODEL = os.environ.get("LH_MODEL", "")
+ACCESS_TOKEN = os.environ.get("LH_ACCESS_TOKEN", "")
+MAX_SESSION_COST = float(os.environ.get("LH_MAX_SESSION_COST", "0.50"))
+
+
+def authenticate(username: str, password: str) -> bool:
+    return password == ACCESS_TOKEN
+
+
+def save_uploaded_files(files: list[str]) -> Path:
+    workspace = Path(tempfile.mkdtemp(prefix="lh-workspace-"))
+    for file_path in files:
+        src = Path(file_path)
+        (workspace / src.name).write_bytes(src.read_bytes())
+    return workspace
+
+
+def format_stats(trace: object) -> str:
+    cost_str = f"${trace.cost:.4f}" if trace.cost else "n/a"
+    cached = trace.cached_tokens
+    cache_str = f" ({cached} cached)" if cached else ""
+    scratchpad = len(trace.scratch_files)
+    model_name = trace.model.split("/")[-1] if trace.model else ""
+    stats = (
+        f"*{model_name}"
+        f" · {trace.prompt_tokens + trace.completion_tokens:,} tokens{cache_str}"
+        f" · {len(trace.tool_calls)} tool calls"
+        f" · {trace.wall_time_s:.1f}s"
+        f" · {cost_str}"
+    )
+    if scratchpad:
+        stats += f" · {scratchpad} scratchpad files"
+    return stats + "*"
+
+
+def chat(
+    message: str,
+    history: list[dict],
+    files: list[str] | None,
+    workspace_path: str,
+    scratch_path: str,
+    session_cost: float,
+):
+    if not MODEL:
+        yield (
+            "Error: LH_MODEL not set.",
+            workspace_path,
+            scratch_path,
+            session_cost,
+        )
+        return
+
+    if session_cost >= MAX_SESSION_COST:
+        yield (
+            f"Session cost limit reached (${session_cost:.2f} / "
+            f"${MAX_SESSION_COST:.2f}). Start a new session.",
+            workspace_path,
+            scratch_path,
+            session_cost,
+        )
+        return
+
+    # Set up workspace from uploaded files (first message only)
+    workspace = Path(workspace_path) if workspace_path else None
+    if files and not workspace:
+        workspace = save_uploaded_files(files)
+        workspace_path = str(workspace)
+
+    # Set up scratchpad (once per session)
+    if not scratch_path:
+        scratch_path = tempfile.mkdtemp(prefix="lh-scratch-")
+    scratch_dir = Path(scratch_path)
+
+    # Build messages from Gradio history
+    system_prompt = build_system_prompt(base_prompt="", workspace=workspace)
+    messages: list[Message] = [{"role": "system", "content": system_prompt}]
+    messages.extend({"role": e["role"], "content": e["content"]} for e in history)
+    messages.append({"role": "user", "content": message})
+
+    # Run agent loop with E2B sandbox
+    start = time.monotonic()
+    agent_run = run_agent_loop(
+        model=MODEL,
+        messages=messages,
+        tools=TOOL_DEFINITIONS,
+        completion=litellm.completion,
+        workspace=workspace,
+        scratch_dir=scratch_dir,
+        sandbox_fn=e2b_run_python,
+    )
+
+    tool_call_count = 0
+    try:
+        for event in agent_run:
+            if isinstance(event, ToolCallEvent):
+                tool_call_count += 1
+                status = f"*Exploring documents ({tool_call_count} tool calls)...*"
+                yield status, workspace_path, scratch_path, session_cost
+            elif isinstance(event, ToolResultEvent):
+                continue
+            else:
+                cost = agent_run.trace.cost or 0
+                session_cost += cost
+    except Exception as exc:
+        yield (
+            f"Error: {exc}",
+            workspace_path,
+            scratch_path,
+            session_cost,
+        )
+        return
+
+    trace = agent_run.trace
+    trace.wall_time_s = round(time.monotonic() - start, 2)
+    answer = trace.answer or "(no answer)"
+    stats = format_stats(trace)
+
+    yield (
+        f"{answer}\n\n---\n{stats}",
+        workspace_path,
+        scratch_path,
+        session_cost,
+    )
+
+
+def build_app() -> gr.Blocks:
+    with gr.Blocks(title="Document Explorer", theme=gr.themes.Soft()) as demo:
+        gr.Markdown(
+            "# Document Explorer\n"
+            "Upload text or CSV files, then ask questions. "
+            "The model explores your documents by writing and running Python code."
+        )
+
+        workspace_state = gr.State("")
+        scratch_state = gr.State("")
+        cost_state = gr.State(0.0)
+
+        with gr.Accordion("Upload documents", open=True):
+            file_upload = gr.File(
+                label="Text, CSV, Markdown, or JSON files",
+                file_count="multiple",
+                file_types=[".txt", ".csv", ".md", ".json"],
+            )
+
+        chatbot = gr.Chatbot(height=500)
+        msg = gr.Textbox(
+            placeholder="Ask a question about your documents...",
+            label="",
+            show_label=False,
+        )
+
+        def respond(
+            message, history, files, workspace_path, scratch_path, session_cost
+        ):
+            history = history or []
+            history.append({"role": "user", "content": message})
+
+            for response, wp, sp, sc in chat(
+                message, history[:-1], files, workspace_path, scratch_path, session_cost
+            ):
+                history_with_response = [
+                    *history,
+                    {"role": "assistant", "content": response},
+                ]
+                yield history_with_response, "", wp, sp, sc
+
+        msg.submit(
+            respond,
+            inputs=[
+                msg,
+                chatbot,
+                file_upload,
+                workspace_state,
+                scratch_state,
+                cost_state,
+            ],
+            outputs=[
+                chatbot,
+                msg,
+                workspace_state,
+                scratch_state,
+                cost_state,
+            ],
+        )
+
+    return demo
+
+
+if __name__ == "__main__":
+    if not ACCESS_TOKEN:
+        print("WARNING: LH_ACCESS_TOKEN not set — app is unprotected")
+
+    app = build_app()
+    app.launch(auth=authenticate if ACCESS_TOKEN else None)

requirements.txt

@@ -0,0 +1,4 @@
+a-simple-llm-harness @ git+https://github.com/chuckfinca/a-simple-llm-harness.git
+e2b-code-interpreter>=2.5
+gradio>=5.0
+python-dotenv

sandbox_e2b.py

@@ -0,0 +1,100 @@
+"""E2B-based sandbox for cloud deployment without Docker."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from e2b_code_interpreter import Sandbox
+
+from llm_harness.sandbox import TIMEOUT_SECONDS, _truncate
+
+# Reuse a sandbox across tool calls within a session.
+# The caller manages the lifecycle via create/close.
+_active_sandbox: Sandbox | None = None
+
+
+def get_or_create_sandbox(
+    workspace: Path | None = None,
+    scratch_dir: Path | None = None,
+) -> Sandbox:
+    """Get the active sandbox, creating one if needed and uploading workspace files."""
+    global _active_sandbox
+    if _active_sandbox is not None:
+        return _active_sandbox
+
+    _active_sandbox = Sandbox.create(timeout=300)
+
+    # Create workspace and scratchpad directories in user-writable home
+    _active_sandbox.commands.run("mkdir -p /home/user/workspace /home/user/scratchpad")
+    # Symlink to expected paths
+    _active_sandbox.commands.run(
+        "ln -sf /home/user/workspace /workspace; "
+        "ln -sf /home/user/scratchpad /scratchpad",
+        user="root",
+    )
+
+    # Upload workspace files
+    if workspace is not None:
+        for file_path in workspace.iterdir():
+            if file_path.is_file():
+                _active_sandbox.files.write(
+                    f"/home/user/workspace/{file_path.name}",
+                    file_path.read_bytes(),
+                )
+
+    return _active_sandbox
+
+
+def close_sandbox() -> None:
+    global _active_sandbox
+    if _active_sandbox is not None:
+        _active_sandbox.kill()
+        _active_sandbox = None
+
+
+def run_python(
+    code: str,
+    *,
+    workspace: Path | None = None,
+    scratch_dir: Path | None = None,
+    timeout: int = TIMEOUT_SECONDS,
+) -> str:
+    """Execute Python code in an E2B sandbox. Same interface as sandbox.run_python."""
+    sandbox = get_or_create_sandbox(workspace, scratch_dir)
+
+    try:
+        execution = sandbox.run_code(code, timeout=timeout)
+
+        stdout = "\n".join(
+            line if isinstance(line, str) else line.text
+            for line in execution.logs.stdout
+        )
+        stderr = "\n".join(
+            line if isinstance(line, str) else line.text
+            for line in execution.logs.stderr
+        )
+
+        if execution.error:
+            stderr += f"\n{execution.error.name}: {execution.error.value}"
+            exit_code = 1
+        else:
+            exit_code = 0
+
+        return json.dumps(
+            {
+                "stdout": _truncate(stdout),
+                "stderr": _truncate(stderr),
+                "exit_code": exit_code,
+                "timed_out": False,
+            }
+        )
+    except TimeoutError:
+        return json.dumps(
+            {
+                "stdout": "",
+                "stderr": "Execution timed out.",
+                "exit_code": -1,
+                "timed_out": True,
+            }
+        )