Building Multi-Agent Systems with CrewAI - Orchestrating AI Teams

1from crewai import Agent
2from langchain_openai import ChatOpenAI
3
4# Initialize LLM
5llm = ChatOpenAI(model="gpt-4")
6
7# Create a researcher agent
8researcher = Agent(
9    role='Senior Research Analyst',
10    goal='Uncover cutting-edge developments in AI and data science',
11    backstory="""You are an expert research analyst with a keen eye for
12    emerging trends and technologies. You have a PhD in Computer Science
13    and years of experience in the AI industry.""",
14    verbose=True,
15    allow_delegation=False,
16    llm=llm,
17    max_iter=5
18)
19
20# Create a writer agent
21writer = Agent(
22    role='Tech Content Strategist',
23    goal='Craft compelling content on tech advancements',
24    backstory="""You are a renowned content creator known for making complex
25    technical topics accessible and engaging. Your articles are widely read
26    and shared in the tech community.""",
27    verbose=True,
28    allow_delegation=True,
29    llm=llm
30)
31
32# Create an editor agent
33editor = Agent(
34    role='Content Quality Assurance Specialist',
35    goal='Ensure all content meets the highest standards',
36    backstory="""You are a meticulous editor with an eye for detail and a
37    passion for quality. You've edited for top tech publications and know
38    what makes content truly exceptional.""",
39    verbose=True,
40    allow_delegation=False,
41    llm=llm
42)
43

1from crewai import Task
2
3research_task = Task(
4    description="""Conduct comprehensive research on the latest trends in
5    {topic}. Identify key developments, major players, and potential
6    future directions. Your final answer should be a detailed report
7    with citations and sources.""",
8    agent=researcher,
9    expected_output="A detailed research report with at least 10 key findings"
10)
11
12writing_task = Task(
13    description="""Using the research report, create an engaging blog post
14    about {topic}. The post should be approximately 800 words, include
15    relevant examples, and be accessible to a general tech audience.""",
16    agent=writer,
17    expected_output="An 800-word blog post in markdown format",
18    context=[research_task]  # Depends on research task
19)
20
21editing_task = Task(
22    description="""Review the blog post for accuracy, clarity, and
23    engagement. Fix any grammatical errors, improve flow, and ensure
24    technical accuracy. Provide the final polished version.""",
25    agent=editor,
26    expected_output="A polished, publication-ready blog post",
27    context=[writing_task]  # Depends on writing task
28)
29

1from crewai import Crew, Process
2
3# Create the crew
4content_crew = Crew(
5    agents=[researcher, writer, editor],
6    tasks=[research_task, writing_task, editing_task],
7    process=Process.sequential,  # Tasks executed in order
8    verbose=True
9)
10
11# Execute the crew
12result = content_crew.kickoff(inputs={"topic": "quantum computing"})
13print(result)
14

1manager = Agent(
2    role='Project Manager',
3    goal='Efficiently coordinate the team to produce high-quality content',
4    backstory="""You are an experienced project manager who excels at
5    coordinating teams and ensuring deadlines are met.""",
6    llm=llm,
7    allow_delegation=True
8)
9
10hierarchical_crew = Crew(
11    agents=[researcher, writer, editor],
12    tasks=[research_task, writing_task, editing_task],
13    process=Process.hierarchical,
14    manager_llm=llm,
15    verbose=True
16)
17
18result = hierarchical_crew.kickoff(inputs={"topic": "AI ethics"})
19

1from langchain.tools import Tool
2from langchain_community.tools import DuckDuckGoSearchRun
3
4# Web search tool
5search = DuckDuckGoSearchRun()
6
7# Custom database query tool
8def query_database(query: str) -> str:
9    """Query the internal knowledge base"""
10    # Your database logic here
11    return f"Database results for: {query}"
12
13db_tool = Tool(
14    name="DatabaseQuery",
15    func=query_database,
16    description="Query the internal knowledge base for company information"
17)
18
19# Agent with tools
20researcher_with_tools = Agent(
21    role='Senior Research Analyst',
22    goal='Conduct thorough research using all available resources',
23    backstory="""Expert researcher with access to multiple data sources.""",
24    tools=[search, db_tool],
25    llm=llm,
26    verbose=True
27)
28

