Building a CrewAI Application

1# Create project structure
2mkdir -p content_crew/src
3mkdir -p content_crew/config
4mkdir -p content_crew/output
5mkdir -p content_crew/tests
6
7# Project layout
8content_crew/
9├── src/
10│   ├── __init__.py
11│   ├── agents.py        # Agent definitions
12│   ├── tasks.py         # Task definitions
13│   ├── tools.py         # Custom tools
14│   ├── crew.py          # Crew assembly
15│   └── models.py        # Pydantic models
16├── config/
17│   └── settings.py      # Configuration
18├── output/              # Generated content
19├── tests/
20│   └── test_crew.py
21├── main.py              # Entry point
22└── requirements.txt

1# requirements.txt
2crewai>=0.28.0
3crewai-tools>=0.2.0
4langchain-openai>=0.0.5
5pydantic>=2.0.0
6python-dotenv>=1.0.0
7httpx>=0.25.0

1# config/settings.py
2import os
3from dotenv import load_dotenv
4
5load_dotenv()
6
7
8class Settings:
9    """Application settings."""
10
11    # API Keys
12    OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
13    SERPER_API_KEY: str = os.getenv("SERPER_API_KEY", "")
14
15    # Model Configuration
16    PRIMARY_MODEL: str = "gpt-4o"
17    FAST_MODEL: str = "gpt-4o-mini"
18    TEMPERATURE: float = 0.7
19
20    # Output Configuration
21    OUTPUT_DIR: str = "./output"
22    VERBOSE: bool = True
23
24    # Crew Settings
25    MAX_ITERATIONS: int = 15
26    MEMORY_ENABLED: bool = True
27
28
29settings = Settings()

1# src/agents.py
2from crewai import Agent
3from langchain_openai import ChatOpenAI
4from crewai_tools import SerperDevTool, WebsiteSearchTool
5from .tools import ContentAnalysisTool, SEOTool
6from config.settings import settings
7
8
9def create_llm(model: str = None, temperature: float = None) -> ChatOpenAI:
10    """Create configured LLM instance."""
11    return ChatOpenAI(
12        model=model or settings.PRIMARY_MODEL,
13        temperature=temperature or settings.TEMPERATURE,
14        api_key=settings.OPENAI_API_KEY
15    )
16
17
18class ContentAgents:
19    """Factory for content creation agents."""
20
21    def __init__(self):
22        self.llm = create_llm()
23        self.fast_llm = create_llm(model=settings.FAST_MODEL)
24
25    def research_lead(self) -> Agent:
26        """Lead researcher who coordinates research efforts."""
27        return Agent(
28            role="Lead Research Analyst",
29            goal="""Conduct comprehensive research on assigned topics,
30            gathering data from reliable sources and identifying key trends""",
31            backstory="""You are a veteran research analyst with 15 years of
32            experience in technology journalism. You have a keen eye for
33            identifying important trends and can quickly separate signal from
34            noise. You're known for finding unique angles that others miss
35            and for always verifying your sources.
36
37            You approach research systematically:
38            1. Start with broad searches to understand the landscape
39            2. Drill down into specific subtopics
40            3. Look for contrarian viewpoints
41            4. Verify facts from multiple sources
42            5. Identify the most newsworthy elements""",
43            tools=[
44                SerperDevTool(),
45                WebsiteSearchTool()
46            ],
47            llm=self.llm,
48            allow_delegation=True,
49            verbose=settings.VERBOSE,
50            max_iter=settings.MAX_ITERATIONS
51        )
52
53    def trend_analyst(self) -> Agent:
54        """Analyst who identifies trends and insights."""
55        return Agent(
56            role="Trend Analysis Specialist",
57            goal="""Analyze research data to identify patterns, trends,
58            and actionable insights for content creation""",
59            backstory="""You are a data-driven analyst who excels at
60            pattern recognition. With a background in data science and
61            market research, you can spot emerging trends before they
62            become mainstream.
63
64            Your analytical approach:
65            1. Quantify and categorize findings
66            2. Identify correlations and patterns
67            3. Separate lasting trends from temporary fads
68            4. Predict future developments
69            5. Provide evidence-based recommendations""",
70            tools=[ContentAnalysisTool()],
71            llm=self.llm,
72            allow_delegation=False,
73            verbose=settings.VERBOSE
74        )
75
76    def content_strategist(self) -> Agent:
77        """Strategist who plans content structure."""
78        return Agent(
79            role="Content Strategy Director",
80            goal="""Develop content strategies that maximize engagement
81            and provide value to readers""",
82            backstory="""You are a content strategist with experience at
83            top tech publications. You understand what makes content
84            compelling and know how to structure articles for maximum
85            impact and readability.
86
87            Your strategy process:
88            1. Define the target audience
89            2. Identify the key message
90            3. Structure the narrative arc
91            4. Plan for engagement hooks
92            5. Optimize for SEO and shareability""",
93            tools=[SEOTool()],
94            llm=self.llm,
95            allow_delegation=True,
96            verbose=settings.VERBOSE
97        )
98
99    def content_writer(self) -> Agent:
100        """Writer who produces the final content."""
101        return Agent(
102            role="Senior Technical Writer",
103            goal="""Create engaging, well-researched content that
104            educates and inspires readers""",
105            backstory="""You are an award-winning technical writer known
106            for making complex topics accessible. You have written for
107            major tech publications and have a talent for storytelling
108            that keeps readers engaged.
109
110            Your writing approach:
111            1. Hook readers with a compelling opening
112            2. Use clear, jargon-free language
113            3. Support claims with data and examples
114            4. Include practical takeaways
115            5. End with a memorable conclusion""",
116            llm=self.llm,
117            allow_delegation=False,
118            verbose=settings.VERBOSE
119        )
120
121    def editor(self) -> Agent:
122        """Editor who reviews and polishes content."""
123        return Agent(
124            role="Senior Content Editor",
125            goal="""Ensure all content is polished, accurate, and
126            optimized for the target audience""",
127            backstory="""You are a meticulous editor with decades of
128            experience at major publications. You catch errors others
129            miss and have a gift for tightening prose while preserving
130            the author's voice.
131
132            Your editing process:
133            1. Check factual accuracy
134            2. Improve clarity and flow
135            3. Tighten the prose
136            4. Verify consistency
137            5. Final polish for publication""",
138            llm=self.llm,
139            allow_delegation=False,
140            verbose=settings.VERBOSE
141        )

