Session 4: CrewAI Team Orchestration - Enterprise Multi-Agent Collaboration¶

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileReadTool

# Research specialist with search tools

researcher = Agent(
    role='Research Specialist',
    goal='Gather comprehensive information on assigned topics',
    backstory='Expert researcher with access to web search capabilities',
    tools=[SerperDevTool()],
    verbose=True
)

# Content creation specialist

writer = Agent(
    role='Content Writer', 
    goal='Create engaging, well-structured content',
    backstory='Professional writer skilled in various content formats',
    verbose=True
)

# Quality assurance editor

editor = Agent(
    role='Content Editor',
    goal='Review and refine content for quality and accuracy',
    backstory='Experienced editor with attention to detail',
    verbose=True
)

# Detailed role configuration

data_analyst = Agent(
    role='Data Analyst',
    goal='Analyze data and extract meaningful insights',
    backstory='''You are a senior data analyst with 10 years of experience 
                 in statistical analysis and data visualization. You excel at 
                 finding patterns and trends in complex datasets.''',
    tools=[],  # Add analysis tools as needed
    allow_delegation=True,  # Can delegate tasks to other agents
    verbose=True,
    max_iter=3,  # Maximum iterations for complex tasks
    memory=True  # Remember previous interactions
)

# Sequential collaboration - agents work one after another

def create_content_team():
    return Crew(
        agents=[researcher, writer, editor],
        process=Process.sequential,  # One agent at a time
        verbose=True,
        memory=True
    )

# Hierarchical pattern requires a manager agent

def create_hierarchical_team():
    manager = Agent(
        role='Team Manager',
        goal='Coordinate team activities and ensure quality output',
        backstory='Experienced project manager with technical background',
        allow_delegation=True
    )

    return Crew(
        agents=[manager, researcher, writer, editor],
        process=Process.hierarchical,
        manager_llm='gpt-4',  # Manager uses more capable model
        verbose=True
    )

def create_research_crew():
    """Create a research-focused crew"""

    # Primary researcher - leads the research effort
    lead_researcher = Agent(
        role='Lead Researcher',
        goal='Conduct thorough research and coordinate findings',
        backstory='''Senior researcher with expertise in multiple domains.
                     Known for comprehensive analysis and clear communication.''',
        tools=[SerperDevTool(), FileReadTool()],
        allow_delegation=True,
        verbose=True
    )

    # Specialized fact-checker for verification
    fact_checker = Agent(
        role='Fact Checker',
        goal='Verify accuracy of research findings',
        backstory='''Detail-oriented professional specializing in fact verification
                     and source validation.''',
        tools=[SerperDevTool()],
        verbose=True
    )

    # Report synthesizer for final output
    synthesizer = Agent(
        role='Research Synthesizer', 
        goal='Combine research into coherent, actionable insights',
        backstory='''Expert at connecting disparate information sources and
                     creating comprehensive reports.''',
        verbose=True
    )

    return [lead_researcher, fact_checker, synthesizer]

def create_research_tasks(topic: str):
    """Define tasks for research crew"""

    # Primary research task
    research_task = Task(
        description=f'''Research the topic: {topic}

        Requirements:
        1. Find at least 5 credible sources
        2. Identify key trends and patterns
        3. Note any controversial aspects
        4. Provide source citations

        Output: Structured research findings with sources''',
        agent=lead_researcher,
        expected_output='Comprehensive research report with citations'
    )

    # Fact verification task builds on research findings
    verification_task = Task(
        description=f'''Verify the accuracy of research findings on: {topic}

        Requirements:
        1. Cross-check major claims against multiple sources
        2. Flag any questionable information
        3. Assess source credibility
        4. Rate overall research quality (1-10)

        Output: Fact-checking report with accuracy ratings''',
        agent=fact_checker,
        expected_output='Verification report with accuracy assessment'
    )

    # Synthesis task combines verified research into final output
    synthesis_task = Task(
        description=f'''Synthesize research and verification into final report

        Requirements:
        1. Combine verified research findings
        2. Highlight key insights and implications
        3. Identify areas for further investigation
        4. Create executive summary

        Output: Final research report with recommendations''',
        agent=synthesizer,
        expected_output='Executive summary and detailed findings report'
    )

    return [research_task, verification_task, synthesis_task]

def assemble_research_crew(topic: str):
    """Assemble and configure the complete crew"""

    # Get agents and tasks
    agents = create_research_crew()
    tasks = create_research_tasks(topic)

    # Create the crew with optimization settings
    crew = Crew(
        agents=agents,
        tasks=tasks,
        process=Process.sequential,
        verbose=True,
        memory=True,
        cache=True,  # Cache results for efficiency
        max_rpm=10,  # Rate limiting
        share_crew=False  # Privacy setting
    )

    return crew

# Usage example

topic = "Impact of AI on software development"
research_crew = assemble_research_crew(topic)
result = research_crew.kickoff()

