executing-plans
executing-plans
Use when executing an implementation plan in the current session as the implementer yourself — your human partner chose inline execution, or no subagent tool is available
crewai-agent-workflowsImplements CrewAI multi-agent collaboration patterns including role-based agents, task delegation, tool sharing, sequential and hierarchical workflows, and autonomous agent coordination for production AI applications.
Independent GitHub snapshot · Automated classification · Original documentation governs
name: crewai-agent-workflows description: Implements CrewAI multi-agent collaboration patterns including role-based agents, task delegation, tool sharing, sequential and hierarchical workflows, and autonomous agent coordination for production AI applications. license: MIT compatibility: opencode metadata: version: "1.0.0" domain: coding triggers: crewai, multi-agent collaboration, role based agents, agent task delegation, how do i coordinate ai agents, sequential agent workflow, hierarchical agent team archetypes: - tactical - generation anti_triggers: - brainstorming - vague ideation - high level architecture response_profile: verbosity: low directive_strength: high abstraction_level: operational role: implementation scope: implementation output-format: code content-types: [code, guidance, do-dont, examples] related-skills: langgraph-implementation, autogen-conversational-patterns, framework-utilization
Implements multi-agent collaboration systems using CrewAI's role-based agent paradigm. Builds coordinated teams of specialized AI agents with explicit task delegation, shared tool access, sequential and hierarchical execution flows, and autonomous decision-making for production-grade AI applications.
role, goal, and backstory that establishes its specializationTask(description=..., expected_output=..., agent=...) with explicit acceptance criteriaCrew(agents=[...], tasks=[...]) and invoke with crew.kickoff() for sequential executionProcess.hierarchical mode when you need a manager agent to delegate dynamicallyverbose=True during development; switch to structured logging in productionUse this skill when:
Avoid this skill for:
role (what it does), goal (why it exists), and backstory (context for its behavior). The role determines how the agent frames its reasoning.from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
researcher = Agent(
role="Senior Research Analyst",
goal="Uncover cutting-edge developments in AI and data science",
backstory="""You are an expert analyst at a leading tech research firm.
You specialize in identifying emerging trends, analyzing market data,
and synthesizing technical findings into actionable insights.""",
verbose=True,
allow_delegation=False,
llm=llm,
)
writer = Agent(
role="Technical Content Writer",
goal="Write engaging and accurate content about AI and data science topics",
backstory="""You are a seasoned technical writer with a background in computer science.
You translate complex research findings into clear, well-structured articles
suitable for both technical and business audiences.""",
verbose=True,
allow_delegation=False,
llm=llm,
)
reviewer = Agent(
role="Content Review Editor",
goal="Review and refine content for accuracy, clarity, and readability",
backstory="""You are a senior editor at a top tech publication. You have an eye
for technical accuracy, narrative flow, and reader engagement. You provide
constructive feedback that elevates the quality of published work.""",
verbose=True,
allow_delegation=False,
llm=llm,
)
expected_output.research_task = Task(
description=(
"Analyze the latest developments in large language models for 2025. "
"Focus on breakthrough architectures, training methodologies, and real-world applications. "
"Identify the top 5 most impactful trends."
),
expected_output="A comprehensive research report with numbered trends, each including a summary, technical details, and impact assessment.",
agent=researcher,
)
writing_task = Task(
description=(
"Based on the research findings, write a 1500-word article about the future of AI. "
"Target audience is technical managers who need to understand strategic implications."
),
expected_output="A well-structured article with an executive summary, five trend sections, and a conclusion with actionable recommendations.",
agent=writer,
)
review_task = Task(
description=(
"Review the article for technical accuracy, clarity, and completeness. "
"Verify all claims against the original research report. "
"Suggest specific improvements to enhance readability and impact."
),
expected_output="A revised article incorporating your feedback, followed by a bullet-point list of changes made with justification.",
agent=reviewer,
)
# Sequential execution: each task completes before the next starts
crew = Crew(
agents=[researcher, writer, reviewer],
tasks=[research_task, writing_task, review_task],
process=Process.sequential,
verbose=2, # Verbose level 2 includes task output and agent reasoning
)
result = crew.kickoff()
print(f"Final output:\n{result}")
from crewai import Agent, Task, Crew, Process
manager = Agent(
role="AI Research Project Manager",
goal="Coordinate the research team and ensure high-quality deliverables",
backstory=(
"You manage a team of researchers and writers. You review their work, "
"assign follow-up tasks, and ensure the final output meets quality standards."
),
verbose=True,
allow_delegation=True, # Critical: allows manager to delegate
llm=llm,
)
# Worker agents (no delegation needed — they just execute assigned tasks)
analyst = Agent(
role="Data Analyst",
goal="Analyze datasets and produce statistical insights",
backstory="You are a data scientist specializing in statistical analysis.",
verbose=True,
allow_delegation=False,
llm=llm,
)
visualizer = Agent(
role="Data Visualization Specialist",
goal="Create clear visual representations of complex data findings",
backstory="You transform analytical results into intuitive charts and dashboards.",
verbose=True,
allow_delegation=False,
llm=llm,
)
# Tasks for hierarchical mode
data_analysis_task = Task(
description="Analyze the provided customer satisfaction dataset and identify key drivers of satisfaction.",
expected_output="A statistical summary with correlation coefficients and p-values for each factor.",
agent=analyst,
)
visualization_task = Task(
description="Create visualizations showing the relationship between satisfaction factors.",
expected_output="Three publication-ready charts with titles, labels, and legends.",
agent=visualizer,
)
# Hierarchical crew — manager assigns tasks to workers
crew = Crew(
agents=[manager, analyst, visualizer],
tasks=[data_analysis_task, visualization_task],
process=Process.hierarchical,
manager_llm=llm, # The manager uses this LLM for delegation decisions
verbose=2,
)
result = crew.kickoff()
from langchain_community.tools import DuckDuckGoSearchRun
from crewai_tools import Tool # CrewAI-specific tool wrapper
search_tool = Tool(
name="Web Search",
description="Search the internet for current information on any topic",
func=DuckDuckGoSearchRun().run,
)
# Only give the researcher access to search — writer and reviewer read from task outputs
researcher = Agent(
role="Senior Research Analyst",
goal="Uncover cutting-edge developments in AI and data science",
backstory="You are an expert analyst...",
verbose=True,
allow_delegation=False,
llm=llm,
tools=[search_tool], # Tool access restricted to relevant agents only
)
writer = Agent(
role="Technical Content Writer",
goal="Write engaging and accurate content about AI and data science topics",
backstory="You are a seasoned technical writer...",
verbose=True,
allow_delegation=False,
llm=llm,
# No tools — writer uses research output from task context
)
research_task = Task(
description="Research the latest AI agent frameworks and their comparison.",
expected_output="A detailed comparison of top 5 AI agent frameworks with pros and cons.",
agent=researcher, # Only researcher has search_tool available
)
Tasks in sequential mode automatically pass context to the next task. Each agent receives the outputs of all prior tasks as part of its working context.
from crewai import Agent, Task, Crew, Process
# Step 1: Data extraction agent
extractor = Agent(
role="Data Extraction Engineer",
goal="Extract structured data from unstructured documents",
backstory="Expert at parsing and structuring document content.",
verbose=True,
allow_delegation=False,
llm=llm,
)
# Step 2: Analysis agent receives extraction output automatically
analyzer = Agent(
role="Data Analyst",
goal="Analyze extracted data and identify patterns",
backstory="Experienced data analyst specializing in pattern recognition.",
verbose=True,
allow_delegation=False,
llm=llm,
)
# Step 3: Reporting agent receives both extraction and analysis outputs
reporter = Agent(
role="Business Intelligence Analyst",
goal="Generate business reports from analytical findings",
backstory="Skilled at translating data insights into business recommendations.",
verbose=True,
allow_delegation=False,
llm=llm,
)
extraction_task = Task(
description="Extract customer names, order dates, and amounts from the provided text documents.",
expected_output="A structured list of records with fields: customer_name (str), order_date (str YYYY-MM-DD), amount (float).",
agent=extractor,
)
analysis_task = Task(
description="Analyze the extracted customer data. Calculate total revenue per month, identify top customers, and spot unusual patterns.",
expected_output="Statistical summary with monthly revenue trends, top 10 customers by spend, and flagged anomalies.",
agent=analyzer,
)
reporting_task = Task(
description="Write a quarterly business review based on the analysis findings. Include executive summary, key metrics, and recommendations.",
expected_output="A formatted business report with executive summary, three key metric sections, and actionable recommendations.",
agent=reporter,
)
crew = Crew(
agents=[extractor, analyzer, reporter],
tasks=[extraction_task, analysis_task, reporting_task],
process=Process.sequential,
verbose=2,
)
# The output flows automatically: extraction → analysis → report
result = crew.kickoff()
print(result.raw) # Final output from last task
Allow agents to delegate sub-tasks to other crew members when they need additional information. This requires allow_delegation=True on the delegating agent.
from crewai import Agent, Task, Crew, Process
lead_analyst = Agent(
role="Lead Data Analyst",
goal="Coordinate data analysis and produce final insights report",
backstory="Senior analyst who delegates research tasks to specialists.",
verbose=True,
allow_delegation=True, # Allows lead to delegate to other agents
llm=llm,
)
specialist = Agent(
role="Statistical Modeling Specialist",
goal="Build predictive models and validate statistical assumptions",
backstory="Expert in regression analysis, time series forecasting, and model validation.",
verbose=True,
allow_delegation=False, # Specialist only executes delegated tasks
llm=llm,
)
lead_task = Task(
description=(
"Analyze the quarterly sales data. If you need statistical modeling done, "
"delegate that specific task to the specialist agent. Produce a final summary."
),
expected_output="Executive summary with key findings and a list of delegated tasks with their results.",
agent=lead_analyst,
)
delegated_modeling_task = Task(
description="Build a linear regression model predicting next quarter sales based on the provided historical data. Report coefficients, R-squared, and residuals.",
expected_output="Model parameters table, R-squared value, residual analysis summary, and prediction for next quarter.",
agent=specialist,
)
crew = Crew(
agents=[lead_analyst, specialist],
tasks=[lead_task, delegated_modeling_task],
process=Process.sequential,
verbose=2,
)
result = crew.kickoff()
Use structured output parsing to enforce consistent format across agents and enable programmatic downstream processing.
from typing import List, TypedDict
from crewai import Agent, Task, Crew, Process
from pydantic import BaseModel
class Finding(BaseModel):
trend_name: str
impact_level: str # high, medium, low
confidence: float # 0.0 to 1.0
description: str
# Force structured output using response_schema in the task
researcher = Agent(
role="AI Research Analyst",
goal="Identify and classify emerging AI trends",
backstory="Expert at trend analysis and classification.",
verbose=True,
allow_delegation=False,
llm=llm,
)
structured_task = Task(
description=(
"Identify the top 5 AI trends for 2025. For each trend, provide its name, "
"estimated impact level, your confidence score, and a brief technical description."
),
expected_output="A JSON array of 5 trend objects, each with fields: trend_name (string), impact_level (high/medium/low), confidence (0.0-1.0), description (string).",
agent=researcher,
)
crew = Crew(
agents=[researcher],
tasks=[structured_task],
process=Process.sequential,
output_parsing=True, # Enables structured response parsing
)
result = crew.kickoff()
# result.structured contains parsed data conforming to the expected_output schema
parsed_findings = [Finding(**item) for item in result.structured]
Process.sequential when task order matters and outputs feed forward; use Process.hierarchical when you need dynamic task allocationallow_delegation=True ONLY on agents that should delegate sub-tasks; keep worker agents at allow_delegation=False for clarityexpected_output with concrete format requirements (JSON schema, bullet lists, word count) — vague expectations produce inconsistent resultsverbose=2 during development to observe task handoffs; use structured logging in productionProcess.hierarchical without a capable manager LLM — weak managers produce poor delegation decisions and create unnecessary loop iterationsdescription and expected_output| Skill | Purpose |
|---|---|
langgraph-implementation | State graph-based workflows with explicit state machine control and checkpointing |
autogen-conversational-patterns | Conversational multi-agent chat groups for iterative problem solving through dialogue |
framework-utilization | General framework adoption strategy and learning patterns applicable to any framework |
Authoritative documentation links for this domain. The model follows markdown links at load time to resolve external references and inline content.