Decentralized Semantic Optimization (DSO)¶

E = {
    exp_1: {
        "text": "When solving equations, verify by substitution",
        "embedding": [0.123, -0.456, ...],  # 384-dim
        "confidence": 0.87,
        "domain": "math",
        "epoch": 3,
        "usage_count": 42
    },
    exp_2: {...},
    ...
}

# Vanilla GRPO (parameter updates)
advantages = [(reward_i - mean) / std for reward in rewards]
θ = θ + α * ∇J_GRPO(θ, advantages)  # Gradient update

# Active Semantic Optimization (experience updates)
outputs = [π_θ(o_i | q, E) for i in range(G)]  # Inject experiences
advantages = [(reward_i - mean) / std for reward in rewards]
semantic_advantage = LLM.extract_insight(outputs, advantages)  # Natural language
E = E ∪ {semantic_advantage}  # Update experience library
# θ unchanged - model frozen!

# Original embedding (384-dim float32)
embedding = [0.123, -0.456, 0.789, ...]  # 1.5KB

# BitDelta quantization
signs = sign(embedding)  # +1/-1 (1-bit each)
scale = mean(abs(embedding))  # Single float32

# Compressed representation
compressed = {
    "signs": bits(signs),  # 48 bytes (384 bits)
    "scale": scale         # 4 bytes
}
# Total: 52 bytes vs 1536 bytes = 29.5× compression!

class ActiveSemanticOptimizer:
    def __init__(self, model, experience_lib):
        self.model = model  # Frozen base model
        self.E = experience_lib
        self.bitdelta = BitDeltaQuantizer()

    def optimize_step(self, query, ground_truth):
        # 1. Generate rollouts with current experiences
        context = self.E.format_for_prompt(query)
        outputs = [self.model.generate(query, context) 
                   for _ in range(GROUP_SIZE)]

        # 2. Compute rewards
        rewards = [evaluate(o, ground_truth) for o in outputs]

        # 3. Extract semantic advantage
        if std(rewards) > 0:  # Skip homogeneous groups
            advantage = extract_semantic_insight(outputs, rewards)

            # 4. Add to experience library
            exp_id = self.E.add_experience(
                text=advantage,
                confidence=std(rewards),
                domain=classify(query)
            )

            # 5. Compress embedding with BitDelta
            embedding = self.E.embeddings[exp_id]
            signs, scale = self.bitdelta.quantize_delta(
                embedding, 
                baseline=torch.zeros_like(embedding)
            )
            self.E.compressed[exp_id] = (signs, scale)

        # 6. Model parameters unchanged!
        return self.E

# Traditional Federated Learning
for epoch in epochs:
    # Each node
    local_gradients = compute_gradients(local_data)

    # Aggregate (server)
    global_gradients = aggregate(all_local_gradients)

    # Update model
    θ = θ - α * global_gradients

# Decentralized Semantic Optimization (DSO)
for epoch in epochs:
    # Each node
    local_experiences = extract_semantic_advantages(local_interactions)
    compressed_exp = bitdelta_compress(local_experiences)

    # Share to Zoo Network
    publish_to_network(compressed_exp, merkle_proof, signature)

    # Aggregate (decentralized voting)
    global_experiences = byzantine_robust_aggregate(all_node_experiences)

    # Update experience library (no parameter updates!)
    E_global = E_global ∪ global_experiences

