Skip to content

Multi-Agent Systems

Multi-agent coordination, field coupling, and collective memory.

MultiAgentConfig

MultiAgentConfig dataclass

Configuration for multi-agent system.

Source code in src/rotalabs_ftms/agents/multi_agent.py 
18
19
20
21
22
23
24
25
26
27
28
@dataclass
class MultiAgentConfig:
    """Configuration for multi-agent system."""
    num_agents: int = 4
    coupling_config: CouplingConfig = None
    field_config: FieldConfig = None
    collective_pool_size: int = 10000
    sync_interval: float = 1.0  # seconds
    enable_specialization: bool = True
    consensus_threshold: float = 0.6
    max_workers: int = 4  # for parallel processing

MultiAgentSystem

MultiAgentSystem

Orchestrates multiple FTCS agents with collective intelligence.

Source code in src/rotalabs_ftms/agents/multi_agent.py 
 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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
class MultiAgentSystem:
    """Orchestrates multiple FTCS agents with collective intelligence."""

    def __init__(self, config: MultiAgentConfig = None):
        self.config = config or MultiAgentConfig()
        self.logger = logging.getLogger(__name__)

        # Initialize components
        self.agents: Dict[str, FTCSAgent] = {}
        self.field_coupler = FieldCoupler(self.config.coupling_config)
        self.collective_pool = CollectiveMemoryPool(
            max_memories=self.config.collective_pool_size,
            consensus_threshold=self.config.consensus_threshold
        )

        # Threading for async operations
        self.executor = ThreadPoolExecutor(max_workers=self.config.max_workers)
        self.sync_thread = None
        self.running = False
        self.lock = threading.RLock()

        # Agent specializations
        self.agent_specializations: Dict[str, str] = {}

        # Performance tracking
        self.performance_stats = {
            "total_queries": 0,
            "collective_hits": 0,
            "sync_cycles": 0,
            "avg_response_time": 0
        }

        # Initialize agents
        self._initialize_agents()

    def _initialize_agents(self):
        """Create and register initial agents."""

        for i in range(self.config.num_agents):
            agent_id = f"agent_{i}"

            # Create agent with unique characteristics
            agent_config = AgentConfig(
                memory_field_shape=(100, 100),  # Smaller field for multi-agent
                diffusion_rate=0.005 * (1 + 0.1 * np.random.randn()),
                temperature=0.08 * (1 + 0.1 * np.random.randn()),
                importance_decay_rate=0.1,
                use_proper_embeddings=False,  # Faster for demos
                embedding_dim=100  # Match field dimension to avoid mismatch
            )

            agent = FTCSAgent(
                agent_id=agent_id,
                config=agent_config
            )

            self.add_agent(agent_id, agent)

    def add_agent(self, agent_id: str, agent: FTCSAgent, 
                  position: Optional[tuple] = None,
                  specialization: Optional[str] = None):
        """Add an agent to the system."""
        with self.lock:
            self.agents[agent_id] = agent

            # Register with field coupler
            self.field_coupler.register_agent(agent_id, agent.memory_field, position)

            # Set specialization if provided
            if specialization:
                self.agent_specializations[agent_id] = specialization

            self.logger.info(f"Added agent {agent_id} to system")

    def remove_agent(self, agent_id: str):
        """Remove an agent from the system."""
        with self.lock:
            if agent_id in self.agents:
                del self.agents[agent_id]
                self.field_coupler.unregister_agent(agent_id)

                if agent_id in self.agent_specializations:
                    del self.agent_specializations[agent_id]

                self.logger.info(f"Removed agent {agent_id} from system")

    def start(self):
        """Start the multi-agent system."""
        with self.lock:
            if self.running:
                return

            self.running = True
            self.sync_thread = threading.Thread(target=self._sync_loop, daemon=True)
            self.sync_thread.start()

            self.logger.info("Multi-agent system started")

    def stop(self):
        """Stop the multi-agent system."""
        with self.lock:
            self.running = False

        if self.sync_thread:
            self.sync_thread.join(timeout=5)

        self.executor.shutdown(wait=True)
        self.logger.info("Multi-agent system stopped")

    def _sync_loop(self):
        """Background synchronization loop."""
        while self.running:
            try:
                # Perform field coupling
                self.field_coupler.couple_fields(dt=0.1)

                # Share high-importance memories
                self._share_important_memories()

                # Update specializations if enabled
                if self.config.enable_specialization:
                    self._update_specializations()

                self.performance_stats["sync_cycles"] += 1

            except Exception as e:
                self.logger.error(f"Error in sync loop: {e}")

            time.sleep(self.config.sync_interval)

    def _share_important_memories(self):
        """Share high-importance memories to collective pool."""
        with self.lock:
            for agent_id, agent in self.agents.items():
                # Get recent high-importance memories
                recent_memories = agent.get_recent_memories(
                    time_window=self.config.sync_interval * 2,
                    importance_threshold=0.7
                )

                # Propose to collective pool
                for memory in recent_memories:
                    self.collective_pool.propose_memory(
                        agent_id=agent_id,
                        content=memory['content'],
                        embedding=memory['embedding'],
                        importance=memory['importance']
                    )

    def _update_specializations(self):
        """Update agent specializations based on their memory patterns."""
        # Analyze each agent's memory distribution
        specialization_scores = {}

        with self.lock:
            for agent_id, agent in self.agents.items():
                # Get memory statistics
                stats = agent.get_memory_statistics()

                # Simple specialization detection based on memory clusters
                if 'dominant_topics' in stats:
                    specialization_scores[agent_id] = stats['dominant_topics']

        # Update specializations
        # This is simplified - in practice would use more sophisticated clustering
        for agent_id, topics in specialization_scores.items():
            if topics:
                self.agent_specializations[agent_id] = topics[0]  # Most dominant topic

    def collective_query(self, query: str, context: Optional[Dict] = None) -> Dict[str, Any]:
        """Process a query using collective intelligence."""
        start_time = time.time()

        with self.lock:
            self.performance_stats["total_queries"] += 1

            # Get query embedding (simplified - would use real embeddings)
            query_embedding = self._get_embedding(query)

            # First, check collective pool
            collective_memories = self.collective_pool.retrieve_memories(
                query_embedding, 
                k=5,
                consensus_only=True
            )

            if collective_memories:
                self.performance_stats["collective_hits"] += 1

            # Query all agents in parallel
            futures = {}
            for agent_id, agent in self.agents.items():
                # Route to specialized agents if applicable
                if agent_id in self.agent_specializations:
                    specialization = self.agent_specializations[agent_id]
                    # Simple keyword matching - would be more sophisticated
                    if specialization.lower() not in query.lower():
                        continue

                future = self.executor.submit(
                    self._query_agent,
                    agent, 
                    query,
                    context,
                    collective_memories
                )
                futures[agent_id] = future

            # Collect responses
            responses = {}
            for agent_id, future in futures.items():
                try:
                    responses[agent_id] = future.result(timeout=5)
                except Exception as e:
                    self.logger.error(f"Error querying agent {agent_id}: {e}")

            # Build consensus response
            consensus_response = self._build_consensus(responses, collective_memories)

            # Update performance stats
            response_time = time.time() - start_time
            self.performance_stats["avg_response_time"] = (
                (self.performance_stats["avg_response_time"] * 
                 (self.performance_stats["total_queries"] - 1) + response_time) /
                self.performance_stats["total_queries"]
            )

            return {
                "response": consensus_response,
                "agent_responses": responses,
                "collective_memories": [(m_id, m.content) for m_id, m in collective_memories],
                "response_time": response_time,
                "participating_agents": list(responses.keys())
            }

    def _query_agent(self, agent: FTCSAgent, query: str, context: Optional[Dict],
                    collective_memories: List[tuple]) -> Dict[str, Any]:
        """Query a single agent."""
        # Inject collective memories into agent's context
        enriched_context = context or {}
        enriched_context['collective_memories'] = [
            {"content": m.content, "importance": m.importance}
            for _, m in collective_memories
        ]

        # Process query
        try:
            if hasattr(agent, 'process_query'):
                response = agent.process_query(query)
            else:
                # Fallback for basic agent
                response = agent.query(query, enriched_context)

            return {
                "response": response,
                "confidence": 0.8,  # Would be computed based on memory quality
                "sources": len(collective_memories)
            }
        except Exception as e:
            self.logger.error(f"Error in agent query: {e}")
            return {
                "response": "",
                "confidence": 0.0,
                "error": str(e)
            }

    def _build_consensus(self, agent_responses: Dict[str, Dict],
                        collective_memories: List[tuple]) -> str:
        """Build consensus response from multiple agents."""
        if not agent_responses:
            return "No agents available to process query."

        # Filter valid responses
        valid_responses = [
            (agent_id, resp) for agent_id, resp in agent_responses.items()
            if resp.get("confidence", 0) > 0.5 and "error" not in resp
        ]

        if not valid_responses:
            return "Unable to generate confident response."

        # Simple consensus: combine responses with confidence weighting
        # In practice, would use more sophisticated consensus building
        if len(valid_responses) == 1:
            return valid_responses[0][1]["response"]

        # Combine multiple responses
        response_parts = []
        total_confidence = 0

        for agent_id, resp in valid_responses:
            confidence = resp.get("confidence", 0.5)
            response_parts.append(f"[{agent_id}]: {resp['response']}")
            total_confidence += confidence

        # Build combined response
        consensus = "Based on collective intelligence:\n\n"
        consensus += "\n\n".join(response_parts)

        if collective_memories:
            consensus += f"\n\nSupported by {len(collective_memories)} collective memories."

        return consensus

    def _get_embedding(self, text: str) -> jnp.ndarray:
        """Get embedding for text (simplified)."""
        # In practice, would use real embedding model
        # For now, return random embedding
        return jnp.array(np.random.randn(768))

    def demonstrate_collective_problem_solving(self, problem: str) -> Dict[str, Any]:
        """Demonstrate collective problem-solving capabilities."""
        self.logger.info(f"Collective problem solving: {problem}")

        # Phase 1: Individual exploration
        exploration_results = {}
        with self.lock:
            for agent_id, agent in self.agents.items():
                # Each agent explores the problem independently
                agent.store_memory(
                    f"Exploring problem: {problem}",
                    {"problem": problem, "phase": "exploration"}
                )
                exploration_results[agent_id] = f"Agent {agent_id} exploring: {problem}"

        # Let fields evolve and couple
        time.sleep(self.config.sync_interval * 2)

        # Phase 2: Share insights through coupling
        self._share_important_memories()

        # Phase 3: Collective synthesis
        collective_response = self.collective_query(
            f"Based on our collective exploration, how should we solve: {problem}",
            {"phase": "synthesis"}
        )

        # Analyze emergent behavior
        coupling_stats = self.field_coupler.get_coupling_statistics()
        pool_stats = self.collective_pool.get_statistics()

        return {
            "problem": problem,
            "exploration_results": exploration_results,
            "collective_solution": collective_response,
            "emergent_metrics": {
                "field_convergence": 1.0 - coupling_stats.get("field_divergence", 1.0),
                "consensus_rate": pool_stats.get("consensus_rate", 0),
                "collective_memories": pool_stats.get("total_memories", 0),
                "agent_collaboration": pool_stats.get("avg_contributors", 1)
            }
        }

    def get_system_statistics(self) -> Dict[str, Any]:
        """Get comprehensive system statistics."""
        with self.lock:
            coupling_stats = self.field_coupler.get_coupling_statistics()
            pool_stats = self.collective_pool.get_statistics()

            # Agent statistics
            agent_stats = {}
            for agent_id, agent in self.agents.items():
                contribution = self.collective_pool.get_agent_contribution(agent_id)
                agent_stats[agent_id] = {
                    "specialization": self.agent_specializations.get(agent_id, "generalist"),
                    "contribution": contribution,
                    "memory_count": agent.get_memory_count()
                }

            return {
                "system": {
                    "num_agents": len(self.agents),
                    "topology": coupling_stats.get("topology", "unknown"),
                    "sync_cycles": self.performance_stats["sync_cycles"],
                    "running": self.running
                },
                "performance": self.performance_stats.copy(),
                "coupling": coupling_stats,
                "collective_pool": pool_stats,
                "agents": agent_stats
            }

