Spaces:
Sleeping
Sleeping
| """Operon Adaptive Assembly Dashboard — template selection, assembly, and experience pool.""" | |
| from __future__ import annotations | |
| from dataclasses import dataclass, field | |
| from typing import Any | |
| import gradio as gr | |
| # --------------------------------------------------------------------------- | |
| # Inline simulation types | |
| # --------------------------------------------------------------------------- | |
| class TaskFingerprint: | |
| task_shape: str | |
| tool_count: int | |
| subtask_count: int | |
| required_roles: tuple[str, ...] | |
| tags: tuple[str, ...] = () | |
| class PatternTemplate: | |
| template_id: str | |
| name: str | |
| topology: str | |
| stage_count: int | |
| fingerprint: TaskFingerprint | |
| tags: tuple[str, ...] = () | |
| success_rate: float | None = None | |
| class ExperienceRecord: | |
| stage_name: str | |
| signal_category: str | |
| intervention_kind: str | |
| outcome_success: bool | |
| # --------------------------------------------------------------------------- | |
| # Preset library | |
| # --------------------------------------------------------------------------- | |
| TEMPLATES = [ | |
| PatternTemplate("t1", "Sequential Review Pipeline", "reviewer_gate", 2, | |
| TaskFingerprint("sequential", 2, 2, ("writer", "reviewer")), | |
| ("content", "review")), | |
| PatternTemplate("t2", "Enterprise Analysis Organism", "skill_organism", 3, | |
| TaskFingerprint("sequential", 4, 3, ("researcher", "strategist")), | |
| ("enterprise", "analysis")), | |
| PatternTemplate("t3", "Research Swarm", "specialist_swarm", 4, | |
| TaskFingerprint("parallel", 5, 4, ("analyst", "writer")), | |
| ("research",)), | |
| PatternTemplate("t4", "Single Worker", "single_worker", 1, | |
| TaskFingerprint("sequential", 1, 1, ("worker",)), | |
| ("simple",)), | |
| ] | |
| def _jaccard(a: tuple[str, ...], b: tuple[str, ...]) -> float: | |
| sa, sb = set(a), set(b) | |
| union = sa | sb | |
| return len(sa & sb) / len(union) if union else 1.0 | |
| def _score(tmpl: PatternTemplate, fp: TaskFingerprint) -> float: | |
| shape = 1.0 if tmpl.fingerprint.task_shape == fp.task_shape else 0.0 | |
| tool_prox = 1.0 / (1.0 + abs(tmpl.fingerprint.tool_count - fp.tool_count)) | |
| sub_prox = 1.0 / (1.0 + abs(tmpl.fingerprint.subtask_count - fp.subtask_count)) | |
| role_sim = _jaccard(tmpl.fingerprint.required_roles, fp.required_roles) | |
| tag_sim = _jaccard(tmpl.tags, fp.tags) | |
| sr = tmpl.success_rate if tmpl.success_rate is not None else 0.5 | |
| return 0.30 * shape + 0.15 * tool_prox + 0.15 * sub_prox + 0.20 * role_sim + 0.10 * tag_sim + 0.10 * sr | |
| # --------------------------------------------------------------------------- | |
| # Presets | |
| # --------------------------------------------------------------------------- | |
| PRESETS = { | |
| "Enterprise Analysis": TaskFingerprint("sequential", 3, 3, ("researcher", "strategist"), ("enterprise",)), | |
| "Content Review": TaskFingerprint("sequential", 2, 2, ("writer", "reviewer"), ("content",)), | |
| "Parallel Research": TaskFingerprint("parallel", 5, 4, ("analyst", "writer"), ("research",)), | |
| "Simple Task": TaskFingerprint("sequential", 1, 1, ("worker",), ("simple",)), | |
| } | |
| EXPERIENCE_PRESETS = { | |
| "Fresh (no experience)": [], | |
| "Warm (escalate works for epistemic)": [ | |
| ExperienceRecord("router", "epistemic", "escalate", True), | |
| ExperienceRecord("router", "epistemic", "escalate", True), | |
| ExperienceRecord("router", "epistemic", "retry", False), | |
| ], | |
| "Mixed (retry and escalate both tried)": [ | |
| ExperienceRecord("planner", "epistemic", "escalate", True), | |
| ExperienceRecord("planner", "epistemic", "retry", True), | |
| ExperienceRecord("planner", "somatic", "halt", True), | |
| ExperienceRecord("router", "epistemic", "retry", False), | |
| ], | |
| } | |
| # --------------------------------------------------------------------------- | |
| # Callbacks | |
| # --------------------------------------------------------------------------- | |
| def _run_selection(preset_name): | |
| if preset_name not in PRESETS: | |
| return "Select a preset." | |
| fp = PRESETS[preset_name] | |
| scored = [(t, round(_score(t, fp), 4)) for t in TEMPLATES] | |
| scored.sort(key=lambda x: x[1], reverse=True) | |
| rows = "" | |
| for i, (t, s) in enumerate(scored): | |
| bg = "#1a2a1a" if i == 0 else "transparent" | |
| badge = ' <span style="color:#6a6;font-size:12px">SELECTED</span>' if i == 0 else "" | |
| rows += f"""<tr style="background:{bg}"> | |
| <td style="padding:10px;font-weight:{'600' if i==0 else '400'}">{t.name}{badge}</td> | |
| <td style="padding:10px">{t.topology}</td> | |
| <td style="padding:10px">{t.stage_count} stages</td> | |
| <td style="padding:10px;font-weight:600">{s:.4f}</td> | |
| </tr>""" | |
| fp_html = f"""<div style="margin-bottom:16px;padding:12px;background:#111;border-radius:6px;font-size:14px"> | |
| <strong>Fingerprint:</strong> shape={fp.task_shape}, tools={fp.tool_count}, subtasks={fp.subtask_count}, roles={fp.required_roles}, tags={fp.tags} | |
| </div>""" | |
| table = f"""<table style="width:100%;border-collapse:collapse;font-size:14px"> | |
| <thead><tr style="border-bottom:2px solid #333"> | |
| <th style="text-align:left;padding:8px">Template</th> | |
| <th style="text-align:left;padding:8px">Topology</th> | |
| <th style="text-align:left;padding:8px">Stages</th> | |
| <th style="text-align:left;padding:8px">Score</th> | |
| </tr></thead> | |
| <tbody>{rows}</tbody> | |
| </table>""" | |
| return fp_html + table | |
| def _run_experience(exp_preset): | |
| if exp_preset not in EXPERIENCE_PRESETS: | |
| return "Select a preset." | |
| records = EXPERIENCE_PRESETS[exp_preset] | |
| if not records: | |
| return '<p style="color:#888">Empty experience pool — watcher uses rule-based decisions only.</p>' | |
| rows = "" | |
| for r in records: | |
| color = "#4a4" if r.outcome_success else "#a44" | |
| rows += f"""<tr> | |
| <td style="padding:8px">{r.stage_name}</td> | |
| <td style="padding:8px">{r.signal_category}</td> | |
| <td style="padding:8px">{r.intervention_kind}</td> | |
| <td style="padding:8px;color:{color}">{'success' if r.outcome_success else 'failed'}</td> | |
| </tr>""" | |
| # Compute recommendation | |
| success_by_kind: dict[str, int] = {} | |
| for r in records: | |
| if r.outcome_success: | |
| success_by_kind[r.intervention_kind] = success_by_kind.get(r.intervention_kind, 0) + 1 | |
| rec = max(success_by_kind, key=lambda k: success_by_kind[k]) if success_by_kind else "none" | |
| table = f"""<table style="width:100%;border-collapse:collapse;font-size:14px;margin-bottom:16px"> | |
| <thead><tr style="border-bottom:2px solid #333"> | |
| <th style="text-align:left;padding:8px">Stage</th> | |
| <th style="text-align:left;padding:8px">Signal</th> | |
| <th style="text-align:left;padding:8px">Intervention</th> | |
| <th style="text-align:left;padding:8px">Outcome</th> | |
| </tr></thead> | |
| <tbody>{rows}</tbody> | |
| </table>""" | |
| rec_html = f'<div style="padding:12px;background:#0a1a2a;border:1px solid #1a2a3a;border-radius:6px;font-size:14px"><strong style="color:#6af">Recommendation:</strong> {rec.upper()}</div>' | |
| return table + rec_html | |
| # --------------------------------------------------------------------------- | |
| # App | |
| # --------------------------------------------------------------------------- | |
| def build_app(): | |
| with gr.Blocks(title="Operon Adaptive Assembly", theme=gr.themes.Base()) as demo: | |
| gr.Markdown("# Operon Adaptive Assembly\nTemplate selection, organism construction, and experience-driven intervention.") | |
| with gr.Tab("Template Selection"): | |
| preset = gr.Dropdown(choices=list(PRESETS.keys()), label="Task Preset", value="Enterprise Analysis") | |
| btn = gr.Button("Select Template") | |
| out = gr.HTML() | |
| btn.click(_run_selection, inputs=[preset], outputs=[out]) | |
| with gr.Tab("Experience Pool"): | |
| exp_preset = gr.Dropdown(choices=list(EXPERIENCE_PRESETS.keys()), label="Experience Preset", value="Fresh (no experience)") | |
| btn2 = gr.Button("Show Pool & Recommendation") | |
| out2 = gr.HTML() | |
| btn2.click(_run_experience, inputs=[exp_preset], outputs=[out2]) | |
| return demo | |
| if __name__ == "__main__": | |
| app = build_app() | |
| app.launch() | |