Upload folder using huggingface_hub

inference.py

@@ -8,22 +8,17 @@ MANDATORY
     HF_TOKEN       Your Hugging Face / API key.
     IMAGE_NAME     The name of the local image to use for the environment if you are using
                    from_docker_image() method.
-
 - Defaults are set only for API_BASE_URL and MODEL_NAME
     (and should reflect your active inference setup):
     API_BASE_URL = os.getenv("API_BASE_URL", "https://router.huggingface.co/v1")
     MODEL_NAME = os.getenv("MODEL_NAME", "Qwen/Qwen2.5-72B-Instruct")
-
 - The inference script must be named `inference.py` and placed in the root directory of the project
 - Participants must use OpenAI Client for all LLM calls using above variables
-
 STDOUT FORMAT
 - The script must emit exactly three line types to stdout, in this order:
-
     [START] task=<task_name> env=<benchmark> model=<model_name>
     [STEP]  step=<n> action=<action_str> reward=<0.00> done=<true|false> error=<msg|null>
     [END]   success=<true|false> steps=<n> score=<score> rewards=<r1,r2,...,rn>
-
   Rules:
     - One [START] line at episode begin.
     - One [STEP] line per step, immediately after env.step() returns.
@@ -37,7 +32,6 @@ STDOUT FORMAT
 
 import asyncio
 import json
-import math
 import os
 import textwrap
 from typing import Any, Dict, List, Optional
@@ -66,13 +60,13 @@ SUCCESS_FIDELITY = 0.80  # minimum fidelity for "success"
 
 # Task configurations
 TASKS = [
-       {"id": "easy", "name": "bell_state", "max_steps": 15},
-        {"id": "medium", "name": "ghz_state", "max_steps": 15},
-    {"id": "hard", "name": "unitary_approx", "max_steps": 15},
-       {"id": "efficient", "name": "imperfect_efficient", "max_steps": 15},
-       {"id": "noisy", "name": "noise_dominant", "max_steps": 15},
-       {"id": "budget", "name": "budget_optimization", "max_steps": 15},
-    #  {"id": "approx", "name": "approximate_target", "max_steps": 15},
+    {"id": "easy", "name": "bell_state", "max_steps": 15},
+    {"id": "medium", "name": "ghz_state", "max_steps": 15},
+    #{"id": "hard", "name": "unitary_approx", "max_steps": 15},
+    {"id": "efficient", "name": "imperfect_efficient", "max_steps": 15},
+    {"id": "noisy", "name": "noise_dominant", "max_steps": 15},
+    {"id": "budget", "name": "budget_optimization", "max_steps": 15},
+    {"id": "approx", "name": "approximate_target", "max_steps": 15},
 ]
 
 
@@ -105,93 +99,31 @@ def log_end(success: bool, steps: int, score: float, rewards: List[float]) -> No
 # System prompt for the LLM
 # ---------------------------------------------------------------------------
 
