Spaces:
Runtime error
Runtime error
| import os | |
| import logging | |
| from logos.manifold_state import ManifoldState | |
| logger = logging.getLogger("ConnectorAgent") | |
| class ConnectorAgent: | |
| """ | |
| Protocol 6: Connector Agent | |
| Weaves the "Prime Network" by connecting tokens based on Domain Potentiality. | |
| Uses "Canonical Separation" to link Inner/Prime/Outer domains appropriately. | |
| """ | |
| def __init__(self): | |
| self.manifold = ManifoldState() | |
| def connect_swarm(self): | |
| """Builds edges in the Manifold Graph.""" | |
| graph = self.manifold.state.get("graph", {}) | |
| if not graph: | |
| logger.warning("No graph to connect. Run Tokenizer first.") | |
| return 0 | |
| nodes = graph.get("nodes", []) | |
| if not nodes: | |
| logger.warning("No nodes in graph.") | |
| return 0 | |
| # Assign 3D Geometry (Manifold Folding) | |
| for node in nodes: | |
| if "position" not in node.get("geometry", {}): | |
| node["geometry"]["position"] = self._calculate_3d_position(node) | |
| edges = [] | |
| logger.info(f"Connecting {len(nodes)} nodes in Potentiality Space...") | |
| # O(N^2) connection scan (acceptable for local repo size) | |
| for i in range(len(nodes)): | |
| n1 = nodes[i] | |
| for j in range(i + 1, len(nodes)): | |
| n2 = nodes[j] | |
| weight = self._calculate_affinity(n1, n2) | |
| if weight > 0: | |
| edges.append({ | |
| "source": n1["id"], | |
| "target": n2["id"], | |
| "weight": weight, | |
| "type": "resonance" | |
| }) | |
| graph["edges"] = edges | |
| self.manifold.save() | |
| logger.info(f"Weaved {len(edges)} connections & folded geometry.") | |
| return len(edges) | |
| def _calculate_3d_position(self, node): | |
| """ | |
| Determines (x, y, z) based on Prime Topology. | |
| Z-axis = Domain Depth (Outer=10, Prime=5, Inner=0). | |
| X/Y-axis = Hash Resonance (Pseudo-random scatter). | |
| """ | |
| import math | |
| domain = node["geometry"]["domain"] | |
| h = node["geometry"]["hash"] | |
| # Z-Axis: Hierarchy | |
| z = 0 | |
| if domain == "OUTER_SHELL": z = 10.0 | |
| elif domain == "PRIME_CHANNEL": z = 5.0 | |
| elif domain == "INNER_SHELL": z = 0.0 | |
| # X/Y-Axis: Spiral Distribution based on Hash | |
| # Use simple polar -> cartesian mapping | |
| angle = (h % 360) * (math.pi / 180) | |
| radius = (h % 100) / 10.0 | |
| x = radius * math.cos(angle) | |
| y = radius * math.sin(angle) | |
| return {"x": round(x, 2), "y": round(y, 2), "z": round(z, 2)} | |
| def _calculate_affinity(self, n1, n2): | |
| """ | |
| Determines connection strength (W_Lo/W_Hi canonical logic). | |
| """ | |
| weight = 0.0 | |
| # 1. Physical Affinity (Same Directory) | |
| # Represents "Local Connectivity" (W_Lo) | |
| dir1 = os.path.dirname(n1["path"]) | |
| dir2 = os.path.dirname(n2["path"]) | |
| if dir1 == dir2: | |
| weight += 0.5 | |
| # 2. Domain Hierarchy (Nested Domains) | |
| # Represents "Hyper Connectivity" (W_Hi) via Prime Resonance | |
| d1 = n1["geometry"]["domain"] | |
| d2 = n2["geometry"]["domain"] | |
| domains = {d1, d2} | |
| # Inner <-> Prime: Structure is controlled by Logic (Strong Link) | |
| if "INNER_SHELL" in domains and "PRIME_CHANNEL" in domains: | |
| weight += 0.4 | |
| # Prime <-> Outer: Logic organizes Entropy (Medium Link) | |
| if "PRIME_CHANNEL" in domains and "OUTER_SHELL" in domains: | |
| weight += 0.3 | |
| # Inner <-> Outer: Direct entropy injection (Weak/Chaos Link) | |
| if "INNER_SHELL" in domains and "OUTER_SHELL" in domains: | |
| weight += 0.1 | |
| # Same Domain Resonance (e.g. Logic <-> Logic) | |
| if d1 == d2: | |
| weight += 0.2 | |
| if weight >= 1.0: weight = 0.99 | |
| return round(weight, 2) | |
| if __name__ == "__main__": | |
| # Test Run | |
| agent = ConnectorAgent() | |
| count = agent.connect_swarm() | |
| print(f"Connector finished. Edges: {count}") | |