add_agent(agent_id: str, agent: FTCSAgent, position: Optional[tuple] = None, specialization: Optional[str] = None)

Add an agent to the system.

Source code in src/rotalabs_ftms/agents/multi_agent.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def add_agent(self, agent_id: str, agent: FTCSAgent, 
              position: Optional[tuple] = None,
              specialization: Optional[str] = None):
    """Add an agent to the system."""
    with self.lock:
        self.agents[agent_id] = agent

        # Register with field coupler
        self.field_coupler.register_agent(agent_id, agent.memory_field, position)

        # Set specialization if provided
        if specialization:
            self.agent_specializations[agent_id] = specialization

        self.logger.info(f"Added agent {agent_id} to system")

remove_agent(agent_id: str)

Remove an agent from the system.

Source code in src/rotalabs_ftms/agents/multi_agent.py 
105
106
107
108
109
110
111
112
113
114
115
def remove_agent(self, agent_id: str):
    """Remove an agent from the system."""
    with self.lock:
        if agent_id in self.agents:
            del self.agents[agent_id]
            self.field_coupler.unregister_agent(agent_id)

            if agent_id in self.agent_specializations:
                del self.agent_specializations[agent_id]

            self.logger.info(f"Removed agent {agent_id} from system")

start()

Start the multi-agent system.

Source code in src/rotalabs_ftms/agents/multi_agent.py 
117
118
119
120
121
122
123
124
125
126
127
def start(self):
    """Start the multi-agent system."""
    with self.lock:
        if self.running:
            return

        self.running = True
        self.sync_thread = threading.Thread(target=self._sync_loop, daemon=True)
        self.sync_thread.start()

        self.logger.info("Multi-agent system started")

stop()

Stop the multi-agent system.

Source code in src/rotalabs_ftms/agents/multi_agent.py 
129
130
131
132
133
134
135
136
137
138
def stop(self):
    """Stop the multi-agent system."""
    with self.lock:
        self.running = False

    if self.sync_thread:
        self.sync_thread.join(timeout=5)

    self.executor.shutdown(wait=True)
    self.logger.info("Multi-agent system stopped")

collective_query(query: str, context: Optional[Dict] = None) -> Dict[str, Any]

Process a query using collective intelligence.

Source code in src/rotalabs_ftms/agents/multi_agent.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def collective_query(self, query: str, context: Optional[Dict] = None) -> Dict[str, Any]:
    """Process a query using collective intelligence."""
    start_time = time.time()

    with self.lock:
        self.performance_stats["total_queries"] += 1

        # Get query embedding (simplified - would use real embeddings)
        query_embedding = self._get_embedding(query)

        # First, check collective pool
        collective_memories = self.collective_pool.retrieve_memories(
            query_embedding, 
            k=5,
            consensus_only=True
        )

        if collective_memories:
            self.performance_stats["collective_hits"] += 1

        # Query all agents in parallel
        futures = {}
        for agent_id, agent in self.agents.items():
            # Route to specialized agents if applicable
            if agent_id in self.agent_specializations:
                specialization = self.agent_specializations[agent_id]
                # Simple keyword matching - would be more sophisticated
                if specialization.lower() not in query.lower():
                    continue

            future = self.executor.submit(
                self._query_agent,
                agent, 
                query,
                context,
                collective_memories
            )
            futures[agent_id] = future

        # Collect responses
        responses = {}
        for agent_id, future in futures.items():
            try:
                responses[agent_id] = future.result(timeout=5)
            except Exception as e:
                self.logger.error(f"Error querying agent {agent_id}: {e}")

        # Build consensus response
        consensus_response = self._build_consensus(responses, collective_memories)

        # Update performance stats
        response_time = time.time() - start_time
        self.performance_stats["avg_response_time"] = (
            (self.performance_stats["avg_response_time"] * 
             (self.performance_stats["total_queries"] - 1) + response_time) /
            self.performance_stats["total_queries"]
        )

        return {
            "response": consensus_response,
            "agent_responses": responses,
            "collective_memories": [(m_id, m.content) for m_id, m in collective_memories],
            "response_time": response_time,
            "participating_agents": list(responses.keys())
        }

demonstrate_collective_problem_solving(problem: str) -> Dict[str, Any]

Demonstrate collective problem-solving capabilities.

Source code in src/rotalabs_ftms/agents/multi_agent.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
def demonstrate_collective_problem_solving(self, problem: str) -> Dict[str, Any]:
    """Demonstrate collective problem-solving capabilities."""
    self.logger.info(f"Collective problem solving: {problem}")

    # Phase 1: Individual exploration
    exploration_results = {}
    with self.lock:
        for agent_id, agent in self.agents.items():
            # Each agent explores the problem independently
            agent.store_memory(
                f"Exploring problem: {problem}",
                {"problem": problem, "phase": "exploration"}
            )
            exploration_results[agent_id] = f"Agent {agent_id} exploring: {problem}"

    # Let fields evolve and couple
    time.sleep(self.config.sync_interval * 2)

    # Phase 2: Share insights through coupling
    self._share_important_memories()

    # Phase 3: Collective synthesis
    collective_response = self.collective_query(
        f"Based on our collective exploration, how should we solve: {problem}",
        {"phase": "synthesis"}
    )

    # Analyze emergent behavior
    coupling_stats = self.field_coupler.get_coupling_statistics()
    pool_stats = self.collective_pool.get_statistics()

    return {
        "problem": problem,
        "exploration_results": exploration_results,
        "collective_solution": collective_response,
        "emergent_metrics": {
            "field_convergence": 1.0 - coupling_stats.get("field_divergence", 1.0),
            "consensus_rate": pool_stats.get("consensus_rate", 0),
            "collective_memories": pool_stats.get("total_memories", 0),
            "agent_collaboration": pool_stats.get("avg_contributors", 1)
        }
    }

get_system_statistics() -> Dict[str, Any]

Get comprehensive system statistics.

Source code in src/rotalabs_ftms/agents/multi_agent.py 
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
def get_system_statistics(self) -> Dict[str, Any]:
    """Get comprehensive system statistics."""
    with self.lock:
        coupling_stats = self.field_coupler.get_coupling_statistics()
        pool_stats = self.collective_pool.get_statistics()

        # Agent statistics
        agent_stats = {}
        for agent_id, agent in self.agents.items():
            contribution = self.collective_pool.get_agent_contribution(agent_id)
            agent_stats[agent_id] = {
                "specialization": self.agent_specializations.get(agent_id, "generalist"),
                "contribution": contribution,
                "memory_count": agent.get_memory_count()
            }

        return {
            "system": {
                "num_agents": len(self.agents),
                "topology": coupling_stats.get("topology", "unknown"),
                "sync_cycles": self.performance_stats["sync_cycles"],
                "running": self.running
            },
            "performance": self.performance_stats.copy(),
            "coupling": coupling_stats,
            "collective_pool": pool_stats,
            "agents": agent_stats
        }

