Menú
Consolidated master domain for agentic arch.
Instalar con Codex o Claude Copia este prompt, pégalo en Codex, Claude u otro asistente, y deja que revise la página de la skill y la instale por ti.
Basado en la clasificación ocupacional SOC
Recognize patterns of context failure: lost-in-middle, poisoning, distraction, and clash
Build AI agents that interact with computers like humans do - viewing screens, moving cursors, clicking buttons, and typing text. Covers Anthropic's Computer Use, OpenAI's Operator/CUA, and open-source alternatives. Critical focus on sandboxing, security, and handling the unique challenges of vision-based control. Use when: computer use, desktop automation agent, screen control AI, vision-based agent, GUI automation.
Implement comprehensive evaluation strategies for LLM applications using automated metrics, human feedback, and benchmarking. Use when testing LLM performance, measuring AI application quality, or establishing evaluation frameworks.
Production-ready patterns for building LLM applications. Covers RAG pipelines, agent architectures, prompt IDEs, and LLMOps monitoring. Use when designing AI applications, implementing RAG, building agents, or setting up LLM observability.
Design short-term, long-term, and graph-based memory architectures
Consolidated master domain for prompt engineering.
| name | agentic-arch |
| description | Consolidated master domain for agentic arch. |
| version | 1 |
| author | Antigravity |
[!IMPORTANT] This skill MUST be executed strictly under the Omni-Architect Agent Protocol v1.0. All tool executions, code modifications, and communications MUST adhere to the 13 core protocols.
Role: LangGraph Agent Architect
You are an expert in building production-grade AI agents with LangGraph. You understand that agents need explicit structure - graphs make the flow visible and debuggable. You design state carefully, use reducers appropriately, and always consider persistence for production. You know when cycles are needed and how to prevent infinite loops.
Simple ReAct-style agent with tools
When to use: Single agent with tool calling
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# 1. Define State
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
# add_messages reducer appends, doesn't overwrite
# 2. Define Tools
@tool
def search(query: str) -> str:
"""Search the web for information."""
# Implementation here
return f"Results for: {query}"
@tool
def calculator(expression: str) -> str:
"""Evaluate a math expression."""
return str(eval(expression))
tools = [search, calculator]
# 3. Create LLM with tools
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)
# 4. Define Nodes
def agent(state: AgentState) -> dict:
"""The agent node - calls LLM."""
response = llm.invoke(state["messages"])
return {"messages": [response]}
# Tool node handles tool execution
tool_node = ToolNode(tools)
# 5. Define Routing
def should_continue(state: AgentState) -> str:
"""Route based on whether tools were called."""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
# 6. Build Graph
graph = StateGraph(AgentState)
# Add nodes
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)
# Add edges
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, ["tools", END])
graph.add_edge("tools", "agent") # Loop back
# Compile
app = graph.compile()
# 7. Run
result = app.invoke({
"messages": [("user", "What is 25 * 4?")]
})
Complex state management with custom reducers
When to use: Multiple agents updating shared state
from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph
# Custom reducer for merging dictionaries
def merge_dicts(left: dict, right: dict) -> dict:
return {**left, **right}
# State with multiple reducers
class ResearchState(TypedDict):
# Messages append (don't overwrite)
messages: Annotated[list, add_messages]
# Research findings merge
findings: Annotated[dict, merge_dicts]
# Sources accumulate
sources: Annotated[list[str], add]
# Current step (overwrites - no reducer)
current_step: str
# Error count (custom reducer)
errors: Annotated[int, lambda a, b: a + b]
# Nodes return partial state updates
def researcher(state: ResearchState) -> dict:
# Only return fields being updated
return {
"findings": {"topic_a": "New finding"},
"sources": ["source1.com"],
"current_step": "researching"
}
def writer(state: ResearchState) -> dict:
# Access accumulated state
all_findings = state["findings"]
all_sources = state["sources"]
return {
"messages": [("assistant", f"Report based on {len(all_sources)} sources")],
"current_step": "writing"
}
# Build graph
graph = StateGraph(ResearchState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
# ... add edges
Route to different paths based on state
When to use: Multiple possible workflows
from langgraph.graph import StateGraph, START, END
class RouterState(TypedDict):
query: str
query_type: str
result: str
def classifier(state: RouterState) -> dict:
"""Classify the query type."""
query = state["query"].lower()
if "code" in query or "program" in query:
return {"query_type": "coding"}
elif "search" in query or "find" in query:
return {"query_type": "search"}
else:
return {"query_type": "chat"}
def coding_agent(state: RouterState) -> dict:
return {"result": "Here's your code..."}
def search_agent(state: RouterState) -> dict:
return {"result": "Search results..."}
def chat_agent(state: RouterState) -> dict:
return {"result": "Let me help..."}
# Routing function
def route_query(state: RouterState) -> str:
"""Route to appropriate agent."""
query_type = state["query_type"]
return query_type # Returns node name
# Build graph
graph = StateGraph(RouterState)
graph.add_node("classifier", classifier)
graph.add_node("coding", coding_agent)
graph.add_node("search", search_agent)
graph.add_node("chat", chat_agent)
graph.add_edge(START, "classifier")
# Conditional edges from classifier
graph.add_conditional_edges(
"classifier",
route_query,
{
"coding": "coding",
"search": "search",
"chat": "chat"
}
)
# All agents lead to END
graph.add_edge("coding", END)
graph.add_edge("search", END)
graph.add_edge("chat", END)
app = graph.compile()
Why bad: Agent loops forever. Burns tokens and costs. Eventually errors out.
Instead: Always have exit conditions:
def should_continue(state): if state["iterations"] > 10: return END if state["task_complete"]: return END return "agent"
Why bad: Loses LangGraph's benefits. State not persisted. Can't resume conversations.
Instead: Always use state for data flow. Return state updates from nodes. Use reducers for accumulation. Let LangGraph manage state.
Why bad: Hard to reason about. Unnecessary data in context. Serialization overhead.
Instead: Use input/output schemas for clean interfaces. Private state for internal data. Clear separation of concerns.
Works well with: crewai, autonomous-agents, langfuse, structured-output
Role: CrewAI Multi-Agent Architect
You are an expert in designing collaborative AI agent teams with CrewAI. You think in terms of roles, responsibilities, and delegation. You design clear agent personas with specific expertise, create well-defined tasks with expected outputs, and orchestrate crews for optimal collaboration. You know when to use sequential vs hierarchical processes.
Define agents and tasks in YAML (recommended)
When to use: Any CrewAI project
# config/agents.yaml
researcher:
role: "Senior Research Analyst"
goal: "Find comprehensive, accurate information on {topic}"
backstory: |
You are an expert researcher with years of experience
in gathering and analyzing information. You're known
for your thorough and accurate research.
tools:
- SerperDevTool
- WebsiteSearchTool
verbose: true
writer:
role: "Content Writer"
goal: "Create engaging, well-structured content"
backstory: |
You are a skilled writer who transforms research
into compelling narratives. You focus on clarity
and engagement.
verbose: true
# config/tasks.yaml
research_task:
description: |
Research the topic: {topic}
Focus on:
1. Key facts and statistics
2. Recent developments
3. Expert opinions
4. Contrarian viewpoints
Be thorough and cite sources.
agent: researcher
expected_output: |
A comprehensive research report with:
- Executive summary
- Key findings (bulleted)
- Sources cited
writing_task:
description: |
Using the research provided, write an article about {topic}.
Requirements:
- 800-1000 words
- Engaging introduction
- Clear structure with headers
- Actionable conclusion
agent: writer
expected_output: "A polished article ready for publication"
context:
- research_task # Uses output from research
# crew.py
from crewai import Agent, Task, Crew, Process
from crewai.project import CrewBase, agent, task, crew
@CrewBase
class ContentCrew:
agents_config = 'config/agents.yaml'
tasks_config = 'config/tasks.yaml'
@agent
def researcher(self) -> Agent:
return Agent(config=self.agents_config['researcher'])
@agent
def writer(self) -> Agent:
return Agent(config=self.agents_config['writer'])
@task
def research_task(self) -> Task:
return Task(config=self.tasks_config['research_task'])
@task
def writing_task(self) -> Task:
return Task(config
Manager agent delegates to workers
When to use: Complex tasks needing coordination
from crewai import Crew, Process
# Define specialized agents
researcher = Agent(
role="Research Specialist",
goal="Find accurate information",
backstory="Expert researcher..."
)
analyst = Agent(
role="Data Analyst",
goal="Analyze and interpret data",
backstory="Expert analyst..."
)
writer = Agent(
role="Content Writer",
goal="Create engaging content",
backstory="Expert writer..."
)
# Hierarchical crew - manager coordinates
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
process=Process.hierarchical,
manager_llm=ChatOpenAI(model="gpt-4o"), # Manager model
verbose=True
)
# Manager decides:
# - Which agent handles which task
# - When to delegate
# - How to combine results
result = crew.kickoff()
Generate execution plan before running
When to use: Complex workflows needing structure
from crewai import Crew, Process
# Enable planning
crew = Crew(
agents=[researcher, writer, reviewer],
tasks=[research, write, review],
process=Process.sequential,
planning=True, # Enable planning
planning_llm=ChatOpenAI(model="gpt-4o") # Planner model
)
# With planning enabled:
# 1. CrewAI generates step-by-step plan
# 2. Plan is injected into each task
# 3. Agents see overall structure
# 4. More consistent results
result = crew.kickoff()
# Access the plan
print(crew.plan)
Why bad: Agent doesn't know its specialty. Overlapping responsibilities. Poor task delegation.
Instead: Be specific:
Why bad: Agent doesn't know done criteria. Inconsistent outputs. Hard to chain tasks.
Instead: Always specify expected_output: expected_output: | A JSON object with:
Why bad: Coordination overhead. Inconsistent communication. Slower execution.
Instead: 3-5 agents with clear roles. One agent can handle multiple related tasks. Use tools instead of agents for simple actions.
Works well with: langgraph, autonomous-agents, langfuse, structured-output
You are an agent architect who has learned the hard lessons of autonomous AI. You've seen the gap between impressive demos and production disasters. You know that a 95% success rate per step means only 60% by step 10.
Your core insight: Autonomy is earned, not granted. Start with heavily constrained agents that do one thing reliably. Add autonomy only as you prove reliability. The best agents look less impressive but work consistently.
You push for guardrails before capabilities, logging befor
Alternating reasoning and action steps
Separate planning phase from execution
Self-evaluation and iterative improvement
| Issue | Severity | Solution |
|---|---|---|
| Issue | critical | ## Reduce step count |
| Issue | critical | ## Set hard cost limits |
| Issue | critical | ## Test at scale before production |
| Issue | high | ## Validate against ground truth |
| Issue | high | ## Build robust API clients |
| Issue | high | ## Least privilege principle |
| Issue | medium | ## Track context usage |
| Issue | medium | ## Structured logging |
Works well with: agent-tool-builder, agent-memory-systems, multi-agent-orchestration, agent-evaluation
Orchestration through Claude Code's built-in Agent Tool
This skill enables coordinating multiple specialized agents through Claude Code's native agent system. Unlike external scripts, this approach keeps all orchestration within Claude's control.
✅ Good for:
❌ Not for:
Use the security-auditor agent to review authentication
First, use the explorer-agent to discover project structure.
Then, use the backend-specialist to review API endpoints.
Finally, use the test-engineer to identify test gaps.
Use the frontend-specialist to analyze React components.
Based on those findings, have the test-engineer generate component tests.
Resume agent [agentId] and continue with additional requirements.
Agents: explorer-agent → [domain-agents] → synthesis
1. explorer-agent: Map codebase structure
2. security-auditor: Security posture
3. backend-specialist: API quality
4. frontend-specialist: UI/UX patterns
5. test-engineer: Test coverage
6. Synthesize all findings
Agents: affected-domain-agents → test-engineer
1. Identify affected domains (backend? frontend? both?)
2. Invoke relevant domain agents
3. test-engineer verifies changes
4. Synthesize recommendations
Agents: security-auditor → penetration-tester → synthesis
1. security-auditor: Configuration and code review
2. penetration-tester: Active vulnerability testing
3. Synthesize with prioritized remediation
| Agent | Expertise | Trigger Phrases |
|---|---|---|
orchestrator | Coordination | "comprehensive", "multi-perspective" |
security-auditor | Security | "security", "auth", "vulnerabilities" |
penetration-tester | Security Testing | "pentest", "red team", "exploit" |
backend-specialist | Backend | "API", "server", "Node.js", "Express" |
frontend-specialist | Frontend | "React", "UI", "components", "Next.js" |
test-engineer | Testing | "tests", "coverage", "TDD" |
devops-engineer | DevOps | "deploy", "CI/CD", "infrastructure" |
database-architect | Database | "schema", "Prisma", "migrations" |
mobile-developer | Mobile | "React Native", "Flutter", "mobile" |
api-designer | API Design | "REST", "GraphQL", "OpenAPI" |
debugger | Debugging | "bug", "error", "not working" |
explorer-agent | Discovery | "explore", "map", "structure" |
documentation-writer | Documentation | "write docs", "create README", "generate API docs" |
performance-optimizer | Performance | "slow", "optimize", "profiling" |
project-planner | Planning | "plan", "roadmap", "milestones" |
seo-specialist | SEO | "SEO", "meta tags", "search ranking" |
game-developer | Game Development | "game", "Unity", "Godot", "Phaser" |
These work alongside custom agents:
| Agent | Model | Purpose |
|---|---|---|
| Explore | Haiku | Fast read-only codebase search |
| Plan | Sonnet | Research during plan mode |
| General-purpose | Sonnet | Complex multi-step modifications |
Use Explore for quick searches, custom agents for domain expertise.
After all agents complete, synthesize:
## Orchestration Synthesis
### Task Summary
[What was accomplished]
### Agent Contributions
| Agent | Finding |
|-------|---------|
| security-auditor | Found X |
| backend-specialist | Identified Y |
### Consolidated Recommendations
1. **Critical**: [Issue from Agent A]
2. **Important**: [Issue from Agent B]
3. **Nice-to-have**: [Enhancement from Agent C]
### Action Items
- [ ] Fix critical security issue
- [ ] Refactor API endpoint
- [ ] Add missing tests
You are an agent architect who has learned the hard lessons of autonomous AI. You've seen the gap between impressive demos and production disasters. You know that a 95% success rate per step means only 60% by step 10.
Your core insight: Autonomy is earned, not granted. Start with heavily constrained agents that do one thing reliably. Add autonomy only as you prove reliability. The best agents look less impressive but work consistently.
You push for guardrails before capabilities, logging befor
Alternating reasoning and action steps
Separate planning phase from execution
Self-evaluation and iterative improvement
| Issue | Severity | Solution |
|---|---|---|
| Issue | critical | ## Reduce step count |
| Issue | critical | ## Set hard cost limits |
| Issue | critical | ## Test at scale before production |
| Issue | high | ## Validate against ground truth |
| Issue | high | ## Build robust API clients |
| Issue | high | ## Least privilege principle |
| Issue | medium | ## Track context usage |
| Issue | medium | ## Structured logging |
Works well with: agent-tool-builder, agent-memory-systems, multi-agent-orchestration, agent-evaluation
Design patterns for building autonomous coding agents, inspired by Cline and OpenAI Codex.
Use this skill when:
┌─────────────────────────────────────────────────────────────┐
│ AGENT LOOP │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Think │───▶│ Decide │───▶│ Act │ │
│ │ (Reason) │ │ (Plan) │ │ (Execute)│ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ ▲ │ │
│ │ ┌──────────┐ │ │
│ └─────────│ Observe │◀─────────┘ │
│ │ (Result) │ │
│ └──────────┘ │
└─────────────────────────────────────────────────────────────┘
class AgentLoop:
def __init__(self, llm, tools, max_iterations=50):
self.llm = llm
self.tools = {t.name: t for t in tools}
self.max_iterations = max_iterations
self.history = []
def run(self, task: str) -> str:
self.history.append({"role": "user", "content": task})
for i in range(self.max_iterations):
# Think: Get LLM response with tool options
response = self.llm.chat(
messages=self.history,
tools=self._format_tools(),
tool_choice="auto"
)
# Decide: Check if agent wants to use a tool
if response.tool_calls:
for tool_call in response.tool_calls:
# Act: Execute the tool
result = self._execute_tool(tool_call)
# Observe: Add result to history
self.history.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
})
else:
# No more tool calls = task complete
return response.content
return "Max iterations reached"
def _execute_tool(self, tool_call) -> Any:
tool = self.tools[tool_call.name]
args = json.loads(tool_call.arguments)
return tool.execute(**args)
class MultiModelAgent:
"""
Use different models for different purposes:
- Fast model for planning
- Powerful model for complex reasoning
- Specialized model for code generation
"""
def __init__(self):
self.models = {
"fast": "gpt-3.5-turbo", # Quick decisions
"smart": "gpt-4-turbo", # Complex reasoning
"code": "claude-3-sonnet", # Code generation
}
def select_model(self, task_type: str) -> str:
if task_type == "planning":
return self.models["fast"]
elif task_type == "analysis":
return self.models["smart"]
elif task_type == "code":
return self.models["code"]
return self.models["smart"]
class Tool:
"""Base class for agent tools"""
@property
def schema(self) -> dict:
"""JSON Schema for the tool"""
return {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": self._get_parameters(),
"required": self._get_required()
}
}
def execute(self, **kwargs) -> ToolResult:
"""Execute the tool and return result"""
raise NotImplementedError
class ReadFileTool(Tool):
name = "read_file"
description = "Read the contents of a file from the filesystem"
def _get_parameters(self):
return {
"path": {
"type": "string",
"description": "Absolute path to the file"
},
"start_line": {
"type": "integer",
"description": "Line to start reading from (1-indexed)"
},
"end_line": {
"type": "integer",
"description": "Line to stop reading at (inclusive)"
}
}
def _get_required(self):
return ["path"]
def execute(self, path: str, start_line: int = None, end_line: int = None) -> ToolResult:
try:
with open(path, 'r') as f:
lines = f.readlines()
if start_line and end_line:
lines = lines[start_line-1:end_line]
return ToolResult(
success=True,
output="".join(lines)
)
except FileNotFoundError:
return ToolResult(
success=False,
error=f"File not found: {path}"
)
CODING_AGENT_TOOLS = {
# File operations
"read_file": "Read file contents",
"write_file": "Create or overwrite a file",
"edit_file": "Make targeted edits to a file",
"list_directory": "List files and folders",
"search_files": "Search for files by pattern",
# Code understanding
"search_code": "Search for code patterns (grep)",
"get_definition": "Find function/class definition",
"get_references": "Find all references to a symbol",
# Terminal
"run_command": "Execute a shell command",
"read_output": "Read command output",
"send_input": "Send input to running command",
# Browser (optional)
"open_browser": "Open URL in browser",
"click_element": "Click on page element",
"type_text": "Type text into input",
"screenshot": "Capture screenshot",
# Context
"ask_user": "Ask the user a question",
"search_web": "Search the web for information"
}
class EditFileTool(Tool):
"""
Precise file editing with conflict detection.
Uses search/replace pattern for reliable edits.
"""
name = "edit_file"
description = "Edit a file by replacing specific content"
def execute(
self,
path: str,
search: str,
replace: str,
expected_occurrences: int = 1
) -> ToolResult:
"""
Args:
path: File to edit
search: Exact text to find (must match exactly, including whitespace)
replace: Text to replace with
expected_occurrences: How many times search should appear (validation)
"""
with open(path, 'r') as f:
content = f.read()
# Validate
actual_occurrences = content.count(search)
if actual_occurrences != expected_occurrences:
return ToolResult(
success=False,
error=f"Expected {expected_occurrences} occurrences, found {actual_occurrences}"
)
if actual_occurrences == 0:
return ToolResult(
success=False,
error="Search text not found in file"
)
# Apply edit
new_content = content.replace(search, replace)
with open(path, 'w') as f:
f.write(new_content)
return ToolResult(
success=True,
output=f"Replaced {actual_occurrences} occurrence(s)"
)
class PermissionLevel(Enum):
# Fully automatic - no user approval needed
AUTO = "auto"
# Ask once per session
ASK_ONCE = "ask_once"
# Ask every time
ASK_EACH = "ask_each"
# Never allow
NEVER = "never"
PERMISSION_CONFIG = {
# Low risk - can auto-approve
"read_file": PermissionLevel.AUTO,
"list_directory": PermissionLevel.AUTO,
"search_code": PermissionLevel.AUTO,
# Medium risk - ask once
"write_file": PermissionLevel.ASK_ONCE,
"edit_file": PermissionLevel.ASK_ONCE,
# High risk - ask each time
"run_command": PermissionLevel.ASK_EACH,
"delete_file": PermissionLevel.ASK_EACH,
# Dangerous - never auto-approve
"sudo_command": PermissionLevel.NEVER,
"format_disk": PermissionLevel.NEVER
}
class ApprovalManager:
def __init__(self, ui, config):
self.ui = ui
self.config = config
self.session_approvals = {}
def request_approval(self, tool_name: str, args: dict) -> bool:
level = self.config.get(tool_name, PermissionLevel.ASK_EACH)
if level == PermissionLevel.AUTO:
return True
if level == PermissionLevel.NEVER:
self.ui.show_error(f"Tool '{tool_name}' is not allowed")
return False
if level == PermissionLevel.ASK_ONCE:
if tool_name in self.session_approvals:
return self.session_approvals[tool_name]
# Show approval dialog
approved = self.ui.show_approval_dialog(
tool=tool_name,
args=args,
risk_level=self._assess_risk(tool_name, args)
)
if level == PermissionLevel.ASK_ONCE:
self.session_approvals[tool_name] = approved
return approved
def _assess_risk(self, tool_name: str, args: dict) -> str:
"""Analyze specific call for risk level"""
if tool_name == "run_command":
cmd = args.get("command", "")
if any(danger in cmd for danger in ["rm -rf", "sudo", "chmod"]):
return "HIGH"
return "MEDIUM"
class SandboxedExecution:
"""
Execute code/commands in isolated environment
"""
def __init__(self, workspace_dir: str):
self.workspace = workspace_dir
self.allowed_commands = ["npm", "python", "node", "git", "ls", "cat"]
self.blocked_paths = ["/etc", "/usr", "/bin", os.path.expanduser("~")]
def validate_path(self, path: str) -> bool:
"""Ensure path is within workspace"""
real_path = os.path.realpath(path)
workspace_real = os.path.realpath(self.workspace)
return real_path.startswith(workspace_real)
def validate_command(self, command: str) -> bool:
"""Check if command is allowed"""
cmd_parts = shlex.split(command)
if not cmd_parts:
return False
base_cmd = cmd_parts[0]
return base_cmd in self.allowed_commands
def execute_sandboxed(self, command: str) -> ToolResult:
if not self.validate_command(command):
return ToolResult(
success=False,
error=f"Command not allowed: {command}"
)
# Execute in isolated environment
result = subprocess.run(
command,
shell=True,
cwd=self.workspace,
capture_output=True,
timeout=30,
env={
**os.environ,
"HOME": self.workspace, # Isolate home directory
}
)
return ToolResult(
success=result.returncode == 0,
output=result.stdout.decode(),
error=result.stderr.decode() if result.returncode != 0 else None
)
class BrowserTool:
"""
Browser automation for agents using Playwright/Puppeteer.
Enables visual debugging and web testing.
"""
def __init__(self, headless: bool = True):
self.browser = None
self.page = None
self.headless = headless
async def open_url(self, url: str) -> ToolResult:
"""Navigate to URL and return page info"""
if not self.browser:
self.browser = await playwright.chromium.launch(headless=self.headless)
self.page = await self.browser.new_page()
await self.page.goto(url)
# Capture state
screenshot = await self.page.screenshot(type='png')
title = await self.page.title()
return ToolResult(
success=True,
output=f"Loaded: {title}",
metadata={
"screenshot": base64.b64encode(screenshot).decode(),
"url": self.page.url
}
)
async def click(self, selector: str) -> ToolResult:
"""Click on an element"""
try:
await self.page.click(selector, timeout=5000)
await self.page.wait_for_load_state("networkidle")
screenshot = await self.page.screenshot()
return ToolResult(
success=True,
output=f"Clicked: {selector}",
metadata={"screenshot": base64.b64encode(screenshot).decode()}
)
except TimeoutError:
return ToolResult(
success=False,
error=f"Element not found: {selector}"
)
async def type_text(self, selector: str, text: str) -> ToolResult:
"""Type text into an input"""
await self.page.fill(selector, text)
return ToolResult(success=True, output=f"Typed into {selector}")
async def get_page_content(self) -> ToolResult:
"""Get accessible text content of the page"""
content = await self.page.evaluate("""
() => {
// Get visible text
const walker = document.createTreeWalker(
document.body,
NodeFilter.SHOW_TEXT,
null,
false
);
let text = '';
while (walker.nextNode()) {
const node = walker.currentNode;
if (node.textContent.trim()) {
text += node.textContent.trim() + '\\n';
}
}
return text;
}
""")
return ToolResult(success=True, output=content)
class VisualAgent:
"""
Agent that uses screenshots to understand web pages.
Can identify elements visually without selectors.
"""
def __init__(self, llm, browser):
self.llm = llm
self.browser = browser
async def describe_page(self) -> str:
"""Use vision model to describe current page"""
screenshot = await self.browser.screenshot()
response = self.llm.chat([
{
"role": "user",
"content": [
{"type": "text", "text": "Describe this webpage. List all interactive elements you see."},
{"type": "image", "data": screenshot}
]
}
])
return response.content
async def find_and_click(self, description: str) -> ToolResult:
"""Find element by visual description and click it"""
screenshot = await self.browser.screenshot()
# Ask vision model to find element
response = self.llm.chat([
{
"role": "user",
"content": [
{
"type": "text",
"text": f"""
Find the element matching: "{description}"
Return the approximate coordinates as JSON: {{"x": number, "y": number}}
"""
},
{"type": "image", "data": screenshot}
]
}
])
coords = json.loads(response.content)
await self.browser.page.mouse.click(coords["x"], coords["y"])
return ToolResult(success=True, output=f"Clicked at ({coords['x']}, {coords['y']})")
class ContextManager:
"""
Manage context provided to the agent.
Inspired by Cline's @-mention patterns.
"""
def __init__(self, workspace: str):
self.workspace = workspace
self.context = []
def add_file(self, path: str) -> None:
"""@file - Add file contents to context"""
with open(path, 'r') as f:
content = f.read()
self.context.append({
"type": "file",
"path": path,
"content": content
})
def add_folder(self, path: str, max_files: int = 20) -> None:
"""@folder - Add all files in folder"""
for root, dirs, files in os.walk(path):
for file in files[:max_files]:
file_path = os.path.join(root, file)
self.add_file(file_path)
def add_url(self, url: str) -> None:
"""@url - Fetch and add URL content"""
response = requests.get(url)
content = html_to_markdown(response.text)
self.context.append({
"type": "url",
"url": url,
"content": content
})
def add_problems(self, diagnostics: list) -> None:
"""@problems - Add IDE diagnostics"""
self.context.append({
"type": "diagnostics",
"problems": diagnostics
})
def format_for_prompt(self) -> str:
"""Format all context for LLM prompt"""
parts = []
for item in self.context:
if item["type"] == "file":
parts.append(f"## File: {item['path']}\n```\n{item['content']}\n```")
elif item["type"] == "url":
parts.append(f"## URL: {item['url']}\n{item['content']}")
elif item["type"] == "diagnostics":
parts.append(f"## Problems:\n{json.dumps(item['problems'], indent=2)}")
return "\n\n".join(parts)
class CheckpointManager:
"""
Save and restore agent state for long-running tasks.
"""
def __init__(self, storage_dir: str):
self.storage_dir = storage_dir
os.makedirs(storage_dir, exist_ok=True)
def save_checkpoint(self, session_id: str, state: dict) -> str:
"""Save current agent state"""
checkpoint = {
"timestamp": datetime.now().isoformat(),
"session_id": session_id,
"history": state["history"],
"context": state["context"],
"workspace_state": self._capture_workspace(state["workspace"]),
"metadata": state.get("metadata", {})
}
path = os.path.join(self.storage_dir, f"{session_id}.json")
with open(path, 'w') as f:
json.dump(checkpoint, f, indent=2)
return path
def restore_checkpoint(self, checkpoint_path: str) -> dict:
"""Restore agent state from checkpoint"""
with open(checkpoint_path, 'r') as f:
checkpoint = json.load(f)
return {
"history": checkpoint["history"],
"context": checkpoint["context"],
"workspace": self._restore_workspace(checkpoint["workspace_state"]),
"metadata": checkpoint["metadata"]
}
def _capture_workspace(self, workspace: str) -> dict:
"""Capture relevant workspace state"""
# Git status, file hashes, etc.
return {
"git_ref": subprocess.getoutput(f"cd {workspace} && git rev-parse HEAD"),
"git_dirty": subprocess.getoutput(f"cd {workspace} && git status --porcelain")
}
from mcp import Server, Tool
class MCPAgent:
"""
Agent that can dynamically discover and use MCP tools.
'Add a tool that...' pattern from Cline.
"""
def __init__(self, llm):
self.llm = llm
self.mcp_servers = {}
self.available_tools = {}
def connect_server(self, name: str, config: dict) -> None:
"""Connect to an MCP server"""
server = Server(config)
self.mcp_servers[name] = server
# Discover tools
tools = server.list_tools()
for tool in tools:
self.available_tools[tool.name] = {
"server": name,
"schema": tool.schema
}
async def create_tool(self, description: str) -> str:
"""
Create a new MCP server based on user description.
'Add a tool that fetches Jira tickets'
"""
# Generate MCP server code
code = self.llm.generate(f"""
Create a Python MCP server with a tool that does:
{description}
Use the FastMCP framework. Include proper error handling.
Return only the Python code.
""")
# Save and install
server_name = self._extract_name(description)
path = f"./mcp_servers/{server_name}/server.py"
with open(path, 'w') as f:
f.write(code)
# Hot-reload
self.connect_server(server_name, {"path": path})
return f"Created tool: {server_name}"
Master orchestrator, peer-to-peer, and hierarchical multi-agent architectures
Use this skill when working with master orchestrator, peer-to-peer, and hierarchical multi-agent architectures.
Multi-agent architectures distribute work across multiple language model instances, each with its own context window. When designed well, this distribution enables capabilities beyond single-agent limits. When designed poorly, it introduces coordination overhead that negates benefits. The critical insight is that sub-agents exist primarily to isolate context, not to anthropomorphize role division.
Activate this skill when:
Multi-agent systems address single-agent context limitations through distribution. Three dominant patterns exist: supervisor/orchestrator for centralized control, peer-to-peer/swarm for flexible handoffs, and hierarchical for layered abstraction. The critical design principle is context isolation—sub-agents exist primarily to partition context rather than to simulate organizational roles.
Effective multi-agent systems require explicit coordination protocols, consensus mechanisms that avoid sycophancy, and careful attention to failure modes including bottlenecks, divergence, and error propagation.
The Context Bottleneck Single agents face inherent ceilings in reasoning capability, context management, and tool coordination. As tasks grow more complex, context windows fill with accumulated history, retrieved documents, and tool outputs. Performance degrades according to predictable patterns: the lost-in-middle effect, attention scarcity, and context poisoning.
Multi-agent architectures address these limitations by partitioning work across multiple context windows. Each agent operates in a clean context focused on its subtask. Results aggregate at a coordination layer without any single context bearing the full burden.
The Token Economics Reality Multi-agent systems consume significantly more tokens than single-agent approaches. Production data shows:
| Architecture | Token Multiplier | Use Case |
|---|---|---|
| Single agent chat | 1× baseline | Simple queries |
| Single agent with tools | ~4× baseline | Tool-using tasks |
| Multi-agent system | ~15× baseline | Complex research/coordination |
Research on the BrowseComp evaluation found that three factors explain 95% of performance variance: token usage (80% of variance), number of tool calls, and model choice. This validates the multi-agent approach of distributing work across agents with separate context windows to add capacity for parallel reasoning.
Critically, upgrading to better models often provides larger performance gains than doubling token budgets. Claude Sonnet 4.5 showed larger gains than doubling tokens on earlier Sonnet versions. GPT-5.2's thinking mode similarly outperforms raw token increases. This suggests model selection and multi-agent architecture are complementary strategies.
The Parallelization Argument Many tasks contain parallelizable subtasks that a single agent must execute sequentially. A research task might require searching multiple independent sources, analyzing different documents, or comparing competing approaches. A single agent processes these sequentially, accumulating context with each step.
Multi-agent architectures assign each subtask to a dedicated agent with a fresh context. All agents work simultaneously, then return results to a coordinator. The total real-world time approaches the duration of the longest subtask rather than the sum of all subtasks.
The Specialization Argument Different tasks benefit from different agent configurations: different system prompts, different tool sets, different context structures. A general-purpose agent must carry all possible configurations in context. Specialized agents carry only what they need.
Multi-agent architectures enable specialization without combinatorial explosion. The coordinator routes to specialized agents; each agent operates with lean context optimized for its domain.
Pattern 1: Supervisor/Orchestrator The supervisor pattern places a central agent in control, delegating to specialists and synthesizing results. The supervisor maintains global state and trajectory, decomposes user objectives into subtasks, and routes to appropriate workers.
User Query -> Supervisor -> [Specialist, Specialist, Specialist] -> Aggregation -> Final Output
When to use: Complex tasks with clear decomposition, tasks requiring coordination across domains, tasks where human oversight is important.
Advantages: Strict control over workflow, easier to implement human-in-the-loop interventions, ensures adherence to predefined plans.
Disadvantages: Supervisor context becomes bottleneck, supervisor failures cascade to all workers, "telephone game" problem where supervisors paraphrase sub-agent responses incorrectly.
The Telephone Game Problem and Solution LangGraph benchmarks found supervisor architectures initially performed 50% worse than optimized versions due to the "telephone game" problem where supervisors paraphrase sub-agent responses incorrectly, losing fidelity.
The fix: implement a forward_message tool allowing sub-agents to pass responses directly to users:
def forward_message(message: str, to_user: bool = True):
"""
Forward sub-agent response directly to user without supervisor synthesis.
Use when:
- Sub-agent response is final and complete
- Supervisor synthesis would lose important details
- Response format must be preserved exactly
"""
if to_user:
return {"type": "direct_response", "content": message}
return {"type": "supervisor_input", "content": message}
With this pattern, swarm architectures slightly outperform supervisors because sub-agents respond directly to users, eliminating translation errors.
Implementation note: Implement direct pass-through mechanisms allowing sub-agents to pass responses directly to users rather than through supervisor synthesis when appropriate.
Pattern 2: Peer-to-Peer/Swarm The peer-to-peer pattern removes central control, allowing agents to communicate directly based on predefined protocols. Any agent can transfer control to any other through explicit handoff mechanisms.
def transfer_to_agent_b():
return agent_b # Handoff via function return
agent_a = Agent(
name="Agent A",
functions=[transfer_to_agent_b]
)
When to use: Tasks requiring flexible exploration, tasks where rigid planning is counterproductive, tasks with emergent requirements that defy upfront decomposition.
Advantages: No single point of failure, scales effectively for breadth-first exploration, enables emergent problem-solving behaviors.
Disadvantages: Coordination complexity increases with agent count, risk of divergence without central state keeper, requires robust convergence constraints.
Implementation note: Define explicit handoff protocols with state passing. Ensure agents can communicate their context needs to receiving agents.
Pattern 3: Hierarchical Hierarchical structures organize agents into layers of abstraction: strategic, planning, and execution layers. Strategy layer agents define goals and constraints; planning layer agents break goals into actionable plans; execution layer agents perform atomic tasks.
Strategy Layer (Goal Definition) -> Planning Layer (Task Decomposition) -> Execution Layer (Atomic Tasks)
When to use: Large-scale projects with clear hierarchical structure, enterprise workflows with management layers, tasks requiring both high-level planning and detailed execution.
Advantages: Mirrors organizational structures, clear separation of concerns, enables different context structures at different levels.
Disadvantages: Coordination overhead between layers, potential for misalignment between strategy and execution, complex error propagation.
The primary purpose of multi-agent architectures is context isolation. Each sub-agent operates in a clean context window focused on its subtask without carrying accumulated context from other subtasks.
Isolation Mechanisms Full context delegation: For complex tasks where the sub-agent needs complete understanding, the planner shares its entire context. The sub-agent has its own tools and instructions but receives full context for its decisions.
Instruction passing: For simple, well-defined subtasks, the planner creates instructions via function call. The sub-agent receives only the instructions needed for its specific task.
File system memory: For complex tasks requiring shared state, agents read and write to persistent storage. The file system serves as the coordination mechanism, avoiding context bloat from shared state passing.
Isolation Trade-offs Full context delegation provides maximum capability but defeats the purpose of sub-agents. Instruction passing maintains isolation but limits sub-agent flexibility. File system memory enables shared state without context passing but introduces latency and consistency challenges.
The right choice depends on task complexity, coordination needs, and acceptable latency.
The Voting Problem Simple majority voting treats hallucinations from weak models as equal to reasoning from strong models. Without intervention, multi-agent discussions devolve into consensus on false premises due to inherent bias toward agreement.
Weighted Voting Weight agent votes by confidence or expertise. Agents with higher confidence or domain expertise carry more weight in final decisions.
Debate Protocols Debate protocols require agents to critique each other's outputs over multiple rounds. Adversarial critique often yields higher accuracy on complex reasoning than collaborative consensus.
Trigger-Based Intervention Monitor multi-agent interactions for specific behavioral markers. Stall triggers activate when discussions make no progress. Sycophancy triggers detect when agents mimic each other's answers without unique reasoning.
Different frameworks implement these patterns with different philosophies. LangGraph uses graph-based state machines with explicit nodes and edges. AutoGen uses conversational/event-driven patterns with GroupChat. CrewAI uses role-based process flows with hierarchical crew structures.
Failure: Supervisor Bottleneck The supervisor accumulates context from all workers, becoming susceptible to saturation and degradation.
Mitigation: Implement output schema constraints so workers return only distilled summaries. Use checkpointing to persist supervisor state without carrying full history.
Failure: Coordination Overhead Agent communication consumes tokens and introduces latency. Complex coordination can negate parallelization benefits.
Mitigation: Minimize communication through clear handoff protocols. Batch results where possible. Use asynchronous communication patterns.
Failure: Divergence Agents pursuing different goals without central coordination can drift from intended objectives.
Mitigation: Define clear objective boundaries for each agent. Implement convergence checks that verify progress toward shared goals. Use time-to-live limits on agent execution.
Failure: Error Propagation Errors in one agent's output propagate to downstream agents that consume that output.
Mitigation: Validate agent outputs before passing to consumers. Implement retry logic with circuit breakers. Use idempotent operations where possible.
Example 1: Research Team Architecture
Supervisor
├── Researcher (web search, document retrieval)
├── Analyzer (data analysis, statistics)
├── Fact-checker (verification, validation)
└── Writer (report generation, formatting)
Example 2: Handoff Protocol
def handle_customer_request(request):
if request.type == "billing":
return transfer_to(billing_agent)
elif request.type == "technical":
return transfer_to(technical_agent)
elif request.type == "sales":
return transfer_to(sales_agent)
else:
return handle_general(request)
This skill builds on context-fundamentals and context-degradation. It connects to:
Internal reference:
Related skills in this collection:
External resources:
Created: 2025-12-20 Last Updated: 2025-12-20 Author: Agent Skills for Context Engineering Contributors Version: 1.0.0
Orchestration through Claude Code's built-in Agent Tool
This skill enables coordinating multiple specialized agents through Claude Code's native agent system. Unlike external scripts, this approach keeps all orchestration within Claude's control.
✅ Good for:
❌ Not for:
Use the security-auditor agent to review authentication
First, use the explorer-agent to discover project structure.
Then, use the backend-specialist to review API endpoints.
Finally, use the test-engineer to identify test gaps.
Use the frontend-specialist to analyze React components.
Based on those findings, have the test-engineer generate component tests.
Resume agent [agentId] and continue with additional requirements.
Agents: explorer-agent → [domain-agents] → synthesis
1. explorer-agent: Map codebase structure
2. security-auditor: Security posture
3. backend-specialist: API quality
4. frontend-specialist: UI/UX patterns
5. test-engineer: Test coverage
6. Synthesize all findings
Agents: affected-domain-agents → test-engineer
1. Identify affected domains (backend? frontend? both?)
2. Invoke relevant domain agents
3. test-engineer verifies changes
4. Synthesize recommendations
Agents: security-auditor → penetration-tester → synthesis
1. security-auditor: Configuration and code review
2. penetration-tester: Active vulnerability testing
3. Synthesize with prioritized remediation
| Agent | Expertise | Trigger Phrases |
|---|---|---|
orchestrator | Coordination | "comprehensive", "multi-perspective" |
security-auditor | Security | "security", "auth", "vulnerabilities" |
penetration-tester | Security Testing | "pentest", "red team", "exploit" |
backend-specialist | Backend | "API", "server", "Node.js", "Express" |
frontend-specialist | Frontend | "React", "UI", "components", "Next.js" |
test-engineer | Testing | "tests", "coverage", "TDD" |
devops-engineer | DevOps | "deploy", "CI/CD", "infrastructure" |
database-architect | Database | "schema", "Prisma", "migrations" |
mobile-developer | Mobile | "React Native", "Flutter", "mobile" |
api-designer | API Design | "REST", "GraphQL", "OpenAPI" |
debugger | Debugging | "bug", "error", "not working" |
explorer-agent | Discovery | "explore", "map", "structure" |
documentation-writer | Documentation | "write docs", "create README", "generate API docs" |
performance-optimizer | Performance | "slow", "optimize", "profiling" |
project-planner | Planning | "plan", "roadmap", "milestones" |
seo-specialist | SEO | "SEO", "meta tags", "search ranking" |
game-developer | Game Development | "game", "Unity", "Godot", "Phaser" |
These work alongside custom agents:
| Agent | Model | Purpose |
|---|---|---|
| Explore | Haiku | Fast read-only codebase search |
| Plan | Sonnet | Research during plan mode |
| General-purpose | Sonnet | Complex multi-step modifications |
Use Explore for quick searches, custom agents for domain expertise.
After all agents complete, synthesize:
## Orchestration Synthesis
### Task Summary
[What was accomplished]
### Agent Contributions
| Agent | Finding |
|-------|---------|
| security-auditor | Found X |
| backend-specialist | Identified Y |
### Consolidated Recommendations
1. **Critical**: [Issue from Agent A]
2. **Important**: [Issue from Agent B]
3. **Nice-to-have**: [Enhancement from Agent C]
### Action Items
- [ ] Fix critical security issue
- [ ] Refactor API endpoint
- [ ] Add missing tests
Role: LangGraph Agent Architect
You are an expert in building production-grade AI agents with LangGraph. You understand that agents need explicit structure - graphs make the flow visible and debuggable. You design state carefully, use reducers appropriately, and always consider persistence for production. You know when cycles are needed and how to prevent infinite loops.
Simple ReAct-style agent with tools
When to use: Single agent with tool calling
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# 1. Define State
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
# add_messages reducer appends, doesn't overwrite
# 2. Define Tools
@tool
def search(query: str) -> str:
"""Search the web for information."""
# Implementation here
return f"Results for: {query}"
@tool
def calculator(expression: str) -> str:
"""Evaluate a math expression."""
return str(eval(expression))
tools = [search, calculator]
# 3. Create LLM with tools
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)
# 4. Define Nodes
def agent(state: AgentState) -> dict:
"""The agent node - calls LLM."""
response = llm.invoke(state["messages"])
return {"messages": [response]}
# Tool node handles tool execution
tool_node = ToolNode(tools)
# 5. Define Routing
def should_continue(state: AgentState) -> str:
"""Route based on whether tools were called."""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
# 6. Build Graph
graph = StateGraph(AgentState)
# Add nodes
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)
# Add edges
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, ["tools", END])
graph.add_edge("tools", "agent") # Loop back
# Compile
app = graph.compile()
# 7. Run
result = app.invoke({
"messages": [("user", "What is 25 * 4?")]
})
Complex state management with custom reducers
When to use: Multiple agents updating shared state
from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph
# Custom reducer for merging dictionaries
def merge_dicts(left: dict, right: dict) -> dict:
return {**left, **right}
# State with multiple reducers
class ResearchState(TypedDict):
# Messages append (don't overwrite)
messages: Annotated[list, add_messages]
# Research findings merge
findings: Annotated[dict, merge_dicts]
# Sources accumulate
sources: Annotated[list[str], add]
# Current step (overwrites - no reducer)
current_step: str
# Error count (custom reducer)
errors: Annotated[int, lambda a, b: a + b]
# Nodes return partial state updates
def researcher(state: ResearchState) -> dict:
# Only return fields being updated
return {
"findings": {"topic_a": "New finding"},
"sources": ["source1.com"],
"current_step": "researching"
}
def writer(state: ResearchState) -> dict:
# Access accumulated state
all_findings = state["findings"]
all_sources = state["sources"]
return {
"messages": [("assistant", f"Report based on {len(all_sources)} sources")],
"current_step": "writing"
}
# Build graph
graph = StateGraph(ResearchState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
# ... add edges
Route to different paths based on state
When to use: Multiple possible workflows
from langgraph.graph import StateGraph, START, END
class RouterState(TypedDict):
query: str
query_type: str
result: str
def classifier(state: RouterState) -> dict:
"""Classify the query type."""
query = state["query"].lower()
if "code" in query or "program" in query:
return {"query_type": "coding"}
elif "search" in query or "find" in query:
return {"query_type": "search"}
else:
return {"query_type": "chat"}
def coding_agent(state: RouterState) -> dict:
return {"result": "Here's your code..."}
def search_agent(state: RouterState) -> dict:
return {"result": "Search results..."}
def chat_agent(state: RouterState) -> dict:
return {"result": "Let me help..."}
# Routing function
def route_query(state: RouterState) -> str:
"""Route to appropriate agent."""
query_type = state["query_type"]
return query_type # Returns node name
# Build graph
graph = StateGraph(RouterState)
graph.add_node("classifier", classifier)
graph.add_node("coding", coding_agent)
graph.add_node("search", search_agent)
graph.add_node("chat", chat_agent)
graph.add_edge(START, "classifier")
# Conditional edges from classifier
graph.add_conditional_edges(
"classifier",
route_query,
{
"coding": "coding",
"search": "search",
"chat": "chat"
}
)
# All agents lead to END
graph.add_edge("coding", END)
graph.add_edge("search", END)
graph.add_edge("chat", END)
app = graph.compile()
Why bad: Agent loops forever. Burns tokens and costs. Eventually errors out.
Instead: Always have exit conditions:
def should_continue(state): if state["iterations"] > 10: return END if state["task_complete"]: return END return "agent"
Why bad: Loses LangGraph's benefits. State not persisted. Can't resume conversations.
Instead: Always use state for data flow. Return state updates from nodes. Use reducers for accumulation. Let LangGraph manage state.
Why bad: Hard to reason about. Unnecessary data in context. Serialization overhead.
Instead: Use input/output schemas for clean interfaces. Private state for internal data. Clear separation of concerns.
Works well with: crewai, autonomous-agents, langfuse, structured-output
Role: CrewAI Multi-Agent Architect
You are an expert in designing collaborative AI agent teams with CrewAI. You think in terms of roles, responsibilities, and delegation. You design clear agent personas with specific expertise, create well-defined tasks with expected outputs, and orchestrate crews for optimal collaboration. You know when to use sequential vs hierarchical processes.
Define agents and tasks in YAML (recommended)
When to use: Any CrewAI project
# config/agents.yaml
researcher:
role: "Senior Research Analyst"
goal: "Find comprehensive, accurate information on {topic}"
backstory: |
You are an expert researcher with years of experience
in gathering and analyzing information. You're known
for your thorough and accurate research.
tools:
- SerperDevTool
- WebsiteSearchTool
verbose: true
writer:
role: "Content Writer"
goal: "Create engaging, well-structured content"
backstory: |
You are a skilled writer who transforms research
into compelling narratives. You focus on clarity
and engagement.
verbose: true
# config/tasks.yaml
research_task:
description: |
Research the topic: {topic}
Focus on:
1. Key facts and statistics
2. Recent developments
3. Expert opinions
4. Contrarian viewpoints
Be thorough and cite sources.
agent: researcher
expected_output: |
A comprehensive research report with:
- Executive summary
- Key findings (bulleted)
- Sources cited
writing_task:
description: |
Using the research provided, write an article about {topic}.
Requirements:
- 800-1000 words
- Engaging introduction
- Clear structure with headers
- Actionable conclusion
agent: writer
expected_output: "A polished article ready for publication"
context:
- research_task # Uses output from research
# crew.py
from crewai import Agent, Task, Crew, Process
from crewai.project import CrewBase, agent, task, crew
@CrewBase
class ContentCrew:
agents_config = 'config/agents.yaml'
tasks_config = 'config/tasks.yaml'
@agent
def researcher(self) -> Agent:
return Agent(config=self.agents_config['researcher'])
@agent
def writer(self) -> Agent:
return Agent(config=self.agents_config['writer'])
@task
def research_task(self) -> Task:
return Task(config=self.tasks_config['research_task'])
@task
def writing_task(self) -> Task:
return Task(config
Manager agent delegates to workers
When to use: Complex tasks needing coordination
from crewai import Crew, Process
# Define specialized agents
researcher = Agent(
role="Research Specialist",
goal="Find accurate information",
backstory="Expert researcher..."
)
analyst = Agent(
role="Data Analyst",
goal="Analyze and interpret data",
backstory="Expert analyst..."
)
writer = Agent(
role="Content Writer",
goal="Create engaging content",
backstory="Expert writer..."
)
# Hierarchical crew - manager coordinates
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
process=Process.hierarchical,
manager_llm=ChatOpenAI(model="gpt-4o"), # Manager model
verbose=True
)
# Manager decides:
# - Which agent handles which task
# - When to delegate
# - How to combine results
result = crew.kickoff()
Generate execution plan before running
When to use: Complex workflows needing structure
from crewai import Crew, Process
# Enable planning
crew = Crew(
agents=[researcher, writer, reviewer],
tasks=[research, write, review],
process=Process.sequential,
planning=True, # Enable planning
planning_llm=ChatOpenAI(model="gpt-4o") # Planner model
)
# With planning enabled:
# 1. CrewAI generates step-by-step plan
# 2. Plan is injected into each task
# 3. Agents see overall structure
# 4. More consistent results
result = crew.kickoff()
# Access the plan
print(crew.plan)
Why bad: Agent doesn't know its specialty. Overlapping responsibilities. Poor task delegation.
Instead: Be specific:
Why bad: Agent doesn't know done criteria. Inconsistent outputs. Hard to chain tasks.
Instead: Always specify expected_output: expected_output: | A JSON object with:
Why bad: Coordination overhead. Inconsistent communication. Slower execution.
Instead: 3-5 agents with clear roles. One agent can handle multiple related tasks. Use tools instead of agents for simple actions.
Works well with: langgraph, autonomous-agents, langfuse, structured-output