Coverage for agentos/swarm/coordinator.py: 25%
652 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""
2v1.9.8: Smart Swarm Coordinator with tool registry + intelligent routing.
4Full intelligence stack:
5- TaskDecomposer: decompose complex tasks into sub-task DAGs
6- ResultFusion: LLM-as-Judge aggregation with confidence scoring
7- EvalFeedbackLoop: execute → evaluate → retry → converge
8- CodeSandbox: safe code generation & execution with test cases
9- HumanLoop: human-in-the-loop breakpoints for approval/intervention
10- AgentMonitor: quality gates + self-monitoring pipeline
11- ExecutionTrace: full span-tree observability + bottleneck detection
12- AgentMemory: three-tier memory (working/short-term/long-term) + context window
13- ToolRegistry: schema-based tool catalog with versioning, capabilities, search
14- ToolRouter: intelligent tool selection with LLM + semantic matching
15- ToolExecutor: safe tool execution with validation, rate limiting, destructive confirmation
16"""
18from __future__ import annotations
20import asyncio
21import time
22import uuid
23from collections import defaultdict
24from dataclasses import dataclass, field
25from enum import Enum
26from typing import Any, Callable, Optional
28from agentos.core.di import Agent
29from agentos.swarm.task_decomposer import TaskDecomposer, Decomposition
30from agentos.swarm.result_fusion import ResultFusion, FusedResult
31from agentos.swarm.eval_feedback_loop import EvalFeedbackLoop, LoopResult, RetryConfig
32from agentos.swarm.code_sandbox import CodeSandbox, SandboxResult, TestCase, CodeFeedbackExtractor
33from agentos.swarm.human_loop import (
34 HITLManager, BreakpointType, HumanDecision,
35)
36from agentos.swarm.agent_monitor import (
37 AgentMonitor, QualityGate, MonitorReport, output_not_empty, no_error_output,
38)
39from agentos.swarm.execution_trace import (
40 ExecutionTrace, TraceEvent, TraceCollector,
41)
42from agentos.swarm.agent_memory import (
43 AgentMemory,
44)
45from agentos.swarm.tool_registry import (
46 ToolRegistry, ToolRouter, ToolExecutor, ToolSchema, ToolParam,
47 ToolCategory, RoutingDecision, RoutingContext, create_tool,
48)
49from agentos.security.guard import (
50 GuardPipeline, create_strict_guard,
51)
54class SwarmTopology(str, Enum):
55 """Swarm topology types."""
56 STAR = "star" # Central coordinator
57 RING = "ring" # Circular message passing
58 MESH = "mesh" # All-to-all communication
59 TREE = "tree" # Hierarchical structure
60 DAG = "dag" # Workflow-based dependencies
61 HYBRID = "hybrid" # Dynamic topology switching
64class ExecutionMode(str, Enum):
65 """Execution strategy for the coordinator."""
66 RAW = "raw" # Original topology-only execution
67 SMART = "smart" # Decompose → Execute DAG → Fuse
68 FEEDBACK = "feedback" # Smart + eval feedback loop
71@dataclass
72class AgentRole:
73 """Agent 角色定义。"""
74 name: str
75 goal: str
76 backstory: str = ""
77 tools: list[str] = field(default_factory=list)
78 model: str = "auto"
79 temperature: float = 0.7
80 allow_delegation: bool = True
81 verbose: bool = False
84class MessageBus:
85 """Agent 间消息总线 — 黑板模式。"""
87 def __init__(self):
88 self._messages: list[dict] = []
89 self._subscribers: dict[str, list[Callable]] = {}
90 self._shared_memory: dict[str, Any] = {}
92 def publish(self, sender: str, topic: str, data: dict):
93 msg = {"sender": sender, "topic": topic, "data": data}
94 self._messages.append(msg)
95 if topic in self._subscribers:
96 for cb in self._subscribers[topic]:
97 cb(msg)
99 def subscribe(self, topic: str, callback: Callable):
100 self._subscribers.setdefault(topic, []).append(callback)
102 @property
103 def messages(self) -> list[dict]:
104 return self._messages
106 @property
107 def shared_memory(self) -> dict[str, Any]:
108 return self._shared_memory
111@dataclass
112class SwarmMessage:
113 """
114 Message in swarm communication.
116 Attributes:
117 id: Unique identifier
118 sender: Sender agent name
119 receiver: Receiver agent name (None = broadcast)
120 content: Message content
121 metadata: Additional metadata
122 timestamp: Message timestamp
123 """
124 id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
125 sender: str = ""
126 receiver: Optional[str] = None
127 content: Any = None
128 metadata: dict[str, Any] = field(default_factory=dict)
129 timestamp: float = field(default_factory=time.time)
131 def to_dict(self) -> dict[str, Any]:
132 """Convert to dict."""
133 return {
134 "id": self.id,
135 "sender": self.sender,
136 "receiver": self.receiver,
137 "content": self.content,
138 "metadata": self.metadata,
139 "timestamp": self.timestamp,
140 }
143@dataclass
144class SwarmResult:
145 """
146 Result of swarm execution.
148 Attributes:
149 id: Unique identifier
150 topology: Swarm topology
151 mode: Execution mode used
152 outputs: Agent outputs
153 messages: Communication messages
154 duration: Execution duration
155 success: Whether execution succeeded
156 fused: ResultFusion output (smart mode only)
157 decomposition: Task decomposition used (smart mode only)
158 feedback_loop: Feedback loop result (feedback mode only)
159 """
160 id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
161 topology: SwarmTopology = SwarmTopology.STAR
162 mode: ExecutionMode = ExecutionMode.RAW
163 outputs: dict[str, Any] = field(default_factory=dict)
164 messages: list[SwarmMessage] = field(default_factory=list)
165 duration: float = 0.0
166 success: bool = True
167 fused: Optional[FusedResult] = None
168 decomposition: Optional[Decomposition] = None
169 feedback_loop: Optional[LoopResult] = None
171 def to_dict(self) -> dict[str, Any]:
172 """Convert to dict."""
173 d: dict[str, Any] = {
174 "id": self.id,
175 "topology": self.topology.value,
176 "mode": self.mode.value,
177 "outputs": self.outputs,
178 "messages": [m.to_dict() for m in self.messages],
179 "duration": f"{self.duration:.2f}s",
180 "success": self.success,
181 }
182 if self.fused:
183 d["fused"] = {
184 "action": self.fused.action,
185 "confidence": self.fused.confidence,
186 "reason": self.fused.reason,
187 }
188 if self.decomposition:
189 d["decomposition"] = {
190 "sub_tasks": [st.to_dict() for st in self.decomposition.sub_tasks],
191 "total_steps": self.decomposition.total_steps,
192 }
193 if self.feedback_loop:
194 d["feedback_loop"] = {
195 "attempts": self.feedback_loop.attempts,
196 "best_score": self.feedback_loop.best_score,
197 "converged": self.feedback_loop.converged,
198 "duration": f"{self.feedback_loop.duration:.2f}s",
199 }
200 return d
203# ── Swarm Agent Role Enum (v1.16.2, migrated from orchestration/swarm_coordinator.py) ─
205class SwarmAgentRole(str, Enum):
206 """Role of an agent within a swarm (enum-based, distinct from AgentRole dataclass)."""
207 COORDINATOR = "coordinator"
208 WORKER = "worker"
209 REVIEWER = "reviewer"
210 OBSERVER = "observer"
211 SPECIALIST = "specialist"
214class TaskPriority(str, Enum):
215 """Priority level for swarm tasks."""
216 CRITICAL = "critical"
217 HIGH = "high"
218 MEDIUM = "medium"
219 LOW = "low"
222class TaskStatus(str, Enum):
223 """Execution status of a swarm task."""
224 PENDING = "pending"
225 ASSIGNED = "assigned"
226 RUNNING = "running"
227 COMPLETED = "completed"
228 FAILED = "failed"
229 CANCELED = "canceled"
232@dataclass
233class SwarmAgentInfo:
234 """Metadata about a swarm agent (v1.16.2, migrated from orchestration)."""
235 agent_id: str
236 role: SwarmAgentRole
237 capabilities: list[str] = field(default_factory=list)
238 model: str = ""
239 max_concurrency: int = 3
240 current_load: int = 0
241 is_alive: bool = True
242 last_heartbeat: float = 0.0
243 metadata: dict[str, Any] = field(default_factory=dict)
245 @property
246 def is_available(self) -> bool:
247 return self.is_alive and self.current_load < self.max_concurrency
250@dataclass
251class SwarmTask:
252 """A task to be executed by the swarm (v1.16.2, migrated from orchestration)."""
253 task_id: str
254 description: str
255 priority: TaskPriority = TaskPriority.MEDIUM
256 status: TaskStatus = TaskStatus.PENDING
257 assigned_to: str = ""
258 required_capabilities: list[str] = field(default_factory=list)
259 parent_task_id: str = ""
260 dependencies: list[str] = field(default_factory=list)
261 result: Any = None
262 error: str = ""
263 started_at: float = 0.0
264 completed_at: float = 0.0
265 retry_count: int = 0
266 max_retries: int = 3
267 metadata: dict[str, Any] = field(default_factory=dict)
269 @property
270 def is_ready(self) -> bool:
271 return self.status == TaskStatus.PENDING and not self.dependencies
273 @property
274 def duration_ms(self) -> float:
275 if self.completed_at and self.started_at:
276 return (self.completed_at - self.started_at) * 1000
277 return 0.0
280# ── Dynamic Task Allocator ──────────────────────────────────────
282class TaskAllocator:
283 """Workload-aware dynamic task allocation (v1.16.2, migrated from orchestration).
285 Considers: capabilities, load, priority, affinity.
286 """
288 def __init__(self):
289 self._assignments: dict[str, str] = {}
291 def allocate(
292 self,
293 task: SwarmTask,
294 agents: list[SwarmAgentInfo],
295 ) -> Optional[str]:
296 available = [a for a in agents if a.is_available]
297 if not available:
298 return None
300 scored: list[tuple[SwarmAgentInfo, float]] = []
301 for agent in available:
302 score = 0.0
304 if task.required_capabilities:
305 match = len(set(task.required_capabilities) & set(agent.capabilities))
306 total = len(task.required_capabilities)
307 score += (match / total) * 50 if total > 0 else 25
309 score -= agent.current_load * 10
311 if agent.role == SwarmAgentRole.SPECIALIST:
312 if any(cap in agent.capabilities for cap in task.required_capabilities):
313 score += 20
315 if task.parent_task_id and self._assignments.get(task.parent_task_id) == agent.agent_id:
316 score += 15
318 scored.append((agent, score))
320 scored.sort(key=lambda x: x[1], reverse=True)
322 if scored and scored[0][1] > 0:
323 best = scored[0][0]
324 self._assignments[task.task_id] = best.agent_id
325 return best.agent_id
327 return available[0].agent_id if available else None
330# ── Conflict Resolver ───────────────────────────────────────────
332class ConflictType(str, Enum):
333 """Type of conflict between agent outputs."""
334 FACTUAL = "factual"
335 METHODOLOGICAL = "methodological"
336 OUTPUT = "output"
337 RESOURCE = "resource"
340class ConflictResolver:
341 """Detect and resolve conflicts between agents (v1.16.2, migrated from orchestration).
343 Strategies: majority vote, weighted vote, ranked choice, escalation.
344 """
346 def __init__(self):
347 self._conflict_log: list[dict] = []
349 def detect_conflict(
350 self,
351 agent_outputs: dict[str, Any],
352 expected_type: str = "text",
353 ) -> list[dict]:
354 conflicts = []
355 agents = list(agent_outputs.keys())
356 if len(agents) < 2:
357 return conflicts
359 outputs = list(agent_outputs.values())
361 if all(isinstance(o, str) for o in outputs):
362 for i in range(len(outputs)):
363 for j in range(i + 1, len(outputs)):
364 similarity = self._text_similarity(outputs[i], outputs[j])
365 if similarity < 0.3:
366 conflicts.append({
367 "type": ConflictType.OUTPUT.value,
368 "agents": [agents[i], agents[j]],
369 "similarity": similarity,
370 "outputs": {agents[i]: outputs[i][:200], agents[j]: outputs[j][:200]},
371 })
373 elif all(isinstance(o, (int, float)) for o in outputs):
374 values = outputs
375 mean_val = sum(values) / len(values)
376 for i, val in enumerate(values):
377 if abs(val - mean_val) / max(abs(mean_val), 1) > 0.5:
378 conflicts.append({
379 "type": ConflictType.FACTUAL.value,
380 "agents": [agents[i]],
381 "value": val,
382 "mean": mean_val,
383 "deviation": abs(val - mean_val) / max(abs(mean_val), 1),
384 })
386 return conflicts
388 def resolve(
389 self,
390 agent_outputs: dict[str, Any],
391 weights: dict[str, float] | None = None,
392 strategy: str = "majority",
393 expected_type: str = "text",
394 ) -> dict[str, Any]:
395 if len(agent_outputs) == 1:
396 agent_id = list(agent_outputs.keys())[0]
397 return {"output": agent_outputs[agent_id], "method": "single_agent", "conflict": False}
399 outputs = list(agent_outputs.values())
401 if all(isinstance(o, str) for o in outputs):
402 return self._resolve_text(agent_outputs, weights, strategy)
403 elif all(isinstance(o, (int, float)) for o in outputs):
404 return self._resolve_numeric(agent_outputs, weights, strategy)
405 else:
406 return {"output": outputs[0], "method": "first", "conflict": True}
408 def _resolve_text(self, outputs: dict[str, str], weights: dict[str, float] | None, strategy: str) -> dict:
409 if strategy == "majority":
410 votes: dict[str, list[str]] = defaultdict(list)
411 agent_ids = list(outputs.keys())
412 for i, a1 in enumerate(agent_ids):
413 best_match = a1
414 best_sim = 0
415 for j, a2 in enumerate(agent_ids):
416 if i == j:
417 continue
418 sim = self._text_similarity(outputs[a1], outputs[a2])
419 if sim > best_sim:
420 best_sim = sim
421 best_match = a2
422 key = outputs[best_match][:50]
423 votes[key].append(a1)
425 winning_key = max(votes, key=lambda k: len(votes[k]))
426 winning_agent = votes[winning_key][0]
427 return {
428 "output": outputs[winning_agent],
429 "method": "majority",
430 "votes": {k: len(v) for k, v in votes.items()},
431 "conflict": len(votes) > 1,
432 }
434 elif strategy == "weighted":
435 if not weights:
436 return self._resolve_text(outputs, weights, "majority")
437 best_agent = max(weights, key=weights.get)
438 return {"output": outputs.get(best_agent, list(outputs.values())[0]), "method": "weighted", "conflict": False}
440 else:
441 return {"output": list(outputs.values())[0], "method": "first", "conflict": False}
443 def _resolve_numeric(self, outputs: dict[str, float], weights: dict[str, float] | None, strategy: str) -> dict:
444 values = list(outputs.values())
445 agents = list(outputs.keys())
447 if strategy == "weighted" and weights:
448 total_weight = sum(weights.get(a, 1.0) for a in agents)
449 weighted = sum(weights.get(a, 1.0) * outputs[a] for a in agents) / total_weight
450 return {"output": weighted, "method": "weighted_average", "conflict": False}
451 else:
452 avg = sum(values) / len(values)
453 return {"output": avg, "method": "average", "conflict": False}
455 def _text_similarity(self, a: str, b: str) -> float:
456 if a == b:
457 return 1.0
458 tokens_a = set(a.lower().split())
459 tokens_b = set(b.lower().split())
460 if not tokens_a or not tokens_b:
461 return 0.0
462 intersection = tokens_a & tokens_b
463 union = tokens_a | tokens_b
464 return len(intersection) / len(union) if union else 0.0
467class SmartSwarmCoordinator:
468 """
469 v1.9.4: Multi-agent coordination with intelligent orchestration.
471 Upgrades the coordinator with:
472 - TaskDecomposer: LLM-driven sub-task DAG decomposition
473 - ResultFusion: LLM-as-Judge aggregation with confidence scoring
474 - EvalFeedbackLoop: execute → evaluate → retry → converge
476 Usage:
477 coordinator = SmartSwarmCoordinator(topology=SwarmTopology.MESH)
478 coordinator.register(agent1)
479 coordinator.register(agent2)
481 # Smart mode with decomposition + fusion
482 result = await coordinator.smart_execute("complex research task")
484 # Feedback mode with evaluation retry loop
485 result = await coordinator.execute_with_feedback(
486 task, expected_output, scorer
487 )
488 """
490 def __init__(
491 self,
492 topology: SwarmTopology = SwarmTopology.STAR,
493 max_rounds: int = 10,
494 execution_mode: ExecutionMode = ExecutionMode.SMART,
495 decomposer: TaskDecomposer | None = None,
496 fusion: ResultFusion | None = None,
497 feedback_loop: EvalFeedbackLoop | None = None,
498 sandbox: CodeSandbox | None = None,
499 hitl_manager: HITLManager | None = None,
500 monitor: AgentMonitor | None = None,
501 trace_collector: TraceCollector | None = None,
502 memory: AgentMemory | None = None,
503 tool_registry: ToolRegistry | None = None,
504 tool_router: ToolRouter | None = None,
505 tool_executor: ToolExecutor | None = None,
506 guard: GuardPipeline | None = None,
507 ):
508 """
509 Initialize smart swarm coordinator.
511 Args:
512 topology: Swarm topology
513 max_rounds: Maximum communication rounds
514 execution_mode: Default execution mode
515 decomposer: TaskDecomposer instance (created if None)
516 fusion: ResultFusion instance (created if None)
517 feedback_loop: EvalFeedbackLoop instance (created if None)
518 sandbox: CodeSandbox instance for code execution (created if None)
519 hitl_manager: HITLManager for human-in-the-loop (created if None)
520 monitor: AgentMonitor for quality gating (created if None)
521 trace_collector: TraceCollector for execution traces (created if None)
522 memory: AgentMemory for layered memory (created if None)
523 tool_registry: ToolRegistry for tool catalog (created if None)
524 tool_router: ToolRouter for intelligent tool selection (created if None)
525 tool_executor: ToolExecutor for safe tool execution (created if None)
526 guard: GuardPipeline for input/output safety filtering (created if None)
527 """
528 self.topology = topology
529 self.max_rounds = max_rounds
530 self.execution_mode = execution_mode
531 self._agents: dict[str, Agent[Any, Any]] = {}
532 self._message_queue: list[SwarmMessage] = []
534 self.decomposer = decomposer or TaskDecomposer()
535 self.fusion = fusion or ResultFusion()
536 self.feedback = feedback_loop or EvalFeedbackLoop()
537 self.sandbox = sandbox or CodeSandbox()
538 self.hitl = hitl_manager or HITLManager()
539 self.monitor = monitor or AgentMonitor()
540 self.tracer = trace_collector or TraceCollector()
541 self.memory = memory or AgentMemory()
542 self.tool_registry = tool_registry or ToolRegistry()
543 self.tool_router = tool_router or ToolRouter(self.tool_registry)
544 self.tool_executor = tool_executor or ToolExecutor(self.tool_registry)
545 self.guard = guard or create_strict_guard()
547 # Original topology methods bound for backward compatibility
548 self._topo_handlers = {
549 SwarmTopology.STAR: self._execute_star,
550 SwarmTopology.RING: self._execute_ring,
551 SwarmTopology.MESH: self._execute_mesh,
552 SwarmTopology.TREE: self._execute_tree,
553 }
555 # ── Agent management ──────────────────────────────────────────
557 def register(self, agent: Agent[Any, Any]) -> None:
558 self._agents[agent.name] = agent
560 def unregister(self, agent_name: str) -> bool:
561 if agent_name in self._agents:
562 del self._agents[agent_name]
563 return True
564 return False
566 def get_agent(self, agent_name: str) -> Optional[Agent[Any, Any]]:
567 return self._agents.get(agent_name)
569 def list_agents(self) -> list[str]:
570 return list(self._agents.keys())
572 # ── Execution API ─────────────────────────────────────────────
574 async def execute(
575 self,
576 task: Any,
577 mode: ExecutionMode | None = None,
578 **metadata,
579 ) -> SwarmResult:
580 """Execute a task. Delegates to smart_execute or raw topology."""
581 mode = mode or self.execution_mode
582 if mode == ExecutionMode.SMART:
583 return await self.smart_execute(task, **metadata)
584 return await self._execute_raw(task, **metadata)
586 async def smart_execute(
587 self,
588 task: Any,
589 _trace: ExecutionTrace | None = None,
590 **metadata,
591 ) -> SwarmResult:
592 """Smart execution: decompose → execute DAG → fuse.
594 Uses ExecutionTrace for observability when tracer is available.
596 Args:
597 task: Task description (string or structured)
598 _trace: Optional trace to attach (auto-created if self.tracer exists)
599 **metadata: Additional metadata
601 Returns:
602 SwarmResult with fused output and decomposition trace
603 """
604 start_time = time.time()
605 task_str = str(task)
607 # Step 0: Security guard — input filtering
608 guard_result = self.guard.process_input(task_str)
609 if guard_result.blocked:
610 result = SwarmResult(
611 topology=self.topology,
612 mode=ExecutionMode.SMART,
613 output=f"[BLOCKED] Input rejected by guard: {guard_result.blocked_by}. Reason: {', '.join(guard_result.warnings)}",
614 completed=False,
615 )
616 return result
617 if guard_result.final_content != task_str:
618 task_str = guard_result.final_content # PII-redacted version
620 # Trace setup
621 trace = _trace
622 if trace is None and self.tracer is not None:
623 trace = ExecutionTrace(task_name=task_str[:80])
624 self.tracer.add(trace)
626 if trace:
627 root = trace.start_span(TraceEvent.TASK_START, name="smart_execute", data={"task": task_str})
629 result = SwarmResult(
630 topology=self.topology,
631 mode=ExecutionMode.SMART,
632 )
634 agent_names = self.list_agents()
636 # Step 0: Load memory context
637 self.memory.set_task(task_str)
638 memory_context = self.memory.get_context(query=task_str) if self.memory else ""
640 # Step 1: Decompose
641 if trace:
642 dspan = trace.start_span(TraceEvent.DECOMPOSE, name="decompose")
643 decomp = self.decomposer.decompose(task_str, agents=agent_names)
644 result.decomposition = decomp
645 if trace and dspan:
646 trace.end_span(dspan.id, status="done", data={"sub_tasks": len(decomp.sub_tasks)})
648 # Step 2: Execute sub-tasks in dependency order
649 sub_outputs: dict[str, dict[str, Any]] = {}
650 completed: set[str] = set()
652 for _round in range(self.max_rounds):
653 ready = [
654 st for st in decomp.sub_tasks
655 if st.status == "pending"
656 and all(dep in completed for dep in st.depends_on)
657 ]
658 if not ready:
659 break
661 for st in ready:
662 st.status = "running"
664 if trace:
665 stspan = trace.start_span(TraceEvent.SUBTASK_START, name=st.description[:60], data={"id": st.id})
667 # Build context from dependencies
668 context = task_str
669 if memory_context:
670 context = f"{memory_context}\n\n[Current Task]\n{task_str}"
671 if st.depends_on:
672 dep_contexts = []
673 for dep_id in st.depends_on:
674 dep_outputs = sub_outputs.get(dep_id, {})
675 for name, out in dep_outputs.items():
676 dep_contexts.append(f"[{name}]: {str(out)[:300]}")
677 if dep_contexts:
678 context = f"{context}\n\nPrevious results:\n" + "\n".join(dep_contexts)
680 # Execute with all agents on this sub-task
681 topo_result = await self._execute_raw(context, **metadata)
682 sub_outputs[st.id] = topo_result.outputs
683 st.output = topo_result.outputs
684 st.status = "done" if topo_result.success else "failed"
685 completed.add(st.id)
687 # Store sub-task result in memory
688 self.memory.remember(
689 content=f"SubTask [{st.description}]: {json.dumps(topo_result.outputs, default=str)[:500]}",
690 role="assistant",
691 importance=0.6,
692 metadata={"subtask_id": st.id, "status": st.status},
693 )
695 if trace and stspan:
696 st_status = "done" if st.status == "done" else "failed"
697 trace.end_span(stspan.id, status=st_status, data={"output_keys": list(topo_result.outputs.keys())})
699 # Step 3: Fuse results from final sub-tasks
700 if trace:
701 fspan = trace.start_span(TraceEvent.FUSE, name="fuse_results")
703 final_subtasks = [
704 st for st in decomp.sub_tasks
705 if st.status == "done" and st.id not in {
706 s.id for s in decomp.sub_tasks
707 if any(d == st.id for d in s.depends_on)
708 }
709 ]
710 if final_subtasks:
711 all_final: dict[str, Any] = {}
712 for st in final_subtasks:
713 if st.output:
714 all_final.update(st.output)
715 if all_final:
716 fused = self.fusion.fuse(task_str, all_final)
717 result.fused = fused
718 result.outputs = all_final
719 result.success = fused.confidence >= 0.3
721 if not result.outputs and sub_outputs:
722 all_outputs: dict[str, Any] = {}
723 for st_outputs in sub_outputs.values():
724 all_outputs.update(st_outputs)
725 if all_outputs:
726 fused = self.fusion.fuse(task_str, all_outputs)
727 result.fused = fused
728 result.outputs = all_outputs
729 result.success = fused.confidence >= 0.3
731 if trace and fspan:
732 trace.end_span(fspan.id, status="done", data={"confidence": result.fused.confidence if result.fused else 0})
734 result.duration = time.time() - start_time
736 if trace and root:
737 trace.end_span(root.id, status="done" if result.success else "failed")
739 # Output guard — filter agent output before returning to user
740 if result.outputs:
741 guarded_outputs: dict[str, Any] = {}
742 for key, value in result.outputs.items():
743 output_str = str(value)
744 output_guard = self.guard.process_output(output_str)
745 if output_guard.blocked:
746 guarded_outputs[key] = f"[BLOCKED by guard: {output_guard.blocked_by}]"
747 elif output_guard.final_content != output_str:
748 guarded_outputs[key] = output_guard.final_content
749 else:
750 guarded_outputs[key] = value
751 result.outputs = guarded_outputs
753 return result
755 async def execute_with_feedback(
756 self,
757 task: Any,
758 expected_output: str = "",
759 scoring_strategy: str = "general",
760 retry_config: RetryConfig | None = None,
761 **metadata,
762 ) -> SwarmResult:
763 """Execution with eval-driven feedback loop.
765 Args:
766 task: Task description
767 expected_output: Reference for scoring
768 scoring_strategy: Scoring strategy (qa/code/summary/translation)
769 retry_config: Retry configuration
770 **metadata: Additional metadata
772 Returns:
773 SwarmResult with feedback_loop trace
774 """
775 start_time = time.time()
776 result = SwarmResult(
777 topology=self.topology,
778 mode=ExecutionMode.FEEDBACK,
779 )
781 task_str = str(task)
783 # Build executor that uses smart_execute
784 async def executor(t: str) -> Any:
785 r = await self.smart_execute(t, **metadata)
786 fused = r.fused
787 if fused and fused.merged:
788 content = fused.merged
789 # If it's a dict with agent outputs, stringify
790 if isinstance(content, dict):
791 parts = []
792 for k, v in content.items():
793 if v and not isinstance(v, dict):
794 parts.append(str(v))
795 return "\n".join(parts) if parts else str(content)
796 return str(content)
797 return str(r.outputs)
799 # Wire scorer if available
800 scorer = None
801 try:
802 from agentos.evaluation.scorers import CompositeScorerV2
803 scorer = CompositeScorerV2()
804 except Exception:
805 pass
807 feedback = EvalFeedbackLoop(
808 scorer=scorer,
809 config=retry_config or RetryConfig(max_retries=3),
810 )
812 loop_result = await feedback.run(
813 task=task_str,
814 executor=executor,
815 expected=expected_output,
816 strategy=scoring_strategy,
817 )
819 result.feedback_loop = loop_result
820 result.outputs = {"final": str(loop_result.final_output) if loop_result.final_output else ""}
821 result.success = loop_result.converged
822 result.duration = time.time() - start_time
823 return result
825 # ── Raw topology execution (backward compatible) ──────────────
827 async def _execute_raw(
828 self,
829 task: Any,
830 **metadata,
831 ) -> SwarmResult:
832 """Original topology-only execution."""
833 handler = self._topo_handlers.get(self.topology)
834 if handler is None:
835 raise ValueError(f"Unknown topology: {self.topology}")
836 return await handler(task, metadata)
838 # ── Star Topology ─────────────────────────────────────────────
840 async def _execute_star(
841 self,
842 task: Any,
843 metadata: dict[str, Any],
844 ) -> SwarmResult:
845 result = SwarmResult(topology=SwarmTopology.STAR, mode=ExecutionMode.RAW)
846 for agent_name, agent in self._agents.items():
847 try:
848 message = SwarmMessage(
849 sender="coordinator",
850 receiver=agent_name,
851 content=task,
852 metadata=metadata,
853 )
854 result.messages.append(message)
855 output = await agent.invoke(task, **metadata)
856 result.outputs[agent_name] = output
857 response = SwarmMessage(
858 sender=agent_name,
859 receiver="coordinator",
860 content=output,
861 )
862 result.messages.append(response)
863 except Exception as e:
864 result.outputs[agent_name] = {"error": str(e)}
865 result.success = False
866 return result
868 # ── Ring Topology ─────────────────────────────────────────────
870 async def _execute_ring(
871 self,
872 task: Any,
873 metadata: dict[str, Any],
874 ) -> SwarmResult:
875 result = SwarmResult(topology=SwarmTopology.RING, mode=ExecutionMode.RAW)
876 agent_names = list(self._agents.keys())
877 if not agent_names:
878 return result
880 current_input = task
881 for i, agent_name in enumerate(agent_names):
882 agent = self._agents[agent_name]
883 next_agent = agent_names[(i + 1) % len(agent_names)]
884 try:
885 output = await agent.invoke(current_input, **metadata)
886 result.outputs[agent_name] = output
887 message = SwarmMessage(
888 sender=agent_name,
889 receiver=next_agent,
890 content=output,
891 )
892 result.messages.append(message)
893 current_input = output
894 except Exception as e:
895 result.outputs[agent_name] = {"error": str(e)}
896 result.success = False
897 return result
899 # ── Mesh Topology ─────────────────────────────────────────────
901 async def _execute_mesh(
902 self,
903 task: Any,
904 metadata: dict[str, Any],
905 ) -> SwarmResult:
906 result = SwarmResult(topology=SwarmTopology.MESH, mode=ExecutionMode.RAW)
907 tasks_ = []
908 for agent_name, agent in self._agents.items():
909 tasks_.append(self._execute_agent_mesh(agent, task, metadata, result))
910 await asyncio.gather(*tasks_, return_exceptions=True)
911 for sender_name, output in result.outputs.items():
912 for receiver_name in self._agents.keys():
913 if sender_name != receiver_name:
914 message = SwarmMessage(
915 sender=sender_name,
916 receiver=receiver_name,
917 content=output,
918 )
919 result.messages.append(message)
920 return result
922 async def _execute_agent_mesh(
923 self,
924 agent: Agent[Any, Any],
925 task: Any,
926 metadata: dict[str, Any],
927 result: SwarmResult,
928 ) -> None:
929 try:
930 output = await agent.invoke(task, **metadata)
931 result.outputs[agent.name] = output
932 except Exception as e:
933 result.outputs[agent.name] = {"error": str(e)}
934 result.success = False
936 # ── Tree Topology ─────────────────────────────────────────────
938 async def _execute_tree(
939 self,
940 task: Any,
941 metadata: dict[str, Any],
942 ) -> SwarmResult:
943 result = SwarmResult(topology=SwarmTopology.TREE, mode=ExecutionMode.RAW)
944 agent_names = list(self._agents.keys())
945 if not agent_names:
946 return result
948 root_name = agent_names[0]
949 root_agent = self._agents[root_name]
950 try:
951 root_output = await root_agent.invoke(task, **metadata)
952 result.outputs[root_name] = root_output
953 except Exception as e:
954 result.outputs[root_name] = {"error": str(e)}
955 result.success = False
956 return result
958 children = agent_names[1:]
959 for child_name in children:
960 child_agent = self._agents[child_name]
961 message = SwarmMessage(
962 sender=root_name,
963 receiver=child_name,
964 content=root_output,
965 )
966 result.messages.append(message)
967 try:
968 child_output = await child_agent.invoke(root_output, **metadata)
969 result.outputs[child_name] = child_output
970 response = SwarmMessage(
971 sender=child_name,
972 receiver=root_name,
973 content=child_output,
974 )
975 result.messages.append(response)
976 except Exception as e:
977 result.outputs[child_name] = {"error": str(e)}
978 result.success = False
979 return result
981 # ── Messaging ─────────────────────────────────────────────────
983 def send_message(
984 self,
985 sender: str,
986 receiver: Optional[str],
987 content: Any,
988 **metadata,
989 ) -> SwarmMessage:
990 message = SwarmMessage(
991 sender=sender,
992 receiver=receiver,
993 content=content,
994 metadata=metadata,
995 )
996 self._message_queue.append(message)
997 return message
999 def get_messages(
1000 self,
1001 receiver: Optional[str] = None,
1002 ) -> list[SwarmMessage]:
1003 if receiver:
1004 return [
1005 m for m in self._message_queue
1006 if m.receiver == receiver or m.receiver is None
1007 ]
1008 return self._message_queue.copy()
1010 def clear_messages(self) -> None:
1011 self._message_queue.clear()
1013 # ── Code Sandbox Execution (v1.9.5) ───────────────────────────
1015 async def execute_code(
1016 self,
1017 code: str,
1018 func_name: str = "",
1019 test_cases: list[TestCase] | None = None,
1020 setup_code: str = "",
1021 sandbox: CodeSandbox | None = None,
1022 max_retries: int = 3,
1023 code_generator: Callable[[str, list[str]], str] | None = None,
1024 ) -> SandboxResult:
1025 """Execute code in sandbox with test cases and feedback-driven retry.
1027 Supports code generation: if code_generator is provided and initial run
1028 fails, it will use the feedback extractor to guide re-generation.
1030 Args:
1031 code: Code to execute (or initial code if using generator)
1032 func_name: Function name to test
1033 test_cases: Test cases for validation
1034 setup_code: Setup code (imports, fixtures)
1035 sandbox: Custom sandbox instance
1036 max_retries: Max retry attempts with code generation
1037 code_generator: Callable(spec, feedback_suggestions) → new_code
1039 Returns:
1040 SandboxResult with execution details and test outcomes
1041 """
1042 sb = sandbox or self.sandbox
1044 result = sb.run(code, func_name, test_cases, setup_code)
1046 # If initial run succeeded, we're done
1047 if result.all_passed:
1048 return result
1050 # Feedback-driven retry loop
1051 for attempt in range(1, max_retries + 1):
1052 if not code_generator:
1053 break
1055 suggestions = CodeFeedbackExtractor.extract(result)
1056 if not suggestions:
1057 break
1059 # Generate improved code
1060 spec = f"Function: {func_name}, Test cases: {len(test_cases or [])}"
1061 try:
1062 new_code = code_generator(spec, suggestions)
1063 except Exception:
1064 break
1066 if not new_code or new_code == code:
1067 break
1069 code = new_code
1070 result = sb.run(code, func_name, test_cases, setup_code)
1072 if result.all_passed:
1073 break
1075 if attempt == max_retries:
1076 break # Don't overwrite last result
1078 return result
1080 # ── HITL-Enhanced Execution (v1.9.5) ──────────────────────────
1082 async def smart_execute_with_hitl(
1083 self,
1084 task: Any,
1085 hitl: HITLManager | None = None,
1086 **metadata,
1087 ) -> SwarmResult:
1088 """Smart execution with human-in-the-loop breakpoints.
1090 Same as smart_execute but pauses at configurable checkpoints:
1091 - Before each sub-task (if hitl.break_on_every_task)
1092 - On sub-task failure (if hitl.break_on_failure)
1093 - On low-confidence fusion (if config threshold met)
1095 Args:
1096 task: Task description
1097 hitl: HITLManager instance (uses self.hitl if None)
1098 **metadata: Additional metadata
1100 Returns:
1101 SwarmResult with fused output
1102 """
1103 hitl_mgr = hitl or self.hitl
1104 start_time = time.time()
1105 result = SwarmResult(
1106 topology=self.topology,
1107 mode=ExecutionMode.SMART,
1108 )
1110 task_str = str(task)
1111 agent_names = self.list_agents()
1113 # Step 1: Decompose
1114 decomp = self.decomposer.decompose(task_str, agents=agent_names)
1115 result.decomposition = decomp
1117 # Step 2: Execute sub-tasks with HITL gates
1118 sub_outputs: dict[str, dict[str, Any]] = {}
1119 completed: set[str] = set()
1120 aborted = False
1122 for _round in range(self.max_rounds):
1123 if aborted:
1124 break
1126 ready = [
1127 st for st in decomp.sub_tasks
1128 if st.status == "pending"
1129 and all(dep in completed for dep in st.depends_on)
1130 ]
1131 if not ready:
1132 break
1134 for st in ready:
1135 # HITL: check before executing sub-task
1136 if hitl_mgr.config.break_on_every_task:
1137 decision, feedback = await hitl_mgr.request_decision(
1138 bp_type=BreakpointType.BEFORE_TASK,
1139 task_id=st.id,
1140 message=f"Execute sub-task: {st.description}?",
1141 context={"task": task_str, "sub_task": st.description},
1142 options=["approve", "abort", "modify"],
1143 )
1144 if decision == HumanDecision.ABORT:
1145 aborted = True
1146 break
1147 if decision == HumanDecision.MODIFY and feedback:
1148 st.description = f"{st.description} [modified: {feedback}]"
1150 st.status = "running"
1152 # Build context from dependencies
1153 context = task_str
1154 if st.depends_on:
1155 dep_contexts = []
1156 for dep_id in st.depends_on:
1157 dep_outputs = sub_outputs.get(dep_id, {})
1158 for name, out in dep_outputs.items():
1159 dep_contexts.append(f"[{name}]: {str(out)[:300]}")
1160 if dep_contexts:
1161 context = f"{task_str}\n\nPrevious results:\n" + "\n".join(dep_contexts)
1163 # Execute
1164 topo_result = await self._execute_raw(context, **metadata)
1165 sub_outputs[st.id] = topo_result.outputs
1166 st.output = topo_result.outputs
1167 st.status = "done" if topo_result.success else "failed"
1168 completed.add(st.id)
1170 # HITL: check on failure
1171 if not topo_result.success:
1172 decision, feedback = await hitl_mgr.should_break_on_failure(
1173 task_id=st.id,
1174 error=topo_result.error or "Unknown error",
1175 attempt=1,
1176 )
1177 if decision == HumanDecision.ABORT:
1178 aborted = True
1179 break
1180 if decision == HumanDecision.MODIFY and feedback:
1181 st.description = f"{st.description} [retry with: {feedback}]"
1182 st.status = "pending" # Re-queue for retry
1183 completed.discard(st.id)
1184 del sub_outputs[st.id]
1186 if aborted:
1187 result.success = False
1188 result.error = "Aborted by human"
1189 return result
1191 # Step 3: Fuse results
1192 final_subtasks = [
1193 st for st in decomp.sub_tasks
1194 if st.status == "done" and st.id not in {
1195 s.id for s in decomp.sub_tasks
1196 if any(d == st.id for d in s.depends_on)
1197 }
1198 ]
1199 if final_subtasks:
1200 all_final: dict[str, Any] = {}
1201 for st in final_subtasks:
1202 if st.output:
1203 all_final.update(st.output)
1204 if all_final:
1205 fused = self.fusion.fuse(task_str, all_final)
1206 result.fused = fused
1207 result.outputs = all_final
1209 # HITL: check low confidence
1210 if fused.confidence < hitl_mgr.config.break_on_low_confidence:
1211 decision, feedback = await hitl_mgr.should_break_on_result(
1212 task_id="final",
1213 output=all_final,
1214 confidence=fused.confidence,
1215 )
1216 if decision == HumanDecision.ABORT:
1217 result.success = False
1218 result.error = "Aborted by human at final result"
1219 return result
1220 if decision == HumanDecision.REJECT:
1221 result.success = False
1222 result.error = f"Rejected: {feedback}"
1223 return result
1225 result.success = fused.confidence >= 0.3
1227 if not result.outputs and sub_outputs:
1228 all_outputs: dict[str, Any] = {}
1229 for st_outputs in sub_outputs.values():
1230 all_outputs.update(st_outputs)
1231 if all_outputs:
1232 fused = self.fusion.fuse(task_str, all_outputs)
1233 result.fused = fused
1234 result.outputs = all_outputs
1235 result.success = fused.confidence >= 0.3
1237 result.duration = time.time() - start_time
1238 return result
1240 # ── Monitored Execution (v1.9.6) ─────────────────────────────
1242 async def monitor_execute(
1243 self,
1244 task: Any,
1245 quality_gates: list[QualityGate] | None = None,
1246 fallback_fn: Callable[[], Any] | None = None,
1247 **metadata,
1248 ) -> tuple[Any, MonitorReport]:
1249 """Execute with automatic quality gating.
1251 Runs smart_execute through the AgentMonitor pipeline. If gates fail,
1252 automatically retries or falls back based on gate configuration.
1254 Args:
1255 task: Task description
1256 quality_gates: Custom quality gates (uses monitor defaults if None)
1257 fallback_fn: Fallback function if all gates fail
1258 **metadata: Additional metadata
1260 Returns:
1261 Tuple of (final_output, MonitorReport)
1262 """
1263 # Configure monitor with custom gates if provided
1264 monitor = self.monitor
1265 if quality_gates:
1266 monitor = AgentMonitor(
1267 max_retries=self.monitor.max_retries,
1268 default_fallback=self.monitor.default_fallback,
1269 )
1270 monitor.add_gates(quality_gates)
1271 elif not self.monitor._gates:
1272 # Default gates if none configured
1273 monitor = AgentMonitor()
1274 monitor.add_gates([
1275 output_not_empty(),
1276 no_error_output(),
1277 ])
1279 # Track latency for latency gates
1280 start = time.time()
1282 async def execute_fn() -> Any:
1283 result = await self.smart_execute(task, **metadata)
1284 fused = result.fused
1285 if fused and fused.merged:
1286 return fused.merged
1287 return result.outputs
1289 output, report = await monitor.monitor_execution(
1290 task_fn=execute_fn,
1291 task_name=str(task)[:80],
1292 context={"_latency_ms": 0},
1293 fallback_fn=fallback_fn,
1294 )
1296 # Inject actual latency
1297 elapsed = (time.time() - start) * 1000
1298 for gate in report.gates:
1299 gate.data["_latency_ms"] = elapsed
1301 return output, report
1303 # ── Tool Registry Convenience Methods ─────────────────────────
1305 def register_tool(
1306 self,
1307 name: str,
1308 description: str,
1309 handler: Callable,
1310 category: ToolCategory = ToolCategory.CUSTOM,
1311 params: list[ToolParam] | None = None,
1312 capabilities: list[str] | None = None,
1313 tags: list[str] | None = None,
1314 is_destructive: bool = False,
1315 rate_limit: int = 0,
1316 **kwargs,
1317 ) -> ToolSchema:
1318 """Register a tool in the coordinator's tool registry."""
1319 tool = create_tool(
1320 name=name, description=description, handler=handler,
1321 category=category, params=params or [],
1322 capabilities=capabilities or [], tags=tags or [],
1323 is_destructive=is_destructive, rate_limit=rate_limit, **kwargs,
1324 )
1325 return self.tool_registry.register(tool)
1327 def find_tool(self, query: str, top_k: int = 5) -> list[tuple[ToolSchema, float]]:
1328 """Search for tools matching a natural language query."""
1329 return self.tool_registry.search(query, top_k=top_k)
1331 def route_tool(self, task: str, **ctx_kwargs) -> RoutingDecision:
1332 """Route a task to the best matching tool."""
1333 context = RoutingContext(task=task, **ctx_kwargs)
1334 return self.tool_router.route(context)
1336 def execute_tool(self, tool_name: str, params: dict[str, Any] | None = None, force: bool = False) -> Any:
1337 """Execute a registered tool safely."""
1338 return self.tool_executor.execute(tool_name, params, force=force)
1341# ── Backward-compatible alias ─────────────────────────────────────
1342SwarmCoordinator = SmartSwarmCoordinator