commit

app.py

@@ -0,0 +1,342 @@
+import os
+import re
+import json
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple, Any
+
+import gradio as gr
+
+
+# ----------------------
+# Data loading utilities
+# ----------------------
+
+HERE = Path(__file__).parent
+DATASOURCE_TXT = HERE / "datasource.txt"
+STATIC_DATA_DIR = HERE / "static_data"
+
+
+def _slugify(text: str) -> str:
+    text = text.strip().lower()
+    text = re.sub(r"[^a-z0-9\s_-]+", "", text)
+    text = re.sub(r"[\s_-]+", "-", text).strip("-")
+    return text or "uncategorized"
+
+
+def _titleize_slug(slug: str) -> str:
+    return re.sub(r"[-_]+", " ", slug).title()
+
+
+def _read_text(path: Path) -> Optional[str]:
+    try:
+        return path.read_text(encoding="utf-8")
+    except Exception:
+        return None
+
+
+def _try_parse_json(text: str) -> Optional[dict]:
+    try:
+        return json.loads(text)
+    except Exception:
+        return None
+
+
+def _try_parse_yaml(text: str) -> Optional[dict]:
+    try:
+        import yaml  # type: ignore
+
+        return yaml.safe_load(text)
+    except Exception:
+        return None
+
+
+def _extract_field(obj: dict, keys: List[str]) -> Optional[str]:
+    for k in keys:
+        if k in obj and isinstance(obj[k], str) and obj[k].strip():
+            return obj[k].strip()
+    # common nested forms, like { system: { content: "..." } } or messages: [{role: system, content: "..."}]
+    if "system" in obj and isinstance(obj["system"], dict):
+        for k in ("content", "prompt", "text"):
+            v = obj["system"].get(k)
+            if isinstance(v, str) and v.strip():
+                return v.strip()
+    if "messages" in obj and isinstance(obj["messages"], list):
+        for m in obj["messages"]:
+            if isinstance(m, dict) and m.get("role") == "system":
+                content = m.get("content")
+                if isinstance(content, str) and content.strip():
+                    return content.strip()
+    return None
+
+
+def _extract_agent_from_obj(obj: dict, fallback_category: str, source_path: Path) -> Optional[dict]:
+    # Heuristics to recognize agent-like configs.
+    name = _extract_field(obj, ["name", "agent_name", "title", "id"])
+    description = _extract_field(obj, ["description", "desc", "about", "summary"])
+    system_prompt = _extract_field(
+        obj,
+        [
+            "system_prompt",
+            "prompt",
+            "instructions",
+            "system",
+            "system_instructions",
+            "system_text",
+        ],
+    )
+
+    if not (name and system_prompt):
+        return None
+
+    category = (
+        _extract_field(obj, ["category", "group", "type"]) or fallback_category or "uncategorized"
+    )
+    category_slug = _slugify(category)
+    agent_id = _slugify(f"{category_slug}-{name}")
+
+    return {
+        "id": agent_id,
+        "name": name,
+        "description": description or "",
+        "system_prompt": system_prompt,
+        "category": category_slug,
+        "source": str(source_path.relative_to(HERE)),
+    }
+
+
+def _scan_static_data(root: Path) -> List[dict]:
+    agents: List[dict] = []
+    patterns = ["**/*.json", "**/*.yaml", "**/*.yml"]
+    for pattern in patterns:
+        for fp in root.glob(pattern):
+            if not fp.is_file():
+                continue
+            text = _read_text(fp)
+            if not text:
+                continue
+            data = _try_parse_json(text)
+            if data is None:
+                data = _try_parse_yaml(text)
+            if not isinstance(data, dict):
+                continue
+            # derive category from parent folder name as a fallback
+            fallback_category = fp.parent.name
+            agent = _extract_agent_from_obj(data, fallback_category, fp)
+            if agent:
+                agents.append(agent)
+    return agents
+
+
+def _maybe_snapshot_download_from_hf(url: str, target_dir: Path) -> Optional[Path]:
+    # Attempt to fetch dataset to local static_data using huggingface_hub.
+    try:
+        from huggingface_hub import snapshot_download
+
+        # Accept full URL like https://huggingface.co/datasets/owner/name
+        m = re.match(r"https?://huggingface.co/datasets/([^/]+/[^/]+)", url.strip())
+        if not m:
+            return None
+        repo_id = m.group(1)
+        target_dir.mkdir(parents=True, exist_ok=True)
+        local_path = snapshot_download(repo_id=repo_id, repo_type="dataset")
+        # Mirror files into target_dir for predictable path
+        src = Path(local_path)
+        for p in src.rglob("*"):
+            if p.is_file():
+                rel = p.relative_to(src)
+                dest = target_dir / rel
+                dest.parent.mkdir(parents=True, exist_ok=True)
+                try:
+                    dest.write_bytes(p.read_bytes())
+                except Exception:
+                    pass
+        return target_dir
+    except Exception:
+        # No network or hub not installed; just skip
+        return None
+
+
+def load_agents() -> Tuple[Dict[str, Any], List[dict], List[str]]:
+    """
+    Returns (catalog_by_category, agents, warnings)
+    catalog_by_category: {category_slug: {label: str, agents: [agent_id, ...]}}
+    agents: list of agent dicts
+    warnings: list of warning strings for UI
+    """
+    warnings: List[str] = []
+    agents: List[dict] = []
+
+    # 1) Prefer local static_data if present
+    if STATIC_DATA_DIR.exists():
+        agents = _scan_static_data(STATIC_DATA_DIR)
+    else:
+        # 2) Try to download dataset indicated by datasource.txt
+        url = _read_text(DATASOURCE_TXT) or ""
+        if url.strip():
+            maybe_dir = _maybe_snapshot_download_from_hf(url.strip(), STATIC_DATA_DIR)
+            if maybe_dir and maybe_dir.exists():
+                agents = _scan_static_data(maybe_dir)
+            else:
+                warnings.append(
+                    "Dataset fetch unavailable. Add a local 'static_data' folder with agent configs."
+                )
+        else:
+            warnings.append("No datasource URL found; using fallback sample data.")
+
+    # 3) Fallback sample if nothing found
+    if not agents:
+        agents = [
+            {
+                "id": "code-assist-starter",
+                "name": "Code Assist Starter",
+                "description": "A simple code generation assistant for boilerplate tasks.",
+                "system_prompt": (
+                    "You are a helpful coding agent. Generate concise, correct code and explain key steps."
+                ),
+                "category": "code-assist",
+                "source": "sample",
+            },
+            {
+                "id": "docs-navigator",
+                "name": "Docs Navigator",
+                "description": "Answers questions using project docs and summarizes APIs.",
+                "system_prompt": (
+                    "Act as a technical writer. Read provided docs and produce accurate, concise answers with citations when possible."
+                ),
+                "category": "documentation",
+                "source": "sample",
+            },
+        ]
+        warnings.append(
+            "Showing sample data. Add 'static_data' with JSON/YAML agent configs to replace."
+        )
+
+    # Build catalog
+    catalog: Dict[str, Dict[str, Any]] = {}
+    for a in agents:
+        cat = a.get("category") or "uncategorized"
+        cat_slug = _slugify(str(cat))
+        if cat_slug not in catalog:
+            catalog[cat_slug] = {"label": _titleize_slug(cat_slug), "agents": []}
+        catalog[cat_slug]["agents"].append(a["id"])
+
+    # Sort categories and agents by display name
+    catalog = dict(sorted(catalog.items(), key=lambda kv: kv[1]["label"]))
+    name_by_id = {a["id"]: a["name"] for a in agents}
+    for c in catalog.values():
+        c["agents"].sort(key=lambda aid: name_by_id.get(aid, aid).lower())
+
+    return catalog, agents, warnings
+
+
+# ----------------------
+# Gradio application
+# ----------------------
+
+
+def build_ui():
+    catalog, agents, warnings = load_agents()
+    agent_by_id = {a["id"]: a for a in agents}
+
+    # Initial selections
+    first_cat = next(iter(catalog.keys())) if catalog else None
+    first_agent = catalog[first_cat]["agents"][0] if first_cat and catalog[first_cat]["agents"] else None
+
+    def categories_choices():
+        return [(c["label"], slug) for slug, c in catalog.items()]
+
+    def agents_choices(cat_slug: Optional[str]):
+        if not cat_slug or cat_slug not in catalog:
+            return []
+        return [
+            (agent_by_id[aid]["name"], aid)
+            for aid in catalog[cat_slug]["agents"]
+            if aid in agent_by_id
+        ]
+
+    def on_category_change(cat_slug: str):
+        choices = agents_choices(cat_slug)
+        value = choices[0][1] if choices else None
+        return gr.update(choices=choices, value=value), *on_agent_change(value)
+
+    def on_agent_change(agent_id: Optional[str]):
+        if not agent_id or agent_id not in agent_by_id:
+            return (
+                "Select an agent",
+                gr.update(value="", visible=True),
+                gr.update(value="", visible=True),
+                gr.update(value="", visible=False),
+            )
+        a = agent_by_id[agent_id]
+        header = f"# {a['name']}"
+        desc = a.get("description", "")
+        prompt = a.get("system_prompt", "")
+        src = a.get("source", "")
+        footer = f"Source: {src}" if src else ""
+        return (
+            header,
+            gr.update(value=desc, visible=True),
+            gr.update(value=prompt, visible=True),
+            gr.update(value=footer, visible=bool(footer)),
+        )
+
+    with gr.Blocks(title="Code Gen Agents Network") as demo:
+        with gr.Row():
+            with gr.Column(scale=1, min_width=280):
+                with gr.Sidebar():
+                    gr.Markdown("## Categories")
+                    dd_category = gr.Dropdown(
+                        label="Category",
+                        choices=categories_choices(),
+                        value=first_cat,
+                    )
+                    dd_agent = gr.Dropdown(
+                        label="Agent",
+                        choices=agents_choices(first_cat),
+                        value=first_agent,
+                    )
+                    if warnings:
+                        with gr.Accordion("Notes", open=False):
+                            gr.Markdown("\n".join(f"- {w}" for w in warnings))
+
+            with gr.Column(scale=3):
+                md_header = gr.Markdown(visible=True)
+                tb_desc = gr.Textbox(
+                    label="Description",
+                    lines=4,
+                    show_copy_button=True,
+                )
+                tb_prompt = gr.Textbox(
+                    label="System Prompt",
+                    lines=12,
+                    show_copy_button=True,
+                )
+                md_footer = gr.Markdown(visible=False)
+
+        # Wire events
+        dd_category.change(
+            on_category_change,
+            inputs=[dd_category],
+            outputs=[dd_agent, md_header, tb_desc, tb_prompt, md_footer],
+        )
+        dd_agent.change(
+            on_agent_change,
+            inputs=[dd_agent],
+            outputs=[md_header, tb_desc, tb_prompt, md_footer],
+        )
+
+        # Initialize content
+        header, desc_upd, prompt_upd, footer_upd = on_agent_change(first_agent)
+        md_header.value = header
+        tb_desc.value = desc_upd.value if isinstance(desc_upd, gr.Update) else ""
+        tb_prompt.value = prompt_upd.value if isinstance(prompt_upd, gr.Update) else ""
+        md_footer.value = footer_upd.value if isinstance(footer_upd, gr.Update) else ""
+        md_footer.visible = bool(footer_upd.value) if isinstance(footer_upd, gr.Update) else False
+
+    return demo
+
+
+if __name__ == "__main__":
+    demo = build_ui()
+    demo.launch()
+

datasource.txt

@@ -0,0 +1 @@
+https://huggingface.co/datasets/danielrosehill/Code-Gen-Agents-0925