1# src/tasks.py
2from crewai import Task
3from pydantic import BaseModel, Field
4from typing import List
5from .agents import ContentAgents
6
7
8class ResearchFindings(BaseModel):
9    """Structured research output."""
10    topic: str
11    key_findings: List[str]
12    sources: List[str]
13    trends: List[str]
14    data_points: List[dict]
15    unique_angles: List[str]
16
17
18class ContentOutline(BaseModel):
19    """Structured content outline."""
20    title: str
21    subtitle: str
22    target_audience: str
23    key_message: str
24    sections: List[dict]
25    seo_keywords: List[str]
26    estimated_reading_time: int
27
28
29class BlogPost(BaseModel):
30    """Final blog post structure."""
31    title: str
32    subtitle: str
33    content: str
34    meta_description: str
35    tags: List[str]
36
37
38class ContentTasks:
39    """Factory for content creation tasks."""
40
41    def __init__(self, agents: ContentAgents):
42        self.agents = agents
43
44    def research_task(self, topic: str) -> Task:
45        """Task for comprehensive research."""
46        return Task(
47            description=f"""Conduct comprehensive research on: {topic}
48
49            Requirements:
50            1. Search for recent articles, studies, and data
51            2. Identify at least 5 key findings
52            3. Find 3-5 unique angles or perspectives
53            4. Gather relevant statistics and data points
54            5. Document all sources with URLs
55
56            Focus on finding information that is:
57            - Recent (preferably from the last 6 months)
58            - From authoritative sources
59            - Unique or underreported
60            - Actionable for readers""",
61            expected_output="""A comprehensive research report containing:
62            - Executive summary of findings
63            - List of key findings with supporting evidence
64            - Identified trends with analysis
65            - Collection of relevant data points
66            - List of unique angles for content
67            - All sources with URLs""",
68            agent=self.agents.research_lead(),
69            output_pydantic=ResearchFindings
70        )
71
72    def analysis_task(self, research_task: Task) -> Task:
73        """Task for analyzing research findings."""
74        return Task(
75            description="""Analyze the research findings to extract
76            actionable insights and identify the most compelling narrative.
77
78            Requirements:
79            1. Categorize findings by theme
80            2. Identify the strongest trends
81            3. Find connections between data points
82            4. Determine the most compelling angle
83            5. Recommend the content focus""",
84            expected_output="""An analysis report containing:
85            - Categorized themes with insights
86            - Top 3 trends with supporting evidence
87            - Key connections and patterns
88            - Recommended content focus with justification
89            - Suggested data visualizations""",
90            agent=self.agents.trend_analyst(),
91            context=[research_task]
92        )
93
94    def strategy_task(self, research_task: Task, analysis_task: Task) -> Task:
95        """Task for content strategy development."""
96        return Task(
97            description="""Develop a comprehensive content strategy and outline.
98
99            Requirements:
100            1. Define the target audience clearly
101            2. Craft a compelling headline and subtitle
102            3. Structure the content for maximum impact
103            4. Plan engagement hooks throughout
104            5. Identify SEO keywords and optimization opportunities""",
105            expected_output="""A content strategy document containing:
106            - Target audience profile
107            - Headline options (3-5)
108            - Content outline with sections
109            - Engagement hooks for each section
110            - SEO keyword recommendations
111            - Estimated reading time""",
112            agent=self.agents.content_strategist(),
113            context=[research_task, analysis_task],
114            output_pydantic=ContentOutline
115        )
116
117    def writing_task(
118        self,
119        research_task: Task,
120        analysis_task: Task,
121        strategy_task: Task
122    ) -> Task:
123        """Task for writing the content."""
124        return Task(
125            description="""Write a compelling blog post based on the
126            research, analysis, and strategy.
127
128            Requirements:
129            1. Follow the content outline structure
130            2. Write in a clear, engaging style
131            3. Include data and examples
132            4. Use subheadings for scannability
133            5. Include a compelling conclusion with takeaways
134
135            Target length: 1500-2000 words""",
136            expected_output="""A complete blog post in markdown format:
137            - Compelling headline
138            - Engaging introduction
139            - Well-structured body sections
140            - Data and examples integrated
141            - Strong conclusion with takeaways""",
142            agent=self.agents.content_writer(),
143            context=[research_task, analysis_task, strategy_task]
144        )
145
146    def editing_task(self, writing_task: Task) -> Task:
147        """Task for editing and polishing content."""
148        return Task(
149            description="""Edit and polish the blog post for publication.
150
151            Requirements:
152            1. Check all facts and figures
153            2. Improve clarity and flow
154            3. Tighten prose (remove redundancy)
155            4. Ensure consistent tone
156            5. Add meta description and tags
157
158            Maintain the author's voice while improving quality.""",
159            expected_output="""A publication-ready blog post:
160            - Polished content in markdown
161            - Meta description (150-160 characters)
162            - 5-7 relevant tags
163            - Any factual corrections noted""",
164            agent=self.agents.editor(),
165            context=[writing_task],
166            output_pydantic=BlogPost,
167            output_file="output/blog_post.md"
168        )

