Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/AI_Brain / services/quantum_harness.py |
| Date | Not specified in source |
| Repository | Janady13/AI_Brain |
| Revision | 5cd0bbf744b0aee9445913337e852889d9a8098e |
| Branch | codex/build-online-github-ani-7v4kug |
| Path or archive member | services/quantum_harness.py |
| Original SHA-256 | 0c70f9ae4122632afae143bdf1c98fc22360092a0afc8c4f862cc640ceabc4aa |
| Git blob SHA | 9c1962151e2790592ce2bf23510ebfe7b4fa6f0c |
| Reading-copy SHA-256 | 0c70f9ae4122632afae143bdf1c98fc22360092a0afc8c4f862cc640ceabc4aa |
Recovered source transcript
Download complete source text ↓155 lines · 5.1 KB
"""Utilities for executing large-scale quantum stress harnesses locally.
The harness coordinates repeated simulator runs so operators can approximate
continuous quantum readiness without needing to connect to live hardware. It
relies on the marketing ANI's quantum plans and the upgraded mini simulator,
providing deterministic results by default with an optional Monte Carlo mode.
"""
from __future__ import annotations
import math
import time
from dataclasses import asdict, dataclass
from typing import List
from services.seeder_api.intelligence import (
MiniQuantumSimulator,
QuantumStressTestPlan,
)
@dataclass(slots=True)
class QuantumNodeObservation:
"""Snapshot of a single node's ability to host agent populations."""
node_index: int
average_measurement: float
agents_supported: int
nodes_required_for_target: int
compression_ratio: float
continuous: bool
class QuantumDeploymentHarness:
"""Executes repeated quantum stress simulations across virtual nodes."""
def __init__(
self,
simulator: MiniQuantumSimulator | None = None,
) -> None:
self._simulator = simulator or MiniQuantumSimulator()
def _run_observation(
self,
plan: QuantumStressTestPlan,
node_index: int,
target_agents: int,
shots: int | None,
monte_carlo: bool,
) -> QuantumNodeObservation:
measurements: List[float] = []
compression: List[float] = []
agent_support = []
for stress in plan.stress_tests:
measurement = self._simulator.run(
stress.gate_sequence,
shots=shots,
monte_carlo=monte_carlo,
)
measurements.append(measurement)
compression.append(stress.compression_ratio)
agent_support.append(max(1, int(round(target_agents * measurement))))
if measurements:
average_measurement = sum(measurements) / len(measurements)
else:
average_measurement = 0.0
if agent_support:
agents_supported = max(1, sum(agent_support) // len(agent_support))
else:
agents_supported = 1
nodes_required = max(1, math.ceil(target_agents / max(1, agents_supported)))
compression_ratio = sum(compression) / len(compression) if compression else 0.0
continuous = average_measurement >= 0.5 and plan.continuous_operation
return QuantumNodeObservation(
node_index=node_index,
average_measurement=round(average_measurement, 4),
agents_supported=agents_supported,
nodes_required_for_target=nodes_required,
compression_ratio=round(compression_ratio, 4),
continuous=continuous,
)
def stress(
self,
plan: QuantumStressTestPlan,
*,
nodes: int = 8,
target_agents: int = 1_000_000,
iterations: int = 3,
delay_seconds: float = 0.0,
shots: int | None = None,
monte_carlo: bool = False,
) -> dict:
"""Run repeated quantum observations and summarise the rollout readiness."""
history: List[List[QuantumNodeObservation]] = []
for iteration in range(max(1, iterations)):
observations: List[QuantumNodeObservation] = []
for node_index in range(max(1, nodes)):
observations.append(
self._run_observation(
plan,
node_index=node_index,
target_agents=target_agents,
shots=shots,
monte_carlo=monte_carlo,
)
)
if delay_seconds:
time.sleep(delay_seconds)
history.append(observations)
flattened: List[QuantumNodeObservation] = [
obs for iteration in history for obs in iteration
]
avg_agents = sum(obs.agents_supported for obs in flattened)
total_samples = len(flattened)
average_agents = avg_agents // max(1, total_samples)
peak_nodes = (
max(obs.nodes_required_for_target for obs in flattened)
if flattened
else 1
)
sustained = all(obs.continuous for obs in flattened) if flattened else False
return {
"plan": {
"simulator": plan.simulator,
"provider": plan.provider,
"agents_per_node": plan.agents_per_node,
"nodes_for_one_million": plan.nodes_for_one_million,
"compression_index": plan.compression_index,
"shots": plan.shots,
"continuous_operation": plan.continuous_operation,
},
"iterations": [
[asdict(observation) for observation in observations]
for observations in history
],
"summary": {
"average_agents_supported": average_agents,
"peak_nodes_required": peak_nodes,
"sustained_operation": sustained,
},
}
__all__ = ["QuantumDeploymentHarness", "QuantumNodeObservation"]