1from crewai import Agent, Task, Crew
2from crewai.memory import EntityMemory, LongTermMemory
3
4# Create agents with memory
5analyst = Agent(
6    role='Data Analyst',
7    goal='Analyze data and identify trends',
8    memory=True,  # Enable memory
9    llm=llm
10)
11
12# Crew with memory capabilities
13memory_crew = Crew(
14    agents=[analyst, researcher, writer],
15    tasks=[analysis_task, research_task, writing_task],
16    memory=True,  # Enable crew-level memory
17    verbose=True
18)
19

1# Define specialized support agents
2triage_agent = Agent(
3    role='Support Ticket Triage Specialist',
4    goal='Quickly categorize and prioritize customer issues',
5    backstory="""You excel at understanding customer problems and
6    routing them to the right team.""",
7    llm=llm
8)
9
10technical_agent = Agent(
11    role='Senior Technical Support Engineer',
12    goal='Resolve technical issues efficiently',
13    backstory="""You have deep technical knowledge and can solve
14    complex problems.""",
15    tools=[kb_search_tool, ticket_system_tool],
16    llm=llm
17)
18
19communication_agent = Agent(
20    role='Customer Communication Specialist',
21    goal='Provide clear, empathetic responses to customers',
22    backstory="""You excel at translating technical solutions into
23    customer-friendly language.""",
24    llm=llm
25)
26
27# Define support workflow
28triage_task = Task(
29    description="Analyze the customer issue: {issue} and categorize it",
30    agent=triage_agent,
31    expected_output="Issue category and priority level"
32)
33
34resolution_task = Task(
35    description="Develop a solution for the categorized issue",
36    agent=technical_agent,
37    context=[triage_task],
38    expected_output="Technical solution steps"
39)
40
41response_task = Task(
42    description="Draft a customer-friendly response with the solution",
43    agent=communication_agent,
44    context=[resolution_task],
45    expected_output="Customer response email"
46)
47
48support_crew = Crew(
49    agents=[triage_agent, technical_agent, communication_agent],
50    tasks=[triage_task, resolution_task, response_task],
51    process=Process.sequential
52)
53
54# Handle a support ticket
55response = support_crew.kickoff(inputs={
56    "issue": "Customer cannot log in after password reset"
57})
58

1# SEO and topic research
2seo_specialist = Agent(
3    role='SEO Specialist',
4    goal='Identify high-value topics and keywords',
5    tools=[keyword_research_tool, competitor_analysis_tool],
6    llm=llm
7)
8
9# Content creation
10content_creator = Agent(
11    role='Content Creator',
12    goal='Create engaging, SEO-optimized content',
13    llm=llm
14)
15
16# Visual design
17designer = Agent(
18    role='Visual Designer',
19    goal='Create compelling visuals for content',
20    tools=[image_generation_tool, design_tool],
21    llm=llm
22)
23
24# Distribution
25marketing_agent = Agent(
26    role='Marketing Specialist',
27    goal='Optimize content distribution across channels',
28    tools=[social_media_tool, email_tool],
29    llm=llm
30)
31
32# Content pipeline tasks
33seo_task = Task(
34    description="Research and identify top 5 topics for {niche}",
35    agent=seo_specialist,
36    expected_output="List of 5 topics with keyword analysis"
37)
38
39content_task = Task(
40    description="Create comprehensive articles for identified topics",
41    agent=content_creator,
42    context=[seo_task],
43    expected_output="5 complete articles in markdown"
44)
45
46design_task = Task(
47    description="Create featured images and graphics for articles",
48    agent=designer,
49    context=[content_task],
50    expected_output="Image URLs and alt text for each article"
51)
52
53distribution_task = Task(
54    description="Create distribution plan and schedule posts",
55    agent=marketing_agent,
56    context=[content_task, design_task],
57    expected_output="Distribution schedule and social media posts"
58)
59
60content_pipeline = Crew(
61    agents=[seo_specialist, content_creator, designer, marketing_agent],
62    tasks=[seo_task, content_task, design_task, distribution_task],
63    process=Process.sequential,
64    verbose=True
65)
66
67result = content_pipeline.kickoff(inputs={"niche": "AI development"})
68