-SYSTEM_PROMPT = textwrap.dedent("""
-You are an expert quantum circuit optimization agent.
-
-You are NOT a pattern generator.
-You MUST behave as a decision-making optimizer that improves the circuit step-by-step.
-
-Your goal:
-Maximize fidelity to the target while minimizing circuit depth and noise impact.
-
----
-
-AVAILABLE ACTIONS (respond with valid JSON only):
-
-1. ADD:
-{"action_type": "ADD", "gate": "<H|X|CNOT|RX|RZ>", "qubits": [<int>, ...], "parameter": <float or null>}
-
-2. REMOVE:
-{"action_type": "REMOVE", "qubits": []}
-
-3. SWAP:
-{"action_type": "SWAP", "qubits": [q1, q2]}
-
-4. PARAM:
-{"action_type": "PARAM", "qubits": [], "parameter": <float>}
-
-5. STOP:
-{"action_type": "STOP", "qubits": []}
-
----
-
-CRITICAL DECISION PROCESS (YOU MUST FOLLOW THIS INTERNALLY):
-
-Step 1: Understand the target
-- What gates are required? (e.g., CNOT, RX, RZ)
-- Are parameters needed? (angles like π/3, π/4)
-
-Step 2: Analyze current circuit
-- What is already correct?
-- What is missing?
-- What is wrong or unnecessary?
-
-Step 3: Evaluate improvement
-- Which action will MOST increase fidelity?
-- Will adding this gate help or hurt under noise?
-- Is circuit becoming too deep?
-
-Step 4: Choose action strategically
-- Prefer actions that move toward target structure
-- Avoid random or repeated gates
-- Use parametric gates (RX, RZ) when precise tuning is required
-
----
-
-STRICT RULES:
-
-- DO NOT repeat the same useless pattern (e.g., H → CNOT blindly).
-- DO NOT use STOP unless:
-  - Fidelity is already high (>0.95), OR
-  - No further improvement is possible.
-- For unitary/parametric tasks:
-  - You MUST use RX/RZ gates with meaningful parameters.
-  - Avoid using H unless clearly useful.
-- Prefer minimal circuits (depth matters due to noise).
-- If a step reduces fidelity, consider REMOVE.
-
----
-
-ANTI-PATTERN WARNING (VERY IMPORTANT):
-
-If you find yourself repeating:
-H → CNOT → STOP
-
-You are WRONG.
-
-Re-evaluate the task and choose a better action.
-
----
-
-OUTPUT FORMAT:
-
-Return ONLY the best action as JSON:
-{
-  "action_type": "...",
-  "gate": "...",
-  "qubits": [...],
-  "parameter": ...
-}
+SYSTEM_PROMPT = textwrap.dedent("""\
+    You are an expert quantum circuit designer. You are interacting with an RL environment
+    that lets you build quantum circuits step by step.
+    AVAILABLE ACTIONS (respond with valid JSON only):
+    1. ADD a gate:
+       {"action_type": "ADD", "gate": "<H|X|CNOT|RX|RZ>", "qubits": [<int>, ...], "parameter": <float or null>}
+       - H, X: single-qubit gates, qubits = [q]
+       - CNOT: two-qubit gate, qubits = [control, target]
+       - RX, RZ: parametric single-qubit gates, qubits = [q], parameter = angle in radians
+    2. REMOVE the last gate:
+       {"action_type": "REMOVE", "qubits": []}
+       Note: Yields +0.05 bonus if depth decreases with no fidelity loss. Yields -0.05 penalty if fidelity drops.
+    3. SWAP two qubits:
+       {"action_type": "SWAP", "qubits": [q1, q2]}
+       Note: Yields -0.02 penalty per SWAP since it represents real hardware routing overhead.
+    4. PARAM -- tune the last parametric gate:
+       {"action_type": "PARAM", "qubits": [], "parameter": <float>}
+    5. STOP -- end the episode:
+       {"action_type": "STOP", "qubits": []}
+    RULES:
+    - Respond with ONLY the JSON object, nothing else.
+    - Maximize fidelity to the target state while minimizing circuit depth.
+    - Consider REMOVE to revert a bad step to avoid depth penalties, or SWAP to map qubits.
+    - Respect any hardware connectivity constraints described in the observation.
+    - All episodes end at 15 steps max.
 """)
 
 
@@ -199,41 +131,24 @@ Return ONLY the best action as JSON:
 # LLM interaction
 # ---------------------------------------------------------------------------
 
-def build_user_prompt(obs_data, step, history):
-    history_block = "\n".join(history[-3:]) if history else "None"
-
-    return textwrap.dedent(f"""
-    You are optimizing a quantum circuit.
-
-    TARGET:
-    {obs_data.get('target_description')}
-
-    CURRENT STATE:
-    - Fidelity: {obs_data.get('fidelity'):.4f}
-    - Score: {obs_data.get('score'):.4f}
-    - Depth: {obs_data.get('depth')}
-    - Steps remaining: {obs_data.get('max_steps') - obs_data.get('steps_taken')}
-
-    CURRENT CIRCUIT:
-    {json.dumps(obs_data.get('circuit_gates'))}
-
-    RECENT HISTORY:
-    {history_block}
-
-    THINK BEFORE ACTING:
-
-    1. What gates are REQUIRED for the target?
-    2. What is MISSING in current circuit?
-    3. Will next action INCREASE fidelity?
-
-    CRITICAL:
-    - Avoid repeating useless patterns (H spam)
-    - Use RY/RZ for parametric/unitary tasks
-    - Do NOT STOP unless fidelity > 0.9
-
-    Now choose the BEST next action.
-
-    Return ONLY JSON.
+def build_user_prompt(obs_data: Dict[str, Any], step: int, history: List[str]) -> str:
+    """Build a user prompt from the current observation."""
+    history_block = "\n".join(history[-5:]) if history else "None"
+    return textwrap.dedent(f"""\
+        Step: {step}
+        Task: {obs_data.get('target_description', 'unknown')}
+        Qubits: {obs_data.get('num_qubits', 2)}
+        Current fidelity: {obs_data.get('fidelity', 0.0):.4f}
+        Current depth: {obs_data.get('depth', 0)}
+        Gate count: {obs_data.get('gate_count', 0)}
+        Noise estimate: {obs_data.get('noise_estimate', 0.0):.4f}
+        Score: {obs_data.get('score', 0.0):.4f}
+        Valid actions: {obs_data.get('valid_actions', [])}
+        Steps remaining: {obs_data.get('max_steps', 20) - obs_data.get('steps_taken', 0)}
+        Current circuit gates: {json.dumps(obs_data.get('circuit_gates', []))}
+        Recent history:
+        {history_block}
+        Choose your next action (JSON only):
     """)
 
 
@@ -274,82 +189,17 @@ def get_model_action(
         return action_dict
     except Exception as exc:
         print(f"[DEBUG] Model request/parse failed: {exc}", flush=True)
-        # Fallback: try CNOT gate on qubits 0,1
-        return {"action_type": "ADD", "gate": "CNOT", "qubits": [0, 1]}
-
-def validate_action(action_dict, obs_data, prev_reward, step, is_hard=False):
-    # ❌ prevent oscillation
-    if prev_reward < 0 and action_dict.get("action_type") == "ADD":
-        return False
-
-    # Prevent useless PARAM actions without an existing parameterized gate
-    if action_dict.get("action_type") == "PARAM":
-        if not any(g["gate"] in ["RX", "RZ"] for g in obs_data.get("circuit_gates", [])):
-            return False
-
-    # Prevent repeated same gate
-    last_gate = obs_data.get("circuit_gates")[-1]["gate"] if obs_data.get("circuit_gates") else None
-    if action_dict.get("gate") == last_gate:
-        return False
-
-    # Lock good structure after decent fidelity is achieved
-    if obs_data.get("fidelity", 0.0) > 0.7 and action_dict.get("gate") == "CNOT":
-        return False
-
-    # Block early STOP
-    if step < 4 and action_dict.get("action_type") == "STOP":
-        return False
-
-    # Block useless H spam in later steps
-    if step > 2 and action_dict.get("gate") == "H":
-        return False
-
-    # For hard tasks, block H gates
-    if is_hard and action_dict.get("gate") == "H":
-        return False
-
-    return True
-
-
-def enforce_ghz_structure(action_dict, obs_data, step):
-    if "ghz" in obs_data.get("target_description", "").lower():
-        if step == 1:
-            return {"action_type": "ADD", "gate": "H", "qubits": [0]}
-    return action_dict
-
-
-def parse_parameter(param):
-    if param is None:
-        return None
-
-    if isinstance(param, (int, float)):
-        return float(param)
-
-    if isinstance(param, str):
-        param = param.replace(" ", "").lower()
-
-        # handle pi expressions
-        if "pi" in param:
-            try:
-                param = param.replace("pi", f"{math.pi}")
-                return float(eval(param))
-            except:
-                return None
-
-        try:
-            return float(param)
-        except:
-            return None
-
-    return None
+        # Fallback: try a simple H gate on qubit 0
+        return {"action_type": "ADD", "gate": "H", "qubits": [0]}
 
 
 def dict_to_action(action_dict: Dict[str, Any]) -> QuantumAction:
+    """Convert a dict to a QuantumAction."""
     return QuantumAction(
         action_type=action_dict.get("action_type", "STOP"),
         gate=action_dict.get("gate"),
         qubits=action_dict.get("qubits", []),
-        parameter=parse_parameter(action_dict.get("parameter")),
+        parameter=action_dict.get("parameter"),
     )
 
 
@@ -372,9 +222,6 @@ async def run_task(
     steps_taken = 0
     score = 0.0
     success = False
-    best_fidelity = 0.0
-    no_improve_steps = 0
-    prev_reward = 0.0
 
     log_start(task=task_name, env=BENCHMARK, model=MODEL_NAME)
 
@@ -399,22 +246,8 @@ async def run_task(
             if result.done:
                 break
 
-            # Stop before chaos if repeated negative reward after step 8
-            if step > 8 and prev_reward < 0:
-                action_dict = {"action_type": "STOP", "qubits": []}
-            elif no_improve_steps >= 2:
-                action_dict = {"action_type": "STOP", "qubits": []}
-            else:
-                # Get action from LLM with retry and validation
-                is_hard = task_id in ["hard", "approx"]
-                for _ in range(3):
-                    action_dict = get_model_action(client, obs_data, step, history)
-                    action_dict = enforce_ghz_structure(action_dict, obs_data, step)
-                    # Parameter nudging for RX/RZ
-                    if action_dict.get("gate") in ["RX", "RZ"] and action_dict.get("parameter") is None:
-                        action_dict["parameter"] = 1.0
-                    if validate_action(action_dict, obs_data, prev_reward, step, is_hard):
-                        break
+            # Get action from LLM
+            action_dict = get_model_action(client, obs_data, step, history)
             action_str = json.dumps(action_dict, separators=(",", ":"))
 
             # Convert to QuantumAction and step
@@ -431,15 +264,6 @@ async def run_task(
 
             log_step(step=step, action=action_str, reward=reward, done=done, error=error)
 
-            # Update improvement tracking
-            if obs.fidelity > best_fidelity:
-                best_fidelity = obs.fidelity
-                no_improve_steps = 0
-            else:
-                no_improve_steps += 1
-
-            prev_reward = reward
-
             # Update obs_data for next prompt
             obs_data = {
                 "fidelity": obs.fidelity,
@@ -497,4 +321,4 @@ async def main() -> None:
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    asyncio.run(main())