def test_crew_creation():
    """Test that crews are created properly"""
    crew = assemble_research_crew("test topic")

    assert len(crew.agents) == 3
    assert len(crew.tasks) == 3
    assert crew.process == Process.sequential

    print("✅ Crew creation test passed!")

def test_crew_execution():
    """Test basic crew execution"""
    crew = assemble_research_crew("Python programming")

    # This would normally run the actual crew
    # For testing, we just verify structure
    assert crew is not None
    assert hasattr(crew, 'kickoff')

    print("✅ Crew execution test passed!")

def create_hierarchical_workflow():
    """Create a hierarchical crew with delegation"""

    # Team manager with delegation capabilities
    project_manager = Agent(
        role='Project Manager',
        goal='Coordinate team activities and ensure deliverable quality',
        backstory='''Experienced project manager with technical background.
                     Expert at resource allocation and timeline management.''',
        allow_delegation=True,
        verbose=True
    )

    # Backend specialist
    backend_dev = Agent(
        role='Backend Developer',
        goal='Design and implement server-side functionality',
        backstory='Senior backend developer specializing in scalable systems.',
        verbose=True
    )

    # Frontend specialist
    frontend_dev = Agent(
        role='Frontend Developer', 
        goal='Create user interfaces and user experiences',
        backstory='UI/UX focused developer with modern framework expertise.',
        verbose=True
    )

    # Complex project task requiring delegation
    project_task = Task(
        description='''Plan and coordinate development of a web application

        Requirements:
        1. Define system architecture
        2. Assign development tasks
        3. Coordinate between frontend and backend
        4. Ensure integration testing
        5. Prepare deployment plan

        Use delegation to assign specific tasks to team members.''',
        agent=project_manager,
        expected_output='Complete project plan with task assignments'
    )

    return Crew(
        agents=[project_manager, backend_dev, frontend_dev],
        tasks=[project_task],
        process=Process.hierarchical,
        manager_llm='gpt-4',
        verbose=True
    )

def process_crew_results(result):
    """Process and analyze crew results"""

    # Extract key information
    summary = {
        'total_tasks': len(result.tasks_output) if hasattr(result, 'tasks_output') else 0,
        'completion_status': 'completed' if result else 'failed',
        'output_length': len(str(result)),
        'key_insights': []
    }

    # Analyze result content for key insights
    result_text = str(result)
    if 'recommendation' in result_text.lower():
        summary['key_insights'].append('Contains recommendations')
    if 'analysis' in result_text.lower():
        summary['key_insights'].append('Includes analysis')

    return summary

# Usage example

crew = create_hierarchical_workflow()
result = crew.kickoff()
analysis = process_crew_results(result)
print(f"Result Analysis: {analysis}")

# Memory-enabled communication

def create_memory_enabled_crew():
    """Crew with enhanced memory and communication"""

    crew = Crew(
        agents=agents,
        tasks=tasks,
        process=Process.sequential,
        memory=True,  # Enable memory
        verbose=True,
        embedder={
            "provider": "openai",
            "config": {"model": "text-embedding-3-small"}
        }
    )

    return crew

# Cross-task information sharing

task_with_context = Task(
    description='''Build upon previous research findings.

    Context: Use information gathered by the research team.
    Review their findings and expand upon the most promising areas.

    Requirements:
    1. Reference previous research explicitly
    2. Build upon existing findings
    3. Identify gaps or areas for deeper investigation''',
    context=[previous_research_task],  # Reference to earlier task
    agent=analysis_agent
)

def create_optimized_crew():
    """Create performance-optimized crew"""

    crew = Crew(
        agents=agents,
        tasks=tasks,
        process=Process.sequential,

        # Performance optimizations
        cache=True,           # Cache intermediate results
        max_rpm=30,          # Increase rate limit
        memory=True,         # Enable memory for context

        embedder={           # Efficient embeddings
            "provider": "openai",
            "config": {"model": "text-embedding-3-small"}
        },

        # Resource management
        max_execution_time=300,  # 5 minute timeout
        step_callback=lambda step: print(f"Step completed: {step}")
    )

    return crew

import time

def monitor_crew_execution(crew, task_description):
    """Monitor crew execution with basic metrics"""

    start_time = time.time()

    print(f"Starting crew execution: {task_description}")
    result = crew.kickoff()

    end_time = time.time()
    execution_time = end_time - start_time

    print(f"⏱️ Execution time: {execution_time:.2f} seconds")
    print(f"📊 Result length: {len(str(result))} characters")
    print(f"✅ Crew execution completed")

    return result

# Performance best practices organized by category

optimization_tips = {
    'agent_design': [
        'Use specific, focused roles',
        'Provide clear backstories and goals',
        'Limit tool sets to essentials'
    ],
    'task_design': [
        'Write clear, specific descriptions',
        'Set realistic expectations',
        'Use context to connect related tasks'
    ],
    'crew_configuration': [
        'Enable caching for repeated operations',
        'Use memory for context continuity',
        'Set appropriate rate limits'
    ]
}