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