agent-swarm-orchestration

Name: agent-swarm-orchestration
Author: TerminalSkills

114stars13forks

How to install agent-swarm-orchestration?

terminalskills/skills/agent-swarm-orchestration

$npx -y skills add TerminalSkills/skills --skill agent-swarm-orchestration

Installs into the current project.

Summary

Coordinate multiple AI agents working together on complex tasks — routing, handoffs, consensus, memory sharing, and quality gates. Use when tasks involve building multi-agent systems, coordinating specialist agents in a pipeline, implementing agent-to-agent communication, designi

Files · 1

View on GitHub

SKILL.md

1	# Agent Swarm Orchestration
2
3	## Overview
4
5	Coordinate multiple AI agents working together on complex tasks. Design topologies, implement routing, handle handoffs, share memory, and enforce quality gates.
6
7	## Instructions
8
9	### Why multi-agent?
10
11	Single-agent limitations: context window fills up, generalist performance degrades on specialist tasks, no parallel execution, single point of failure. Multi-agent benefits: focused expertise per agent, parallel subtasks, quality agents review others' work, failed agents retry without losing all progress.
12
13	### Topologies
14
15	```
16	Pipeline (sequential):
17	Task → Agent A → Agent B → Agent C → Result
18	Best for: Linear workflows (Spec → Code → Test → Deploy)
19
20	Hierarchical (manager + workers):
21	Orchestrator
22	/ \| \
23	Coder Tester Reviewer
24	Best for: Complex tasks decomposing into independent subtasks
25
26	Hub-and-spoke (router):
27	┌→ Specialist A
28	Router → Specialist B
29	└→ Specialist C
30	Best for: Task classification and routing to the right expert
31	```
32
33	### Orchestrator pattern
34
35	```python
36	# orchestrator.py — Central coordinator managing agent pipeline
37
38	from dataclasses import dataclass, field
39	from enum import Enum
40
41	class AgentRole(Enum):
42	PLANNER = "planner"
43	CODER = "coder"
44	REVIEWER = "reviewer"
45	TESTER = "tester"
46
47	@dataclass
48	class AgentTask:
49	id: str
50	role: AgentRole
51	input_data: dict
52	output_data: dict = field(default_factory=dict)
53	status: str = "pending"
54	retries: int = 0
55	max_retries: int = 3
56
57	class Orchestrator:
58	def __init__(self, agents: dict[AgentRole, 'Agent']):
59	self.agents = agents
60	self.tasks: list[AgentTask] = []
61	self.context: dict = {} # Shared memory
62
63	async def run_pipeline(self, spec: str) -> dict:
64	plan = await self._run_agent(AgentRole.PLANNER, {"spec": spec})
65	self.context["plan"] = plan
66
67	for subtask in plan.get("subtasks", []):
68	result = await self._run_agent(AgentRole.CODER, {
69	"task": subtask, "plan": plan
70	})
71	review = await self._run_agent(AgentRole.REVIEWER, {
72	"code": result, "requirements": subtask
73	})
74	retries = 0
75	while not review.get("approved") and retries < 3:
76	result = await self._run_agent(AgentRole.CODER, {
77	"task": subtask, "previous_attempt": result,
78	"feedback": review.get("feedback")
79	})
80	review = await self._run_agent(AgentRole.REVIEWER, {
81	"code": result, "requirements": subtask
82	})
83	retries += 1
84	self.context[f"subtask_{subtask['id']}"] = result
85
86	tests = await self._run_agent(AgentRole.TESTER, {"code": self.context})
87	return {"plan": plan, "results": self.context, "tests": tests}
88
89	async def _run_agent(self, role: AgentRole, input_data: dict) -> dict:
90	agent = self.agents[role]
91	task = AgentTask(id=f"{role.value}_{len(self.tasks)}", role=role, input_data=input_data)
92	self.tasks.append(task)
93	try:
94	task.status = "running"
95	result = await agent.execute(input_data)
96	task.output_data = result
97	task.status = "completed"
98	return result
99	except Exception:
100	task.status = "failed"
101	if task.retries < task.max_retries:
102	task.retries += 1
103	return await self._run_agent(role, input_data)
104	raise
105	```
106
107	### Router pattern
108
109	```python
110	# router.py — Classify and route tasks to specialists
111
112	class TaskRouter:
113	ROUTING_PROMPT = """Classify this task and select the best agent:
114	Task: {task}
115	Available agents: {agents}
116	Return JSON: {{"agent": "name", "confidence": 0.0-1.0, "reasoning": "why"}}"""
117
118	def __init__(self, agents: dict[str, 'Agent']):
119	self.agents = agents
120
121	async def route(self, task: str) -> dict:
122	agent_descriptions = "\n".join(
123	f"- {name}: {agent.description}" for name, agent in self.agents.items()
124	)
125	routing = await self._classify(task, agent_descriptions)
126	return await self.agents[routing["agent"]].execute({"task": task})
127	``

People who liked this also viewed

6 picks

Type

agent-swarm-orchestration

114stars13forks

View on GitHub

How to install agent-swarm-orchestration?

terminalskills/skills/agent-swarm-orchestration

$npx -y skills add TerminalSkills/skills --skill agent-swarm-orchestration

Installs into the current project.

Summary

Files · 1

View on GitHub

SKILL.md

1	# Agent Swarm Orchestration
2
3	## Overview
4
5	Coordinate multiple AI agents working together on complex tasks. Design topologies, implement routing, handle handoffs, share memory, and enforce quality gates.
6
7	## Instructions
8
9	### Why multi-agent?
10
11	Single-agent limitations: context window fills up, generalist performance degrades on specialist tasks, no parallel execution, single point of failure. Multi-agent benefits: focused expertise per agent, parallel subtasks, quality agents review others' work, failed agents retry without losing all progress.
12
13	### Topologies
14
15	```
16	Pipeline (sequential):
17	Task → Agent A → Agent B → Agent C → Result
18	Best for: Linear workflows (Spec → Code → Test → Deploy)
19
20	Hierarchical (manager + workers):
21	Orchestrator
22	/ \| \
23	Coder Tester Reviewer
24	Best for: Complex tasks decomposing into independent subtasks
25
26	Hub-and-spoke (router):
27	┌→ Specialist A
28	Router → Specialist B
29	└→ Specialist C
30	Best for: Task classification and routing to the right expert
31	```
32
33	### Orchestrator pattern
34
35	```python
36	# orchestrator.py — Central coordinator managing agent pipeline
37
38	from dataclasses import dataclass, field
39	from enum import Enum
40
41	class AgentRole(Enum):
42	PLANNER = "planner"
43	CODER = "coder"
44	REVIEWER = "reviewer"
45	TESTER = "tester"
46
47	@dataclass
48	class AgentTask:
49	id: str
50	role: AgentRole
51	input_data: dict
52	output_data: dict = field(default_factory=dict)
53	status: str = "pending"
54	retries: int = 0
55	max_retries: int = 3
56
57	class Orchestrator:
58	def __init__(self, agents: dict[AgentRole, 'Agent']):
59	self.agents = agents
60	self.tasks: list[AgentTask] = []
61	self.context: dict = {} # Shared memory
62
63	async def run_pipeline(self, spec: str) -> dict:
64	plan = await self._run_agent(AgentRole.PLANNER, {"spec": spec})
65	self.context["plan"] = plan
66
67	for subtask in plan.get("subtasks", []):
68	result = await self._run_agent(AgentRole.CODER, {
69	"task": subtask, "plan": plan
70	})
71	review = await self._run_agent(AgentRole.REVIEWER, {
72	"code": result, "requirements": subtask
73	})
74	retries = 0
75	while not review.get("approved") and retries < 3:
76	result = await self._run_agent(AgentRole.CODER, {
77	"task": subtask, "previous_attempt": result,
78	"feedback": review.get("feedback")
79	})
80	review = await self._run_agent(AgentRole.REVIEWER, {
81	"code": result, "requirements": subtask
82	})
83	retries += 1
84	self.context[f"subtask_{subtask['id']}"] = result
85
86	tests = await self._run_agent(AgentRole.TESTER, {"code": self.context})
87	return {"plan": plan, "results": self.context, "tests": tests}
88
89	async def _run_agent(self, role: AgentRole, input_data: dict) -> dict:
90	agent = self.agents[role]
91	task = AgentTask(id=f"{role.value}_{len(self.tasks)}", role=role, input_data=input_data)
92	self.tasks.append(task)
93	try:
94	task.status = "running"
95	result = await agent.execute(input_data)
96	task.output_data = result
97	task.status = "completed"
98	return result
99	except Exception:
100	task.status = "failed"
101	if task.retries < task.max_retries:
102	task.retries += 1
103	return await self._run_agent(role, input_data)
104	raise
105	```
106
107	### Router pattern
108
109	```python
110	# router.py — Classify and route tasks to specialists
111
112	class TaskRouter:
113	ROUTING_PROMPT = """Classify this task and select the best agent:
114	Task: {task}
115	Available agents: {agents}
116	Return JSON: {{"agent": "name", "confidence": 0.0-1.0, "reasoning": "why"}}"""
117
118	def __init__(self, agents: dict[str, 'Agent']):
119	self.agents = agents
120
121	async def route(self, task: str) -> dict:
122	agent_descriptions = "\n".join(
123	f"- {name}: {agent.description}" for name, agent in self.agents.items()
124	)
125	routing = await self._classify(task, agent_descriptions)
126	return await self.agents[routing["agent"]].execute({"task": task})
127	``

People who liked this also viewed

6 picks

Type