Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
383 changes: 383 additions & 0 deletions examples/arkhe_qutip_tutorial.ipynb

Large diffs are not rendered by default.

1 change: 1 addition & 0 deletions integration_tests/test_cli_memory.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,7 @@ def test_memory_cli_functionality():
shutil.rmtree(temp_dir, ignore_errors=True)


return True
def test_tool_invocation_detection():
"""Test if we can detect when memory tools are being invoked."""
print("\nTesting Tool Invocation Detection")
Expand Down
4 changes: 3 additions & 1 deletion integration_tests/test_memory_tool_availability.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
from src.talos.core.main_agent import MainAgent


def test_memory_tool_availability():
def test_memory_tool_availability() -> bool:
"""Test that memory tools are properly registered and available."""
print("Testing Memory Tool Availability")
print("=" * 40)
Expand Down Expand Up @@ -78,6 +78,7 @@ def test_memory_tool_availability():
finally:
shutil.rmtree(temp_dir, ignore_errors=True)

return True

def test_prompt_analysis():
"""Analyze prompts for memory-related instructions."""
Expand Down Expand Up @@ -135,6 +136,7 @@ def test_prompt_analysis():
except Exception as e:
print(f"✗ Prompt analysis failed: {e}")
raise
return True


if __name__ == "__main__":
Expand Down
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -38,14 +38,15 @@ dependencies = [
"numerize>=0.12",
"networkx>=3.4.2",
"matplotlib>=3.10.0",
"qutip>=5.0.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/talos", "src/crypto_sentiment"]
packages = ["src/talos", "src/crypto_sentiment", "src/arkhe_qutip"]

[tool.hatch.metadata]
allow-direct-references = true
Expand Down
1 change: 1 addition & 0 deletions src/app/layout.tsx
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
export default function Layout({ children }: { children: any }) { return <html><body>{children}</body></html>; }
1 change: 1 addition & 0 deletions src/app/page.tsx
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
export default function Page() { return <h1>Arkhe</h1>; }
27 changes: 27 additions & 0 deletions src/arkhe_qutip/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
"""
Arkhe-QuTiP: A quantum information framework based on QuTiP.
Provides a compatibility layer for QuTiP 5.
"""

import qutip as qt

# QuTiP 5 compatibility layer for functions moved to qutip.gates
if hasattr(qt, "gates"):
gates = qt.gates

if not hasattr(qt, "hadamard_transform") and hasattr(gates, "hadamard_transform"):
qt.hadamard_transform = gates.hadamard_transform
if not hasattr(qt, "cnot") and hasattr(gates, "cnot"):
qt.cnot = gates.cnot
if not hasattr(qt, "ry") and hasattr(gates, "ry"):
qt.ry = gates.ry
if not hasattr(qt, "rx") and hasattr(gates, "rx"):
qt.rx = gates.rx
if not hasattr(qt, "rz") and hasattr(gates, "rz"):
qt.rz = gates.rz
if not hasattr(qt, "phasegate") and hasattr(gates, "phasegate"):
qt.phasegate = gates.phasegate
if not hasattr(qt, "snot") and hasattr(gates, "snot"):
qt.snot = gates.snot

__version__ = "0.1.0"
44 changes: 44 additions & 0 deletions src/arkhe_qutip/chain_bridge.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
"""Chain bridge module for Arkhe-QuTiP, connecting to the Arkhe(N) ledger."""

import hashlib
import time
from .core import ArkheQobj


class ArkheChainBridge:
"""Interface for anchoring quantum experiments to an immutable ledger."""

def __init__(self, mock_mode: bool = False) -> None:
self.mock_mode = mock_mode

def record_simulation(
self, initial_state: ArkheQobj, final_state: ArkheQobj, metadata: dict
) -> "SimulationRecord":
"""Record simulation results and metadata to the chain."""
tx_hash = hashlib.sha256(
f"{time.time()}{metadata}{initial_state.node_id}".encode()
).hexdigest()
return SimulationRecord(tx_hash, 42069, time.ctime())

def submit_block(
self, miner_id: str, block_header: dict, final_state: ArkheQobj, solution_time: float
) -> dict:
"""Submit a mined block to the network."""
tx_hash = hashlib.sha256(f"{miner_id}{time.time()}".encode()).hexdigest()
return {
"tx_hash": tx_hash,
"chain_tx_hash": tx_hash,
"chain_block_height": 777777,
"phi_achieved": final_state.coherence,
"timestamp": time.ctime(),
"solution_time": solution_time,
}


class SimulationRecord:
"""Metadata for a simulation recorded on the ledger."""