Field Coupling

CouplingTopology

CouplingTopology

Bases: Enum

Types of agent coupling topologies.

Source code in src/rotalabs_ftms/agents/coupling.py 
15
16
17
18
19
20
class CouplingTopology(Enum):
    """Types of agent coupling topologies."""
    NEAREST_NEIGHBOR = "nearest_neighbor"
    SMALL_WORLD = "small_world"
    FULLY_CONNECTED = "fully_connected"
    ADAPTIVE = "adaptive"

CouplingConfig

CouplingConfig dataclass

Configuration for field coupling.

Source code in src/rotalabs_ftms/agents/coupling.py 
23
24
25
26
27
28
29
30
31
32
@dataclass
class CouplingConfig:
    """Configuration for field coupling."""
    topology: CouplingTopology = CouplingTopology.FULLY_CONNECTED
    coupling_strength: float = 0.1
    sync_interval: float = 1.0  # seconds
    importance_threshold: float = 0.5
    max_coupling_distance: int = 2  # for nearest neighbor
    small_world_probability: float = 0.1  # for small world
    adaptive_threshold: float = 0.7  # similarity threshold for adaptive

FieldCoupler

FieldCoupler

Manages field coupling between multiple agents.

Source code in src/rotalabs_ftms/agents/coupling.py 
 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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
class FieldCoupler:
    """Manages field coupling between multiple agents."""

    def __init__(self, config: CouplingConfig = None):
        self.config = config or CouplingConfig()
        self.logger = logging.getLogger(__name__)

        # Agent registry
        self.agents: Dict[str, MemoryField] = {}
        self.agent_positions: Dict[str, Tuple[int, int]] = {}  # For spatial topologies
        self.coupling_matrix: Optional[jnp.ndarray] = None

        # Coupling functions
        self._couple_fields = jit(self._couple_fields_impl)
        self._compute_coupling_term = jit(self._compute_coupling_term_impl)

    def register_agent(self, agent_id: str, field: MemoryField, 
                      position: Optional[Tuple[int, int]] = None):
        """Register an agent for field coupling."""
        self.agents[agent_id] = field
        if position:
            self.agent_positions[agent_id] = position

        # Rebuild coupling matrix
        self._build_coupling_matrix()

    def unregister_agent(self, agent_id: str):
        """Remove an agent from coupling."""
        if agent_id in self.agents:
            del self.agents[agent_id]
            if agent_id in self.agent_positions:
                del self.agent_positions[agent_id]
            self._build_coupling_matrix()

    def _build_coupling_matrix(self):
        """Build coupling strength matrix based on topology."""
        n_agents = len(self.agents)
        if n_agents == 0:
            self.coupling_matrix = None
            return

        agent_ids = list(self.agents.keys())
        matrix = np.zeros((n_agents, n_agents))

        if self.config.topology == CouplingTopology.FULLY_CONNECTED:
            # All agents coupled with equal strength
            matrix = np.ones((n_agents, n_agents)) * self.config.coupling_strength
            np.fill_diagonal(matrix, 0)  # No self-coupling

        elif self.config.topology == CouplingTopology.NEAREST_NEIGHBOR:
            # Couple based on spatial distance
            for i, id1 in enumerate(agent_ids):
                for j, id2 in enumerate(agent_ids):
                    if i != j and id1 in self.agent_positions and id2 in self.agent_positions:
                        pos1 = self.agent_positions[id1]
                        pos2 = self.agent_positions[id2]
                        distance = np.sqrt((pos1[0] - pos2[0])**2 + (pos1[1] - pos2[1])**2)
                        if distance <= self.config.max_coupling_distance:
                            matrix[i, j] = self.config.coupling_strength / (1 + distance)

        elif self.config.topology == CouplingTopology.SMALL_WORLD:
            # Start with nearest neighbor, add random long-range connections
            self._build_coupling_matrix()  # Start with nearest neighbor
            matrix = self.coupling_matrix.copy()

            # Add random long-range connections
            rng = np.random.default_rng(42)
            for i in range(n_agents):
                for j in range(i + 1, n_agents):
                    if matrix[i, j] == 0 and rng.random() < self.config.small_world_probability:
                        strength = self.config.coupling_strength * 0.5  # Weaker long-range
                        matrix[i, j] = strength
                        matrix[j, i] = strength

        elif self.config.topology == CouplingTopology.ADAPTIVE:
            # Coupling strength based on field similarity
            # Start with weak uniform coupling
            matrix = np.ones((n_agents, n_agents)) * self.config.coupling_strength * 0.1
            np.fill_diagonal(matrix, 0)

        self.coupling_matrix = jnp.array(matrix)

    def couple_fields(self, dt: float = 0.1) -> Dict[str, jnp.ndarray]:
        """Perform one step of field coupling between all agents."""
        if len(self.agents) < 2 or self.coupling_matrix is None:
            return {}

        agent_ids = list(self.agents.keys())
        fields = [self.agents[aid].field for aid in agent_ids]

        # Stack all fields
        stacked_fields = jnp.stack(fields)

        # Compute coupling updates
        coupled_fields = self._couple_fields(
            stacked_fields, 
            self.coupling_matrix,
            dt
        )

        # Update agent fields
        updates = {}
        for i, agent_id in enumerate(agent_ids):
            self.agents[agent_id].field = coupled_fields[i]
            updates[agent_id] = coupled_fields[i]

        # Update adaptive topology if needed
        if self.config.topology == CouplingTopology.ADAPTIVE:
            self._update_adaptive_coupling(stacked_fields)

        return updates

    def _couple_fields_impl(self, fields: jnp.ndarray, coupling_matrix: jnp.ndarray, 
                           dt: float) -> jnp.ndarray:
        """JAX implementation of field coupling."""
        n_agents = fields.shape[0]

        # Compute coupling terms for each agent
        coupling_terms = vmap(
            lambda i: self._compute_coupling_term_impl(
                fields[i], fields, coupling_matrix[i]
            )
        )(jnp.arange(n_agents))

        # Apply coupling
        return fields + dt * coupling_terms

    def _compute_coupling_term_impl(self, field_i: jnp.ndarray, all_fields: jnp.ndarray,
                                   coupling_strengths: jnp.ndarray) -> jnp.ndarray:
        """Compute coupling term for a single agent."""
        # Reshape coupling strengths for broadcasting
        strengths = coupling_strengths[:, None, None]

        # Compute differences weighted by coupling strength
        differences = all_fields - field_i[None, :, :]
        weighted_diff = differences * strengths

        # Sum over all agents
        return jnp.sum(weighted_diff, axis=0)

    def _update_adaptive_coupling(self, fields: jnp.ndarray):
        """Update coupling matrix based on field similarity."""
        n_agents = fields.shape[0]
        new_matrix = np.zeros((n_agents, n_agents))

        # Compute pairwise similarities
        for i in range(n_agents):
            for j in range(i + 1, n_agents):
                # Use normalized dot product as similarity
                field_i = fields[i].flatten()
                field_j = fields[j].flatten()

                norm_i = jnp.linalg.norm(field_i)
                norm_j = jnp.linalg.norm(field_j)

                if norm_i > 0 and norm_j > 0:
                    similarity = jnp.dot(field_i, field_j) / (norm_i * norm_j)

                    if similarity > self.config.adaptive_threshold:
                        strength = self.config.coupling_strength * float(similarity)
                        new_matrix[i, j] = strength
                        new_matrix[j, i] = strength

        # Smooth transition: blend old and new matrices
        alpha = 0.1  # adaptation rate
        self.coupling_matrix = (1 - alpha) * self.coupling_matrix + alpha * jnp.array(new_matrix)

    def get_collective_field(self) -> Optional[jnp.ndarray]:
        """Compute the collective field (average of all agent fields)."""
        if not self.agents:
            return None

        fields = [agent.field for agent in self.agents.values()]
        return jnp.mean(jnp.stack(fields), axis=0)

    def get_coupling_statistics(self) -> Dict:
        """Get statistics about current coupling state."""
        if not self.agents or self.coupling_matrix is None:
            return {}

        collective_field = self.get_collective_field()

        # Compute field divergence (how different agents are)
        fields = jnp.stack([agent.field for agent in self.agents.values()])
        divergences = jnp.std(fields, axis=0).mean()

        # Coupling strength statistics
        non_zero_coupling = self.coupling_matrix[self.coupling_matrix > 0]

        return {
            "num_agents": len(self.agents),
            "topology": self.config.topology.value,
            "avg_coupling_strength": float(non_zero_coupling.mean()) if len(non_zero_coupling) > 0 else 0,
            "max_coupling_strength": float(non_zero_coupling.max()) if len(non_zero_coupling) > 0 else 0,
            "field_divergence": float(divergences),
            "collective_energy": float(jnp.sum(collective_field)),
            "connectivity": float(len(non_zero_coupling)) / (len(self.agents) * (len(self.agents) - 1))
        }

    def broadcast_memory(self, source_agent: str, memory: jnp.ndarray, 
                        importance: float) -> Set[str]:
        """Broadcast a high-importance memory from one agent to others."""
        if source_agent not in self.agents:
            return set()

        if importance < self.config.importance_threshold:
            return set()

        # Get coupling strengths for source agent
        source_idx = list(self.agents.keys()).index(source_agent)
        coupling_strengths = self.coupling_matrix[source_idx]

        # Inject memory into coupled agents
        injected_agents = set()
        for i, (agent_id, agent_field) in enumerate(self.agents.items()):
            if i != source_idx and coupling_strengths[i] > 0:
                # Scale injection by coupling strength
                injection_strength = coupling_strengths[i] * importance

                # Inject memory into agent's field
                # This is a simplified injection - in practice, would use MemoryField.inject_memory
                agent_field.field = agent_field.field.at[:].add(
                    memory * injection_strength
                )
                injected_agents.add(agent_id)

        return injected_agents

