""" app.py — Legawa Gradio Space for Build Small Hackathon. Runs the 4 agent workflows (analis_ruu, peneliti, penyusun, surat) inside a Gradio web UI instead of the Typer CLI. Default LLM backend is HF Inference API (zero-config demo); users can override in Settings. """ from __future__ import annotations import io import os import sys import tempfile import threading import time import traceback from pathlib import Path from typing import Callable, Iterator # Ensure the src/ package is importable on HF Spaces _src = Path(__file__).resolve().parent / "src" if _src.exists() and str(_src) not in sys.path: sys.path.insert(0, str(_src)) import gradio as gr from rich.console import Console from legawa.agents import analis_ruu, peneliti, penyusun, surat from legawa.tools.cache import CachingPasalClient from legawa.tools.pasal import PasalClient from legawa.tools.ethics import ethics_verify # ── Default HF Inference API config (zero-config demo) ────────────────── # Uses huggingface_hub's InferenceClient (works reliably on HF Spaces). # Users can override via the Settings tab to use custom endpoints. HF_BIG_MODEL = os.environ.get("HF_BIG_MODEL", "Qwen/Qwen3.5-9B") HF_SMALL_MODEL = os.environ.get("HF_SMALL_MODEL", "Qwen/Qwen3.5-9B") HF_TOKEN = os.environ.get("HF_TOKEN", "") BUILD_INFO = "Build Small Hackathon 2026 · legawa v0.1" RUU_EXAMPLE = """RUU Perlindungan Data Pribadi Kesehatan Pasal 1 Data kesehatan pasien wajib dilindungi oleh fasilitas pelayanan kesehatan dan penyelenggara sistem elektronik kesehatan. Pasal 2 Setiap rumah sakit wajib meminta persetujuan tertulis sebelum membagikan data pasien kepada pihak ketiga. Pasal 3 Pemerintah daerah wajib menyediakan kanal pengaduan bagi pasien yang data kesehatannya disalahgunakan.""" SURAT_EXAMPLE = """Yth. Anggota DPRD, Saya warga Kelurahan Sukamaju. Sudah tiga bulan saluran drainase di depan rumah kami tersumbat dan menyebabkan banjir setiap hujan. Kami sudah melapor ke RT dan kelurahan, tetapi belum ada tindak lanjut. Mohon bantuan agar dinas terkait segera turun mengecek dan membersihkan saluran tersebut. Hormat kami, Warga RW 04""" SURAT_MBG_EXAMPLE = """Yth. Bapak/Ibu Anggota DPR RI, Saya Sulastri, orang tua murid SDN 2 Karangrejo. Hari Selasa lalu, 27 anak di sekolah \ kami mengalami mual dan muntah setelah menyantap menu Makan Bergizi Gratis dari dapur \ SPPG kecamatan. Tiga anak sampai dirawat di puskesmas. Menurut guru, lauk ayam yang \ dibagikan berbau asam karena makanan baru tiba di sekolah lewat pukul 11 padahal \ dimasak sejak subuh. Kami para orang tua sebenarnya mendukung program MBG karena sangat membantu, tetapi \ kami takut kejadian ini terulang. Kami mohon Bapak/Ibu memperjuangkan: (1) pemeriksaan \ standar keamanan pangan dapur SPPG di daerah kami, (2) kejelasan tanggung jawab dan \ biaya pengobatan anak-anak yang keracunan, dan (3) jadwal distribusi yang tidak \ membiarkan makanan terlalu lama di perjalanan. Hormat kami, Sulastri, perwakilan paguyuban orang tua murid SDN 2 Karangrejo""" def _llm_label(llm: object) -> str: """Return the model label for both HFLLM and OpenAI-compatible LLM objects.""" if hasattr(llm, "model_id"): return str(getattr(llm, "model_id")) cfg = getattr(llm, "cfg", None) if cfg is not None and hasattr(cfg, "model"): return str(cfg.model) return "model" def _is_hf_default(url_or_model: str) -> bool: """True if this is a model ID (no ://) or a default HF Inference API endpoint.""" return "://" not in url_or_model or "huggingface.co/models/" in url_or_model def _model_id_from_url(url: str) -> str: """Extract model ID from HF Inference API URL.""" # URL format: https://api-inference.huggingface.co/models/{model_id}/v1 if "/models/" in url: return url.split("/models/")[1].split("/v1")[0] return url # ── Bootstrap: create settings + pool given user overrides ────────────── def build_pool( big_url: str = "", big_key: str = "", big_model: str = "", small_url: str = "", small_key: str = "", small_model: str = "", pasal_token: str = "", temperature: float = 0.3, max_tokens: int = 4096, strict_citations: bool = True, ) -> tuple: """Build an LLM pool + CachingPasalClient from user-provided overrides. Uses HFLLMPool (InferenceClient) for HF endpoints, LLMPool (OpenAI client) for custom endpoints. Falls through to env vars / HF defaults for anything left blank. """ from datetime import date # Resolve Pasal token: user input → env var → empty pasal_token = pasal_token or os.environ.get("PASAL_API_TOKEN", "") # Resolve BIG endpoint: user input → env var → HF default resolved_big_url = big_url or os.environ.get("LLM_BIG_URL", "") resolved_big_key = big_key or os.environ.get("LLM_BIG_API_KEY", HF_TOKEN) resolved_big_model = big_model or os.environ.get("LLM_BIG_MODEL", HF_BIG_MODEL) # Resolve SMALL endpoint: user input → env var → HF default resolved_small_url = small_url or os.environ.get("LLM_SMALL_URL", "") resolved_small_key = small_key or os.environ.get("LLM_SMALL_API_KEY", HF_TOKEN) resolved_small_model = small_model or os.environ.get("LLM_SMALL_MODEL", HF_SMALL_MODEL) run_date = os.environ.get("LEGAWA_RUN_DATE", date.today().isoformat()) # Decide which backend to use if not resolved_big_url or _is_hf_default(resolved_big_url): # --- HF Inference Client (default, works reliably) --- from hf_llm import HFLLMPool big_mid = _model_id_from_url(resolved_big_url) if resolved_big_url else resolved_big_model small_mid = _model_id_from_url(resolved_small_url) if resolved_small_url else resolved_small_model pool = HFLLMPool(big_mid, small_mid, token=resolved_big_key) pool.settings.run_date = run_date pool.settings.corpus_watermark = os.environ.get("PASAL_CORPUS_WATERMARK", "") pool.settings.strict_citations = strict_citations else: # --- OpenAI client (custom endpoint, e.g. llama.cpp) --- from legawa.config import LLMConfig, Settings big_cfg = LLMConfig( base_url=resolved_big_url, api_key=resolved_big_key, model=resolved_big_model, temperature=temperature, max_tokens=max_tokens, ) small_cfg = LLMConfig( base_url=resolved_small_url, api_key=resolved_small_key, model=resolved_small_model, temperature=temperature, max_tokens=max_tokens, ) override_settings = Settings( pasal_token=pasal_token, pasal_base_url=os.environ.get("PASAL_BASE_URL", "https://pasal.id/api/v1"), big=big_cfg, small=small_cfg, run_date=run_date, corpus_watermark=os.environ.get("PASAL_CORPUS_WATERMARK", ""), strict_citations=strict_citations, ) from legawa.llm import LLMPool pool = LLMPool(override_settings) raw = PasalClient( _pasal_settings(pasal_token) ) pasal = CachingPasalClient(raw) return pool, pasal def _pasal_settings(pasal_token: str) -> Settings: """Build a minimal Settings just for PasalClient.""" from legawa.config import LLMConfig, Settings dummy = LLMConfig(base_url="", api_key="", model="", temperature=0.3, max_tokens=4096) return Settings( pasal_token=pasal_token, pasal_base_url=os.environ.get("PASAL_BASE_URL", "https://pasal.id/api/v1"), big=dummy, small=dummy, run_date="", corpus_watermark="", strict_citations=False, ) return pool, pasal # ── Live agent-activity streaming ──────────────────────────────────────── # Agents report progress via rich Console. Each run gets its own capturing # console; lines are streamed to a gr.Chatbot so the multi-agent pipeline # (query expansion → probes → search → synthesis → ethics) is visible live. AGENT_LABELS = { "peneliti": "🔎 Peneliti Hukum", "penyusun": "📝 Penyusun Naskah", "surat": "📬 Agen Surat", "analis_ruu": "📄 Analis RUU", "analis": "📄 Analis RUU", "etika": "⚖️ Verifikator Etika & HAM", } ORCHESTRATOR = "🧭 Orkestrator" # Gradio 6 dropped Chatbot(type=...) — "messages" is the only/default format. _CHATBOT_KW = {} if int(gr.__version__.split(".")[0]) >= 6 else {"type": "messages"} def _activity_chatbot() -> gr.Chatbot: return gr.Chatbot(label="🤖 Aktivitas Agen (live)", height=260, **_CHATBOT_KW) def _label_for(line: str) -> tuple[str, str]: head, sep, rest = line.partition(":") if sep and head.strip().lower() in AGENT_LABELS: return AGENT_LABELS[head.strip().lower()], rest.strip() return ORCHESTRATOR, line.strip() def _stream_run(work: Callable[[Console], str]) -> Iterator[tuple[list[dict], str]]: """Run a pipeline in a thread, yielding (activity_messages, final_markdown).""" buf = io.StringIO() console = Console(file=buf, force_terminal=False, no_color=True, width=200) result: dict[str, str] = {} def runner() -> None: try: result["output"] = work(console) except Exception as e: # noqa: BLE001 result["error"] = str(e) traceback.print_exc() thread = threading.Thread(target=runner, daemon=True) thread.start() messages: list[dict] = [] seen = 0 while True: alive = thread.is_alive() text = buf.getvalue() chunk = text[seen:] lines = chunk.splitlines(keepends=True) for line in lines: if not line.endswith("\n") and alive: break # hold back a partial trailing line seen += len(line) stripped = line.strip() if not stripped: continue label, msg = _label_for(stripped) messages.append({"role": "assistant", "content": f"**{label}** · {msg}"}) if not alive: break yield messages, "⏳ *Agen sedang bekerja...*" time.sleep(0.4) thread.join() if "error" in result: messages.append({"role": "assistant", "content": f"**{ORCHESTRATOR}** · ❌ gagal: {result['error']}"}) yield messages, f"**Error:** {result['error']}" else: messages.append({"role": "assistant", "content": f"**{ORCHESTRATOR}** · ✅ selesai"}) yield messages, result.get("output", "") # ── Agent wrappers (called by Gradio) ─────────────────────────────────── def agent_analyze( source: str, big_url: str, big_key: str, small_url: str, small_key: str, pasal_token: str, ): if not source.strip(): yield [], "Masukkan teks RUU atau upload file PDF." return def work(console: Console) -> str: console.print("orkestrator: memuat model & koneksi, menyerahkan ke Analis RUU") pool, pasal = build_pool( big_url=big_url, big_key=big_key, small_url=small_url, small_key=small_key, pasal_token=pasal_token, ) try: result = analis_ruu.analyze(pool, pasal, source, console=console) console.print("etika: verifikasi 4 nilai demokrasi & HAM") return ethics_verify(result.output, pool.small) finally: pasal.close() yield from _stream_run(work) def agent_research( topic: str, big_url: str, big_key: str, small_url: str, small_key: str, pasal_token: str, ): if not topic.strip(): yield [], "Masukkan topik riset hukum." return def work(console: Console) -> str: console.print("orkestrator: memuat model & koneksi, menyerahkan ke Peneliti Hukum") pool, pasal = build_pool( big_url=big_url, big_key=big_key, small_url=small_url, small_key=small_key, pasal_token=pasal_token, ) try: output = peneliti.research(pool, pasal, topic, console=console) console.print("etika: verifikasi 4 nilai demokrasi & HAM") return ethics_verify(output, pool.small) finally: pasal.close() yield from _stream_run(work) def agent_draft( kind: str, topic: str, extra_instructions: str, with_research: bool, big_url: str, big_key: str, small_url: str, small_key: str, pasal_token: str, ): if not topic.strip(): yield [], "Masukkan topik." return def work(console: Console) -> str: console.print("orkestrator: memuat model & koneksi, menyerahkan ke Penyusun Naskah") pool, pasal = build_pool( big_url=big_url, big_key=big_key, small_url=small_url, small_key=small_key, pasal_token=pasal_token, ) try: output = penyusun.draft( pool, pasal, kind, topic, with_research=with_research, extra_instructions=extra_instructions or None, console=console, ) console.print("etika: verifikasi 4 nilai demokrasi & HAM") return ethics_verify(output, pool.small) finally: pasal.close() yield from _stream_run(work) def agent_surat( surat_text: str, verify_law: bool, big_url: str, big_key: str, small_url: str, small_key: str, pasal_token: str, ): if not surat_text.strip(): yield [], "Masukkan teks surat konstituen." return def work(console: Console) -> str: console.print("orkestrator: memuat model & koneksi, menyerahkan ke Agen Surat") pool, pasal = build_pool( big_url=big_url, big_key=big_key, small_url=small_url, small_key=small_key, pasal_token=pasal_token, ) try: result = surat.reply( pool, pasal, surat_text, verify_law=verify_law, console=console, ) output = surat.format_report(result) console.print("etika: verifikasi 4 nilai demokrasi & HAM") return ethics_verify(output, pool.small) finally: pasal.close() yield from _stream_run(work) def agent_health( big_url: str, big_key: str, small_url: str, small_key: str, pasal_token: str, ) -> str: """Quick connectivity check for all services.""" lines: list[str] = [] pool, pasal = build_pool( big_url=big_url, big_key=big_key, small_url=small_url, small_key=small_key, pasal_token=pasal_token, ) try: # Check BIG LLM try: resp = pool.big.chat( [{"role": "user", "content": "Jawab dengan satu kata: OK"}], max_tokens=10, ) lines.append(f"✅ **BIG LLM** ({_llm_label(pool.big)[:30]}...): {resp.strip()}") except Exception as e: lines.append(f"❌ **BIG LLM**: {e}") # Check SMALL LLM try: resp = pool.small.chat( [{"role": "user", "content": "Jawab dengan satu kata: OK"}], max_tokens=10, ) lines.append(f"✅ **SMALL LLM** ({_llm_label(pool.small)[:30]}...): {resp.strip()}") except Exception as e: lines.append(f"❌ **SMALL LLM**: {e}") # Check pasal.id try: result = pasal.search("ketenagakerjaan", limit=1) count = len(result.get("results", result.get("hits", []))) lines.append(f"✅ **pasal.id**: {count} hasil untuk 'ketenagakerjaan'") except Exception as e: lines.append(f"❌ **pasal.id**: {e}") lines.append(f"\n{BUILD_INFO}") return "\n\n".join(lines) finally: pasal.close() # ── File upload helper for analis_ruu ─────────────────────────────────── def handle_file_upload(file: object | None) -> str: if file is None: return "" path = Path(getattr(file, "name")) if path.suffix.lower() == ".pdf": from pypdf import PdfReader reader = PdfReader(str(path)) return "\n\n".join(page.extract_text() or "" for page in reader.pages) return path.read_text(encoding="utf-8") # ── Build Gradio UI ───────────────────────────────────────────────────── CSS = """ /* Space is compact, judge-friendly, and readable */ .gradio-container { max-width: 1100px !important; margin: 0 auto !important; } .legawa-hero { padding: 1.25rem 1.4rem; border-radius: 18px; background: linear-gradient(135deg, rgba(79,70,229,.16), rgba(16,185,129,.12)); border: 1px solid rgba(99,102,241,.25); margin-bottom: 1rem; } .legawa-hero h1 { margin-top: 0; } .legawa-card { padding: .85rem 1rem; border-radius: 14px; border: 1px solid rgba(148,163,184,.25); background: rgba(148,163,184,.08); } .legawa-card strong { color: #4f46e5; } footer { display: none !important; } .dark table { color: #e0e0e0; } """ def build_app() -> gr.Blocks: with gr.Blocks( css=CSS, title="Legawa — Asisten Legislatif", theme=gr.themes.Soft(), ) as app: gr.HTML( f"""
Backyard AI untuk staf DPR/DPRD: triase surat warga, riset aturan, analisis RUU, dan draf naskah kebijakan dalam menit — bukan hari.
{BUILD_INFO} · 2× Qwen3.5-9B = 18B params total, under the 32B trail limit.