1# src/crew.py
2from crewai import Crew, Process
3from langchain_openai import ChatOpenAI
4from .agents import ContentAgents
5from .tasks import ContentTasks
6from config.settings import settings
7
8
9class ContentCreationCrew:
10    """Orchestrates the content creation workflow."""
11
12    def __init__(self, process: Process = Process.sequential):
13        self.agents_factory = ContentAgents()
14        self.tasks_factory = ContentTasks(self.agents_factory)
15        self.process = process
16
17    def create_crew(self, topic: str) -> Crew:
18        """Create a crew for the given topic."""
19
20        # Create tasks with dependencies
21        research = self.tasks_factory.research_task(topic)
22        analysis = self.tasks_factory.analysis_task(research)
23        strategy = self.tasks_factory.strategy_task(research, analysis)
24        writing = self.tasks_factory.writing_task(research, analysis, strategy)
25        editing = self.tasks_factory.editing_task(writing)
26
27        # Get unique agents (avoid duplicates)
28        agents = [
29            self.agents_factory.research_lead(),
30            self.agents_factory.trend_analyst(),
31            self.agents_factory.content_strategist(),
32            self.agents_factory.content_writer(),
33            self.agents_factory.editor()
34        ]
35
36        tasks = [research, analysis, strategy, writing, editing]
37
38        return Crew(
39            agents=agents,
40            tasks=tasks,
41            process=self.process,
42            verbose=settings.VERBOSE,
43            memory=settings.MEMORY_ENABLED
44        )
45
46    def run(self, topic: str) -> dict:
47        """Execute the content creation pipeline."""
48        crew = self.create_crew(topic)
49
50        try:
51            result = crew.kickoff()
52            return {
53                "success": True,
54                "result": result,
55                "topic": topic
56            }
57        except Exception as e:
58            return {
59                "success": False,
60                "error": str(e),
61                "topic": topic
62            }
63
64
65class HierarchicalContentCrew(ContentCreationCrew):
66    """Content crew with hierarchical management."""
67
68    def __init__(self):
69        super().__init__(process=Process.hierarchical)
70
71    def create_crew(self, topic: str) -> Crew:
72        """Create a hierarchically managed crew."""
73
74        # Same task setup
75        research = self.tasks_factory.research_task(topic)
76        analysis = self.tasks_factory.analysis_task(research)
77        strategy = self.tasks_factory.strategy_task(research, analysis)
78        writing = self.tasks_factory.writing_task(research, analysis, strategy)
79        editing = self.tasks_factory.editing_task(writing)
80
81        agents = [
82            self.agents_factory.research_lead(),
83            self.agents_factory.trend_analyst(),
84            self.agents_factory.content_strategist(),
85            self.agents_factory.content_writer(),
86            self.agents_factory.editor()
87        ]
88
89        tasks = [research, analysis, strategy, writing, editing]
90
91        # Manager LLM for hierarchical coordination
92        manager_llm = ChatOpenAI(
93            model=settings.PRIMARY_MODEL,
94            temperature=0.3  # Lower for more consistent management
95        )
96
97        return Crew(
98            agents=agents,
99            tasks=tasks,
100            process=Process.hierarchical,
101            manager_llm=manager_llm,
102            verbose=settings.VERBOSE,
103            memory=settings.MEMORY_ENABLED
104        )

