multi-agent-collaboration

# Multi-Agent Collaboration Pattern Orchestrates multiple specialized agents working in concert to decompose and solve complex, multi-domain problems that exceed any single agent's capabilities. Loading this skill makes the model design, implement, and validate agent topologies — hierarchical delegation, parallel execution, sequential handoffs, debate/consensus, critic-reviewer loops, and agent-as-tool patterns — choosing the right structure based on task dependencies, latency requirements, and output coherence needs. ## TL;DR Checklist - [ ] Decompose the problem into independent or weakly-coupled sub-tasks before defining agents - [ ] Assign each agent a narrow, non-overlapping role with explicit tools and goals - [ ] Select an orchestration topology (sequential / parallel / hierarchical / debate) based on task dependency graph - [ ] Implement typed inter-agent contracts (Pydantic models) at every communication boundary - [ ] Add quality gates between pipeline stages to catch degraded outputs before they propagate - [ ] Set hard turn limits, token caps, and timeouts for all collaboration loops - [ ] Build a result synthesizer that merges, deduplicates, and resolves conflicts from multiple agents --- ## When to Use Use this skill when: - A task requires multiple distinct domains of expertise (e.g., research + analysis + writing) that no single agent can perform well in isolation - You need parallel processing across independent sub-tasks with result synthesis at the end - The problem benefits from debate or consensus — agents with varied perspectives evaluating options before converging on a decision - A hierarchical structure would help: a manager agent delegating to worker agents based on their tool access or plugin capabilities - An expert team is needed (researcher, writer, editor, reviewer) collaborating to produce a complex output like a report, codebase, or creative campaign - Sequential handoffs are natural — one agent's output becomes the next agent's input in a multi-stage pipeline - Customer support escalation flows require routing from front-line agents to specialists based on problem complexity - You need a critic-reviewer loop where one agent creates and another critically assesses for correctness, compliance, quality, or security --- ## When NOT to Use Avoid this skill for: - **Single-domain tasks** — If one well-crafted prompt handles the problem, adding agents adds overhead without benefit (use single-agent pattern instead) - **Strictly sequential micro-tasks** — Simple two-step transforms are better handled by chaining prompts rather than defining full agent roles - **Real-time latency-sensitive systems** — Multi-agent orchestration adds round-trip costs; each inter-agent communication is an LLM call - **Tasks with no clear decomposition boundary** — If sub-tasks cannot be defined independently, agents will fight over context and produce incoherent results - **Budget-constrained one-shot queries** — Every additional agent multiplies token costs; only justify when the quality gain outweighs the cost --- ## Core Workflow ### 1. Decompose the Problem into Sub-Tasks Analyze the problem statement and identify natural boundaries where different expertise, tools, or data sources apply. Break the objective into discrete sub-problems, each solvable by a single specialized agent. Avoid over-decomposition (too many tiny tasks) and under-decomposition (tasks still too broad for one agent). Target 2–6 sub-tasks per orchestration. **Checkpoint:** Every sub-task must have a clearly defined input source, output contract, and required tool access. If a sub-task's boundary is fuzzy, merge it with an adjacent task. ### 2. Define Agent Roles with Bounded Responsibilities For each sub-task, define an agent with a narrow role, explicit tools, and a bounded goal. Each agent must have a distinct scope — no two agents should claim the same responsibility. Assign each agent a system prompt that encodes its persona, constraints, and output format. **Checkpoint:** Verify that no two agents overlap in responsibility. Run a "who does this" matrix: for every aspect of the problem, exactly one agent owns it. ### 3. Select Orchestration Topology Choose the interaction model based on task dependencies: | Topology | When to Use | Coordination Model | |----------|-------------|---------------------| | **Sequential Handoffs** | Strict ordering; each stage depends on prior output | Pipeline: A → B → C | | **Parallel Processing** | Independent sub-tasks that merge at end | Fan-out: A + B + C → Merge | | **Hierarchical (Parent-Child)** | Complex goals decomposable by a coordinator | Manager delegates to workers | | **Debate / Consensus** | Multiple perspectives needed before decision | Agents argue, converge on agreement | | **Agent-as-Tool** | One agent needs another as a callable capability | Agent calls sub-agent via tool wrapper | | **Expert Team with Critic-Reviewer** | Output quality must be validated iteratively | Creator → Critic → Revise | **Checkpoint:** Confirm that your topology matches the dependency graph. Sequential for dependencies, parallel for independence, hierarchical for decomposability. ### 4. Implement Inter-Agent Communication Contracts Define typed contracts at every agent boundary using structured output schemas (e.g., Pydantic models). Each agent's output must be parseable by its consumer. Use shared state mechanisms (session state, message queues) or explicit context passing to transfer data between agents. For Google ADK, use `output_key` for simple text responses or `EventActions.state_delta` for complex multi-key updates. **Checkpoint:** Every inter-agent boundary has a typed contract defined and validated. Unparsed outputs cause pipeline failure — never let raw strings cross stage boundaries in production. ### 5. Add Quality Gates and Retry Logic Insert validation checkpoints between stages. Rule-based gates check field presence, type compliance, and schema validity. LLM-based gates rate output quality against criteria. Failed gates trigger retry logic (bounded to max_retries) before escalating to pipeline failure. This prevents degraded outputs from propagating downstream. **Checkpoint:** Every pipeline stage has at least one gate defined. Configure `max_retries ≥ 1` for non-critical stages and `max_retries = 0` for security/compliance gates that must fail fast. ### 6. Synthesize Final Output Merge results from all agents into a coherent final answer. Deduplicate overlapping findings, resolve contradictions using confidence-weighted voting or LLM-based arbitration, and produce a unified output. The synthesizer is the final quality filter — it turns fragmented agent outputs into a single coherent result that meets the original objective. **Checkpoint:** Run the synthesized output against the original objective. If it doesn't fully address the goal, loop back to re-decompose or add specialist agents for missing coverage. --- ## Implementation Patterns ### Pattern 1: Sequential Handoffs (Pipeline) Agents execute in strict order where each stage's output becomes the next stage's input. Use this when output fidelity matters more than latency and task dependencies are explicit. This mirrors the Planning pattern but explicitly involves different agents per stage. ```python from pydantic import BaseModel, field_validator from enum import Enum from typing import Any class PipelineStageStatus(str, Enum): PENDING = "pending" RUNNING = "running" PASSED = "passed" FAILED = "failed" RETRYABLE = "retryable" 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[dict[str, Any], Any | None]] = [] def add_stage( self, agent_spec: dict[str, Any], quality_gate: Any | None = None, ) -> "SequentialPipeline": """Register a pipeline stage with agent spec and optional quality gate. Args: agent_spec: Dict with keys 'role', 'system_prompt', 'max_tokens'. quality_gate: Optional callable(output) -> (passed: bool, score: float). Returns: self for method chaining. """ self._stages.append((agent_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_dict, status_enum). """ 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.get("max_tokens", 4096) ) output = { "content": raw_output, "agent_role": spec["role"], } # Run quality gate if defined for this stage if quality_gate: passed, score = quality_gate(output) if not passed: if retries < self.max_retries: retries += 1 output["quality_feedback"] = ( f"Quality gate scored {score:.2f} " f"(threshold: 0.7). 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"]}, 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', and 'final_output'. """ 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"], "status": status.value, "output": output, }) if status == PipelineStageStatus.FAILED: return { "success": False, "stage_results": stage_results, "final_output": {}, "failure_stage": spec["role"], } 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"], } ``` **CrewAI Sequential Example (from Chapter 7):** ```python import os from dotenv import load_dotenv from crewai import Agent, Task, Crew, Process from langchain_google_genai import ChatGoogleGenerativeAI def build_blog_crew() -> Crew: """Create a sequential Crew for AI trend blog post creation. Researcher finds trends → Writer composes post based on research. Uses Process.sequential to guarantee order of execution. """ load_dotenv() if not os.getenv("GOOGLE_API_KEY"): raise ValueError("GOOGLE_API_KEY not set in .env") llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash") researcher = Agent( role="Senior Research Analyst", goal="Find and summarize the latest trends in AI.", backstory=( "You are an experienced research analyst with a knack for " "identifying key trends and synthesizing information." ), verbose=True, allow_delegation=False, ) writer = Agent( role="Technical Content Writer", goal="Write a clear and engaging blog post based on research findings.", backstory=( "You are a skilled writer who can translate complex technical " "topics into accessible content." ), verbose=True, allow_delegation=False, ) research_task = Task( description=( "Research the top 3 emerging trends in Artificial Intelligence. " "Focus on practical applications and potential impact." ), expected_output=( "A detailed summary of the top 3 AI trends, including key " "points and sources." ), agent=researcher, ) writing_task = Task( description=( "Write a 500-word blog post based on the research findings. " "The post should be engaging and easy for a general audience." ), expected_output="A complete 500-word blog post about the latest AI trends.", agent=writer, context=[research_task], # Writer sees Researcher's output ) return Crew( agents=[researcher, writer], tasks=[research_task, writing_task], process=Process.sequential, llm=llm, verbose=2, ) ``` ### Pattern 2: Hierarchical Parent-Child Delegation (Google ADK) A coordinator agent delegates to specialized sub-agents based on its instructions. This creates a multi-layered organizational structure where higher-level agents oversee lower-level ones — well-suited for problems decomposable into sub-problems managed by specific layers. ```python from google.adk.agents import LlmAgent, BaseAgent from google.adk.agents.invocation_context import InvocationContext from google.adk.events import Event from typing import AsyncGenerator class TaskExecutor(BaseAgent): """A custom non-LLM agent with predefined task behavior. Extends BaseAgent to implement deterministic or tool-based tasks that don't require LLM inference (e.g., data validation, file ops). """ name: str = "TaskExecutor" description: str = "Executes a predefined task without LLM involvement." async def _run_async_impl( self, context: InvocationContext ) -> AsyncGenerator[Event, None]: """Custom implementation logic for the task.""" yield Event(author=self.name, content="Task finished successfully.") def build_hierarchical_team() -> LlmAgent: """Create a parent coordinator with delegated sub-agents. The coordinator routes requests to Greeter or TaskExecutor based on the nature of the incoming request. """ # Child agent 1: Handles greeting/friendly interactions greeter = LlmAgent( name="Greeter", model="gemini-2.0-flash-exp", instruction="You are a friendly greeter.", ) # Child agent 2: Handles deterministic task execution task_doer = TaskExecutor() # Parent coordinator: delegates to children based on intent coordinator = LlmAgent( name="Coordinator", model="gemini-2.0-flash-exp", description=( "A coordinator that can greet users and execute tasks." ), instruction=( "When asked to greet, delegate to the Greeter. " "When asked to perform a task, delegate to the TaskExecutor." ), sub_agents=[greeter, task_doer], ) # ADK automatically establishes parent-child relationships assert greeter.parent_agent == coordinator assert task_doer.parent_agent == coordinator return coordinator # --- BAD vs GOOD Example --- # ❌ BAD: No hierarchy — all agents called flatly with no delegation def bad_flat_agents(llm_client: Any) -> None: """Bad example: Define agents but call them without a coordinator. Each agent operates in isolation with no shared context or delegation.""" pass # No orchestration logic — agents are disconnected # ✅ GOOD: Coordinator routes to specialists via parent-child hierarchy def good_hierarchical_agents() -> LlmAgent: """Good example: A coordinator agent delegates to specialized children. The parent manages task routing, reducing per-agent complexity.""" return build_hierarchical_team() ``` ### Pattern 3: LoopAgent for Iterative Workflows Use LoopAgent when a process must repeat until a termination condition is met (e.g., processing steps that iterate until status="completed", polling operations, or iterative refinement). Combines an LlmAgent with a custom ConditionChecker that evaluates session state to decide whether to continue. ```python import asyncio from typing import AsyncGenerator from google.adk.agents import LoopAgent, LlmAgent, BaseAgent from google.adk.events import Event, EventActions from google.adk.agents.invocation_context import InvocationContext class ConditionChecker(BaseAgent): """Custom agent that checks session state for a 'completed' status. Yields an escalation event to terminate the loop when done, or a continuation event otherwise. """ name: str = "ConditionChecker" description: ( "Checks if a process is complete and signals the loop to stop." ) async def _run_async_impl( self, context: InvocationContext ) -> AsyncGenerator[Event, None]: """Check state and yield an event to continue or stop the loop.""" status = context.session.state.get("status", "pending") is_done = (status == "completed") if is_done: # Escalate to terminate the loop when condition is met yield Event(author=self.name, actions=EventActions(escalate=True)) else: # Yield a simple event to continue the loop yield Event( author=self.name, content="Condition not met, continuing loop.", ) def build_iterative_pipeline(max_iterations: int = 10) -> LoopAgent: """Create an iterative pipeline that repeats until status=completed. Each iteration runs ProcessingStep (LLM-driven task) followed by ConditionChecker (state-based termination decision). Max iterations serves as a safety bound against infinite loops. """ # LLM agent performs one step of the overall process processing_step = LlmAgent( name="ProcessingStep", model="gemini-2.0-flash-exp", instruction=( "You are a step in a longer process. Perform your task. "