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