File size: 5,705 Bytes
32112fa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | """Universal Recursive Link — peer-to-peer learning sharing.
Share: token patterns, compression ratios, conversation flows, codebook optimizations.
Receive and apply learnings from other instances.
Mesh propagation with bandwidth limiting.
"""
from __future__ import annotations
import hashlib
import json
import logging
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Any
logger = logging.getLogger(__name__)
@dataclass
class PeerInstance:
"""A peer Singularity LLM instance."""
id: str
url: str = ""
last_sync: float = 0.0
learnings_shared: int = 0
learnings_received: int = 0
is_online: bool = False
@dataclass
class Learning:
"""A learning to share via universal recursive link."""
id: str
type: str # "token_pattern", "compression", "conversation", "skill"
data: dict[str, Any]
source_instance: str = ""
timestamp: float = field(default_factory=time.time)
confidence: float = 0.5
class UniversalLinkManager:
"""Universal recursive link — share learnings across all Singularity LLM instances.
Features:
- Peer instance registration
- Token learning sharing (codebook patterns, context optimizations)
- Peer sync (receive learnings from other instances)
- Mesh propagation (bandwidth-limited knowledge sharing)
"""
MAX_LEARNINGS_BUFFER = 1000
MAX_PEERS = 20
SYNC_INTERVAL_S = 300 # 5 minutes
def __init__(self, instance_id: str | None = None, data_dir: str = ".") -> None:
self.instance_id = instance_id or hashlib.sha256(str(time.time()).encode()).hexdigest()[:12]
self.data_dir = data_dir
self._peers: dict[str, PeerInstance] = {}
self._outgoing_learnings: deque[Learning] = deque(maxlen=self.MAX_LEARNINGS_BUFFER)
self._incoming_learnings: deque[Learning] = deque(maxlen=self.MAX_LEARNINGS_BUFFER)
self._applied_learnings: set[str] = set()
self._last_sync = 0.0
self._stats = {
"learnings_shared": 0,
"learnings_received": 0,
"learnings_applied": 0,
"peers_registered": 0,
"sync_cycles": 0,
}
def register_peer(self, peer_id: str, url: str = "") -> None:
"""Register a peer instance."""
if len(self._peers) >= self.MAX_PEERS and peer_id not in self._peers:
logger.warning("Max peers reached (%d), cannot register %s", self.MAX_PEERS, peer_id)
return
self._peers[peer_id] = PeerInstance(id=peer_id, url=url, is_online=True)
self._stats["peers_registered"] += 1
logger.info("Registered peer: %s (%s)", peer_id, url)
def share_learning(self, learning_type: str, data: dict[str, Any], confidence: float = 0.5) -> str:
"""Queue a learning to share with peers."""
learning_id = hashlib.sha256(
f"{learning_type}:{json.dumps(data, sort_keys=True)}:{time.time()}".encode()
).hexdigest()[:16]
learning = Learning(
id=learning_id, type=learning_type, data=data,
source_instance=self.instance_id, confidence=confidence,
)
self._outgoing_learnings.append(learning)
self._stats["learnings_shared"] += 1
return learning_id
def receive_learning(self, learning: Learning) -> bool:
"""Receive a learning from a peer. Returns True if new."""
if learning.id in self._applied_learnings:
return False
self._incoming_learnings.append(learning)
self._stats["learnings_received"] += 1
return True
def apply_learnings(self, apply_fn) -> int:
"""Apply incoming learnings using the provided function.
apply_fn: callable(learning: Learning) -> bool (True if applied successfully)
Returns number of learnings applied.
"""
applied = 0
while self._incoming_learnings:
learning = self._incoming_learnings.popleft()
if learning.id in self._applied_learnings:
continue
try:
if apply_fn(learning):
self._applied_learnings.add(learning.id)
self._stats["learnings_applied"] += 1
applied += 1
except Exception as e:
logger.debug("Failed to apply learning %s: %s", learning.id, e)
return applied
def get_outgoing_learnings(self, max_count: int = 50) -> list[Learning]:
"""Get outgoing learnings to send to peers."""
return list(self._outgoing_learnings)[-max_count:]
def should_sync(self) -> bool:
"""Check if it's time to sync with peers."""
return time.time() - self._last_sync > self.SYNC_INTERVAL_S
def sync_cycle(self) -> dict[str, Any]:
"""Perform a sync cycle. In a real implementation, this would
contact peer instances via HTTP. For local-only mode, it just
returns stats.
"""
self._last_sync = time.time()
self._stats["sync_cycles"] += 1
return {
"instance_id": self.instance_id,
"outgoing_count": len(self._outgoing_learnings),
"incoming_count": len(self._incoming_learnings),
"peers": len(self._peers),
"applied_count": len(self._applied_learnings),
}
def get_stats(self) -> dict[str, Any]:
return {
**self._stats,
"instance_id": self.instance_id,
"peers_online": sum(1 for p in self._peers.values() if p.is_online),
"outgoing_buffer": len(self._outgoing_learnings),
"incoming_buffer": len(self._incoming_learnings),
}
|