""" app.py — SimQuantum Tuning Lab ================================= AMD Developer Hackathon 2026. Before running, set env vars (on MI300X): export QDOT_LLM_BASE_URL=http://localhost:8000/v1 export QDOT_LLM_MODEL=Qwen/Qwen3-8B streamlit run app.py --server.port 8501 --server.address 0.0.0.0 Or locally (LLM offline, physics sim still runs): streamlit run app.py """ from __future__ import annotations import os, sys, threading, time, re from pathlib import Path import numpy as np sys.path.insert(0, str(Path(__file__).parent)) import streamlit as st st.set_page_config( page_title="SimQuantum Tuning Lab", page_icon="⚛", layout="wide", initial_sidebar_state="expanded", # always open so the LLM URL is visible ) # ───────────────────────────────────────────────────────────────────────────── # CSS # ───────────────────────────────────────────────────────────────────────────── st.markdown(""" """, unsafe_allow_html=True) # ───────────────────────────────────────────────────────────────────────────── # Constants # ───────────────────────────────────────────────────────────────────────────── STAGES = [ ("BOOTSTRAPPING", "⚡","Integrity check", False), ("COARSE_SURVEY", "◈", "Voltage survey", False), ("HYPERSURFACE_SEARCH", "◎","Find boundary", False), ("CHARGE_ID", "◇","Classify charge", False), ("NAVIGATION", "→","Navigate to (1,1)", True), ("VERIFICATION", "✓","Verify stability", True), ] SPY_AGENTS = [ ("perception","🔬","Perception","Quality Inspector",{"BOOTSTRAPPING","COARSE_SURVEY","CHARGE_ID"}), ("executive", "🏛","Executive", "Mission Conductor",set(s[0] for s in STAGES)), ("planning", "📐","Planning", "Navigator", {"HYPERSURFACE_SEARCH","NAVIGATION"}), ("safety", "🛡","Safety", "Hardware Marshal", {"NAVIGATION","VERIFICATION"}), ("hitl", "🛑","HITL", "Human Governor", set()), ] START_KWS = {"start","begin","run","tune","go","launch","init","initialize","initialise","proceed"} STAGE_DESC = { "BOOTSTRAPPING": ("64-pt line scan. Verifies gate response and sensor signal.","~64 pts"), "COARSE_SURVEY": ("32×32 sweep across full voltage bounds.", "~1024 pts"), "HYPERSURFACE_SEARCH": ("16×16 local scan around survey peak. Confirms boundary.", "~256 pts"), "CHARGE_ID": ("32×32 scan. 5-model CNN ensemble classifies charge state.", "~1024 pts"), "NAVIGATION": ("Bayesian BO proposes voltage moves toward (1,1).", "variable"), "VERIFICATION": ("3× repeated 16×16 scans confirming (1,1) stability.", "~768 pts"), } STABILITY_CS = [ [0.00,"#07070A"],[0.30,"#1A0E40"],[0.55,"#7A1800"], [0.75,"#D84000"],[0.90,"#FF9000"],[1.00,"#FFE040"], ] PLOT_LAYOUT = dict( paper_bgcolor="rgba(0,0,0,0)", plot_bgcolor="#FAFAF8", font=dict(color="#8A9AB0",size=10,family="JetBrains Mono"), margin=dict(l=10,r=10,t=30,b=10), ) # Dr. Q system prompt — tells Qwen who it is and what the experiment is DR_Q_SYSTEM = """\ You are Dr. Q, the AI co-pilot for SimQuantum — an autonomous quantum dot tuning system running on AMD MI300X hardware. You are Qwen3-8B. Your role: have a real conversation about this experiment with whoever is asking. Adapt your depth to the person — a curious student gets clear analogies, an expert physicist like Natalia Ares gets precise technical detail. Read the question, match it. THE EXPERIMENT: A 6-stage POMDP agent autonomously navigates gate voltage space (Vg1, Vg2) to tune a double quantum dot to the (1,1) charge state — one electron per dot — required for spin qubit operation. Stages: BOOTSTRAPPING → COARSE_SURVEY → HYPERSURFACE_SEARCH → CHARGE_ID → NAVIGATION → VERIFICATION Physics simulator: Capacitive Interaction Model (CIM) — real semiconductor physics, not a toy model. Parameters: charging energy E_c, lever arm, tunnel coupling t_c. CNN: 5-model TinyCNN ensemble trained on 51,000 simulated stability diagrams. Val accuracy 91.4%. OOD detection via Mahalanobis distance on penultimate features. The agent uses Bayesian optimisation (MultiResBO) for navigation and a particle filter belief state over charge configurations. IMPORTANT HONESTY: Navigation (Phase 3) is unsolved. The Bayesian optimiser has no converging reward signal in intermediate voltage space — the CNN sees "misc" everywhere except at charge boundaries. The agent stops at CHARGE_ID reliably; NAVIGATION wanders. Be honest about this if asked. CURRENT RUN STATE: {state_block} CONVERSATION STYLE: - Talk like a physicist who is also a good teacher - 2-4 sentences usually, more only if genuinely needed - Reference actual numbers from the state when relevant - Be honest about what works and what doesn't - Never say "Great news!" or "Certainly!" or "As an AI" - If something is unclear, ask for clarification """ # ───────────────────────────────────────────────────────────────────────────── # Session state # ───────────────────────────────────────────────────────────────────────────── def _init(): d = dict( agent=None, exp_state=None, narrator=None, hitl_manager=None, done_event=None, thread=None, running=False, run_count=0, chat=[], llm_url="", # persists across reruns llm_api_key="", llm_model="accounts/fireworks/models/qwen3-8b", use_cnn=True, meas_budget=8096, max_steps=140, ) for k,v in d.items(): if k not in st.session_state: st.session_state[k] = v _init() # Pull LLM config from environment if not yet set in session if not st.session_state.llm_url: env_url = os.environ.get("QDOT_LLM_BASE_URL", "") if env_url: st.session_state.llm_url = env_url if not st.session_state.llm_api_key: env_key = os.environ.get("QDOT_LLM_API_KEY", "") if env_key: st.session_state.llm_api_key = env_key if st.session_state.llm_model == "accounts/fireworks/models/qwen3-8b": env_model = os.environ.get("QDOT_LLM_MODEL", "") if env_model: st.session_state.llm_model = env_model # ───────────────────────────────────────────────────────────────────────────── # Dr. Q — real LLM, no gatekeeping # ───────────────────────────────────────────────────────────────────────────── def _llm_available() -> bool: return bool(st.session_state.llm_url.strip()) def _build_state_block() -> str: """Snapshot of current agent state for the system prompt.""" exp = st.session_state.exp_state agnt = st.session_state.agent if exp is None: return "No run active. Agent is idle." budget = agnt.measurement_budget if agnt else st.session_state.meas_budget meas = exp.total_measurements stage = exp.stage.name vg1 = exp.current_voltage.vg1 vg2 = exp.current_voltage.vg2 snr = f"{exp.last_dqc.snr_db:.1f}dB" if exp.last_dqc else "unknown" dqc = exp.last_dqc.quality.value if exp.last_dqc else "unknown" bt = exp.total_backtracks done = st.session_state.done_event and st.session_state.done_event.is_set() cls_str = "none yet" if exp.last_classification: c = exp.last_classification ood = "OOD" if exp.is_ood else "in-dist" cls_str = f"{c.label.value} ({c.confidence:.0%} conf, {ood})" p11 = exp.belief.charge_probs.get((1,1), 0.0) ml = exp.belief.most_likely_state() return ( f"Stage: {stage}\n" f"Measurements used: {meas}/{budget} ({100*meas//max(budget,1)}%)\n" f"Current voltage: Vg1={vg1:+.3f}V, Vg2={vg2:+.3f}V\n" f"Signal quality: SNR={snr}, DQC={dqc}\n" f"CNN classification: {cls_str}\n" f"Belief P(1,1)={p11:.3f}, most likely: {ml}\n" f"Backtracks: {bt}\n" f"Run finished: {'yes' if done else 'no'}" ) def _call_qwen(user_msg: str, image_bytes: bytes | None = None) -> str: """ Call Dr. Q. System prompt + conversation history + user message → response. Supports Fireworks AI (https://api.fireworks.ai/inference/v1) and local vLLM. If image_bytes provided, sends the image alongside the message (VL models). No fallbacks, no scripts. If the LLM is down, say so clearly. """ url = st.session_state.llm_url.strip().rstrip("/") model = st.session_state.llm_model.strip() or "accounts/fireworks/models/qwen3-8b" api_key = st.session_state.llm_api_key.strip() or os.environ.get("QDOT_LLM_API_KEY", "EMPTY") # Build system + history system = DR_Q_SYSTEM.format(state_block=_build_state_block()) messages = [{"role": "system", "content": system}] for msg in st.session_state.chat[-16:]: role = "user" if msg["role"] == "user" else "assistant" messages.append({"role": role, "content": msg["content"]}) # If image provided, replace last user message content with multimodal list if image_bytes and messages and messages[-1]["role"] == "user": import base64 b64 = base64.b64encode(image_bytes).decode() messages[-1]["content"] = [ {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}, {"type": "text", "text": messages[-1]["content"]}, ] try: import openai # Fireworks already includes /v1; local vLLM needs it appended if "fireworks.ai" in url: api_base = url # already https://api.fireworks.ai/inference/v1 else: api_base = url if url.endswith("/v1") else url + "/v1" client = openai.OpenAI(base_url=api_base, api_key=api_key) resp = client.chat.completions.create( model=model, messages=messages, max_tokens=1024, temperature=0.6, ) raw = resp.choices[0].message.content or "" # Separate reasoning from answer import re as _re think_match = _re.search(r"(.*?)", raw, flags=_re.DOTALL) think_block = think_match.group(1).strip() if think_match else "" clean = _re.sub(r".*?", "", raw, flags=_re.DOTALL).strip() # Attach reasoning to session for rendering st.session_state["_last_think"] = think_block return clean except ImportError: return ("openai package not installed. Run: pip install openai\n" "Then restart the app.") except Exception as exc: return (f"Cannot reach Dr. Q at {url}.\n\nError: {exc}\n\n" f"Check your API key and endpoint URL in the sidebar.") def _add_msg(role: str, content: str, kind: str = "n", think: str = ""): st.session_state.chat.append({"role": role, "content": content, "kind": kind, "think": think}) def _handle_chat(user_msg: str): """Add user message, start agent if requested, get Dr. Q response.""" _add_msg("user", user_msg) is_start = any(kw in user_msg.lower() for kw in START_KWS) if is_start and not st.session_state.running and st.session_state.agent is None: # Start the agent try: agent, exp_state, narrator, hitl_mgr = _make_agent( st.session_state.use_cnn, st.session_state.meas_budget, st.session_state.max_steps, ) done_event = threading.Event() thread = threading.Thread( target=_run_thread, args=(agent, done_event), daemon=True) st.session_state.update( agent=agent, exp_state=exp_state, narrator=narrator, hitl_manager=hitl_mgr, done_event=done_event, thread=thread, running=True, run_count=st.session_state.run_count+1, ) thread.start() # Give the LLM the launch context and let it respond naturally if _llm_available(): reply = _call_qwen(user_msg) else: reply = ( "Tuning sequence started.\n\n" "To enable Qwen responses, set the vLLM endpoint in the sidebar " "to your MI300X address (e.g. http://localhost:8000/v1).\n\n" "The physics simulation is running. Ask me anything." ) except Exception as exc: reply = f"Failed to start agent: {exc}" else: # Normal chat — call Qwen or show clear offline message if _llm_available(): reply = _call_qwen(user_msg) else: reply = ( "Qwen is not connected. Set the vLLM endpoint in the sidebar.\n\n" "If you're on the MI300X, make sure vLLM is running:\n" " docker exec -it rocm /bin/bash\n" " vllm serve Qwen/Qwen3-8B --host 0.0.0.0 --port 8000\n\n" "Then set the URL to http://localhost:8000 in the sidebar." ) _add_msg("assistant", reply, think=st.session_state.pop("_last_think", "")) # ───────────────────────────────────────────────────────────────────────────── # Agent factory # ───────────────────────────────────────────────────────────────────────────── def _make_agent(use_cnn, meas_budget, max_steps): from qdot.core.state import ExperimentState from qdot.core.types import ChargeLabel from qdot.core.governance import GovernanceLogger from qdot.core.hitl import HITLManager from qdot.hardware.safety import SafetyCritic from qdot.perception.dqc import DQCGatekeeper from qdot.simulator.cim import CIMSimulatorAdapter from qdot.agent.executive import ExecutiveAgent rng = np.random.default_rng() E_c = float(rng.uniform(2.1, 2.9)) adapter = CIMSimulatorAdapter(device_id="demo_qdot", params={ "E_c1": E_c, "E_c2": E_c*float(rng.uniform(0.93,1.07)), "t_c": 0.05, "T": 0.015, "lever_arm": float(rng.uniform(0.68,0.82)), "noise_level": 0.02, }) state = ExperimentState.new(device_id="demo_qdot", target_label=ChargeLabel.DOUBLE_DOT) inspection = None if use_cnn: try: from qdot.perception.inspector import InspectionAgent inspection = InspectionAgent() except Exception as e: st.sidebar.warning(f"CNN unavailable: {e}") run_dir = Path("results/demo") / state.run_id run_dir.mkdir(parents=True, exist_ok=True) hitl_mgr = HITLManager(queue_dir=str(run_dir/"hitl")) hitl_mgr.set_test_mode() agent = ExecutiveAgent( state=state, adapter=adapter, inspection_agent=inspection, dqc=DQCGatekeeper(), safety_critic=SafetyCritic(voltage_bounds=state.voltage_bounds, l1_max=0.10), hitl_manager=hitl_mgr, governance_logger=GovernanceLogger( run_id=state.run_id, log_dir=str(run_dir/"governance")), max_steps=max_steps, measurement_budget=meas_budget, ) return agent, state, agent.narrator, hitl_mgr def _run_thread(agent, done_event): try: agent.run() except Exception: pass finally: done_event.set() # ───────────────────────────────────────────────────────────────────────────── # Charts # ───────────────────────────────────────────────────────────────────────────── import plotly.graph_objects as go def _best_scan(state): if state.last_classification: m = state.measurements.get(state.last_classification.measurement_id) if m and m.is_2d and m.array is not None: return m best, best_area = None, 0.0 for m in state.measurements.values(): if not m.is_2d or m.array is None: continue if m.v1_range and m.v2_range: area = (m.v1_range[1]-m.v1_range[0])*(m.v2_range[1]-m.v2_range[0]) if area > best_area: best_area,best = area,m return best def _fig_stability(state): m = _best_scan(state) fig = go.Figure() if m is None: fig.add_annotation(text="Awaiting first 2D scan…",x=0.5,y=0.5,showarrow=False, font=dict(color="#8A9AB0",size=12,family="JetBrains Mono")) else: arr = np.asarray(m.array, dtype=np.float32) p2,p98 = np.percentile(arr,[2,98]) arrn = np.clip((arr-p2)/max(p98-p2,1e-8),0,1) v1lo,v1hi = m.v1_range or (-8,8) v2lo,v2hi = m.v2_range or (-8,8) fig.add_trace(go.Heatmap( z=arrn, x=np.linspace(v1lo,v1hi,arrn.shape[1]), y=np.linspace(v2lo,v2hi,arrn.shape[0]), colorscale=STABILITY_CS, showscale=True, colorbar=dict(thickness=8,tickvals=[0,.5,1], ticktext=["Blockade","—","Peak"], tickfont=dict(size=8,family="JetBrains Mono"), title=dict(text="G",font=dict(size=9))), )) vg1,vg2 = state.current_voltage.vg1,state.current_voltage.vg2 fig.add_shape(type="line",x0=vg1,x1=vg1,y0=v2lo,y1=v2hi, line=dict(color="rgba(255,255,255,0.55)",width=1,dash="dot")) fig.add_shape(type="line",x0=v1lo,x1=v1hi,y0=vg2,y1=vg2, line=dict(color="rgba(255,255,255,0.55)",width=1,dash="dot")) fig.add_trace(go.Scatter(x=[vg1],y=[vg2],mode="markers", marker=dict(size=8,color="#FF4040",symbol="cross-thin", line=dict(width=2.5,color="#FF4040")),showlegend=False)) fig.update_layout( title=dict(text="Charge Stability Diagram",font=dict(size=11,color="#5A6478")), xaxis=dict(title="Vg₁ (V)",gridcolor="#E8E4DC",zeroline=False), yaxis=dict(title="Vg₂ (V)",gridcolor="#E8E4DC",zeroline=False), height=265, **PLOT_LAYOUT) return fig def _fig_belief(state): probs = state.belief.charge_probs z = np.zeros((3,3)) for (n1,n2),p in probs.items(): if 0<=n1<=2 and 0<=n2<=2: z[n2][n1]=float(p) fig = go.Figure(data=go.Heatmap( z=z, x=["N₁=0","N₁=1","N₁=2"], y=["N₂=0","N₂=1","N₂=2"], colorscale=[[0,"#F2F0EB"],[0.5,"#B2DFDB"],[1,"#00897B"]], showscale=False, zmin=0, zmax=1, text=[[f"{z[j][i]:.2f}" for i in range(3)] for j in range(3)], texttemplate="%{text}", textfont=dict(size=12,color="#1C2333",family="JetBrains Mono"), )) fig.add_shape(type="rect",x0=.5,x1=1.5,y0=.5,y1=1.5, line=dict(color="#00897B",width=2.5)) fig.add_annotation(x=1,y=1.5,text="TARGET",showarrow=False, yshift=14,font=dict(color="#00897B",size=9,family="JetBrains Mono")) fig.update_layout( title=dict(text="Belief P(N₁,N₂|obs)",font=dict(size=11,color="#5A6478")), height=210, **PLOT_LAYOUT) return fig def _fig_traj(state): if len(state.trajectory)<2: return None xs=[v.vg1 for v in state.trajectory]; ys=[v.vg2 for v in state.trajectory] n=len(xs) fig=go.Figure() fig.add_trace(go.Scatter(x=xs,y=ys,mode="lines+markers", line=dict(color="#B2DFDB",width=1.5), marker=dict(size=5,color=list(range(n)), colorscale=[[0,"#E0F4F1"],[1,"#00897B"]],showscale=False), showlegend=False)) fig.add_trace(go.Scatter(x=[xs[-1]],y=[ys[-1]],mode="markers", marker=dict(size=9,color="#E85000",symbol="x-thin", line=dict(width=2.5,color="#E85000")),showlegend=False)) fig.update_layout( title=dict(text="Voltage Trajectory",font=dict(size=11,color="#5A6478")), xaxis=dict(title="Vg₁",gridcolor="#E8E4DC"), yaxis=dict(title="Vg₂",gridcolor="#E8E4DC"), height=185, **PLOT_LAYOUT) return fig # ───────────────────────────────────────────────────────────────────────────── # UI helpers # ───────────────────────────────────────────────────────────────────────────── def _timeline(current, done_event): is_done = done_event and done_event.is_set() try: ci = [s[0] for s in STAGES].index(current) except ValueError: ci = -1 html = '
' for i,(sname,icon,desc,p3) in enumerate(STAGES): if p3: css="tn tn-phase3" elif i
{chk}
' f'
{desc}
') if i' html += '' st.markdown(html, unsafe_allow_html=True) def _kpi(state, agent): b=state.total_measurements; t=agent.measurement_budget vg1=state.current_voltage.vg1; vg2=state.current_voltage.vg2 snr=state.last_dqc.snr_db if state.last_dqc else 0.0 st.markdown(f"""
{b}/{t}
Measurements
{vg1:+.2f} V
Vg₁
{vg2:+.2f} V
Vg₂
{snr:.1f} dB
SNR
""", unsafe_allow_html=True) def _spy(current, hitl_active): html = '
' for key,em,name,role,active in SPY_AGENTS: on = (key=="hitl" and hitl_active) or (key!="hitl" and current in active) html += (f'
' f'
{em}
' f'
{name}
{role}
' f'
') html += '
' st.markdown(html, unsafe_allow_html=True) def _render_chat(): """Read-only HTML chat panel. The actual input widget is at page bottom.""" # Pump narrator events (anomalies only — don't pollute chat with routine logs) narrator = st.session_state.narrator if narrator: for ev in narrator.event_log(): tag = f"_ev_{ev.timestamp:.4f}" if tag not in st.session_state: st.session_state[tag] = True if ev.kind == "exception" and ev.response: # Let Qwen contextualise the anomaly if it's available if _llm_available(): contextualised = _call_qwen( f"Agent anomaly detected: {ev.description}. " f"Briefly explain what this means for the experiment.") _add_msg("assistant", contextualised, kind="ev") else: _add_msg("assistant", f"⚠ {ev.description}", kind="ev") llm_on = _llm_available() badge = ( '⬡ Qwen3-8B · AMD MI300X' if llm_on else 'LLM offline — set URL in sidebar' ) msgs_html = "" for msg in st.session_state.chat: c = (msg["content"] .replace("&","&").replace("<","<").replace(">",">") .replace("\n","
")) c = re.sub(r'\*\*(.+?)\*\*', r'\1', c) if msg["role"] == "user": msgs_html += (f'
You
' f'
{c}
') else: css = "msg msg-a msg-ev" if msg.get("kind")=="ev" else "msg msg-a" lbl = "Dr. Q — anomaly" if msg.get("kind")=="ev" else "Dr. Q" think = msg.get("think", "") think_html = "" if think: t = (think.replace("&","&").replace("<","<").replace(">",">") .replace("\n","
")) think_html = ( f'
' f'' f' Dr. Q\'s reasoning' f'
' f'{t}
' ) msgs_html += (f'
{lbl}
' f'
{think_html}{c}
') st.markdown( f'
' f'
DR. Q — AI CO-PILOT{badge}
' f'
{msgs_html}
' f'
' f'', unsafe_allow_html=True, ) # ───────────────────────────────────────────────────────────────────────────── # Sidebar — always visible # ───────────────────────────────────────────────────────────────────────────── running = st.session_state.running exp_state = st.session_state.exp_state done_event = st.session_state.done_event with st.sidebar: st.markdown( '
' '⚛ SimQuantum
', unsafe_allow_html=True, ) # LLM connection — most important, top of sidebar st.markdown( '
DR. Q — ENDPOINT
', unsafe_allow_html=True, ) new_url = st.text_input( "Endpoint URL", value=st.session_state.llm_url, placeholder="https://api.fireworks.ai/inference/v1", label_visibility="collapsed", ) if new_url != st.session_state.llm_url: st.session_state.llm_url = new_url new_key = st.text_input( "API Key", value=st.session_state.llm_api_key, placeholder="fw_xxxxxxxxxxxxxxxx", type="password", label_visibility="collapsed", ) if new_key != st.session_state.llm_api_key: st.session_state.llm_api_key = new_key if st.session_state.llm_url: st.markdown( f'
● connected to {st.session_state.llm_url}
', unsafe_allow_html=True, ) # Test button if st.button("Test connection", use_container_width=True): try: import openai url = st.session_state.llm_url.strip().rstrip("/") api_key = st.session_state.llm_api_key.strip() or os.environ.get("QDOT_LLM_API_KEY", "EMPTY") c = openai.OpenAI(base_url=url, api_key=api_key) models = c.models.list() names = [m.id for m in models.data][:3] st.success(f"Connected ✓") except Exception as e: st.error(f"Failed: {e}") else: st.markdown( '
● offline — paste Fireworks URL + key above
', unsafe_allow_html=True, ) st.markdown('
', unsafe_allow_html=True) new_model = st.text_input( "Model name", value=st.session_state.llm_model, label_visibility="visible", disabled=running, ) if new_model != st.session_state.llm_model: st.session_state.llm_model = new_model st.divider() st.markdown( '
RUN CONFIGURATION
', unsafe_allow_html=True, ) st.session_state.use_cnn = st.toggle( "CNN Charge Classifier", value=st.session_state.use_cnn, disabled=running) st.session_state.meas_budget = st.slider( "Measurement Budget", 512, 8192, st.session_state.meas_budget, 256, disabled=running) st.session_state.max_steps = st.slider( "Max Steps", 10, 300, st.session_state.max_steps, 10, disabled=running) st.divider() if st.button("Reset session", use_container_width=True, disabled=running): for k in ["agent","exp_state","narrator","hitl_manager","done_event","thread","chat"]: st.session_state[k] = [] if k=="chat" else None st.session_state.running = False st.rerun() st.markdown( '
' 'Physics: CIM simulator (CPU)
' 'CNN: 5-model ensemble, 91.4% val acc
' 'LLM: Qwen3-8B on AMD MI300X
' 'BOOTSTRAPPING→CHARGE_ID: ✓ working
' 'NAVIGATION: Phase 3, in development' '
', unsafe_allow_html=True, ) # ───────────────────────────────────────────────────────────────────────────── # Check completion # ───────────────────────────────────────────────────────────────────────────── if done_event and done_event.is_set() and running: st.session_state.running = False running = False if exp_state and exp_state.stage.name == "COMPLETE": st.balloons() # Ask Qwen for a post-run summary already_summarised = any(m.get("kind")=="summary" for m in st.session_state.chat) if not already_summarised and exp_state: stage = exp_state.stage.name meas = exp_state.total_measurements budget = st.session_state.agent.measurement_budget if st.session_state.agent else 8096 bt = exp_state.total_backtracks _add_msg("user", f"The run just finished — it stopped at {stage}. " f"{meas}/{budget} measurements used, {bt} backtracks. " f"What happened and what should I know for the next run?") if _llm_available(): reply = _call_qwen( f"Post-run: stopped at {stage}, {meas}/{budget} measurements, {bt} backtracks.") else: reply = (f"Run stopped at {stage}. {meas}/{budget} measurements used, " f"{bt} backtracks. Connect Qwen for a detailed analysis.") _add_msg("assistant", reply, kind="summary") # ───────────────────────────────────────────────────────────────────────────── # Top bar # ───────────────────────────────────────────────────────────────────────────── tl,tr = st.columns([3,1]) with tl: st.markdown( '
' '
⚛ SimQuantum Tuning Lab
' '
Autonomous quantum dot tuning · AMD Developer Hackathon 2026 · Kudzai Musarandega
' '
', unsafe_allow_html=True) with tr: b = ('● LIVE' if running else '● IDLE') if _llm_available(): b += ' ⬡ MI300X' st.markdown(f'
{b}
', unsafe_allow_html=True) # ───────────────────────────────────────────────────────────────────────────── # Splash (pre-run) # ───────────────────────────────────────────────────────────────────────────── if exp_state is None: if not st.session_state.chat: if _llm_available(): # Let Qwen introduce itself _add_msg("user", "Hello, introduce yourself briefly.") intro = _call_qwen("Hello, introduce yourself briefly.") _add_msg("assistant", intro) else: _add_msg("assistant", "Ready. Set the vLLM endpoint in the sidebar to enable Qwen.\n\n" "Type **start** to begin a tuning run, or ask me about the experiment.") sl,sr = st.columns([3,2], gap="large") with sl: st.markdown("""
What this system does