register_agent(agent_id: str, field: MemoryField, position: Optional[Tuple[int, int]] = None)

Register an agent for field coupling.

Source code in src/rotalabs_ftms/agents/coupling.py 
51
52
53
54
55
56
57
58
59
def register_agent(self, agent_id: str, field: MemoryField, 
                  position: Optional[Tuple[int, int]] = None):
    """Register an agent for field coupling."""
    self.agents[agent_id] = field
    if position:
        self.agent_positions[agent_id] = position

    # Rebuild coupling matrix
    self._build_coupling_matrix()

unregister_agent(agent_id: str)

Remove an agent from coupling.

Source code in src/rotalabs_ftms/agents/coupling.py 
61
62
63
64
65
66
67
def unregister_agent(self, agent_id: str):
    """Remove an agent from coupling."""
    if agent_id in self.agents:
        del self.agents[agent_id]
        if agent_id in self.agent_positions:
            del self.agent_positions[agent_id]
        self._build_coupling_matrix()

couple_fields(dt: float = 0.1) -> Dict[str, jnp.ndarray]

Perform one step of field coupling between all agents.

Source code in src/rotalabs_ftms/agents/coupling.py 
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
def couple_fields(self, dt: float = 0.1) -> Dict[str, jnp.ndarray]:
    """Perform one step of field coupling between all agents."""
    if len(self.agents) < 2 or self.coupling_matrix is None:
        return {}

    agent_ids = list(self.agents.keys())
    fields = [self.agents[aid].field for aid in agent_ids]

    # Stack all fields
    stacked_fields = jnp.stack(fields)

    # Compute coupling updates
    coupled_fields = self._couple_fields(
        stacked_fields, 
        self.coupling_matrix,
        dt
    )

    # Update agent fields
    updates = {}
    for i, agent_id in enumerate(agent_ids):
        self.agents[agent_id].field = coupled_fields[i]
        updates[agent_id] = coupled_fields[i]

    # Update adaptive topology if needed
    if self.config.topology == CouplingTopology.ADAPTIVE:
        self._update_adaptive_coupling(stacked_fields)

    return updates

get_collective_field() -> Optional[jnp.ndarray]

Compute the collective field (average of all agent fields).

Source code in src/rotalabs_ftms/agents/coupling.py 
202
203
204
205
206
207
208
def get_collective_field(self) -> Optional[jnp.ndarray]:
    """Compute the collective field (average of all agent fields)."""
    if not self.agents:
        return None

    fields = [agent.field for agent in self.agents.values()]
    return jnp.mean(jnp.stack(fields), axis=0)

get_coupling_statistics() -> Dict

Get statistics about current coupling state.

Source code in src/rotalabs_ftms/agents/coupling.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def get_coupling_statistics(self) -> Dict:
    """Get statistics about current coupling state."""
    if not self.agents or self.coupling_matrix is None:
        return {}

    collective_field = self.get_collective_field()

    # Compute field divergence (how different agents are)
    fields = jnp.stack([agent.field for agent in self.agents.values()])
    divergences = jnp.std(fields, axis=0).mean()

    # Coupling strength statistics
    non_zero_coupling = self.coupling_matrix[self.coupling_matrix > 0]

    return {
        "num_agents": len(self.agents),
        "topology": self.config.topology.value,
        "avg_coupling_strength": float(non_zero_coupling.mean()) if len(non_zero_coupling) > 0 else 0,
        "max_coupling_strength": float(non_zero_coupling.max()) if len(non_zero_coupling) > 0 else 0,
        "field_divergence": float(divergences),
        "collective_energy": float(jnp.sum(collective_field)),
        "connectivity": float(len(non_zero_coupling)) / (len(self.agents) * (len(self.agents) - 1))
    }

broadcast_memory(source_agent: str, memory: jnp.ndarray, importance: float) -> Set[str]

Broadcast a high-importance memory from one agent to others.

Source code in src/rotalabs_ftms/agents/coupling.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def broadcast_memory(self, source_agent: str, memory: jnp.ndarray, 
                    importance: float) -> Set[str]:
    """Broadcast a high-importance memory from one agent to others."""
    if source_agent not in self.agents:
        return set()

    if importance < self.config.importance_threshold:
        return set()

    # Get coupling strengths for source agent
    source_idx = list(self.agents.keys()).index(source_agent)
    coupling_strengths = self.coupling_matrix[source_idx]

    # Inject memory into coupled agents
    injected_agents = set()
    for i, (agent_id, agent_field) in enumerate(self.agents.items()):
        if i != source_idx and coupling_strengths[i] > 0:
            # Scale injection by coupling strength
            injection_strength = coupling_strengths[i] * importance

            # Inject memory into agent's field
            # This is a simplified injection - in practice, would use MemoryField.inject_memory
            agent_field.field = agent_field.field.at[:].add(
                memory * injection_strength
            )
            injected_agents.add(agent_id)

    return injected_agents

Collective Memory

CollectiveMemoryPool

CollectiveMemoryPool

Manages shared memories across multiple agents.

Source code in src/rotalabs_ftms/agents/collective.py 
 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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
