| |
| """ |
| Minimal FastAPI + Gradio server for the Quantum Circuit Optimization Environment. |
| |
| Deployment spec: |
| - Root endpoint: GET / → health check JSON |
| - Gradio UI: mounted at /ui |
| - Port: 7860 (managed externally by Docker / HF Spaces) |
| - No uvicorn.run() in this file |
| |
| Run locally: |
| uvicorn server.app:app --host 0.0.0.0 --port 7860 |
| """ |
|
|
| import json |
| import logging |
| import os |
| import sys |
|
|
| |
| _HERE = os.path.dirname(os.path.abspath(__file__)) |
| _ROOT = os.path.dirname(_HERE) |
| for _p in (_ROOT, _HERE): |
| if _p not in sys.path: |
| sys.path.insert(0, _p) |
|
|
| from fastapi import FastAPI, Body |
| from fastapi.responses import RedirectResponse |
| import gradio as gr |
| from pydantic import BaseModel |
|
|
| from openenv.core.env_server import create_fastapi_app |
|
|
| from models import ActionType, GateType, QuantumAction, QuantumObservation |
| from .my_env_environment import QuantumCircuitEnvironment |
|
|
| logging.basicConfig(level=logging.INFO) |
| logger = logging.getLogger(__name__) |
|
|
| |
| |
| |
| fastapi_app = create_fastapi_app( |
| lambda: QuantumCircuitEnvironment(seed=42), |
| QuantumAction, |
| QuantumObservation |
| ) |
|
|
| @fastapi_app.get("/web") |
| def web_redirect(): |
| """Redirect /web to root Gradio UI.""" |
| return RedirectResponse(url="/") |
|
|
| @fastapi_app.get("/health") |
| def health(): |
| return {"status": "ok", "message": "Quantum Circuit Optimizer is running"} |
|
|
| @fastapi_app.get("/api/info") |
| def info(): |
| return { |
| "name": "Quantum Circuit Optimization Environment", |
| "version": "1.0.0", |
| "gradio_ui": "Available at root path /" |
| } |
|
|
| class ResetRequest(BaseModel): |
| task_id: str = "easy" |
|
|
| |
| @fastapi_app.post("/reset") |
| def reset_endpoint(body: dict = Body(default={})): |
| """OpenEnv-compliant reset endpoint.""" |
| task_id = body.get("task_id", "easy") |
| temp_env = QuantumCircuitEnvironment(seed=42) |
| obs = temp_env.reset(config={"task_id": task_id}) |
| obs_dict = obs.model_dump() if hasattr(obs, "model_dump") else obs.dict() |
| return JSONResponse(content={"observation": obs_dict}) |
|
|
| @fastapi_app.post("/step") |
| def step_endpoint(action: QuantumAction): |
| """OpenEnv-compliant step endpoint.""" |
| temp_env = QuantumCircuitEnvironment(seed=42) |
| temp_env.reset(config={"task_id": "easy"}) |
| obs = temp_env.step(action) |
| obs_dict = obs.model_dump() if hasattr(obs, "model_dump") else obs.dict() |
| return JSONResponse(content={"observation": obs_dict}) |
|
|
| @fastapi_app.get("/state") |
| def state_endpoint(): |
| """OpenEnv-compliant state endpoint.""" |
| temp_env = QuantumCircuitEnvironment(seed=42) |
| temp_env.reset(config={"task_id": "easy"}) |
| state = temp_env.state |
| state_dict = state.model_dump() if hasattr(state, "model_dump") else state.dict() |
| return JSONResponse(content={"state": state_dict}) |
|
|
| |
| @fastapi_app.post("/api/reset") |
| def api_reset(req: ResetRequest): |
| """Programmatic endpoint to reset the environment.""" |
| global current_obs, step_history |
| current_obs = env.reset(config={"task_id": req.task_id}) |
| step_history = [] |
| return current_obs |
|
|
| @fastapi_app.post("/api/step") |
| def api_step(action: QuantumAction): |
| """Programmatic endpoint to step the environment.""" |
| global current_obs |
| current_obs = env.step(action) |
| return current_obs |
|
|
| @fastapi_app.get("/api/state") |
| def api_state(): |
| """Programmatic endpoint to fetch the full internal state.""" |
| return env.state |
|
|
| |
| |
| |
| env = QuantumCircuitEnvironment(seed=42) |
|
|
| |
| current_obs = None |
| step_history: list = [] |
|
|
|
|
| |
| |
| |
|
|
| def _format_circuit(obs) -> str: |
| """ASCII circuit summary.""" |
| if not obs or not obs.circuit_gates: |
| return "(empty circuit)" |
| lines = [] |
| for i, g in enumerate(obs.circuit_gates): |
| gate = g.get("gate", "?") |
| qubits = g.get("qubits", []) |
| param = g.get("parameter") |
| q_str = ",".join(str(q) for q in qubits) |
| if param is not None: |
| lines.append(f" {i+1:>2}. {gate}({q_str}, θ={param:.4f})") |
| else: |
| lines.append(f" {i+1:>2}. {gate}({q_str})") |
| return "\n".join(lines) |
|
|
|
|
| def _format_scores(obs) -> str: |
| """Plain-text score summary.""" |
| if not obs: |
| return "" |
| meta = obs.metadata or {} |
| fid = round(meta.get("fidelity_score", obs.fidelity), 4) |
| eff = round(meta.get("efficiency_score", 0), 4) |
| noi = round(meta.get("noise_score", 0), 4) |
| con = round(meta.get("constraints_score", 0), 4) |
| agg = round(meta.get("aggregate_score", obs.score), 4) |
| return ( |
| f"Fidelity: {fid}\n" |
| f"Efficiency: {eff}\n" |
| f"Noise: {noi}\n" |
| f"Constraints: {con}\n" |
| f"AGGREGATE: {agg}" |
| ) |
|
|
|
|
| def _status(obs, extra: str = "") -> str: |
| if not obs: |
| return "No environment loaded — click Reset." |
| s = ( |
| f"Task: {obs.task_id} | Qubits: {obs.num_qubits} | " |
| f"Step: {obs.steps_taken}/{obs.max_steps}\n" |
| f"Fidelity: {obs.fidelity:.4f} | Score: {obs.score:.4f} | " |
| f"Reward: {obs.reward:+.4f}" |
| ) |
| if obs.done: |
| s += f"\n>>> EPISODE COMPLETE — final score: {obs.score:.4f} <<<" |
| if extra: |
| s += f"\n{extra}" |
| return s |
|
|
|
|
| |
| |
| |
|
|
| TASK_MAP = { |
| "Bell State (Easy)": "easy", |
| "GHZ State (Medium)": "medium", |
| "Unitary Approx (Hard)": "hard", |
| "Imperfect but Efficient": "efficient", |
| "Noise-Dominant": "noisy", |
| "Budgeted Optimization": "budget", |
| "Approximate Target": "approx", |
| } |
|
|
|
|
| def do_reset(task_name): |
| global current_obs, step_history |
| task_id = TASK_MAP.get(task_name, "easy") |
| current_obs = env.reset(config={"task_id": task_id}) |
| step_history = [] |
| return ( |
| _status(current_obs, f"Reset to: {task_name}"), |
| _format_circuit(current_obs), |
| _format_scores(current_obs), |
| "", |
| ) |
|
|
|
|
| def do_step(action_type, gate_type, qubit_0, qubit_1, parameter): |
| global current_obs, step_history |
|
|
| if current_obs is None: |
| return "ERROR: Reset the environment first!", "(no circuit)", "", "" |
| if current_obs.done: |
| return _status(current_obs), _format_circuit(current_obs), _format_scores(current_obs), "\n".join(step_history) |
|
|
| try: |
| at = ActionType(action_type) |
| gate = None |
| qubits = [] |
| param = None |
|
|
| if at == ActionType.ADD: |
| gate = GateType(gate_type) if gate_type else None |
| if gate in (GateType.CNOT,): |
| qubits = [int(qubit_0), int(qubit_1)] |
| elif gate is not None and gate.value == "SWAP": |
| qubits = [int(qubit_0), int(qubit_1)] |
| else: |
| qubits = [int(qubit_0)] |
| if gate in (GateType.RX, GateType.RZ) and parameter is not None: |
| param = float(parameter) |
|
|
| elif at == ActionType.SWAP: |
| qubits = [int(qubit_0), int(qubit_1)] |
|
|
| elif at == ActionType.PARAM: |
| if parameter is not None: |
| param = float(parameter) |
|
|
| action = QuantumAction(action_type=at, gate=gate, qubits=qubits, parameter=param) |
| current_obs = env.step(action) |
|
|
| except Exception as exc: |
| logger.exception("Step error: %s", exc) |
| return ( |
| f"ACTION ERROR: {exc}", |
| _format_circuit(current_obs) if current_obs else "(no circuit)", |
| _format_scores(current_obs) if current_obs else "", |
| "\n".join(step_history), |
| ) |
|
|
| gate_label = "" |
| if gate: |
| gate_label = f" {gate_type}({','.join(str(q) for q in qubits)})" |
| entry = ( |
| f"Step {current_obs.steps_taken}: {action_type}{gate_label}" |
| f" → fid={current_obs.fidelity:.4f}" |
| f" score={current_obs.score:.4f}" |
| f" reward={current_obs.reward:+.4f}" |
| + (" [DONE]" if current_obs.done else "") |
| ) |
| step_history.append(entry) |
|
|
| return ( |
| _status(current_obs), |
| _format_circuit(current_obs), |
| _format_scores(current_obs), |
| "\n".join(step_history), |
| ) |
|
|
|
|
| |
| |
| |
|
|
| TASK_DESCRIPTIONS = { |
| "Bell State (Easy)": { |
| "description": "Create the famous Bell state: (|00⟩ + |11⟩)/√2", |
| "difficulty": "Easy", |
| "qubits": 2, |
| "noise": "None", |
| "connectivity": "Full", |
| "goal": "Learn basic 2-qubit entanglement with H + CNOT" |
| }, |
| "GHZ State (Medium)": { |
| "description": "Build a 3-qubit GHZ state: (|000⟩ + |111⟩)/√2", |
| "difficulty": "Medium", |
| "qubits": 3, |
| "noise": "Depolarizing", |
| "connectivity": "Linear (0↔1↔2)", |
| "goal": "Handle noise and limited connectivity" |
| }, |
| "Unitary Approx (Hard)": { |
| "description": "Approximate a complex unitary: Ry(π/3) ⊗ Rz(π/4) · CNOT", |
| "difficulty": "Hard", |
| "qubits": 2, |
| "noise": "Thermal", |
| "connectivity": "Restricted", |
| "goal": "Precise parametric gate tuning under noise" |
| }, |
| "Imperfect but Efficient": { |
| "description": "GHZ state with emphasis on circuit efficiency", |
| "difficulty": "Medium", |
| "qubits": 3, |
| "noise": "None", |
| "connectivity": "Full", |
| "goal": "Balance fidelity (60%) vs efficiency (40%)" |
| }, |
| "Noise-Dominant": { |
| "description": "Bell state optimized for noise resilience", |
| "difficulty": "Medium", |
| "qubits": 2, |
| "noise": "Thermal (High)", |
| "connectivity": "Full", |
| "goal": "Minimize gate count to survive noise (40% weight)" |
| }, |
| "Budgeted Optimization": { |
| "description": "GHZ state with strict 3-gate budget", |
| "difficulty": "Medium", |
| "qubits": 3, |
| "noise": "None", |
| "connectivity": "Full", |
| "goal": "Achieve high fidelity with minimal gates" |
| }, |
| "Approximate Target": { |
| "description": "4-qubit GHZ with relaxed fidelity threshold (0.80)", |
| "difficulty": "Medium", |
| "qubits": 4, |
| "noise": "None", |
| "connectivity": "Full", |
| "goal": "Scale to larger circuits with approximate solutions" |
| }, |
| } |
|
|
| GATE_DESCRIPTIONS = { |
| "H": "Hadamard — Creates superposition: |0⟩ → (|0⟩+|1⟩)/√2", |
| "X": "Pauli-X — Bit flip: |0⟩ ↔ |1⟩", |
| "CNOT": "Controlled-NOT — Entangles 2 qubits (requires qubit 0 & 1)", |
| "RX": "X-Rotation — Rotate around X-axis by angle θ (needs parameter)", |
| "RZ": "Z-Rotation — Rotate around Z-axis by angle θ (needs parameter)", |
| } |
|
|
| ACTION_DESCRIPTIONS = { |
| "ADD": "Add a quantum gate to the circuit", |
| "REMOVE": "Remove the last gate from the circuit", |
| "SWAP": "Swap two qubits (requires qubit 0 & 1)", |
| "PARAM": "Tune the parameter of the last parametric gate (RX/RZ)", |
| "STOP": "End the episode and finalize the circuit", |
| } |
|
|
| def create_gradio_app() -> gr.Blocks: |
| with gr.Blocks(title="Quantum Circuit Optimizer", theme=gr.themes.Soft()) as demo: |
|
|
| |
| gr.Markdown( |
| """ |
| # Quantum Circuit Optimizer |
| ### Build quantum circuits step-by-step to maximize fidelity with target states |
| |
| **Goal:** Design circuits that are accurate, efficient, noise-resistant, and hardware-compliant. |
| """ |
| ) |
|
|
| |
| with gr.Accordion("Available Tasks & Challenges", open=False): |
| gr.Markdown( |
| """ |
| Choose a task below to start optimizing. Each task has different constraints: |
| |
| | Task | Difficulty | Qubits | Noise | Connectivity | Focus | |
| |------|------------|--------|-------|--------------|-------| |
| | **Bell State** | Easy | 2 | None | Full | Basic entanglement | |
| | **GHZ State** | Medium | 3 | Depolarizing | Linear | Noise + topology | |
| | **Unitary Approx** | Hard | 2 | Thermal | Restricted | Parametric tuning | |
| | **Efficient** | Medium | 3 | None | Full | Minimize depth/gates | |
| | **Noise-Dominant** | Medium | 2 | High thermal | Full | Noise resilience | |
| | **Budgeted** | Medium | 3 | None | Full | 3-gate limit | |
| | **Approximate** | Medium | 4 | None | Full | Scale to 4 qubits | |
| """ |
| ) |
|
|
| with gr.Accordion("Quick Start Guide", open=False): |
| gr.Markdown( |
| """ |
| ### How to Use: |
| 1. **Select a Task** — Choose from 7 quantum optimization challenges |
| 2. **Reset Environment** — Initialize the task and clear the circuit |
| 3. **Build Your Circuit** — Add gates step-by-step: |
| - Use **H** and **CNOT** for entanglement |
| - Use **RX/RZ** for parametric rotations (set angle in radians) |
| - Use **SWAP** to rearrange qubits for connectivity |
| 4. **Monitor Scores** — Watch fidelity, efficiency, noise, and constraints |
| 5. **Stop** — End the episode when satisfied |
| |
| ### Scoring: |
| - **Fidelity** (40-60%): How close to the target state |
| - **Efficiency** (20%): Circuit depth and gate count |
| - **Noise** (15-25%): Resilience to quantum errors |
| - **Constraints** (10-15%): Hardware connectivity compliance |
| """ |
| ) |
|
|
| gr.Markdown("---") |
|
|
| |
| gr.Markdown("### Step 1: Select Task") |
| with gr.Row(): |
| with gr.Column(scale=3): |
| task_dd = gr.Dropdown( |
| choices=list(TASK_MAP.keys()), |
| value="Bell State (Easy)", |
| label="Choose a Quantum Task", |
| info="Select the optimization challenge you want to solve" |
| ) |
| with gr.Column(scale=1): |
| reset_btn = gr.Button("Reset Environment", variant="primary", size="lg") |
|
|
| |
| task_info = gr.Markdown("") |
| |
| def update_task_info(task_name): |
| info = TASK_DESCRIPTIONS.get(task_name, {}) |
| return f""" |
| **{info.get('difficulty', '')}** | {info.get('qubits', 0)} qubits | Noise: {info.get('noise', 'N/A')} | Connectivity: {info.get('connectivity', 'N/A')} |
| |
| {info.get('description', '')} |
| |
| **Goal:** {info.get('goal', '')} |
| """ |
| |
| task_dd.change(fn=update_task_info, inputs=[task_dd], outputs=[task_info]) |
|
|
| gr.Markdown("---") |
|
|
| |
| gr.Markdown("### Step 2: Build Your Circuit") |
| |
| with gr.Row(): |
| with gr.Column(): |
| action_dd = gr.Dropdown( |
| choices=["ADD", "REMOVE", "SWAP", "PARAM", "STOP"], |
| value="ADD", |
| label="Action Type", |
| info="What operation do you want to perform?" |
| ) |
| action_help = gr.Markdown(ACTION_DESCRIPTIONS["ADD"]) |
| action_dd.change(fn=lambda x: ACTION_DESCRIPTIONS.get(x, ""), inputs=[action_dd], outputs=[action_help]) |
| |
| with gr.Column(): |
| gate_dd = gr.Dropdown( |
| choices=["H", "X", "CNOT", "RX", "RZ"], |
| value="H", |
| label="Gate Type (for ADD action)", |
| info="Which quantum gate to add?" |
| ) |
| gate_help = gr.Markdown(GATE_DESCRIPTIONS["H"]) |
| gate_dd.change(fn=lambda x: GATE_DESCRIPTIONS.get(x, ""), inputs=[gate_dd], outputs=[gate_help]) |
|
|
| with gr.Row(): |
| qubit_0 = gr.Number( |
| value=0, |
| label="Qubit 0", |
| info="Target qubit (or first qubit for 2-qubit gates)", |
| precision=0 |
| ) |
| qubit_1 = gr.Number( |
| value=1, |
| label="Qubit 1", |
| info="Second qubit (for CNOT/SWAP only)", |
| precision=0 |
| ) |
| parameter = gr.Slider( |
| minimum=-3.14159, |
| maximum=3.14159, |
| value=0, |
| step=0.01, |
| label="Parameter (θ in radians)", |
| info="Rotation angle for RX/RZ gates" |
| ) |
|
|
| with gr.Row(): |
| step_btn = gr.Button("Execute Step", variant="primary", size="lg", scale=2) |
| state_btn = gr.Button("Get Full State", variant="secondary", scale=1) |
|
|
| gr.Markdown("---") |
|
|
| |
| gr.Markdown("### Step 3: Monitor Your Progress") |
| |
| with gr.Row(): |
| with gr.Column(): |
| status_out = gr.Textbox( |
| label="Current Status", |
| lines=4, |
| interactive=False, |
| placeholder="Click 'Reset Environment' to start..." |
| ) |
| scores_out = gr.Textbox( |
| label="Score Breakdown", |
| lines=6, |
| interactive=False |
| ) |
| |
| with gr.Column(): |
| circuit_out = gr.Textbox( |
| label="Current Circuit", |
| lines=10, |
| interactive=False, |
| placeholder="Your circuit will appear here..." |
| ) |
|
|
| history_out = gr.Textbox( |
| label="Step History", |
| lines=8, |
| interactive=False, |
| placeholder="Action history will be logged here..." |
| ) |
|
|
| |
| with gr.Accordion("Advanced: Full Environment State", open=False): |
| state_out = gr.JSON(label="Complete State Dump") |
|
|
| |
| gr.Markdown( |
| """ |
| --- |
| **Tips:** |
| - Start simple: H on qubit 0, then CNOT(0,1) creates a Bell state |
| - Watch the fidelity score — aim for >0.90 on easy tasks |
| - Fewer gates = better efficiency and noise scores |
| - Use REMOVE to undo mistakes |
| - Use PARAM to fine-tune RX/RZ angles |
| """ |
| ) |
|
|
| |
| OUTS = [status_out, circuit_out, scores_out, history_out] |
|
|
| reset_btn.click(fn=do_reset, inputs=[task_dd], outputs=OUTS) |
| step_btn.click(fn=do_step, inputs=[action_dd, gate_dd, qubit_0, qubit_1, parameter], outputs=OUTS) |
| |
| def fetch_state(): |
| s = env.state |
| return s.dict() if hasattr(s, "dict") else s |
| |
| state_btn.click(fn=fetch_state, inputs=[], outputs=[state_out]) |
|
|
| |
| demo.load(fn=update_task_info, inputs=[task_dd], outputs=[task_info]) |
|
|
| return demo |
|
|
|
|
| |
| |
| |
| demo = create_gradio_app() |
|
|
| |
| app = gr.mount_gradio_app(fastapi_app, demo, path="/") |
|
|
| |
| |
| |
| def main(): |
| """Entry point for 'uv run server' command.""" |
| import uvicorn |
| uvicorn.run(app, host="0.0.0.0", port=7860) |
|
|
| if __name__ == "__main__": |
| main() |
|
|