1# main.py
2import argparse
3import json
4from datetime import datetime
5from pathlib import Path
6
7from src.crew import ContentCreationCrew, HierarchicalContentCrew
8from config.settings import settings
9
10
11def create_output_dir():
12    """Ensure output directory exists."""
13    Path(settings.OUTPUT_DIR).mkdir(parents=True, exist_ok=True)
14
15
16def save_result(result: dict, topic: str):
17    """Save result to file."""
18    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
19    filename = f"{settings.OUTPUT_DIR}/result_{timestamp}.json"
20
21    with open(filename, "w") as f:
22        json.dump({
23            "topic": topic,
24            "timestamp": timestamp,
25            "result": str(result.get("result", "")),
26            "success": result.get("success", False)
27        }, f, indent=2)
28
29    print(f"Result saved to: {filename}")
30
31
32def main():
33    parser = argparse.ArgumentParser(
34        description="AI Content Creation Crew"
35    )
36    parser.add_argument(
37        "topic",
38        type=str,
39        help="Topic to create content about"
40    )
41    parser.add_argument(
42        "--hierarchical",
43        action="store_true",
44        help="Use hierarchical process instead of sequential"
45    )
46    parser.add_argument(
47        "--verbose",
48        action="store_true",
49        help="Enable verbose output"
50    )
51
52    args = parser.parse_args()
53
54    # Update settings
55    if args.verbose:
56        settings.VERBOSE = True
57
58    # Create output directory
59    create_output_dir()
60
61    print(f"\n{'='*60}")
62    print(f"Content Creation Crew")
63    print(f"Topic: {args.topic}")
64    print(f"Process: {'Hierarchical' if args.hierarchical else 'Sequential'}")
65    print(f"{'='*60}\n")
66
67    # Create and run crew
68    if args.hierarchical:
69        crew = HierarchicalContentCrew()
70    else:
71        crew = ContentCreationCrew()
72
73    result = crew.run(args.topic)
74
75    # Handle result
76    if result["success"]:
77        print(f"\n{'='*60}")
78        print("Content created successfully!")
79        print(f"{'='*60}\n")
80        print(result["result"])
81    else:
82        print(f"\n{'='*60}")
83        print(f"Error: {result['error']}")
84        print(f"{'='*60}\n")
85
86    # Save result
87    save_result(result, args.topic)
88
89
90if __name__ == "__main__":
91    main()