class CollectiveMemoryPool:
    """Manages shared memories across multiple agents."""

    def __init__(self, max_memories: int = 10000, consensus_threshold: float = 0.6):
        self.max_memories = max_memories
        self.consensus_threshold = consensus_threshold
        self.logger = logging.getLogger(__name__)

        # Memory storage
        self.memories: Dict[str, CollectiveMemory] = {}
        self.memory_index: Dict[str, List[str]] = defaultdict(list)  # agent_id -> memory_ids

        # Thread safety
        self.lock = threading.RLock()

        # Statistics
        self.stats = {
            "total_memories": 0,
            "consensus_memories": 0,
            "avg_contributors": 0,
            "memory_conflicts": 0
        }

    def propose_memory(self, agent_id: str, content: str, embedding: jnp.ndarray,
                      importance: float, memory_id: Optional[str] = None) -> str:
        """Propose a memory to the collective pool."""
        with self.lock:
            # Generate memory ID if not provided
            if memory_id is None:
                memory_id = f"{agent_id}_{datetime.now().timestamp()}"

            # Check if similar memory exists
            similar_memory_id = self._find_similar_memory(embedding)

            if similar_memory_id:
                # Update existing memory
                existing = self.memories[similar_memory_id]
                existing.add_contributor(agent_id, importance)

                # Update importance as weighted average
                n = len(existing.contributors)
                existing.importance = ((n - 1) * existing.importance + importance) / n

                # Update embedding as weighted average
                existing.embedding = ((n - 1) * existing.embedding + embedding) / n

                # Track in agent's index
                if agent_id not in self.memory_index:
                    self.memory_index[agent_id] = []
                if similar_memory_id not in self.memory_index[agent_id]:
                    self.memory_index[agent_id].append(similar_memory_id)

                self._update_stats()
                return similar_memory_id
            else:
                # Create new memory
                new_memory = CollectiveMemory(
                    content=content,
                    embedding=embedding,
                    importance=importance
                )
                new_memory.add_contributor(agent_id, importance)

                self.memories[memory_id] = new_memory
                self.memory_index[agent_id].append(memory_id)

                # Evict if necessary
                if len(self.memories) > self.max_memories:
                    self._evict_memory()

                self._update_stats()
                return memory_id

    def _find_similar_memory(self, embedding: jnp.ndarray, threshold: float = 0.9) -> Optional[str]:
        """Find memory with similar embedding."""
        best_similarity = 0.0
        best_id = None

        for mem_id, memory in self.memories.items():
            similarity = self._compute_similarity(embedding, memory.embedding)
            if similarity > threshold and similarity > best_similarity:
                best_similarity = similarity
                best_id = mem_id

        return best_id

    def _compute_similarity(self, emb1: jnp.ndarray, emb2: jnp.ndarray) -> float:
        """Compute cosine similarity between embeddings."""
        norm1 = jnp.linalg.norm(emb1)
        norm2 = jnp.linalg.norm(emb2)

        if norm1 == 0 or norm2 == 0:
            return 0.0

        return float(jnp.dot(emb1, emb2) / (norm1 * norm2))

    def retrieve_memories(self, query_embedding: jnp.ndarray, k: int = 5,
                         agent_id: Optional[str] = None,
                         consensus_only: bool = False) -> List[Tuple[str, CollectiveMemory]]:
        """Retrieve k most relevant memories."""
        with self.lock:
            # Filter memories
            candidates = []
            for mem_id, memory in self.memories.items():
                # Apply consensus filter if requested
                if consensus_only and memory.consensus_score < self.consensus_threshold:
                    continue

                # Apply agent filter if requested
                if agent_id and agent_id not in memory.contributors:
                    continue

                # Compute relevance score
                similarity = self._compute_similarity(query_embedding, memory.embedding)
                relevance = similarity * memory.importance * (1 + memory.consensus_score)

                candidates.append((mem_id, memory, relevance))

            # Sort by relevance and return top k
            candidates.sort(key=lambda x: x[2], reverse=True)

            # Update access counts
            results = []
            for mem_id, memory, _ in candidates[:k]:
                memory.access()
                results.append((mem_id, memory))

            return results

    def get_agent_contribution(self, agent_id: str) -> Dict[str, Any]:
        """Get statistics about an agent's contributions."""
        with self.lock:
            agent_memories = self.memory_index.get(agent_id, [])

            if not agent_memories:
                return {
                    "total_contributions": 0,
                    "avg_importance": 0,
                    "consensus_rate": 0,
                    "unique_contributions": 0
                }

            # Gather statistics
            importances = []
            consensus_count = 0
            unique_count = 0

            for mem_id in agent_memories:
                if mem_id not in self.memories:
                    continue

                memory = self.memories[mem_id]
                importances.append(memory.importance)

                if memory.consensus_score >= self.consensus_threshold:
                    consensus_count += 1

                if len(memory.contributors) == 1:
                    unique_count += 1

            return {
                "total_contributions": len(agent_memories),
                "avg_importance": float(np.mean(importances)) if importances else 0,
                "consensus_rate": consensus_count / len(agent_memories) if agent_memories else 0,
                "unique_contributions": unique_count,
                "collaborative_contributions": len(agent_memories) - unique_count
            }

    def resolve_conflicts(self, memory_ids: List[str]) -> Optional[str]:
        """Resolve conflicts between multiple similar memories."""
        with self.lock:
            if not memory_ids:
                return None

            # Get all memories
            memories = []
            for mem_id in memory_ids:
                if mem_id in self.memories:
                    memories.append((mem_id, self.memories[mem_id]))

            if not memories:
                return None

            # Find memory with highest consensus
            best_id = max(memories, key=lambda x: x[1].consensus_score)[0]

            # Merge other memories into the best one
            best_memory = self.memories[best_id]
            for mem_id, memory in memories:
                if mem_id != best_id:
                    # Transfer contributors
                    for contributor in memory.contributors:
                        if contributor not in best_memory.contributors:
                            best_memory.add_contributor(contributor, memory.importance)

                    # Remove merged memory
                    del self.memories[mem_id]

                    # Update indices
                    for agent_id in self.memory_index:
                        if mem_id in self.memory_index[agent_id]:
                            self.memory_index[agent_id].remove(mem_id)
                            if best_id not in self.memory_index[agent_id]:
                                self.memory_index[agent_id].append(best_id)

            self.stats["memory_conflicts"] += len(memories) - 1
            self._update_stats()
            return best_id

    def _evict_memory(self):
        """Evict least valuable memory."""
        if not self.memories:
            return

        # Score memories by value (importance * consensus * recency)
        memory_scores = []
        now = datetime.now()

        for mem_id, memory in self.memories.items():
            age_factor = 1.0 / (1 + (now - memory.last_accessed).total_seconds() / 3600)  # Hours
            access_factor = 1.0 + np.log1p(memory.access_count)
            score = memory.importance * memory.consensus_score * age_factor * access_factor
            memory_scores.append((mem_id, score))

        # Evict lowest scoring memory
        evict_id = min(memory_scores, key=lambda x: x[1])[0]

        # Remove from all indices
        for agent_id in self.memory_index:
            if evict_id in self.memory_index[agent_id]:
                self.memory_index[agent_id].remove(evict_id)

        del self.memories[evict_id]

    def _update_stats(self):
        """Update pool statistics."""
        if not self.memories:
            self.stats = {
                "total_memories": 0,
                "consensus_memories": 0,
                "avg_contributors": 0,
                "memory_conflicts": self.stats.get("memory_conflicts", 0)
            }
            return

        consensus_count = sum(1 for m in self.memories.values() 
                             if m.consensus_score >= self.consensus_threshold)
        avg_contributors = np.mean([len(m.contributors) for m in self.memories.values()])

        self.stats.update({
            "total_memories": len(self.memories),
            "consensus_memories": consensus_count,
            "avg_contributors": float(avg_contributors),
            "consensus_rate": consensus_count / len(self.memories) if self.memories else 0
        })

    def get_statistics(self) -> Dict[str, Any]:
        """Get current pool statistics."""
        with self.lock:
            return self.stats.copy()

propose_memory(agent_id: str, content: str, embedding: jnp.ndarray, importance: float, memory_id: Optional[str] = None) -> str

Propose a memory to the collective pool.

Source code in src/rotalabs_ftms/agents/collective.py 
 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
def propose_memory(self, agent_id: str, content: str, embedding: jnp.ndarray,
                  importance: float, memory_id: Optional[str] = None) -> str:
    """Propose a memory to the collective pool."""
    with self.lock:
        # Generate memory ID if not provided
        if memory_id is None:
            memory_id = f"{agent_id}_{datetime.now().timestamp()}"

        # Check if similar memory exists
        similar_memory_id = self._find_similar_memory(embedding)

        if similar_memory_id:
            # Update existing memory
            existing = self.memories[similar_memory_id]
            existing.add_contributor(agent_id, importance)

            # Update importance as weighted average
            n = len(existing.contributors)
            existing.importance = ((n - 1) * existing.importance + importance) / n

            # Update embedding as weighted average
            existing.embedding = ((n - 1) * existing.embedding + embedding) / n

            # Track in agent's index
            if agent_id not in self.memory_index:
                self.memory_index[agent_id] = []
            if similar_memory_id not in self.memory_index[agent_id]:
                self.memory_index[agent_id].append(similar_memory_id)

            self._update_stats()
            return similar_memory_id
        else:
            # Create new memory
            new_memory = CollectiveMemory(
                content=content,
                embedding=embedding,
                importance=importance
            )
            new_memory.add_contributor(agent_id, importance)

            self.memories[memory_id] = new_memory
            self.memory_index[agent_id].append(memory_id)

            # Evict if necessary
            if len(self.memories) > self.max_memories:
                self._evict_memory()

            self._update_stats()
            return memory_id

retrieve_memories(query_embedding: jnp.ndarray, k: int = 5, agent_id: Optional[str] = None, consensus_only: bool = False) -> List[Tuple[str, CollectiveMemory]]

Retrieve k most relevant memories.

Source code in src/rotalabs_ftms/agents/collective.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def retrieve_memories(self, query_embedding: jnp.ndarray, k: int = 5,
                     agent_id: Optional[str] = None,
                     consensus_only: bool = False) -> List[Tuple[str, CollectiveMemory]]:
    """Retrieve k most relevant memories."""
    with self.lock:
        # Filter memories
        candidates = []
        for mem_id, memory in self.memories.items():
            # Apply consensus filter if requested
            if consensus_only and memory.consensus_score < self.consensus_threshold:
                continue

            # Apply agent filter if requested
            if agent_id and agent_id not in memory.contributors:
                continue

            # Compute relevance score
            similarity = self._compute_similarity(query_embedding, memory.embedding)
            relevance = similarity * memory.importance * (1 + memory.consensus_score)

            candidates.append((mem_id, memory, relevance))

        # Sort by relevance and return top k
        candidates.sort(key=lambda x: x[2], reverse=True)

        # Update access counts
        results = []
        for mem_id, memory, _ in candidates[:k]:
            memory.access()
            results.append((mem_id, memory))

        return results

get_agent_contribution(agent_id: str) -> Dict[str, Any]

Get statistics about an agent's contributions.

Source code in src/rotalabs_ftms/agents/collective.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def get_agent_contribution(self, agent_id: str) -> Dict[str, Any]:
    """Get statistics about an agent's contributions."""
    with self.lock:
        agent_memories = self.memory_index.get(agent_id, [])

        if not agent_memories:
            return {
                "total_contributions": 0,
                "avg_importance": 0,
                "consensus_rate": 0,
                "unique_contributions": 0
            }

        # Gather statistics
        importances = []
        consensus_count = 0
        unique_count = 0

        for mem_id in agent_memories:
            if mem_id not in self.memories:
                continue

            memory = self.memories[mem_id]
            importances.append(memory.importance)

            if memory.consensus_score >= self.consensus_threshold:
                consensus_count += 1

            if len(memory.contributors) == 1:
                unique_count += 1

        return {
            "total_contributions": len(agent_memories),
            "avg_importance": float(np.mean(importances)) if importances else 0,
            "consensus_rate": consensus_count / len(agent_memories) if agent_memories else 0,
            "unique_contributions": unique_count,
            "collaborative_contributions": len(agent_memories) - unique_count
        }

resolve_conflicts(memory_ids: List[str]) -> Optional[str]

Resolve conflicts between multiple similar memories.

