Upload aiq_bridge.py with huggingface_hub

aiq_bridge.py

@@ -0,0 +1,63 @@
+"""
+NVIDIA AI-Q Blueprint Embedded Bridge for Fiber-MoE Autonomous World Model
+Integrates `/aiq-research` and `/aiq-deploy` directly into the agent substrate.
+"""
+
+import os
+import sys
+import subprocess
+import json
+from pathlib import Path
+
+AIQ_SKILL_DIR = Path(r"C:\Users\gemin\.gemini\config\skills\aiq-research")
+AIQ_SCRIPT = AIQ_SKILL_DIR / "scripts" / "aiq.py"
+DEFAULT_AIQ_SERVER_URL = os.environ.get("AIQ_SERVER_URL", "http://localhost:8000")
+
+class AIQBridge:
+    def __init__(self, server_url: str = DEFAULT_AIQ_SERVER_URL):
+        self.server_url = server_url
+        self.script_path = str(AIQ_SCRIPT)
+
+    def check_health(self) -> dict:
+        """Ping AI-Q server /health endpoint."""
+        cmd = [sys.executable, self.script_path, "health"]
+        env = os.environ.copy()
+        env["AIQ_SERVER_URL"] = self.server_url
+        try:
+            res = subprocess.run(cmd, capture_output=True, text=True, env=env, timeout=10)
+            if res.returncode == 0:
+                try:
+                    return {"status": "healthy", "details": json.loads(res.stdout.strip())}
+                except Exception:
+                    return {"status": "healthy", "raw": res.stdout.strip()}
+            return {"status": "unreachable", "error": res.stderr.strip() or res.stdout.strip()}
+        except Exception as e:
+            return {"status": "error", "error": str(e)}
+
+    def execute_chat(self, query: str) -> dict:
+        """Route shallow query or dispatch async deep research job."""
+        cmd = [sys.executable, self.script_path, "chat", query]
+        env = os.environ.copy()
+        env["AIQ_SERVER_URL"] = self.server_url
+        res = subprocess.run(cmd, capture_output=True, text=True, env=env)
+        try:
+            return json.loads(res.stdout.strip())
+        except Exception:
+            return {"raw_output": res.stdout.strip(), "stderr": res.stderr.strip()}
+
+    def poll_research(self, job_id: str) -> dict:
+        """Poll deep research job until completion."""
+        cmd = [sys.executable, self.script_path, "research_poll", job_id]
+        env = os.environ.copy()
+        env["AIQ_SERVER_URL"] = self.server_url
+        res = subprocess.run(cmd, capture_output=True, text=True, env=env)
+        try:
+            return json.loads(res.stdout.strip())
+        except Exception:
+            return {"raw_output": res.stdout.strip()}
+
+if __name__ == "__main__":
+    bridge = AIQBridge()
+    print("Testing AIQBridge health check:")
+    health = bridge.check_health()
+    print(json.dumps(health, indent=2))