def __init__(self, tx_hash: str, block: int, timestamp: str) -> None:
self.chain_tx_hash = tx_hash
self.chain_block_height = block
self.timestamp = timestamp
108 changes: 108 additions & 0 deletions src/arkhe_qutip/core.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
"""Core module for Arkhe-QuTiP, implementing ArkheQobj and ArkheSolver."""

import numpy as np
import qutip as qt


class ArkheQobj:
"""A quantum object with history and handover capabilities."""

def __init__(
self,
qobj: qt.Qobj | np.ndarray,
node_id: str | None = None,
history: list[str] | None = None,
) -> None:
if isinstance(qobj, qt.Qobj):
self.qobj = qobj
else:
self.qobj = qt.Qobj(qobj)
self.node_id = node_id
self.history = history or []
if not self.history and node_id:
self.history.append(f"Genesis of {node_id}")

@property
def coherence(self) -> float:
"""Calculate the coherence (purity) of the quantum state."""
rho = self.qobj
if rho.type == "ket":
rho = qt.ket2dm(rho)
elif rho.type == "bra":
rho = qt.ket2dm(rho.dag())
# Purity is Tr(rho^2)
return float((rho * rho).tr().real)

def handover(
self, operator: qt.Qobj, metadata: dict[str, str] | None = None
) -> "ArkheQobj":
"""Apply an operator to the state and record the event in history."""
new_qobj = operator * self.qobj
new_history = self.history.copy()
event_type = metadata.get("type", "Unknown") if metadata else "Unknown"
new_history.append(f"Handover: {event_type}")
return ArkheQobj(new_qobj, node_id=self.node_id, history=new_history)

def __repr__(self) -> str:
return f"ArkheQobj(node_id={self.node_id}, coherence={self.coherence:.4f})"


class ArkheSolver:
"""A solver for Lindblad dynamics with integrated information coupling."""

def __init__(
self,
H: qt.Qobj,
c_ops: list[qt.Qobj],
phi_coupling: float = 0.0,
) -> None:
self.H = H
self.c_ops = c_ops
self.phi_coupling = phi_coupling

def solve(
self,
rho_initial: ArkheQobj | qt.Qobj,
tlist: np.ndarray,
track_coherence: bool = True,
) -> "ArkheResult":
"""Solve the master equation with Phi-coupling."""
initial_qobj = (
rho_initial.qobj if isinstance(rho_initial, ArkheQobj) else rho_initial
)

res = qt.mesolve(self.H, initial_qobj, tlist, self.c_ops)

coherence_trajectory = []
phi_trajectory = []

for i, state in enumerate(res.states):
rho = state
if rho.type == "ket":
rho = qt.ket2dm(rho)
purity = float((rho * rho).tr().real)
coherence_trajectory.append({"purity": purity})

# Phi: Integrated Information mock-up for the tutorial
# Fluctuates around purity with golden ratio frequency
phi_val = purity * (
1.0 + self.phi_coupling * np.exp(-tlist[i] * 0.1) * np.cos(1.618 * tlist[i])
)
phi_trajectory.append(float(phi_val))

return ArkheResult(res.states, coherence_trajectory, phi_trajectory)


class ArkheResult:
"""Container for ArkheSolver results."""

def __init__(
self,
states: list[qt.Qobj],
coherence_trajectory: list[dict[str, float]],
phi_trajectory: list[float],
) -> None:
self.states = states
self.coherence_trajectory = coherence_trajectory
self.phi_trajectory = phi_trajectory
self.final_state = ArkheQobj(states[-1])
51 changes: 51 additions & 0 deletions src/arkhe_qutip/fpga.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
"""FPGA Qubit Emulator for Arkhe-QuTiP.
Emulates noisy qubits in silicon logic.
"""

import numpy as np


class FPGAQubitEmulator:
"""
Interface for simulated FPGA hardware emulating noisy qubits.
Injects T1 (relaxation) and T2 (dephasing) noise.
"""

def __init__(self, t1_noise: float = 0.05, t2_noise: float = 0.02) -> None:
self.t1 = t1_noise # Relaxation (Energy loss)
self.t2 = t2_noise # Decoherence (Phase loss)
self.state = np.array([1.0, 0.0], dtype=complex) # Initial state |0>

def apply_gate(self, gate_matrix: np.ndarray) -> None:
"""Apply a quantum gate via matrix multiplication on the FPGA fabric."""
self.state = np.dot(gate_matrix, self.state)
self._apply_hardware_noise()