Source code in src/rotalabs_ftms/agents/collective.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def resolve_conflicts(self, memory_ids: List[str]) -> Optional[str]:
    """Resolve conflicts between multiple similar memories."""
    with self.lock:
        if not memory_ids:
            return None

        # Get all memories
        memories = []
        for mem_id in memory_ids:
            if mem_id in self.memories:
                memories.append((mem_id, self.memories[mem_id]))

        if not memories:
            return None

        # Find memory with highest consensus
        best_id = max(memories, key=lambda x: x[1].consensus_score)[0]

        # Merge other memories into the best one
        best_memory = self.memories[best_id]
        for mem_id, memory in memories:
            if mem_id != best_id:
                # Transfer contributors
                for contributor in memory.contributors:
                    if contributor not in best_memory.contributors:
                        best_memory.add_contributor(contributor, memory.importance)

                # Remove merged memory
                del self.memories[mem_id]

                # Update indices
                for agent_id in self.memory_index:
                    if mem_id in self.memory_index[agent_id]:
                        self.memory_index[agent_id].remove(mem_id)
                        if best_id not in self.memory_index[agent_id]:
                            self.memory_index[agent_id].append(best_id)

        self.stats["memory_conflicts"] += len(memories) - 1
        self._update_stats()
        return best_id

get_statistics() -> Dict[str, Any]

Get current pool statistics.

Source code in src/rotalabs_ftms/agents/collective.py 
297
298
299
300
def get_statistics(self) -> Dict[str, Any]:
    """Get current pool statistics."""
    with self.lock:
        return self.stats.copy()

Coordination

MultiAgentCoordinator

MultiAgentCoordinator

Coordinates memory sharing between multiple FTCS agents.

Enables collective intelligence through field coupling where memories from one agent can influence others' fields.

Source code in src/rotalabs_ftms/agents/coordinator.py 
 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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