def aggregate_community_deltas(
    self,
    community_deltas: Dict[str, Tuple[torch.Tensor, torch.Tensor]],
    weights: Optional[torch.Tensor] = None
) -> Dict[str, Tuple[torch.Tensor, torch.Tensor]]:
    """
    DeltaSoup: Aggregate community improvements with Byzantine-robust averaging.
    """
    aggregated = {}

    for layer_name in self.delta_signs.keys():
        layer_signs = [deltas[layer_name][0] for deltas in community_deltas.values()]
        layer_scales = [deltas[layer_name][1] for deltas in community_deltas.values()]

        if self.config.byzantine_robust:
            # Use median for robustness (resistant to malicious nodes)
            agg_signs = torch.stack(layer_signs).median(dim=0)[0]
            agg_scales = torch.stack(layer_scales).median(dim=0)[0]
        else:
            # Simple averaging
            agg_signs = torch.stack(layer_signs).float().mean(dim=0)
            agg_signs = (agg_signs >= 0).to(torch.int8) * 2 - 1
            agg_scales = torch.stack(layer_scales).mean(dim=0)

        aggregated[layer_name] = (agg_signs, agg_scales)

    return aggregated

class SemanticAggregator:
    def aggregate_experiences(
        self,
        node_experiences: Dict[str, List[Experience]],
        voting_threshold: float = 0.66  # 2/3 majority
    ) -> List[Experience]:
        """
        Aggregate experiences from multiple nodes with quality voting.
        """
        # 1. Collect all unique experiences
        all_experiences = {}
        for node_id, experiences in node_experiences.items():
            for exp in experiences:
                exp_hash = hash(exp.text)
                if exp_hash not in all_experiences:
                    all_experiences[exp_hash] = {
                        'experience': exp,
                        'votes': [],
                        'nodes': []
                    }
                all_experiences[exp_hash]['votes'].append(exp.confidence)
                all_experiences[exp_hash]['nodes'].append(node_id)

        # 2. Byzantine-robust filtering
        accepted_experiences = []
        for exp_hash, data in all_experiences.items():
            # Median confidence across nodes
            median_confidence = torch.tensor(data['votes']).median()

            # Sybil resistance: weight by unique node count
            unique_nodes = len(set(data['nodes']))
            sybil_weight = min(unique_nodes / len(node_experiences), 1.0)

            # Accept if above threshold
            if median_confidence * sybil_weight >= voting_threshold:
                exp = data['experience']
                exp.confidence = median_confidence.item()
                exp.metadata['votes'] = len(data['votes'])
                exp.metadata['nodes'] = unique_nodes
                accepted_experiences.append(exp)

        return accepted_experiences

# src/gym/train/grpo/continuous_learning/dso_local.py
class LocalDSOOptimizer(ActiveSemanticOptimizer):
    """ASO with BitDelta compression for network sharing."""

    def compress_for_network(self) -> Dict[str, Any]:
        """Prepare experiences for network submission."""
        compressed = {}
        for exp_id, exp in self.E.experiences.items():
            # BitDelta quantization
            signs, scale = self.bitdelta.quantize_delta(
                exp.embedding,
                baseline=torch.zeros_like(exp.embedding)
            )
            compressed[exp_id] = {
                'text': exp.text,
                'signs': signs.cpu().numpy().tobytes(),
                'scale': float(scale),
                'confidence': exp.confidence,
                'domain': exp.domain
            }
        return compressed

