Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
State management patterns for multi-agent workflows
Error recovery and retry strategies
Context window budget management
Cost optimization strategies per platform
Platform-specific configs: Claude Code Agent Teams, OpenClaw, CrewAI, AutoGen
When to Use
Building a multi-step AI pipeline that exceeds one agent's context capacity
Parallelizing research, generation, or analysis tasks for speed
Creating specialist agents with defined roles and handoff contracts
Designing fault-tolerant AI workflows for production
Pattern Selection Guide
Is the task sequential (each step needs previous output)?
YES → Sequential Pipeline
NO → Can tasks run in parallel?
YES → Parallel Fan-out/Fan-in
NO → Is there a hierarchy of decisions?
YES → Hierarchical Delegation
NO → Is it event-triggered?
YES → Event-Driven
NO → Need consensus/validation?
YES → Consensus Pattern
Pattern 1: Sequential Pipeline
Use when: Each step depends on the previous output. Research → Draft → Review → Polish.
"You are a research specialist. Given a topic, produce a structured research brief with: key facts, statistics, expert perspectives, and controversy points."
"input"
"research"
"writer"
"You are a senior content writer. Using the research provided, write a compelling 800-word blog post with a clear hook, 3 main sections, and a strong CTA."
"research"
"draft"
"editor"
"You are a copy editor. Review the draft for: clarity, flow, grammar, and SEO. Return the improved version only, no commentary."
"draft"
"final"
Pattern 2: Parallel Fan-out / Fan-in
Use when: Independent tasks that can run concurrently. Research 5 competitors simultaneously.
# parallel_fanout.pyimport asyncio
import os
import anthropic
from typing importAnyasyncdefrun_agent(client, task_name: str, system: str, user: str, model: str | None = None) -> dict:
"""Single async agent call"""
model = model or os.environ["ANTHROPIC_MODEL"]
loop = asyncio.get_event_loop()
def_call():
return client.messages.create(
model=model,
max_tokens=2048,
system=system,
messages=[{"role": "user", "content": user}],
)
response = await loop.run_in_executor(None, _call)
return {
"task": task_name,
"output": response.content[0].text,
"tokens": response.usage.input_tokens + response.usage.output_tokens,
}
asyncdefparallel_research(competitors: list[str], research_type: str) -> dict:
"""Fan-out: research all competitors in parallel. Fan-in: synthesize results."""
client = anthropic.Anthropic()
# FAN-OUT: spawn parallel agent calls
tasks = [
run_agent(
client,
task_name=competitor,
system=f"You are a competitive intelligence analyst. Research {competitor} and provide: pricing, key features, target market, and known weaknesses.",
user=f"Analyze {competitor} for comparison with our product in the {research_type} market.",
)
for competitor in competitors
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Handle failures gracefully
successful = [r for r in results ifnotisinstance(r, Exception)]
failed = [r for r in results ifisinstance(r, Exception)]
if failed:
print(f"Warning: {len(failed)} research tasks failed: {failed}")
# FAN-IN: synthesize
combined_research = "\n\n".join([
f"## {r['task']}\n{r['output']}"for r in successful
])
synthesis = await run_agent(
client,
task_name="synthesizer",
system="You are a strategic analyst. Synthesize competitor research into a concise comparison matrix and strategic recommendations.",
user=f"Synthesize these competitor analyses:\n\n{combined_research}",
model=os.environ["ANTHROPIC_MODEL"],
)
return {
"individual_analyses": successful,
"synthesis": synthesis["output"],
"total_tokens": sum(r["tokens"] for r in successful) + synthesis["tokens"],
}
Pattern 3: Hierarchical Delegation
Use when: Complex tasks with subtask discovery. Orchestrator breaks down work, delegates to specialists.
# hierarchical_delegation.pyimport json
import os
import anthropic
ORCHESTRATOR_SYSTEM = """You are an orchestration agent. Your job is to:
1. Analyze the user's request
2. Break it into subtasks
3. Assign each to the appropriate specialist agent
4. Collect results and synthesize
Available specialists:
- researcher: finds facts, data, and information
- writer: creates content and documents
- coder: writes and reviews code
- analyst: analyzes data and produces insights
Respond with a JSON plan:
{
"subtasks": [
{"id": "1", "agent": "researcher", "task": "...", "depends_on": []},
{"id": "2", "agent": "writer", "task": "...", "depends_on": ["1"]}
]
}"""
SPECIALIST_SYSTEMS = {
"researcher": "You are a research specialist. Find accurate, relevant information and cite sources when possible.",
"writer": "You are a professional writer. Create clear, engaging content in the requested format.",
"coder": "You are a senior software engineer. Write clean, well-commented code with error handling.",
"analyst": "You are a data analyst. Provide structured analysis with evidence-backed conclusions.",
}
classHierarchicalOrchestrator:
def__init__(self):
self.client = anthropic.Anthropic()
defrun(self, user_request: str) -> str:
# 1. Orchestrator creates plan
plan_response = self.client.messages.create(
model=os.environ["ANTHROPIC_MODEL"],
max_tokens=1024,
system=ORCHESTRATOR_SYSTEM,
messages=[{"role": "user", "content": user_request}],
)
plan = json.loads(plan_response.content[0].text)
results = {}
# 2. Execute subtasks respecting dependenciesfor subtask inself._topological_sort(plan["subtasks"]):
context = self._build_context(subtask, results)
specialist = SPECIALIST_SYSTEMS[subtask["agent"]]
result = self.client.messages.create(
model=os.environ["ANTHROPIC_MODEL"],
max_tokens=2048,
system=specialist,
messages=[{"role": "user", "content": f"{context}\n\nTask: {subtask['task']}"}],
)
results[subtask["id"]] = result.content[0].text
# 3. Final synthesis
all_results = "\n\n".join([f"### {k}\n{v}"for k, v in results.items()])
synthesis = self.client.messages.create(
model=os.environ["ANTHROPIC_MODEL"],
max_tokens=2048,
system="Synthesize the specialist outputs into a coherent final response.",
messages=[{"role": "user", "content": f"Original request: {user_request}\n\nSpecialist outputs:\n{all_results}"}],
)
return synthesis.content[0].text
def_build_context(self, subtask: dict, results: dict) -> str:
ifnot subtask.get("depends_on"):
return""
deps = [f"Output from task {dep}:\n{results[dep]}"for dep in subtask["depends_on"] if dep in results]
return"Previous results:\n" + "\n\n".join(deps) if deps else""def_topological_sort(self, subtasks: list) -> list:
# Simple ordered execution respecting depends_on
ordered, remaining = [], list(subtasks)
completed = set()
while remaining:
for task in remaining:
ifall(dep in completed for dep in task.get("depends_on", [])):
ordered.append(task)
completed.add(task["id"])
remaining.remove(task)
breakreturn ordered
Handoff Protocol Template
# Standard handoff context format — use between all agents@dataclassclassAgentHandoff:
"""Structured context passed between agents in a workflow."""
task_id: str
workflow_id: str
step_number: int
total_steps: int# What was done
previous_agent: str
previous_output: str
artifacts: dict# {"filename": "content"} for any files produced# What to do next
current_agent: str
current_task: str
constraints: list[str] # hard rules for this step# Metadata
context_budget_remaining: int# tokens left for this agent
cost_so_far_usd: floatdefto_prompt(self) -> str:
returnf"""
# Agent Handoff — Step {self.step_number}/{self.total_steps}
## Your Task
{self.current_task}
## Constraints
{chr(10).join(f'- {c}'for c in self.constraints)}
## Context from Previous Step ({self.previous_agent})
{self.previous_output[:2000]}{"... [truncated]"iflen(self.previous_output) > 2000else""}
## Context Budget
You have approximately {self.context_budget_remaining} tokens remaining. Be concise.
"""
Error Recovery Patterns
import os
import time
from functools import wraps
defwith_retry(max_attempts=3, backoff_seconds=2, fallback_model=None):
"""Decorator for agent calls with exponential backoff and model fallback."""defdecorator(fn):
@wraps(fn)defwrapper(*args, **kwargs):
last_error = Nonefor attempt inrange(max_attempts):
try:
return fn(*args, **kwargs)
except Exception as e:
last_error = e
if attempt < max_attempts - 1:
wait = backoff_seconds * (2 ** attempt)
print(f"Attempt {attempt+1} failed: {e}. Retrying in {wait}s...")
time.sleep(wait)
# Fall back to cheaper/faster model on rate limitif fallback_model and"rate_limit"instr(e).lower():
kwargs["model"] = fallback_model
raise last_error
return wrapper
return decorator
@with_retry(max_attempts=3, fallback_model=os.environ.get("ANTHROPIC_FALLBACK_MODEL"))defcall_agent(model, system, user):
...
Context Window Budgeting
# Budget context across a multi-step pipeline# Rule: never let any step consume more than 60% of remaining budgetclassContextBudget:
def__init__(self, total_context_tokens: int, reserve_pct: float = 0.2):
# Read the current limit from the selected provider's official model# documentation or API metadata; do not hard-code model generations here.
total = total_context_tokens
self.total = total
self.reserve = int(total * reserve_pct) # keep 20% as bufferself.used = 0 @propertydefremaining(self):
returnself.total - self.reserve - self.used
defallocate(self, step_name: str, requested: int) -> int:
allocated = min(requested, int(self.remaining * 0.6)) # max 60% of remainingprint(f"[Budget] {step_name}: allocated {allocated:,} tokens (remaining: {self.remaining:,})")
return allocated
defconsume(self, tokens_used: int):
self.used += tokens_used
deftruncate_to_budget(text: str, token_budget: int, chars_per_token: float = 4.0) -> str:
"""Rough truncation — use tiktoken for precision."""
char_budget = int(token_budget * chars_per_token)
iflen(text) <= char_budget:
return text
return text[:char_budget] + "\n\n[... truncated to fit context budget ...]"
Cost Optimization Strategies
Strategy
Savings
Tradeoff
Use Haiku for routing/classification
85-90%
Slightly less nuanced judgment
Cache repeated system prompts
50-90%
Requires prompt caching setup
Truncate intermediate outputs
20-40%
May lose detail in handoffs
Batch similar tasks
50%
Latency increases
Use Sonnet for most, Opus for final step only
60-70%
Final quality may improve
Short-circuit on confidence threshold
30-50%
Need confidence scoring
Common Pitfalls
Circular dependencies — agents calling each other in loops; enforce DAG structure at design time
Context bleed — passing entire previous output to every step; summarize or extract only what's needed
No timeout — a stuck agent blocks the whole pipeline; always set max_tokens and wall-clock timeouts
Silent failures — agent returns plausible but wrong output; add validation steps for critical paths
Ignoring cost — 10 parallel Opus calls is $0.50 per workflow; model selection is a cost decision
Over-orchestration — if a single prompt can do it, it should; only add agents when genuinely needed
