""" Mini Coding Agent — Gradio frontend for Hugging Face Spaces. Architecture: Browser → Gradio (this file, Python) → Elixir backend (HTTP) → LLM provider ↘ (or directly to LLM if backend is offline) The Elixir backend (see elixir_backend/) handles the actual agent loop: prompt construction, multi-file project planning, and provider dispatch. This file is purely UI + a thin HTTP client. """ from __future__ import annotations import io import json import os import re import time import zipfile from typing import Iterator import gradio as gr import requests # --------------------------------------------------------------------------- # # Config # --------------------------------------------------------------------------- # ELIXIR_BACKEND_URL = os.getenv("ELIXIR_BACKEND_URL", "http://localhost:4000") REQUEST_TIMEOUT = 180 PROVIDERS = { "OpenAI (gpt-4o-mini)": {"id": "openai", "model": "gpt-4o-mini", "key_hint": "sk-...", "url": "https://platform.openai.com/api-keys"}, "Anthropic (claude-3-5-sonnet)": {"id": "anthropic", "model": "claude-3-5-sonnet-latest", "key_hint": "sk-ant-...", "url": "https://console.anthropic.com/settings/keys"}, "Groq (llama-3.3-70b, free)": {"id": "groq", "model": "llama-3.3-70b-versatile", "key_hint": "gsk_...", "url": "https://console.groq.com/keys"}, } SYSTEM_PROMPT = """You are a senior software engineer acting as a project scaffolding agent. When the user describes a project, output a complete, runnable codebase as multiple files. OUTPUT FORMAT — strictly follow this: For each file, emit a fenced code block whose info string is the file path, e.g.: ```python:src/main.py # code here ``` ```text:README.md # Project ``` Rules: - Use the language identifier before the colon (python, javascript, typescript, html, css, json, text, etc.). - The path AFTER the colon is the file path relative to the project root. - Include a README.md and any config files (requirements.txt, package.json, etc.) needed to run the project. - Keep files focused and small. Prefer 3–8 files over one giant file. - After the files, add a short "How to run" section in plain prose. """ # --------------------------------------------------------------------------- # # Backend client # --------------------------------------------------------------------------- # def call_elixir_backend(provider: str, model: str, api_key: str, history: list[dict]) -> Iterator[str]: """Stream tokens from the Elixir backend. Falls back to direct provider call.""" try: with requests.post( f"{ELIXIR_BACKEND_URL}/api/agent/stream", json={"provider": provider, "model": model, "api_key": api_key, "system": SYSTEM_PROMPT, "messages": history}, stream=True, timeout=REQUEST_TIMEOUT, ) as r: r.raise_for_status() for line in r.iter_lines(decode_unicode=True): if not line or not line.startswith("data: "): continue payload = line[6:] if payload == "[DONE]": return try: chunk = json.loads(payload) except json.JSONDecodeError: continue if "delta" in chunk: yield chunk["delta"] elif "error" in chunk: yield f"\n\n**Backend error:** {chunk['error']}" return except requests.exceptions.RequestException as e: yield f"⚠️ Elixir backend unreachable ({e}). Falling back to direct call.\n\n" yield from call_provider_direct(provider, model, api_key, history) def call_provider_direct(provider: str, model: str, api_key: str, history: list[dict]) -> Iterator[str]: """Fallback: call the LLM provider directly from Python (no streaming, simpler).""" messages = [{"role": "system", "content": SYSTEM_PROMPT}] + history if provider in ("openai", "groq"): url = ("https://api.openai.com/v1/chat/completions" if provider == "openai" else "https://api.groq.com/openai/v1/chat/completions") headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} body = {"model": model, "messages": messages, "stream": True} try: with requests.post(url, headers=headers, json=body, stream=True, timeout=REQUEST_TIMEOUT) as r: r.raise_for_status() for line in r.iter_lines(decode_unicode=True): if not line or not line.startswith("data: "): continue data = line[6:] if data == "[DONE]": return try: delta = json.loads(data)["choices"][0]["delta"].get("content", "") except (json.JSONDecodeError, KeyError, IndexError): continue if delta: yield delta except requests.exceptions.HTTPError as e: yield f"\n\n**Provider error:** {e.response.status_code} {e.response.text[:300]}" except Exception as e: yield f"\n\n**Error:** {e}" elif provider == "anthropic": url = "https://api.anthropic.com/v1/messages" headers = {"x-api-key": api_key, "anthropic-version": "2023-06-01", "Content-Type": "application/json"} body = {"model": model, "max_tokens": 4096, "system": SYSTEM_PROMPT, "messages": history, "stream": True} try: with requests.post(url, headers=headers, json=body, stream=True, timeout=REQUEST_TIMEOUT) as r: r.raise_for_status() for line in r.iter_lines(decode_unicode=True): if not line or not line.startswith("data: "): continue try: evt = json.loads(line[6:]) except json.JSONDecodeError: continue if evt.get("type") == "content_block_delta": delta = evt.get("delta", {}).get("text", "") if delta: yield delta except requests.exceptions.HTTPError as e: yield f"\n\n**Provider error:** {e.response.status_code} {e.response.text[:300]}" except Exception as e: yield f"\n\n**Error:** {e}" # --------------------------------------------------------------------------- # # File extraction # --------------------------------------------------------------------------- # FILE_BLOCK_RE = re.compile( r"```([a-zA-Z0-9_+\-]*?):([^\n`]+)\n(.*?)```", re.DOTALL, ) def extract_files(markdown: str) -> dict[str, str]: """Pull `lang:path\\n...` fenced blocks into a {path: content} dict.""" files: dict[str, str] = {} for _lang, path, body in FILE_BLOCK_RE.findall(markdown): path = path.strip() if path: files[path] = body.rstrip() + "\n" return files def make_zip(files: dict[str, str]) -> str | None: if not files: return None path = "/tmp/generated_project.zip" with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as zf: for name, content in files.items(): zf.writestr(name, content) return path # --------------------------------------------------------------------------- # # Gradio handlers # --------------------------------------------------------------------------- # def chat_fn(message: str, history: list[dict], provider_label: str, api_key: str): if not api_key.strip(): yield history + [ {"role": "user", "content": message}, {"role": "assistant", "content": "🔑 Please paste your API key in the panel on the left first."}, ], gr.update(), gr.update(visible=False) return if not message.strip(): yield history, gr.update(), gr.update(visible=False) return cfg = PROVIDERS[provider_label] convo = history + [{"role": "user", "content": message}] assistant_buf = "" yield convo + [{"role": "assistant", "content": "▍"}], gr.update(), gr.update(visible=False) for tok in call_elixir_backend(cfg["id"], cfg["model"], api_key.strip(), convo): assistant_buf += tok yield convo + [{"role": "assistant", "content": assistant_buf + "▍"}], gr.update(), gr.update(visible=False) time.sleep(0) # cooperative yield final = convo + [{"role": "assistant", "content": assistant_buf}] files = extract_files(assistant_buf) zip_path = make_zip(files) file_summary = "" if files: file_summary = "### 📁 Generated files\n" + "\n".join(f"- `{p}`" for p in files) yield final, file_summary, gr.update(value=zip_path, visible=bool(zip_path)) def clear_chat(): return [], "", gr.update(visible=False) # --------------------------------------------------------------------------- # # UI # --------------------------------------------------------------------------- # CUSTOM_CSS = """ .gradio-container { max-width: 1280px !important; } #title { text-align: center; padding: 8px 0 4px; } #title h1 { background: linear-gradient(90deg,#7c3aed,#06b6d4); -webkit-background-clip: text; -webkit-text-fill-color: transparent; font-size: 2.2rem; margin: 0; } #subtitle { text-align: center; color: #94a3b8; margin-bottom: 12px; } .api-panel { border: 1px solid rgba(124,58,237,.25); border-radius: 14px; padding: 14px; background: linear-gradient(180deg, rgba(124,58,237,.05), rgba(6,182,212,.03)); } footer { display: none !important; } """ with gr.Blocks(theme=gr.themes.Soft(primary_hue="violet", secondary_hue="cyan"), css=CUSTOM_CSS, title="Mini Coding Agent") as demo: gr.HTML('