class MultiAgentCoordinator:
    """
    Coordinates memory sharing between multiple FTCS agents.

    Enables collective intelligence through field coupling where
    memories from one agent can influence others' fields.
    """

    def __init__(self,
                 coordinator_id: str = "coordinator",
                 max_agents: int = 10,
                 enable_async_updates: bool = True):
        """
        Initialize multi-agent coordinator.

        Args:
            coordinator_id: Unique ID for this coordinator
            max_agents: Maximum number of agents to coordinate
            enable_async_updates: Whether to update fields asynchronously
        """
        self.coordinator_id = coordinator_id
        self.max_agents = max_agents
        self.enable_async_updates = enable_async_updates

        # Agent management
        self.agents: Dict[str, FTCSAgent] = {}
        self.agent_groups: Dict[str, AgentGroup] = {}
        self.agent_positions: Dict[str, Tuple[int, int]] = {}  # Virtual positions

        # Coupling configuration
        self.global_coupling_config = CouplingConfig()

        # Threading for async updates
        self.update_lock = threading.Lock()
        self.executor = ThreadPoolExecutor(max_workers=4) if enable_async_updates else None
        self.active_updates: List[Future] = []

        # Statistics
        self.total_couplings = 0
        self.last_coupling_time = time.time()

        self.logger = logging.getLogger(f"MultiAgentCoordinator-{coordinator_id}")

    def add_agent(self, 
                  agent: FTCSAgent,
                  position: Optional[Tuple[int, int]] = None,
                  group_id: Optional[str] = None) -> bool:
        """
        Add an agent to the coordinator.

        Args:
            agent: FTCS agent to add
            position: Virtual position for spatial coupling
            group_id: Optional group to add agent to

        Returns:
            True if successfully added
        """
        if len(self.agents) >= self.max_agents:
            self.logger.warning(f"Maximum agents ({self.max_agents}) reached")
            return False

        if agent.agent_id in self.agents:
            self.logger.warning(f"Agent {agent.agent_id} already exists")
            return False

        with self.update_lock:
            self.agents[agent.agent_id] = agent

            # Assign virtual position if not provided
            if position is None:
                # Arrange agents in a grid
                grid_size = int(np.ceil(np.sqrt(self.max_agents)))
                idx = len(self.agents) - 1
                position = (idx % grid_size, idx // grid_size)

            self.agent_positions[agent.agent_id] = position

            # Add to group if specified
            if group_id:
                self._add_to_group(agent, group_id)

        self.logger.info(f"Added agent {agent.agent_id} at position {position}")
        return True

    def _add_to_group(self, agent: FTCSAgent, group_id: str):
        """Add agent to a group."""
        if group_id not in self.agent_groups:
            self.agent_groups[group_id] = AgentGroup(
                group_id=group_id,
                agents=[],
                coupling_config=CouplingConfig(),
                shared_topics=set()
            )

        if agent not in self.agent_groups[group_id].agents:
            self.agent_groups[group_id].agents.append(agent)

    def create_group(self,
                    group_id: str,
                    agents: List[FTCSAgent],
                    coupling_config: Optional[CouplingConfig] = None,
                    shared_topics: Optional[Set[str]] = None) -> bool:
        """
        Create a group of agents with shared coupling configuration.

        Args:
            group_id: Unique group identifier
            agents: List of agents in the group
            coupling_config: Coupling configuration for this group
            shared_topics: Topics this group focuses on

        Returns:
            True if group created successfully
        """
        if group_id in self.agent_groups:
            self.logger.warning(f"Group {group_id} already exists")
            return False

        # Ensure all agents are registered
        for agent in agents:
            if agent.agent_id not in self.agents:
                self.add_agent(agent)

        self.agent_groups[group_id] = AgentGroup(
            group_id=group_id,
            agents=agents,
            coupling_config=coupling_config or CouplingConfig(),
            shared_topics=shared_topics or set()
        )

        self.logger.info(f"Created group {group_id} with {len(agents)} agents")
        return True

    def couple_agent_fields(self,
                           agent1_id: str,
                           agent2_id: str,
                           coupling_strength: Optional[float] = None) -> bool:
        """
        Couple two agents' memory fields.

        Args:
            agent1_id: First agent ID
            agent2_id: Second agent ID  
            coupling_strength: Override default coupling strength

        Returns:
            True if coupling successful
        """
        if agent1_id not in self.agents or agent2_id not in self.agents:
            self.logger.error("One or both agents not found")
            return False

        agent1 = self.agents[agent1_id]
        agent2 = self.agents[agent2_id]

        if coupling_strength is None:
            coupling_strength = self.global_coupling_config.coupling_strength

        try:
            # Perform field coupling
            self._couple_fields(agent1, agent2, coupling_strength)
            self.total_couplings += 1

            self.logger.debug(f"Coupled {agent1_id} <-> {agent2_id} with strength {coupling_strength}")
            return True

        except Exception as e:
            self.logger.error(f"Coupling failed: {e}")
            return False

    def _couple_fields(self, 
                      agent1: FTCSAgent,
                      agent2: FTCSAgent,
                      coupling_strength: float):
        """
        Perform actual field coupling between agents.

        This allows memories from one agent to influence another's field.
        """
        # Get field dimensions
        field1 = agent1.memory_field.field
        field2 = agent2.memory_field.field

        # Ensure compatible dimensions
        if field1.shape != field2.shape:
            # Resize to smaller dimension
            min_shape = (
                min(field1.shape[0], field2.shape[0]),
                min(field1.shape[1], field2.shape[1])
            )
            field1 = field1[:min_shape[0], :min_shape[1]]
            field2 = field2[:min_shape[0], :min_shape[1]]

        # Calculate field influence
        # Agent 1 influences Agent 2
        influence_1_to_2 = field1 * coupling_strength
        # Agent 2 influences Agent 1  
        influence_2_to_1 = field2 * coupling_strength

        # Apply influences with distance decay if configured
        if self.global_coupling_config.decay_with_distance:
            pos1 = self.agent_positions.get(agent1.agent_id, (0, 0))
            pos2 = self.agent_positions.get(agent2.agent_id, (0, 0))

            distance = np.sqrt((pos1[0] - pos2[0])**2 + (pos1[1] - pos2[1])**2)
            decay_factor = np.exp(-distance / self.global_coupling_config.coupling_radius)

            influence_1_to_2 *= decay_factor
            influence_2_to_1 *= decay_factor

        # Update fields
        # Create new field arrays to avoid modifying during iteration
        new_field1 = agent1.memory_field.field + influence_2_to_1[:agent1.memory_field.field.shape[0], 
                                                                   :agent1.memory_field.field.shape[1]]
        new_field2 = agent2.memory_field.field + influence_1_to_2[:agent2.memory_field.field.shape[0],
                                                                   :agent2.memory_field.field.shape[1]]

        # Apply updates
        agent1.memory_field.field = new_field1
        agent2.memory_field.field = new_field2

    def broadcast_memory(self,
                        source_agent_id: str,
                        content: str,
                        importance: float = 1.0,
                        target_group: Optional[str] = None) -> Dict[str, bool]:
        """
        Broadcast a memory from one agent to others.

        Args:
            source_agent_id: Agent broadcasting the memory
            content: Memory content
            importance: Memory importance
            target_group: Optional group to broadcast to

        Returns:
            Dict of agent_id -> success status
        """
        if source_agent_id not in self.agents:
            self.logger.error(f"Source agent {source_agent_id} not found")
            return {}

        source_agent = self.agents[source_agent_id]

        # Determine target agents
        if target_group and target_group in self.agent_groups:
            target_agents = [
                a for a in self.agent_groups[target_group].agents
                if a.agent_id != source_agent_id
            ]
        else:
            target_agents = [
                a for aid, a in self.agents.items()
                if aid != source_agent_id
            ]

        results = {}

        # Store memory in source agent first
        source_memory_id = source_agent.store_memory(
            content, 
            importance=importance,
            memory_type="broadcast"
        )

        # Broadcast to targets with distance-based importance decay
        source_pos = self.agent_positions.get(source_agent_id, (0, 0))

        for target_agent in target_agents:
            try:
                target_pos = self.agent_positions.get(target_agent.agent_id, (0, 0))
                distance = np.sqrt(
                    (source_pos[0] - target_pos[0])**2 + 
                    (source_pos[1] - target_pos[1])**2
                )

                # Decay importance with distance
                decayed_importance = importance * np.exp(-distance / 10.0)

                # Store in target with metadata about source
                target_agent.store_memory(
                    f"[From {source_agent_id}] {content}",
                    importance=decayed_importance,
                    memory_type="received_broadcast"
                )

                results[target_agent.agent_id] = True

            except Exception as e:
                self.logger.error(f"Failed to broadcast to {target_agent.agent_id}: {e}")
                results[target_agent.agent_id] = False

        self.logger.info(
            f"Broadcast from {source_agent_id} to {len(results)} agents "
            f"({sum(results.values())} successful)"
        )

        return results

    def synchronize_group_memories(self, group_id: str) -> bool:
        """
        Synchronize memories within a group through field coupling.

        Args:
            group_id: Group to synchronize

        Returns:
            True if synchronization successful
        """
        if group_id not in self.agent_groups:
            self.logger.error(f"Group {group_id} not found")
            return False

        group = self.agent_groups[group_id]
        agents = group.agents

        if len(agents) < 2:
            return True  # Nothing to synchronize

        try:
            # Couple all pairs within the group
            coupling_strength = group.coupling_config.coupling_strength

            for i in range(len(agents)):
                for j in range(i + 1, len(agents)):
                    self._couple_fields(agents[i], agents[j], coupling_strength)

            self.logger.info(f"Synchronized {len(agents)} agents in group {group_id}")
            return True

        except Exception as e:
            self.logger.error(f"Group synchronization failed: {e}")
            return False

    def run_coupling_update(self, async_update: bool = None):
        """
        Run a coupling update across all configured agent pairs.

        Args:
            async_update: Override async setting for this update
        """
        if async_update is None:
            async_update = self.enable_async_updates

        if async_update and self.executor:
            # Submit async task
            future = self.executor.submit(self._perform_coupling_update)
            self.active_updates.append(future)

            # Clean up completed futures
            self.active_updates = [f for f in self.active_updates if not f.done()]
        else:
            # Synchronous update
            self._perform_coupling_update()

    def _perform_coupling_update(self):
        """Perform the actual coupling update."""
        start_time = time.time()

        with self.update_lock:
            # Update all groups
            for group_id, group in self.agent_groups.items():
                self.synchronize_group_memories(group_id)

            # Update global couplings based on distance
            if len(self.agents) > 1:
                agents_list = list(self.agents.values())

                for i in range(len(agents_list)):
                    for j in range(i + 1, len(agents_list)):
                        agent1 = agents_list[i]
                        agent2 = agents_list[j]

                        # Check if within coupling radius
                        pos1 = self.agent_positions.get(agent1.agent_id, (0, 0))
                        pos2 = self.agent_positions.get(agent2.agent_id, (0, 0))

                        distance = np.sqrt(
                            (pos1[0] - pos2[0])**2 + 
                            (pos1[1] - pos2[1])**2
                        )

                        if distance <= self.global_coupling_config.coupling_radius:
                            self._couple_fields(
                                agent1, agent2,
                                self.global_coupling_config.coupling_strength
                            )

        self.last_coupling_time = time.time()
        update_time = time.time() - start_time

        self.logger.debug(f"Coupling update completed in {update_time:.3f}s")

    def get_collective_memory(self, 
                            query: str,
                            max_per_agent: int = 3) -> List[Dict[str, Any]]:
        """
        Retrieve memories from all agents for a query.

        Args:
            query: Query string
            max_per_agent: Max memories per agent

        Returns:
            Aggregated memory results with agent attribution
        """
        all_memories = []

        for agent_id, agent in self.agents.items():
            try:
                memories = agent.retrieve_memories(query, max_memories=max_per_agent)

                # Add agent attribution
                for memory in memories:
                    memory["source_agent"] = agent_id
                    all_memories.append(memory)

            except Exception as e:
                self.logger.error(f"Failed to retrieve from {agent_id}: {e}")

        # Sort by relevance score if available
        all_memories.sort(
            key=lambda m: m.get("combined_score", m.get("importance", 0)),
            reverse=True
        )

        return all_memories

    def get_statistics(self) -> Dict[str, Any]:
        """Get coordinator statistics."""
        agent_stats = {}
        for agent_id, agent in self.agents.items():
            agent_stats[agent_id] = agent.get_statistics()

        return {
            "coordinator_id": self.coordinator_id,
            "num_agents": len(self.agents),
            "num_groups": len(self.agent_groups),
            "total_couplings": self.total_couplings,
            "time_since_last_coupling": time.time() - self.last_coupling_time,
            "active_async_updates": len(self.active_updates) if self.executor else 0,
            "agent_statistics": agent_stats,
            "group_sizes": {
                gid: len(group.agents) 
                for gid, group in self.agent_groups.items()
            }
        }

    def shutdown(self):
        """Shutdown the coordinator and cleanup resources."""
        if self.executor:
            # Wait for active updates
            for future in self.active_updates:
                future.result(timeout=5.0)

            self.executor.shutdown(wait=True)

        self.logger.info("Multi-agent coordinator shutdown complete")

__init__(coordinator_id: str = 'coordinator', max_agents: int = 10, enable_async_updates: bool = True)

Initialize multi-agent coordinator.

Parameters:

Name Type Description Default
coordinator_id str

Unique ID for this coordinator

 'coordinator'
max_agents int

Maximum number of agents to coordinate

 10
enable_async_updates bool

Whether to update fields asynchronously

 True
Source code in src/rotalabs_ftms/agents/coordinator.py 
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
def __init__(self,
             coordinator_id: str = "coordinator",
             max_agents: int = 10,
             enable_async_updates: bool = True):
    """
    Initialize multi-agent coordinator.

    Args:
        coordinator_id: Unique ID for this coordinator
        max_agents: Maximum number of agents to coordinate
        enable_async_updates: Whether to update fields asynchronously
    """
    self.coordinator_id = coordinator_id
    self.max_agents = max_agents
    self.enable_async_updates = enable_async_updates

    # Agent management
    self.agents: Dict[str, FTCSAgent] = {}
    self.agent_groups: Dict[str, AgentGroup] = {}
    self.agent_positions: Dict[str, Tuple[int, int]] = {}  # Virtual positions

    # Coupling configuration
    self.global_coupling_config = CouplingConfig()

    # Threading for async updates
    self.update_lock = threading.Lock()
    self.executor = ThreadPoolExecutor(max_workers=4) if enable_async_updates else None
    self.active_updates: List[Future] = []

    # Statistics
    self.total_couplings = 0
    self.last_coupling_time = time.time()

    self.logger = logging.getLogger(f"MultiAgentCoordinator-{coordinator_id}")

add_agent(agent: FTCSAgent, position: Optional[Tuple[int, int]] = None, group_id: Optional[str] = None) -> bool

Add an agent to the coordinator.

Parameters:

Name Type Description Default
agent FTCSAgent

FTCS agent to add

 required
position Optional[Tuple[int, int]]

Virtual position for spatial coupling

 None
group_id Optional[str]

Optional group to add agent to

 None

Returns:

Type Description
bool

True if successfully added
Source code in src/rotalabs_ftms/agents/coordinator.py 
 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
def add_agent(self, 
              agent: FTCSAgent,
              position: Optional[Tuple[int, int]] = None,
              group_id: Optional[str] = None) -> bool:
    """
    Add an agent to the coordinator.

    Args:
        agent: FTCS agent to add
        position: Virtual position for spatial coupling
        group_id: Optional group to add agent to

    Returns:
        True if successfully added
    """
    if len(self.agents) >= self.max_agents:
        self.logger.warning(f"Maximum agents ({self.max_agents}) reached")
        return False

    if agent.agent_id in self.agents:
        self.logger.warning(f"Agent {agent.agent_id} already exists")
        return False

    with self.update_lock:
        self.agents[agent.agent_id] = agent

        # Assign virtual position if not provided
        if position is None:
            # Arrange agents in a grid
            grid_size = int(np.ceil(np.sqrt(self.max_agents)))
            idx = len(self.agents) - 1
            position = (idx % grid_size, idx // grid_size)

        self.agent_positions[agent.agent_id] = position

        # Add to group if specified
        if group_id:
            self._add_to_group(agent, group_id)

    self.logger.info(f"Added agent {agent.agent_id} at position {position}")
    return True

create_group(group_id: str, agents: List[FTCSAgent], coupling_config: Optional[CouplingConfig] = None, shared_topics: Optional[Set[str]] = None) -> bool

Create a group of agents with shared coupling configuration.

Parameters:

Name Type Description Default
group_id str

Unique group identifier

 required
agents List[FTCSAgent]

List of agents in the group

 required
coupling_config Optional[CouplingConfig]

Coupling configuration for this group

 None
shared_topics Optional[Set[str]]

Topics this group focuses on

 None

Returns:

Type Description
bool

True if group created successfully
Source code in src/rotalabs_ftms/agents/coordinator.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def create_group(self,
                group_id: str,
                agents: List[FTCSAgent],
                coupling_config: Optional[CouplingConfig] = None,
                shared_topics: Optional[Set[str]] = None) -> bool:
    """
    Create a group of agents with shared coupling configuration.

    Args:
        group_id: Unique group identifier
        agents: List of agents in the group
        coupling_config: Coupling configuration for this group
        shared_topics: Topics this group focuses on

    Returns:
        True if group created successfully
    """
    if group_id in self.agent_groups:
        self.logger.warning(f"Group {group_id} already exists")
        return False

    # Ensure all agents are registered
    for agent in agents:
        if agent.agent_id not in self.agents:
            self.add_agent(agent)

    self.agent_groups[group_id] = AgentGroup(
        group_id=group_id,
        agents=agents,
        coupling_config=coupling_config or CouplingConfig(),
        shared_topics=shared_topics or set()
    )

    self.logger.info(f"Created group {group_id} with {len(agents)} agents")
    return True

couple_agent_fields(agent1_id: str, agent2_id: str, coupling_strength: Optional[float] = None) -> bool

Couple two agents' memory fields.

Parameters:

Name Type Description Default
agent1_id str

First agent ID

 required
agent2_id str

Second agent ID

required
coupling_strength Optional[float]

Override default coupling strength

 None

Returns:

Type Description
bool

True if coupling successful
Source code in src/rotalabs_ftms/agents/coordinator.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def couple_agent_fields(self,
                       agent1_id: str,
                       agent2_id: str,
                       coupling_strength: Optional[float] = None) -> bool:
    """
    Couple two agents' memory fields.

    Args:
        agent1_id: First agent ID
        agent2_id: Second agent ID  
        coupling_strength: Override default coupling strength

    Returns:
        True if coupling successful
    """
    if agent1_id not in self.agents or agent2_id not in self.agents:
        self.logger.error("One or both agents not found")
        return False

    agent1 = self.agents[agent1_id]
    agent2 = self.agents[agent2_id]

    if coupling_strength is None:
        coupling_strength = self.global_coupling_config.coupling_strength

    try:
        # Perform field coupling
        self._couple_fields(agent1, agent2, coupling_strength)
        self.total_couplings += 1

        self.logger.debug(f"Coupled {agent1_id} <-> {agent2_id} with strength {coupling_strength}")
        return True

    except Exception as e:
        self.logger.error(f"Coupling failed: {e}")
        return False

broadcast_memory(source_agent_id: str, content: str, importance: float = 1.0, target_group: Optional[str] = None) -> Dict[str, bool]

Broadcast a memory from one agent to others.

Parameters:

Name Type Description Default
source_agent_id str

Agent broadcasting the memory

 required
content str

Memory content

 required
importance float

Memory importance

 1.0
target_group Optional[str]

Optional group to broadcast to

 None

Returns:

Type Description
Dict[str, bool]

Dict of agent_id -> success status
Source code in src/rotalabs_ftms/agents/coordinator.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def broadcast_memory(self,
                    source_agent_id: str,
                    content: str,
                    importance: float = 1.0,
                    target_group: Optional[str] = None) -> Dict[str, bool]:
    """
    Broadcast a memory from one agent to others.

    Args:
        source_agent_id: Agent broadcasting the memory
        content: Memory content
        importance: Memory importance
        target_group: Optional group to broadcast to

    Returns:
        Dict of agent_id -> success status
    """
    if source_agent_id not in self.agents:
        self.logger.error(f"Source agent {source_agent_id} not found")
        return {}

    source_agent = self.agents[source_agent_id]

    # Determine target agents
    if target_group and target_group in self.agent_groups:
        target_agents = [
            a for a in self.agent_groups[target_group].agents
            if a.agent_id != source_agent_id
        ]
    else:
        target_agents = [
            a for aid, a in self.agents.items()
            if aid != source_agent_id
        ]

    results = {}

    # Store memory in source agent first
    source_memory_id = source_agent.store_memory(
        content, 
        importance=importance,
        memory_type="broadcast"
    )

    # Broadcast to targets with distance-based importance decay
    source_pos = self.agent_positions.get(source_agent_id, (0, 0))

    for target_agent in target_agents:
        try:
            target_pos = self.agent_positions.get(target_agent.agent_id, (0, 0))
            distance = np.sqrt(
                (source_pos[0] - target_pos[0])**2 + 
                (source_pos[1] - target_pos[1])**2
            )

            # Decay importance with distance
            decayed_importance = importance * np.exp(-distance / 10.0)

            # Store in target with metadata about source
            target_agent.store_memory(
                f"[From {source_agent_id}] {content}",
                importance=decayed_importance,
                memory_type="received_broadcast"
            )

            results[target_agent.agent_id] = True

        except Exception as e:
            self.logger.error(f"Failed to broadcast to {target_agent.agent_id}: {e}")
            results[target_agent.agent_id] = False

    self.logger.info(
        f"Broadcast from {source_agent_id} to {len(results)} agents "
        f"({sum(results.values())} successful)"
    )

    return results

synchronize_group_memories(group_id: str) -> bool

Synchronize memories within a group through field coupling.

Parameters:

Name Type Description Default
group_id str

Group to synchronize

 required

Returns:

Type Description
bool

True if synchronization successful
Source code in src/rotalabs_ftms/agents/coordinator.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def synchronize_group_memories(self, group_id: str) -> bool:
    """
    Synchronize memories within a group through field coupling.

    Args:
        group_id: Group to synchronize

    Returns:
        True if synchronization successful
    """
    if group_id not in self.agent_groups:
        self.logger.error(f"Group {group_id} not found")
        return False

    group = self.agent_groups[group_id]
    agents = group.agents

    if len(agents) < 2:
        return True  # Nothing to synchronize

    try:
        # Couple all pairs within the group
        coupling_strength = group.coupling_config.coupling_strength

        for i in range(len(agents)):
            for j in range(i + 1, len(agents)):
                self._couple_fields(agents[i], agents[j], coupling_strength)

        self.logger.info(f"Synchronized {len(agents)} agents in group {group_id}")
        return True

    except Exception as e:
        self.logger.error(f"Group synchronization failed: {e}")
        return False

run_coupling_update(async_update: bool = None)

Run a coupling update across all configured agent pairs.

Parameters:

Name Type Description Default
async_update bool

Override async setting for this update

 None
Source code in src/rotalabs_ftms/agents/coordinator.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def run_coupling_update(self, async_update: bool = None):
    """
    Run a coupling update across all configured agent pairs.

    Args:
        async_update: Override async setting for this update
    """
    if async_update is None:
        async_update = self.enable_async_updates

    if async_update and self.executor:
        # Submit async task
        future = self.executor.submit(self._perform_coupling_update)
        self.active_updates.append(future)

        # Clean up completed futures
        self.active_updates = [f for f in self.active_updates if not f.done()]
    else:
        # Synchronous update
        self._perform_coupling_update()

get_collective_memory(query: str, max_per_agent: int = 3) -> List[Dict[str, Any]]

Retrieve memories from all agents for a query.

Parameters:

Name Type Description Default
query str

Query string

 required
max_per_agent int

Max memories per agent

 3

Returns:

Type Description
List[Dict[str, Any]]

Aggregated memory results with agent attribution
Source code in src/rotalabs_ftms/agents/coordinator.py 
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
def get_collective_memory(self, 
                        query: str,
                        max_per_agent: int = 3) -> List[Dict[str, Any]]:
    """
    Retrieve memories from all agents for a query.

    Args:
        query: Query string
        max_per_agent: Max memories per agent

    Returns:
        Aggregated memory results with agent attribution
    """
    all_memories = []

    for agent_id, agent in self.agents.items():
        try:
            memories = agent.retrieve_memories(query, max_memories=max_per_agent)

            # Add agent attribution
            for memory in memories:
                memory["source_agent"] = agent_id
                all_memories.append(memory)

        except Exception as e:
            self.logger.error(f"Failed to retrieve from {agent_id}: {e}")

    # Sort by relevance score if available
    all_memories.sort(
        key=lambda m: m.get("combined_score", m.get("importance", 0)),
        reverse=True
    )

    return all_memories

get_statistics() -> Dict[str, Any]

Get coordinator statistics.

Source code in src/rotalabs_ftms/agents/coordinator.py 
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
def get_statistics(self) -> Dict[str, Any]:
    """Get coordinator statistics."""
    agent_stats = {}
    for agent_id, agent in self.agents.items():
        agent_stats[agent_id] = agent.get_statistics()

    return {
        "coordinator_id": self.coordinator_id,
        "num_agents": len(self.agents),
        "num_groups": len(self.agent_groups),
        "total_couplings": self.total_couplings,
        "time_since_last_coupling": time.time() - self.last_coupling_time,
        "active_async_updates": len(self.active_updates) if self.executor else 0,
        "agent_statistics": agent_stats,
        "group_sizes": {
            gid: len(group.agents) 
            for gid, group in self.agent_groups.items()
        }
    }

shutdown()

Shutdown the coordinator and cleanup resources.

Source code in src/rotalabs_ftms/agents/coordinator.py 
488
489
490
491
492
493
494
495
496
497
def shutdown(self):
    """Shutdown the coordinator and cleanup resources."""
    if self.executor:
        # Wait for active updates
        for future in self.active_updates:
            future.result(timeout=5.0)

        self.executor.shutdown(wait=True)

    self.logger.info("Multi-agent coordinator shutdown complete")