1code_reviewer = Agent(
2    role='Senior Code Reviewer',
3    goal='Identify bugs, security issues, and improvement opportunities',
4    tools=[static_analysis_tool, security_scanner],
5    llm=llm
6)
7
8test_engineer = Agent(
9    role='Test Engineer',
10    goal='Design comprehensive test cases',
11    tools=[test_framework_tool],
12    llm=llm
13)
14
15documentation_writer = Agent(
16    role='Technical Documentation Writer',
17    goal='Create clear documentation for code changes',
18    llm=llm
19)
20
21review_task = Task(
22    description="Review the code changes in PR #{pr_number}",
23    agent=code_reviewer,
24    expected_output="Detailed code review with findings"
25)
26
27test_task = Task(
28    description="Create test cases for the reviewed code",
29    agent=test_engineer,
30    context=[review_task],
31    expected_output="Comprehensive test suite"
32)
33
34docs_task = Task(
35    description="Update documentation based on code changes",
36    agent=documentation_writer,
37    context=[review_task],
38    expected_output="Updated documentation in markdown"
39)
40
41qa_crew = Crew(
42    agents=[code_reviewer, test_engineer, documentation_writer],
43    tasks=[review_task, test_task, docs_task],
44    process=Process.sequential
45)
46

1from crewai.telemetry import Telemetry
2
3class CrewMonitor:
4    def __init__(self):
5        self.metrics = {
6            'task_completion_times': [],
7            'agent_interactions': [],
8            'errors': []
9        }
10
11    def track_crew_execution(self, crew: Crew, inputs: dict):
12        start_time = time.time()
13
14        try:
15            result = crew.kickoff(inputs=inputs)
16            execution_time = time.time() - start_time
17
18            self.metrics['task_completion_times'].append({
19                'crew': crew.__class__.__name__,
20                'time': execution_time,
21                'success': True
22            })
23
24            return result
25        except Exception as e:
26            self.metrics['errors'].append({
27                'crew': crew.__class__.__name__,
28                'error': str(e),
29                'timestamp': time.time()
30            })
31            raise
32
33monitor = CrewMonitor()
34result = monitor.track_crew_execution(content_crew, {"topic": "AI"})
35

1from langchain.callbacks import get_openai_callback
2
3def optimize_crew_costs(crew: Crew, inputs: dict):
4    with get_openai_callback() as cb:
5        result = crew.kickoff(inputs=inputs)
6
7        print(f"Total Tokens: {cb.total_tokens}")
8        print(f"Total Cost: ${cb.total_cost:.4f}")
9
10        # Analyze and optimize
11        if cb.total_cost > 1.0:  # $1 threshold
12            print("Warning: High cost detected. Consider:")
13            print("- Using smaller models for simple tasks")
14            print("- Reducing max_iter for agents")
15            print("- Implementing caching")
16
17        return result
18

1# Good: Specific and clear
2data_analyst = Agent(
3    role='Financial Data Analyst',
4    goal='Analyze quarterly financial data and identify trends',
5    backstory='Expert in financial analysis with 10 years experience'
6)
7
8# Avoid: Too broad
9generic_agent = Agent(
10    role='AI Assistant',
11    goal='Help with various tasks',
12    backstory='General AI helper'
13)
14

1# Tasks with clear dependencies
2tasks = [
3    research_task,           # Step 1: Research
4    analysis_task,           # Step 2: Analyze (needs research)
5    writing_task,            # Step 3: Write (needs analysis)
6    review_task              # Step 4: Review (needs writing)
7]
8
9# Each task specifies its context
10analysis_task.context = [research_task]
11writing_task.context = [analysis_task]
12review_task.context = [writing_task]
13

1def execute_crew_safely(crew: Crew, inputs: dict, max_retries: int = 3):
2    for attempt in range(max_retries):
3        try:
4            return crew.kickoff(inputs=inputs)
5        except Exception as e:
6            if attempt == max_retries - 1:
7                raise
8            print(f"Attempt {attempt + 1} failed: {e}. Retrying...")
9            time.sleep(2 ** attempt)  # Exponential backoff
10