paulpas Multi-Agent Orchestration
Orchestrates multi-agent workflows to decompose complex tasks into specialized agent roles with structured coordination patterns. Loading this skill makes the model design, implement, and validate agent teams using sequential pipelines, supervisor-worker delegation, group chat coordination, and parallel execution — choosing the right pattern based on task dependencies, latency requirements, and output coherence needs.
TL;DR Checklist
- Define each agent's role, tools, and bounded responsibility before writing code
- Select orchestration pattern by analyzing task dependency graph (sequential / parallel / hierarchical)
- Enforce typed inputs/outputs via Pydantic models at every agent boundary
- Add quality gates between pipeline stages to reject or retry poor outputs
- Implement a result synthesizer that merges, deduplicates, and resolves conflicts from multiple agents
- Set a maximum turn limit for group chat patterns to prevent infinite loops
When to Use
Use this skill when:
- A task naturally decomposes into distinct subtasks handled by specialized agents (e.g., research → analysis → summary generation)
- You need hierarchical delegation where a supervisor breaks down complex goals and assigns subtasks to workers
- Multiple independent analyses must run in parallel and the results need conflict resolution before producing a final output
- You are building a group chat system where agents take turns contributing expertise with LLM-powered speaker selection
When NOT to Use
Avoid this skill for:
- Single-turn tasks that one agent can complete reliably (overhead of orchestration exceeds benefit)
- Tasks requiring real-time sub-millisecond response times (inter-agent communication latency is non-trivial)
- Scenarios where a single LLM call with good prompting would suffice — multi-agent adds complexity, cost, and failure surface
Core Workflow
1. Define Agent Roles and Bounded Responsibilities
Create distinct agent personas using TypedDict or Pydantic models that declare each agent's role, available tools, and output schema. Each agent must have a narrow, well-defined scope — avoid agents with overlapping responsibilities.
from pydantic import BaseModel, Field
from typing import Protocol, runtime_checkable
import enum
class AgentRole(str, enum.Enum):
RESEARCHER = "researcher"
ANALYST = "analyst"
WRITER = "writer"
REVIEWER = "reviewer"
class AgentSpec(BaseModel):
"""Defines a single agent's role, tools, and output contract."""
role: AgentRole
description: str
system_prompt: str
allowed_tools: list[str] = Field(default_factory=list)
max_tokens: int = 4096
class Config:
frozen = True
# Example agent definitions — narrow, non-overlapping scopes
RESEARCHER_SPEC = AgentSpec(
role=AgentRole.RESEARCHER,
description="Gathers raw information from sources and returns structured findings.",
system_prompt=(
"You are a Researcher. Your job is to collect factual information from provided "
"sources. Return only verified facts with source citations. Do not analyze or "
"synthesize — just extract."
),
allowed_tools=["web_search", "web_fetch", "file_read"],
)
ANALYST_SPEC = AgentSpec(
role=AgentRole.ANALYST,
description="Analyzes research findings and produces structured conclusions with confidence scores.",
system_prompt=(
"You are an Analyst. Use the Researcher's findings to draw conclusions. "
"Assign a confidence score (0.0–1.0) to each conclusion. Flag contradictions."
),
allowed_tools=["code_interpreter", "math_solver"],
)
WRITER_SPEC = AgentSpec(
role=AgentRole.WRITER,
description="Writes a coherent final report from the Analyst's conclusions.",
system_prompt=(
"You are a Writer. Transform analytical conclusions into a well-structured "
"report suitable for executive consumption. Maintain factual accuracy and "
"cite confidence scores."
),
allowed_tools=["template_render"],
)
Checkpoint: Verify that no two agents claim the same responsibility. Each agent's allowed_tools should be a strict subset of what it actually needs — this is your security boundary.
2. Design Orchestration Pattern
Analyze the task dependency graph and select the appropriate pattern:
| Pattern | When to Use | Latency | Coordination Complexity |
|---|---|---|---|
| Sequential Pipeline | Strict ordering, each stage depends on prior output | High (serial) | Low |
| Supervisor-Worker | One complex goal decomposable into independent subtasks | Medium (partial parallel) | Medium |
| Group Chat | Open-ended exploration where agents iteratively build on each other | Variable | High |
| Parallel Execution | Independent tasks that merge at the end | Low (parallel) | Low-Medium |
3. Wire Task Dependencies
Map outputs from one agent to inputs of the next using typed message passing. Use Pydantic models to enforce schema compliance at every boundary. This is where you handle conditional branching — if a pipeline stage produces insufficient quality, route to an retry or escalation path.
from pydantic import BaseModel, field_validator
from typing import Optional
class ResearchOutput(BaseModel):
"""Structured output from the Researcher agent."""
findings: list[dict[str, str]] = Field(
description="List of {fact, source, confidence} dicts"
)
gaps: list[str] = Field(
default_factory=list,
description="Information gaps identified during research"
)
@field_validator("findings")
@classmethod
def validate_findings(cls, v: list[dict]) -> list[dict]:
for finding in v:
if not all(k in finding for k in ("fact", "source")):
raise ValueError(f"Finding missing required keys: {finding}")
return v
class AnalysisOutput(BaseModel):
"""Structured output from the Analyst agent."""
conclusions: list[dict[str, str | float]] = Field(
description="List of {conclusion, confidence, supporting_evidence} dicts"
)
contradictions: list[tuple[str, str]] = Field(
default_factory=list,
description="Pairs of contradictory findings and their resolution"
)
@field_validator("conclusions")
@classmethod
def validate_confidence(cls, v: list[dict]) -> list[dict]:
for c in v:
conf = c.get("confidence", 0.0)
if not (0.0 <= conf <= 1.0):
raise ValueError(f"Confidence must be 0.0-1.0, got {conf}")
return v
class ReportOutput(BaseModel):
"""Final structured report from the Writer agent."""
executive_summary: str
sections: list[dict[str, str]] = Field(
default_factory=list,
description="List of {title, content, confidence} dicts"
)
references: list[str]
@field_validator("sections")
@classmethod
def validate_sections(cls, v: list[dict]) -> list[dict]:
if not v:
raise ValueError("Report must contain at least one section")
return v
Checkpoint: Run each Pydantic model through a validation test with realistic input before deploying to production. Schema mismatches are the #1 cause of pipeline failures.
4. Execute and Collect Results
Run agents according to the chosen pattern. For parallel execution, use concurrent.futures.ThreadPoolExecutor to dispatch independent workers. For sequential pipelines, chain agent calls with inter-stage validation.
5. Synthesize Final Output
Merge results from multiple agents, deduplicate overlapping conclusions, resolve contradictions using confidence-weighted voting, and produce a coherent final answer. The synthesizer is the most critical component — it turns fragmented agent outputs into a unified result.
Implementation Patterns
Pattern 1: Supervisor-Worker Delegation
A supervisor agent decomposes a complex goal into subtasks, dispatches workers via ThreadPoolExecutor, then synthesizes results. Use this when you have one clear objective that can be parallelized.
import asyncio
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Any, Callable
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class Subtask:
"""A discrete unit of work delegated to a worker agent."""
id: str
description: str
required_tools: list[str]
input_data: dict[str, Any] = field(default_factory=dict)
@dataclass
class WorkerResult:
"""Result from a single worker agent execution."""
subtask_id: str
success: bool
output: dict[str, Any] | None
error: str | None = None
latency_ms: float = 0.0
def create_supervisor(
llm_client: Any,
max_workers: int = 4,
timeout_seconds: float = 120.0,
) -> "SupervisorOrchestrator":
"""Factory that creates a SupervisorOrchestrator with configured constraints."""
return SupervisorOrchestrator(
llm_client=llm_client,
max_workers=max_workers,
timeout_seconds=timeout_seconds,
)
class SupervisorOrchestrator:
"""Supervisor-worker pattern: decompose → delegate → synthesize.
The supervisor breaks a high-level goal into subtasks using an LLM call,
dispatches workers in parallel via ThreadPoolExecutor, then merges results.
"""
def __init__(
self,
llm_client: Any,
max_workers: int = 4,
timeout_seconds: float = 120.0,
) -> None:
self.llm_client = llm_client
self.max_workers = min(max_workers, 8) # Hard cap to prevent resource exhaustion
self.timeout_seconds = timeout_seconds
def decompose_task(self, goal: str) -> list[Subtask]:
"""Use LLM to decompose a high-level goal into parallel subtasks.
Args:
goal: The high-level objective to decompose.
Returns:
List of Subtask objects describing work for each worker.
"""
prompt = (
f"Break the following goal into 2-5 independent subtasks that can run "
f"in parallel. Return a JSON array with 'id', 'description', and "
f"'required_tools' for each subtask.\n\nGoal: {goal}"
)
response = self.llm_client.generate(prompt, max_tokens=1024)
# Parse LLM response into Subtask objects
import json
try:
data = json.loads(response.strip())
return [
Subtask(
id=str(item["id"]),
description=item["description"],
required_tools=list(item.get("required_tools", [])),
)
for item in data
]
except (json.JSONDecodeError, KeyError) as exc:
raise ValueError(f"Failed to parse decomposition response: {exc}") from exc
def execute_workers(self, subtasks: list[Subtask]) -> list[WorkerResult]:
"""Dispatch subtasks to workers in parallel and collect results.
Args:
subtasks: List of Subtask objects to execute.
Returns:
List of WorkerResult objects with success/failure status.
"""
def run_worker(subtask: Subtask) -> WorkerResult:
"""Execute a single worker task with error handling and timing."""
start = asyncio.get_event_loop().time() if not hasattr(self, "_sync_mode") else __import__("time").perf_counter()
try:
# Each worker gets a specialized prompt based on its subtask
worker_prompt = (
f"Execute this task: {subtask.description}\n\n"
f"Input data: {subtask.input_data}\n\n"
f"Available tools: {', '.join(subtask.required_tools)}\n\n"
f"Return a JSON object with your results."
)
output = self.llm_client.generate(worker_prompt, max_tokens=2048)
elapsed = __import__("time").perf_counter() - start
return WorkerResult(
subtask_id=subtask.id,
success=True,
output={"raw_response": output},
latency_ms=elapsed * 1000,
)
except Exception as exc:
elapsed = __import__("time").perf_counter() - start
logger.error("Worker %s failed: %s", subtask.id, exc)
return WorkerResult(
subtask_id=subtask.id,
success=False,
output=None,
error=str(exc),
latency_ms=elapsed * 1000,
)
# Dispatch in parallel with a thread pool
results: list[WorkerResult] = []
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
future_to_subtask = {
executor.submit(run_worker, st): st for st in subtasks
}
for future in as_completed(future_to_subtask, timeout=self.timeout_seconds):
try:
results.append(future.result())
except TimeoutError:
subtask = future_to_subtask[future]
logger.warning("Worker %s timed out after %.0fs", subtask.id, self.timeout_seconds)
results.append(WorkerResult(
subtask_id=subtask.id, success=False, output=None,
error=f"Timed out after {self.timeout_seconds}s",
))
return results
def synthesize_results(self, subtasks: list[Subtask], results: list[WorkerResult]) -> dict[str, Any]:
"""Merge worker outputs into a coherent final answer.
Deduplicates overlapping findings, resolves contradictions via
confidence-weighted voting, and produces the unified result.
Args:
subtasks: Original task definitions for context.
results: Completed (or failed) worker results.
Returns:
Dict with 'success', 'summary', 'failed_subtasks', and 'final_output'.
"""
successful = [r for r in results if r.success]
failed = [r for r in results if not r.success]
if not successful:
return {
"success": False,
"summary": "All worker tasks failed.",
"failed_subtasks": [r.subtask_id for r in failed],
"final_output": {},
}
# Collect all raw outputs for synthesis
raw_outputs = []
for r in successful:
if r.output and "raw_response" in r.output:
try:
import json as _json
parsed = _json.loads(r.output["raw_response"])
raw_outputs.append(parsed)
except (json.JSONDecodeError, TypeError):
raw_outputs.append({"text": r.output["raw_response"]})
# Synthesize via LLM — merge fragmented results into coherent output
synthesis_prompt = (
"Merge the following partial results from parallel worker agents "
"into a single coherent report. Deduplicate overlapping findings, "
"resolve contradictions using the highest-confidence answer, and "
"produce a unified output.\n\n"
+ "\n---\n".join(str(o) for o in raw_outputs)
)
synthesis_response = self.llm_client.generate(synthesis_prompt, max_tokens=4096)
return {
"success": len(failed) == 0,
"summary": f"Completed {len(successful)}/{len(subtasks)} tasks successfully.",
"failed_subtasks": [r.subtask_id for r in failed],
"final_output": {"synthesis": synthesis_response},
}
def run(self, goal: str) -> dict[str, Any]:
"""Full supervisor workflow: decompose → delegate → synthesize.
Args:
goal: The high-level objective to accomplish.
Returns:
Dict containing the synthesized final output and metadata.
"""
subtasks = self.decompose_task(goal)
if len(subtasks) < 1 or len(subtasks) > 8:
raise ValueError(f"Decomposition produced {len(subtasks)} subtasks — expected 1-8")
logger.info("Supervisor decomposed goal into %d subtasks", len(subtasks))
results = self.execute_workers(subtasks)
return self.synthesize_results(subtasks, results)
Pattern 2: Sequential Pipeline with Guardrails
Agents execute in strict order. Each stage validates the prior stage's output via Pydantic and applies quality gates before passing to the next stage. Use this when output fidelity matters more than latency.
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
class PipelineStageStatus(str, Enum):
PENDING = "pending"
RUNNING = "running"
PASSED = "passed"
FAILED = "failed"
RETRYABLE = "retryable"
@dataclass
class QualityGate:
"""Defines a validation check that must pass before advancing to the next pipeline stage."""
name: str
threshold: float # Minimum acceptable score (0.0-1.0)
is_mandatory: bool = True
def evaluate(self, output: dict[str, Any]) -> tuple[bool, float]:
"""Run quality check on output. Returns (passed, score)."""
raise NotImplementedError
class LLMQualityGate(QualityGate):
"""Uses an LLM to rate the quality of a stage's output."""
def __init__(self, llm_client: Any, rating_criteria: str) -> None:
super().__init__(name="llm_quality_check", threshold=0.7)
self.llm_client = llm_client
self.rating_criteria = rating_criteria
def evaluate(self, output: dict[str, Any]) -> tuple[bool, float]:
prompt = (
f"Rate the quality of this output on a scale of 0.0 to 1.0 based on: {self.rating_criteria}\n\n"
f"Output: {output.get('content', str(output))}"
)
response = self.llm_client.generate(prompt, max_tokens=256)
try:
score = float(response.strip())
return (score >= self.threshold, score)
except ValueError:
# Fallback: parse embedded score from response
import re
match = re.search(r"(\d+\.?\d*)", response)
if match:
score = min(float(match.group(1)), 1.0)
return (score >= self.threshold, score)
return (False, 0.0)
class SequentialPipeline:
"""Runs agents in strict order with typed message passing and quality gates.
Each stage must pass its quality gate before the next stage begins.
Failed stages can be retried (up to max_retries) or cause pipeline failure.
"""
def __init__(
self,
llm_client: Any,
max_retries: int = 2,
raise_on_failure: bool = True,
) -> None:
self.llm_client = llm_client
self.max_retries = max_retries
self.raise_on_failure = raise_on_failure
self._stages: list[tuple[AgentSpec, QualityGate | None]] = []
def add_stage(
self,
spec: AgentSpec,
quality_gate: QualityGate | None = None,
) -> "SequentialPipeline":
"""Register a pipeline stage with its agent spec and optional quality gate.
Args:
spec: The agent specification for this stage.
quality_gate: Optional validation to run after execution.
Returns:
self for method chaining.
"""
self._stages.append((spec, quality_gate))
return self
def execute_stage(self, stage_index: int, input_data: dict[str, Any]) -> tuple[dict[str, Any], PipelineStageStatus]:
"""Execute a single pipeline stage with retry logic.
Args:
stage_index: Which stage to execute (0-based).
input_data: Input for this stage.
Returns:
Tuple of (output, status) where status indicates pass/fail/retryable.
"""
spec, quality_gate = self._stages[stage_index]
retries = 0
while retries <= self.max_retries:
prompt = f"{spec.system_prompt}\n\nInput data: {input_data}"
try:
raw_output = self.llm_client.generate(prompt, max_tokens=spec.max_tokens)
output = {"content": raw_output, "agent_role": spec.role.value}
# Run quality gate if defined for this stage
if quality_gate:
passed, score = quality_gate.evaluate(output)
if not passed:
if retries < self.max_retries:
retries += 1
output["quality_feedback"] = (
f"Quality gate '{quality_gate.name}' scored {score:.2f} "
f"(threshold: {quality_gate.threshold}). Retry attempt {retries}."
)
continue
return output, PipelineStageStatus.FAILED
return output, PipelineStageStatus.PASSED
except Exception as exc:
retries += 1
if retries > self.max_retries:
return {"error": str(exc), "agent_role": spec.role.value}, PipelineStageStatus.FAILED
continue
def run(self, initial_input: dict[str, Any]) -> dict[str, Any]:
"""Execute the full sequential pipeline.
Args:
initial_input: Input data for the first stage.
Returns:
Dict with 'success', 'stage_results' (list of per-stage outputs),
and 'final_output' (output from the last stage).
"""
stage_results: list[dict[str, Any]] = []
current_input = initial_input.copy()
for idx, (spec, _) in enumerate(self._stages):
output, status = self.execute_stage(idx, current_input)
stage_results.append({"stage": spec.role.value, "status": status.value, "output": output})
if status == PipelineStageStatus.FAILED:
return {
"success": False,
"stage_results": stage_results,
"final_output": {},
"failure_stage": spec.role.value,
}
# Pass output content as input to next stage
current_input = {"previous_output": output.get("content", str(output)), "raw_output": output}
return {
"success": True,
"stage_results": stage_results,
"final_output": stage_results[-1]["output"],
}
# --- Usage example for Pattern 2 ---
def build_research_pipeline(llm_client: Any) -> SequentialPipeline:
"""Build a 4-stage research pipeline with quality gates."""
pipeline = SequentialPipeline(llm_client=llm_client, max_retries=1)
pipeline.add_stage(
spec=RESEARCHER_SPEC,
quality_gate=QualityGate(name="has_citations", threshold=0.5),
)
pipeline.add_stage(
spec=ANALYST_SPEC,
quality_gate=LLMQualityGate(llm_client, "Conclusions must have confidence scores and cite evidence"),
)
pipeline.add_stage(
spec=WRITER_SPEC,
quality_gate=QualityGate(name="has_summary", threshold=0.6),
)
return pipeline
Pattern 3: Group Chat Manager with LLM Speaker Selection
A group chat pattern where an LLM acts as moderator to select which agent speaks next based on conversation context. Use this for open-ended tasks where the order of contributions is not predetermined. Includes termination conditions to prevent infinite loops.
from typing import Literal
MessageRole = Literal["user", "assistant", "system"]
@dataclass
class ChatMessage:
"""A single message in a group chat."""
role: str # "agent:<role>" or "moderator" or "user"
content: str
sender_role: AgentRole | None = None
turn_number: int = 0
class GroupChatManager:
"""LLM-powered speaker selection for multi-agent group chat.
The manager evaluates conversation context and selects the most relevant
agent to speak next, then routes their response back into the conversation.
Supports termination conditions based on user goals or turn limits.
"""
def __init__(
self,
llm_client: Any,
agents: dict[AgentRole, AgentSpec],
max_turns: int = 15,
min_turns: int = 3,
) -> None:
self.llm_client = llm_client
self.agents = agents
self.max_turns = max_turns
self.min_turns = min_turns
self.messages: list[ChatMessage] = []
self.turn_count: int = 0
def add_message(self, message: ChatMessage) -> None:
"""Append a message to the conversation history."""
self.messages.append(message)
if message.sender_role:
self.turn_count += 1
def select_next_speaker(self, context: dict[str, Any] | None = None) -> AgentRole:
"""Ask the LLM which agent should speak next based on conversation context.
Args:
context: Optional additional context (e.g., user goal summary).
Returns:
The AgentRole selected to speak next.
Raises:
ValueError: If no suitable speaker is found or turn limit exceeded.
"""
# Build conversation summary for the speaker selection prompt
recent_messages = self.messages[-6:] # Last 3 turns (each agent sends 1-2 messages)
history_str = "\n".join(
f"[{msg.role}] {msg.content}" for msg in recent_messages
)
available_roles = list(self.agents.keys())
last_speaker = self.messages[-1].sender_role if self.messages else None
prompt = (
"Select which agent should speak next to make the most progress.\n\n"
f"Available agents: {', '.join(r.value for r in available_roles)}\n"
f"Last speaker: {last_speaker.value if last_speaker else 'none'}\n"
f"Conversation history:\n{history_str}\n\n"
)
if context and context.get("goal"):
prompt += f"User goal: {context['goal']}\n"
if context and context.get("termination_check") == True:
prompt += (
"Check if the user's goal has been fully addressed. If yes, respond with 'MODERATOR_END'. "
"Otherwise, select the next agent."
)
prompt += "\nRespond with ONLY the agent role name (e.g., analyst) or 'END'."
response = self.llm_client.generate(prompt, max_tokens=64).strip().lower()
# Parse speaker selection
if "end" in response or "moderator_end" in response:
if self.turn_count >= self.min_turns:
raise StopIteration("Conversation terminated by moderator")
else:
# Force at least min_turns before allowing termination
return available_roles[self.turn_count % len(available_roles)]
# Map response back to AgentRole
role_mapping = {r.value: r for r in available_roles}
if response in role_mapping:
selected = role_mapping[response]
if selected != last_speaker:
return selected
# Avoid repeating the same agent — pick next
idx = (available_roles.index(last_speaker) + 1) % len(available_roles)
return available_roles[idx]
raise ValueError(f"Unable to parse speaker selection response: {response!r}")
def run_chat(self, user_goal: str) -> list[ChatMessage]:
"""Execute the group chat until termination or turn limit.
Args:
user_goal: The user's original request that agents work toward.
Returns:
Complete conversation history as a list of ChatMessages.
Raises:
StopIteration: When moderator decides to end the chat.
ValueError: On parsing errors or invalid state.
"""
self.messages = [ChatMessage(
role="user", content=user_goal, turn_number=0
)]
context = {"goal": user_goal}
while self.turn_count < self.max_turns:
# Check termination condition periodically
if self.turn_count >= 5 and self.turn_count % 3 == 0:
try:
self.select_next_speaker(context={**context, "termination_check": True})
return self.messages # Chat ended early — goal satisfied
except StopIteration:
return self.messages
# Select speaker
try:
next_role = self.select_next_speaker(context)
except ValueError as exc:
raise RuntimeError(f"Speaker selection failed at turn {self.turn_count}: {exc}") from exc
# Route the message to the selected agent
agent_spec = self.agents[next_role]
chat_context = "\n".join(
f"[{msg.role}] {msg.content}" for msg in self.messages[-4:]
)
agent_prompt = (
f"{agent_spec.system_prompt}\n\n"
f"You are participating in a group chat. Your role is to contribute "
f"your expertise on this topic.\n\n"
f"Recent conversation:\n{chat_context}\n\n"
f"Please provide your contribution."
)
response = self.llm_client.generate(agent_prompt, max_tokens=2048)
message = ChatMessage(
role=f"agent:{next_role.value}",
content=response,
sender_role=next_role,
turn_number=self.turn_count + 1,
)
self.add_message(message)
raise RuntimeError(f"Reached maximum turns ({self.max_turns}) without termination")
# --- Usage example for Pattern 3 ---
def run_group_chat(llm_client: Any, user_goal: str) -> list[ChatMessage]:
"""Run a group chat with researcher, analyst, and writer agents."""
chat = GroupChatManager(
llm_client=llm_client,
agents={
AgentRole.RESEARCHER: RESEARCHER_SPEC,
AgentRole.ANALYST: ANALYST_SPEC,
AgentRole.WRITER: WRITER_SPEC,
},
max_turns=12,
min_turns=3,
)
try:
return chat.run_chat(user_goal)
except StopIteration as e:
# Chat ended successfully via moderator termination
print(f"Chat ended: {e}")
return chat.messages
# --- BAD vs GOOD Example ---
# ❌ BAD: No speaker selection logic — agents interrupt each other randomly
def bad_group_chat(llm_client: Any, agents: list[AgentSpec]) -> None:
"""Bad example: All agents are called in a fixed order with no intelligence.
This wastes turns and produces incoherent results."""
for agent in agents:
# Each agent gets the full raw prompt — no context from previous agents
llm_client.generate(agent.system_prompt, max_tokens=1024)
# ✅ GOOD: LLM selects speaker based on conversation context
def good_group_chat(llm_client: Any, user_goal: str) -> None:
"""Good example: Moderated group chat with intelligent speaker selection.
The moderator evaluates who adds the most value at each turn."""
results = run_group_chat(llm_client, user_goal)
# Moderator synthesizes final answer from conversation
final_prompt = (
"Synthesize the following group chat discussion into a final answer.\n\n"
+ "\n".join(f"[{m.role}] {m.content}" for m in results)
)
llm_client.generate(final_prompt, max_tokens=2048)
Constraints
MUST DO
- Enforce typed contracts at every agent boundary — Use Pydantic models to validate inputs and outputs between agents. Never pass raw strings across stage boundaries in production code.
- Set hard limits on turns and tokens — Always configure
max_tokensper agent,max_turnsfor group chats, and timeouts for parallel execution to prevent runaway costs. - Implement a quality gate at each pipeline stage — Use either rule-based checks (field presence, type validation) or LLM-based rating gates to catch degraded outputs before they propagate downstream.
- Deduplicate results in the synthesizer — When merging outputs from multiple workers, detect and remove duplicate findings, keeping only the highest-confidence version of each fact.
- Log agent execution metadata — Record latency, success/failure status, token counts, and quality gate scores for every stage to enable post-hoc analysis and pipeline tuning.
MUST NOT DO
- Never let an agent access tools outside its declared scope — The
allowed_toolslist is a security boundary, not a suggestion. Validate tool access at runtime by checking against the spec. - Never allow infinite loops in group chat — Always enforce
max_turnsand a minimum turn threshold before allowing moderator termination. Never rely on the LLM to self-terminate without safeguards. - Never skip output validation between pipeline stages — Passing unchecked outputs downstream is the primary cause of cascading failures. If a stage produces invalid output, retry or fail fast — never let it propagate.
- Never hardcode agent selection order in supervisor-worker patterns — Use LLM-powered decomposition to create subtasks dynamically based on the specific goal. Static decomposition misses opportunities for parallelization and wastes resources.
- Never use more than 8 agents in a single orchestration — Beyond this threshold, coordination overhead dominates useful work, token costs explode, and failure rates compound unmanageably.
Related Skills
| Skill | Purpose |
|---|---|
tool-use-function-calling |
Each agent needs tools to perform its role. This skill covers how agents discover, call, and parse tool outputs. |
planning-reasoning |
Before orchestrating agents, the supervisor must plan — decompose goals, identify dependencies, and estimate complexity. |
memory-systems |
Multi-agent systems often need shared memory or persistent state across agent interactions (conversation history, cached results). |
Output Template
When this skill is active, the model's output should contain:
- Agent Definitions — Typed specs for each agent with role, tools, and constraints
- Orchestration Code — Complete implementation of the selected pattern (pipeline, supervisor-worker, or group chat)
- Quality Gates — Validation logic at every inter-agent boundary
- Synthesis Logic — How results are merged, deduplicated, and conflict-resolved
- Error Handling — Retry strategies, timeout policies, and failure modes
# Example output structure
orchestrator = create_supervisor(llm_client, max_workers=4)
result = orchestrator.run("Analyze the competitive landscape for AI coding assistants")
# → {success: bool, summary: str, failed_subtasks: list[str], final_output: dict}
Live References
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.