1# src/tools.py
2from crewai.tools import BaseTool
3from pydantic import BaseModel, Field
4from typing import Type
5import httpx
6
7
8class AnalysisInput(BaseModel):
9    """Input for content analysis."""
10    content: str = Field(description="Content to analyze")
11
12
13class ContentAnalysisTool(BaseTool):
14    """Analyze content for patterns and insights."""
15
16    name: str = "Content Analyzer"
17    description: str = "Analyzes content for themes, patterns, and insights"
18    args_schema: Type[BaseModel] = AnalysisInput
19
20    def _run(self, content: str) -> str:
21        """Analyze content and return insights."""
22        # Simple analysis (in production, use NLP libraries)
23        word_count = len(content.split())
24        sentence_count = content.count('.') + content.count('!') + content.count('?')
25        avg_sentence_length = word_count / max(sentence_count, 1)
26
27        # Find frequent themes (simple approach)
28        words = content.lower().split()
29        word_freq = {}
30        for word in words:
31            if len(word) > 5:  # Focus on longer words
32                word_freq[word] = word_freq.get(word, 0) + 1
33
34        top_words = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:10]
35
36        return f"""Content Analysis:
37- Word count: {word_count}
38- Sentences: {sentence_count}
39- Avg sentence length: {avg_sentence_length:.1f} words
40- Top themes: {[w[0] for w in top_words]}"""
41
42
43class SEOInput(BaseModel):
44    """Input for SEO analysis."""
45    content: str = Field(description="Content to analyze for SEO")
46    keywords: str = Field(description="Target keywords (comma-separated)")
47
48
49class SEOTool(BaseTool):
50    """Analyze content for SEO optimization."""
51
52    name: str = "SEO Analyzer"
53    description: str = "Analyzes content for SEO optimization opportunities"
54    args_schema: Type[BaseModel] = SEOInput
55
56    def _run(self, content: str, keywords: str) -> str:
57        """Analyze SEO factors."""
58        content_lower = content.lower()
59        keyword_list = [k.strip().lower() for k in keywords.split(',')]
60
61        analysis = []
62        for keyword in keyword_list:
63            count = content_lower.count(keyword)
64            density = (count * len(keyword.split())) / len(content.split()) * 100
65            analysis.append(f"- '{keyword}': {count} occurrences ({density:.2f}% density)")
66
67        word_count = len(content.split())
68        has_headings = '##' in content or '<h' in content.lower()
69        has_lists = '- ' in content or '* ' in content
70
71        return f"""SEO Analysis:
72Keyword Analysis:
73{chr(10).join(analysis)}
74
75Content Factors:
76- Word count: {word_count} {'(good)' if 1000 < word_count < 2500 else '(consider adjusting)'}
77- Has headings: {'Yes' if has_headings else 'No (add headings)'}
78- Has lists: {'Yes' if has_lists else 'No (consider adding)'}
79
80Recommendations:
811. Ensure keywords appear in headings
822. Add meta description with primary keyword
833. Include internal and external links"""

1# Set environment variables
2export OPENAI_API_KEY="your-key"
3export SERPER_API_KEY="your-key"
4
5# Run with sequential process
6python main.py "The Future of AI Agents in 2025"
7
8# Run with hierarchical process
9python main.py "How LLMs are Transforming Software Development" --hierarchical
10
11# Verbose output
12python main.py "Building Production AI Systems" --verbose