# src/gym/network/dso_aggregator.py
class NetworkDSOAggregator(SemanticAggregator):
    """Decentralized experience aggregation via Zoo Network."""

    def __init__(self, node_id, rpc_url):
        self.node_id = node_id
        self.web3 = Web3(Web3.HTTPProvider(rpc_url))
        self.contract = self.web3.eth.contract(
            address=EXPERIENCE_REGISTRY_ADDRESS,
            abi=EXPERIENCE_REGISTRY_ABI
        )

    def submit_to_network(self, compressed_experiences):
        """Submit compressed experiences to Zoo Network."""
        for exp_id, exp_data in compressed_experiences.items():
            # Upload to IPFS
            ipfs_cid = ipfs_client.add_json(exp_data)

            # Compute Merkle proof
            merkle_root = compute_merkle_root(exp_data)

            # Sign with node key
            signature = self.sign(merkle_root)

            # Submit to contract
            tx = self.contract.functions.submitExperience(
                merkle_root,
                ipfs_cid,
                signature
            ).transact({'from': self.node_id})

            receipt = self.web3.eth.wait_for_transaction_receipt(tx)
            print(f"Experience {exp_id} submitted: {receipt.transactionHash.hex()}")

    def fetch_global_experiences(self):
        """Fetch accepted experiences from network."""
        accepted_hashes = self.contract.functions.getGlobalExperiences().call()

        global_experiences = []
        for merkle_root in accepted_hashes:
            exp_data = self.contract.functions.experiences(merkle_root).call()

            # Download from IPFS
            ipfs_cid = exp_data[1]  # ipfsCid field
            exp_json = ipfs_client.get_json(ipfs_cid)

            # Verify Merkle proof
            if verify_merkle_proof(exp_json, merkle_root):
                # Decompress BitDelta
                signs = torch.tensor(np.frombuffer(exp_json['signs'], dtype=np.int8))
                scale = torch.tensor([exp_json['scale']])
                embedding = signs.float() * scale

                # Create experience
                exp = Experience(
                    text=exp_json['text'],
                    embedding=embedding,
                    confidence=exp_json['confidence'],
                    domain=exp_json['domain']
                )
                global_experiences.append(exp)

        return global_experiences

# src/gym/train/grpo/continuous_learning/dso_trainer.py
class DecentralizedSemanticOptimizationTrainer(ContinuousLearningGRPOTrainer):
    """Unified trainer with local ASO + network DSO."""

    def __init__(self, *args, network_config=None, **kwargs):
        super().__init__(*args, **kwargs)

        # Local ASO
        self.aso = LocalDSOOptimizer(self.model, self.experience_manager)

        # Network DSO (optional)
        if network_config:
            self.network_dso = NetworkDSOAggregator(
                node_id=network_config['node_id'],
                rpc_url=network_config['rpc_url']
            )
        else:
            self.network_dso = None

    def training_step(self, model, inputs):
        # Local ASO optimization
        self.aso.optimize_step(inputs['query'], inputs['ground_truth'])

        # Periodically sync with network
        if self.global_step % self.args.dso_sync_frequency == 0:
            if self.network_dso:
                # Submit local experiences
                compressed = self.aso.compress_for_network()
                self.network_dso.submit_to_network(compressed)

                # Fetch global experiences
                global_exp = self.network_dso.fetch_global_experiences()

                # Merge with local library
                for exp in global_exp:
                    if exp.confidence >= self.args.dso_acceptance_threshold:
                        self.experience_manager.add_experience(
                            text=exp.text,
                            embedding=exp.embedding,
                            confidence=exp.confidence,
                            domain=exp.domain
                        )

                print(f"DSO sync: {len(global_exp)} new experiences added")

        # No parameter updates!
        return torch.tensor(0.0)

# src/gym/hparams/finetuning_args.py

@dataclass
class DSOArguments:
    """Arguments for Decentralized Semantic Optimization"""

    # Enable DSO
    enable_dso: bool = field(
        default=False,
        metadata={"help": "Enable decentralized semantic optimization"}
    )

    # Network configuration
    node_id: str = field(
        default=None,
        metadata={"help": "Node ID on Zoo Network"}
    )

    rpc_url: str = field(
        default="https://rpc.zoo.network",
        metadata={"help": "Zoo Network RPC endpoint"}
    )

    ipfs_gateway: str = field(
        default="https://ipfs.zoo.network",
        metadata={"help": "IPFS gateway for experience storage"}
    )

    # Synchronization
    dso_sync_frequency: int = field(
        default=100,
        metadata={"help": "Steps between network syncs"}
    )

    dso_acceptance_threshold: float = field(
        default=0.7,
        metadata={"help": "Minimum confidence to accept global experiences"}
    )

    # Byzantine robustness
    byzantine_robust: bool = field(
        default=True,
        metadata={"help": "Use Byzantine-robust aggregation"}
    )

    voting_threshold: float = field(
        default=0.66,
        metadata={"help": "DAO voting threshold (2/3 default)"}
    )