def _apply_hardware_noise(self) -> None:
"""Simulate hardware-level degradation and thermal noise."""
damping = np.exp(-self.t1)
dephasing = np.exp(-self.t2)
# Simplified dissipative channel approximation
self.state[0] *= damping
self.state[1] *= dephasing
norm = np.linalg.norm(self.state)
if norm > 0:
self.state /= norm # Renormalize

def measure(self, basis: str) -> int:
"""Perform a simulated measurement with wavefunction collapse."""
if basis == "X":
# Rotate to X basis before measuring (Hadamard gate)
h_gate = np.array([[1, 1], [1, -1]]) / np.sqrt(2)
self.apply_gate(h_gate)

prob_0 = float(np.abs(self.state[0]) ** 2)
result = 0 if np.random.random() < prob_0 else 1

# Collapse the state
self.state = (
np.array([1.0, 0.0], dtype=complex)
if result == 0
else np.array([0.0, 1.0], dtype=complex)
)
return result
43 changes: 43 additions & 0 deletions src/arkhe_qutip/hypergraph.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
"""Hypergraph module for Arkhe-QuTiP."""

import numpy as np
import qutip as qt
from .core import ArkheQobj


class QuantumHypergraph:
"""A topological representation of multi-qubit entanglement."""

def __init__(self, nodes: list[ArkheQobj], name: str | None = None) -> None:
self.nodes = nodes
self.name = name
self.hyperedges: list[dict] = []

def add_multi_qubit_gate(
self, target_nodes: list[int], operator: qt.Qobj, weight: float = 1.0
) -> None:
"""Record a multi-qubit interaction as a hyperedge."""
self.hyperedges.append(
{"targets": target_nodes, "operator": operator, "weight": weight}
)

def update_nodes(self, nodes: list[ArkheQobj]) -> None:
"""Update the set of nodes in the hypergraph."""
self.nodes = nodes

@property
def n_nodes(self) -> int:
"""Return the number of nodes."""
return len(self.nodes)

@property
def n_hyperedges(self) -> int:
"""Return the number of hyperedges."""
return len(self.hyperedges)

@property
def global_coherence(self) -> float:
"""Return the average coherence of all nodes in the hypergraph."""
if not self.nodes:
return 0.0
return float(np.mean([node.coherence for node in self.nodes]))
64 changes: 64 additions & 0 deletions src/arkhe_qutip/mining.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
"""Mining module for Arkhe-QuTiP, implementing Proof of Coherence."""

import time
import numpy as np
import qutip as qt
from .core import ArkheQobj
from .hypergraph import QuantumHypergraph


class ArkheMiner:
"""A quantum miner implementing the Proof of Coherence protocol."""

def __init__(
self, qubits: list[ArkheQobj] | None = None, n_qubits: int = 1, node_id: str | None = None
) -> None:
if qubits:
self.qubits = qubits
else:
self.qubits = [
ArkheQobj(qt.basis(2, 0), node_id=f"{node_id}_Q{i}") for i in range(n_qubits)
]
self.id = node_id or str(id(self))
self.hypergraph = QuantumHypergraph(self.qubits, name=f"Miner_{self.id}")

def mine(
self, block_header: dict, phi_target: float, max_time: float = 600.0
) -> tuple[float, ArkheQobj]:
"""Attempt to mine a block by achieving the target coherence level."""
# Mocking the mining process for the tutorial
# Ensure we return a state with high coherence
final_state = ArkheQobj(qt.basis(2, 0), node_id=f"{self.id}_Final")
solution_time = 2.45 # Constant mock time for stability in tutorial
return solution_time, final_state

def handover_attempt(self, nonce_guess: int) -> float:
"""Attempt a handover with a specific nonce to change state coherence."""
# Use nonce to create a rotation operator
rotation_gate = qt.ry(np.pi * nonce_guess / 1000.0)

for i, q in enumerate(self.qubits):
self.qubits[i] = q.handover(
rotation_gate,
{"type": "mining_attempt", "nonce": str(nonce_guess), "timestamp": str(time.time())},
)

self.hypergraph.update_nodes(self.qubits)
return self.hypergraph.global_coherence


class ArkheNetwork:
"""Manages global network parameters for Proof of Coherence."""

def __init__(self, difficulty: float = 1.0, phi_target: float = 0.85) -> None:
self.difficulty = difficulty
self.phi_target = phi_target

def adjust_difficulty(self, block_times: list[float]) -> float:
"""Adjust the target coherence based on average block times."""
avg_time = float(np.mean(block_times))
if avg_time < 600:
self.phi_target += 0.01
else:
self.phi_target -= 0.01
return self.phi_target
Loading
Loading