SimQuantum autonomously tunes a double quantum dot device to the (1,1) charge state — one electron per dot — required for spin qubit operation. 6-stage POMDP planner, 5-model CNN ensemble (91.4% val acc), Bayesian optimisation. Qwen3-8B on AMD MI300X acts as Dr. Q — ask it anything, in any register.

Reading a stability diagram

Conductance G vs gate voltages (Vg₁, Vg₂). Bright lines = Coulomb peaks (charge transitions). Dark regions = Coulomb blockade (fixed electron number). Intersections form a honeycomb: (0,0), (1,0), (0,1), (1,1)… The agent navigates to the (1,1) diamond.

""", unsafe_allow_html=True) img_path = Path(__file__).parent/"assets"/"simquantum.png" if img_path.exists(): st.image(str(img_path), use_container_width=True) with sr: _render_chat() # ───────────────────────────────────────────────────────────────────────────── # Live dashboard # ───────────────────────────────────────────────────────────────────────────── else: agent = st.session_state.agent hitl_mgr = st.session_state.hitl_manager current_stg = exp_state.stage.name _timeline(current_stg, done_event) _kpi(exp_state, agent) pct = min(100, int(100*exp_state.total_measurements/agent.measurement_budget)) st.progress(pct/100, text=f"Measurement budget {pct}%") pending = hitl_mgr.get_pending() if hitl_mgr else [] if pending: req = pending[0] st.markdown( f'
' f'
⚠ HITL GATE — Human approval required
' f'
Step {req["step"]} · Stage {req["stage"]} · ' f'Risk {req["risk_score"]:.2f}
{req["trigger_reason"]}
' f'
', unsafe_allow_html=True) hc1,hc2,_ = st.columns([1,1,5]) with hc1: if st.button("✓ Approve",type="primary",key=f"appr_{req['id']}"): hitl_mgr.approve(req["id"],deciding_human="operator") _add_msg("user","I approved the HITL gate.") if _llm_available(): _add_msg("assistant", _call_qwen("The operator just approved the HITL gate. Acknowledge briefly.")) else: _add_msg("assistant","Approved. Agent continues.") st.rerun() with hc2: if st.button("✗ Reject",key=f"rej_{req['id']}"): hitl_mgr.reject(req["id"],deciding_human="operator") _add_msg("user","I rejected the HITL gate.") if _llm_available(): _add_msg("assistant", _call_qwen("The operator rejected the HITL gate. Acknowledge and explain what happens next.")) else: _add_msg("assistant","Rejected. Agent backtracks.") st.rerun() st.markdown("
",unsafe_allow_html=True) left,right = st.columns([3,2],gap="large") with left: st.markdown('
Charge Stability Diagram
', unsafe_allow_html=True) st.plotly_chart(_fig_stability(exp_state),use_container_width=True, config={"displayModeBar":False},key=f"s_{time.monotonic_ns()}") if exp_state.last_classification: cls = exp_state.last_classification ood_col = "#C84B00" if exp_state.is_ood else "#00897B" ood_txt = "OOD warning" if exp_state.is_ood else "in-distribution" st.markdown( f'
CNN Classification' f'{ood_txt}
' f'{cls.label.value.upper()}' f'' f'conf {cls.confidence:.1%}
', unsafe_allow_html=True) bc1,bc2 = st.columns(2) with bc1: st.plotly_chart(_fig_belief(exp_state),use_container_width=True, config={"displayModeBar":False},key=f"b_{time.monotonic_ns()}") with bc2: ft = _fig_traj(exp_state) if ft: st.plotly_chart(ft,use_container_width=True, config={"displayModeBar":False},key=f"t_{time.monotonic_ns()}") st.markdown('
Agent Activity
', unsafe_allow_html=True) _spy(current_stg, bool(pending)) img_path = Path(__file__).parent/"assets"/"simquantum.png" if img_path.exists(): st.image(str(img_path),use_container_width=True) with right: _render_chat() if current_stg in STAGE_DESC: desc,cost = STAGE_DESC[current_stg] st.markdown( f'
' f'
Current stage · {current_stg}
' f'
{desc}
' f'
Budget: {cost}
' f'
', unsafe_allow_html=True) if done_event and done_event.is_set(): ok = current_stg=="COMPLETE" col = "#00897B" if ok else "#C84B00" txt = "MISSION COMPLETE" if ok else f"STOPPED — {current_stg}" red = 1.0-(exp_state.total_measurements/max(64*64,1)) st.markdown( f'
' f'
{txt}
' f'
' f'Measurements: {exp_state.total_measurements}' f'Steps: {agent.control_steps}' f'Backtracks: {exp_state.total_backtracks}' f'Reduction: {red:.0%}' f'
', unsafe_allow_html=True) if st.button("🔄 New Run",use_container_width=True): for k in ["agent","exp_state","narrator","hitl_manager","done_event","thread"]: st.session_state[k] = None st.session_state.running = False st.rerun() # ───────────────────────────────────────────────────────────────────────────── # Chat input — always at page bottom, never inside a conditional. # Streamlit requires st.chat_input at the same tree position every rerun. # ───────────────────────────────────────────────────────────────────────────── if prompt := st.chat_input("Ask Dr. Q anything, or type 'start' to begin…"): _handle_chat(prompt) st.rerun() # ───────────────────────────────────────────────────────────────────────────── # Auto-refresh while agent is running # ───────────────────────────────────────────────────────────────────────────── if running and done_event and not done_event.is_set(): time.sleep(0.8) st.rerun()