⚡ Mini Coding Agent

' '
Multi-file project scaffolding · Python frontend · Elixir backend
') with gr.Row(): with gr.Column(scale=1, min_width=280): with gr.Group(elem_classes="api-panel"): gr.Markdown("### 🔐 Provider") provider = gr.Dropdown( choices=list(PROVIDERS.keys()), value="Groq (llama-3.3-70b, free)", label="Model provider", info="Groq is free & fast — great for demos.", ) api_key = gr.Textbox( label="API key", placeholder="Paste your key…", type="password", info="Stays in your browser session. Never stored.", ) key_links = gr.Markdown( "\n".join(f"- [{name}]({cfg['url']})" for name, cfg in PROVIDERS.items()) ) gr.Markdown("### 💡 Try") gr.Examples( examples=[ "Build a Flask todo API with SQLite, including tests.", "Create a small React + Vite landing page for a coffee shop.", "Write a Python CLI that scrapes Hacker News front page to JSON.", "Scaffold a FastAPI app with JWT auth and a users endpoint.", ], inputs=None, # filled below label=None, ) with gr.Column(scale=3): chatbot = gr.Chatbot( type="messages", height=540, show_label=False, avatar_images=(None, "https://api.dicebear.com/7.x/bottts/svg?seed=agent"), render_markdown=True, ) with gr.Row(): msg = gr.Textbox(placeholder="Describe the project you want to build…", show_label=False, scale=8, container=False) send = gr.Button("Send", variant="primary", scale=1) clear = gr.Button("Clear", scale=1) files_md = gr.Markdown("") zip_dl = gr.File(label="Download project (.zip)", visible=False) # Wire examples now that `msg` exists demo.load(lambda: None) # no-op; placeholder send.click(chat_fn, [msg, chatbot, provider, api_key], [chatbot, files_md, zip_dl]).then(lambda: "", None, msg) msg.submit(chat_fn, [msg, chatbot, provider, api_key], [chatbot, files_md, zip_dl]).then(lambda: "", None, msg) clear.click(clear_chat, None, [chatbot, files_md, zip_dl]) if __name__ == "__main__": demo.queue(default_concurrency_limit=4).launch( server_name="0.0.0.0", server_port=int(os.getenv("PORT", 7860)), )