Coverage for agentos/swarm/patterns.py: 35%
158 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 19:15 +0800
1"""
2Enhanced Swarm collaboration patterns.
4Extends the base SwarmCoordinator with broadcast, pipeline, hierarchical,
5and consensus-based collaboration topologies.
6"""
8from __future__ import annotations
10import asyncio
11import time
12from collections.abc import Callable
13from dataclasses import dataclass
14from enum import Enum
15from typing import Any
17from agentos.swarm.coordinator import SwarmCoordinator
20class Topology(Enum):
21 """Swarm collaboration topology."""
23 BROADCAST = "broadcast"
24 """One-to-all: leader broadcasts task, all members respond independently."""
26 PIPELINE = "pipeline"
27 """Sequential chain: each member processes output of previous member."""
29 HIERARCHICAL = "hierarchical"
30 """Tree structure: leader delegates to sub-leaders who manage sub-teams."""
32 CONSENSUS = "consensus"
33 """Voting: all members vote on outputs, majority wins."""
35 ROUND_ROBIN = "round_robin"
36 """Load-balancing: tasks distributed evenly across members."""
39@dataclass
40class CollaborationConfig:
41 """Configuration for swarm collaboration."""
43 topology: Topology = Topology.BROADCAST
44 timeout_per_member: float = 60.0
45 """Max seconds per member invocation."""
47 max_parallel: int = 5
48 """Max concurrent member executions (broadcast/consensus)."""
50 quorum_ratio: float = 0.5
51 """Minimum ratio of members needed for consensus (consensus topology)."""
53 allow_partial_results: bool = True
54 """Return partial results if some members fail."""
57@dataclass
58class MemberResult:
59 """Result from a single swarm member."""
61 member_id: str
62 success: bool
63 output: Any = None
64 error: str | None = None
65 latency_ms: float = 0.0
68@dataclass
69class CollaborationResult:
70 """Aggregated result from a swarm collaboration."""
72 topology: Topology
73 member_results: list[MemberResult]
74 aggregated_output: Any = None
75 total_latency_ms: float = 0.0
76 success_count: int = 0
77 failure_count: int = 0
80class SwarmPatterns:
81 """
82 Higher-order swarm collaboration patterns built on SwarmCoordinator.
84 Supports five topologies: broadcast, pipeline, hierarchical, consensus, round_robin.
85 """
87 def __init__(
88 self,
89 coordinator: SwarmCoordinator,
90 config: CollaborationConfig | None = None,
91 ):
92 self._coordinator = coordinator
93 self._config = config or CollaborationConfig()
95 # ---- Synchronous API ----
97 def collaborate(
98 self,
99 task: str,
100 context: dict[str, Any] | None = None,
101 ) -> CollaborationResult:
102 """
103 Execute collaboration using configured topology.
105 Args:
106 task: The task description to collaborate on.
107 context: Optional context dict passed to all members.
109 Returns:
110 CollaborationResult with individual and aggregated outputs.
111 """
112 t0 = time.perf_counter()
113 topology = self._config.topology
115 dispatchers: dict[Topology, Callable] = {
116 Topology.BROADCAST: self._broadcast,
117 Topology.PIPELINE: self._pipeline,
118 Topology.HIERARCHICAL: self._hierarchical,
119 Topology.CONSENSUS: self._consensus,
120 Topology.ROUND_ROBIN: self._round_robin,
121 }
123 handler = dispatchers.get(topology)
124 if handler is None:
125 raise ValueError(f"Unknown topology: {topology}")
127 result = handler(task, context)
128 result.total_latency_ms = (time.perf_counter() - t0) * 1000
129 result.topology = topology
130 return result
132 def _broadcast(self, task: str, context: dict | None = None) -> CollaborationResult:
133 """Broadcast task to all members, collect all responses."""
134 members = self._coordinator.list_members()
135 results: list[MemberResult] = []
137 for member in members:
138 m_result = self._invoke_member(member, task, context)
139 results.append(m_result)
141 aggregated = [r.output for r in results if r.success]
142 success = sum(1 for r in results if r.success)
143 failure = len(results) - success
145 return CollaborationResult(
146 topology=Topology.BROADCAST,
147 member_results=results,
148 aggregated_output=aggregated,
149 success_count=success,
150 failure_count=failure,
151 )
153 def _pipeline(self, task: str, context: dict | None = None) -> CollaborationResult:
154 """Sequential pipeline: each member processes previous output."""
155 members = self._coordinator.list_members()
156 results: list[MemberResult] = []
157 current_input = task
159 for member in members:
160 m_result = self._invoke_member(member, current_input, context)
161 results.append(m_result)
162 if m_result.success:
163 current_input = str(m_result.output) if m_result.output else current_input
164 elif not self._config.allow_partial_results:
165 break
167 success = sum(1 for r in results if r.success)
168 failure = len(results) - success
170 return CollaborationResult(
171 topology=Topology.PIPELINE,
172 member_results=results,
173 aggregated_output=current_input,
174 success_count=success,
175 failure_count=failure,
176 )
178 def _hierarchical(self, task: str, context: dict | None = None) -> CollaborationResult:
179 """Two-level hierarchy: leader delegates to sub-groups."""
180 members = self._coordinator.list_members()
181 n = len(members)
182 if n < 2:
183 # Fallback to broadcast for small swarms
184 return self._broadcast(task, context)
186 # Split members: first half as sub-leaders, rest as workers
187 split = max(1, n // 2)
188 sub_leaders = members[:split]
189 workers = members[split:]
191 results: list[MemberResult] = []
192 # Step 1: Sub-leaders plan task decomposition
193 plan_task = f"Decompose this task into sub-tasks for {len(workers)} workers: {task}"
194 for leader in sub_leaders:
195 m_result = self._invoke_member(leader, plan_task, context)
196 results.append(m_result)
198 # Step 2: Workers execute sub-tasks
199 sub_tasks = task.split(";") if ";" in task else [task]
200 for i, worker in enumerate(workers):
201 sub_task = sub_tasks[i % len(sub_tasks)]
202 m_result = self._invoke_member(worker, sub_task.strip(), context)
203 results.append(m_result)
205 success = sum(1 for r in results if r.success)
206 failure = len(results) - success
208 return CollaborationResult(
209 topology=Topology.HIERARCHICAL,
210 member_results=results,
211 aggregated_output=[r.output for r in results if r.success],
212 success_count=success,
213 failure_count=failure,
214 )
216 def _consensus(self, task: str, context: dict | None = None) -> CollaborationResult:
217 """Voting: all members vote, majority output wins."""
218 members = self._coordinator.list_members()
219 results: list[MemberResult] = []
220 votes: dict[str, int] = {}
222 for member in members:
223 m_result = self._invoke_member(member, task, context)
224 results.append(m_result)
225 if m_result.success and m_result.output is not None:
226 key = str(m_result.output)
227 votes[key] = votes.get(key, 0) + 1
229 quorum = max(1, int(len(members) * self._config.quorum_ratio))
230 winner = None
231 for output_key, count in votes.items():
232 if count >= quorum:
233 winner = output_key
234 break
236 success = sum(1 for r in results if r.success)
237 failure = len(results) - success
239 return CollaborationResult(
240 topology=Topology.CONSENSUS,
241 member_results=results,
242 aggregated_output=winner or "No consensus reached",
243 success_count=success,
244 failure_count=failure,
245 )
247 def _round_robin(self, task: str, context: dict | None = None) -> CollaborationResult:
248 """Load-balancing: pick next available member."""
249 members = self._coordinator.list_members()
250 if not members:
251 return CollaborationResult(
252 topology=Topology.ROUND_ROBIN,
253 member_results=[],
254 aggregated_output=None,
255 success_count=0,
256 failure_count=0,
257 )
258 # Simple: use first available member (full RR needs state)
259 member = members[0]
260 m_result = self._invoke_member(member, task, context)
261 success = 1 if m_result.success else 0
262 failure = 0 if m_result.success else 1
264 return CollaborationResult(
265 topology=Topology.ROUND_ROBIN,
266 member_results=[m_result],
267 aggregated_output=m_result.output,
268 success_count=success,
269 failure_count=failure,
270 )
272 def _invoke_member(
273 self, member_id: str, task: str, context: dict | None = None
274 ) -> MemberResult:
275 """Invoke a single swarm member with timeout."""
276 t0 = time.perf_counter()
277 try:
278 output = self._coordinator.delegate(
279 member_id=member_id,
280 task=task,
281 context=context or {},
282 )
283 latency = (time.perf_counter() - t0) * 1000
284 return MemberResult(
285 member_id=member_id,
286 success=True,
287 output=output,
288 latency_ms=latency,
289 )
290 except Exception as exc:
291 latency = (time.perf_counter() - t0) * 1000
292 return MemberResult(
293 member_id=member_id,
294 success=False,
295 error=f"{type(exc).__name__}: {exc}",
296 latency_ms=latency,
297 )
299 # ---- Async API ----
301 async def collaborate_async(
302 self,
303 task: str,
304 context: dict[str, Any] | None = None,
305 ) -> CollaborationResult:
306 """Async version using asyncio for broadcast/consensus topologies."""
307 t0 = time.perf_counter()
308 members = self._coordinator.list_members()
310 if self._config.topology == Topology.BROADCAST:
311 semaphore = asyncio.Semaphore(self._config.max_parallel)
313 async def run_one(member_id: str) -> MemberResult:
314 async with semaphore:
315 return await asyncio.to_thread(self._invoke_member, member_id, task, context)
317 results = await asyncio.gather(*[run_one(m) for m in members])
318 member_results = list(results)
319 else:
320 # Other topologies: run sequentially in thread
321 member_results = await asyncio.to_thread(self.collaborate, task, context)
322 if isinstance(member_results, CollaborationResult):
323 member_results = member_results.member_results
325 success = sum(1 for r in member_results if r.success)
326 failure = len(member_results) - success
328 return CollaborationResult(
329 topology=self._config.topology,
330 member_results=member_results,
331 aggregated_output=[r.output for r in member_results if r.success],
332 total_latency_ms=(time.perf_counter() - t0) * 1000,
333 success_count=success,
334 failure_count=failure,
335 )