CrewAI Integration

CrewAI is a cutting-edge framework for orchestrating role-playing, autonomous AI agents. Forjinn's CrewAI integration enables the creation of collaborative AI teams that work together to accomplish complex tasks through role-based cooperation.

Overview

CrewAI in Forjinn provides:

• Multi-agent Collaboration: Teams of specialized AI agents
• Role-Based Architecture: Agents with specific roles and responsibilities
• Task Orchestration: Coordinated task execution across agents
• Hierarchical Management: Structured agent hierarchies
• Process Automation: Automated workflows with agent handoffs
• Memory Sharing: Shared knowledge and context across agents

Core Concepts

Agents

Agents are autonomous AI entities with specific roles, goals, and capabilities.

agent Roles

{
  roles: {
    researcher: {
      description: 'Gathers and analyzes information',
      capabilities: ['web_search', 'data_analysis', 'fact_checking'],
      tools: ['google_search', 'arxiv_search', 'web_scraper']
    },
    writer: {
      description: 'Creates written content and documentation',
      capabilities: ['content_creation', 'editing', 'formatting'],
      tools: ['grammar_check', 'style_guide', 'plagiarism_check']
    },
    analyst: {
      description: 'Performs data analysis and insights',
      capabilities: ['statistical_analysis', 'visualization', 'reporting'],
      tools: ['python_repl', 'data_viz', 'statistical_tools']
    },
    manager: {
      description: 'Coordinates team activities and decisions',
      capabilities: ['task_delegation', 'quality_control', 'decision_making'],
      tools: ['project_management', 'communication', 'approval_workflow']
    }
  }
}

agent Configuration

from crewai import agent

researcher = agent(
    role='Senior Research Analyst',
    goal='Uncover cutting-edge developments in AI and data science',
    backstory="""You work at a leading tech think tank.
    Your expertise lies in identifying emerging trends.
    You have a knack for dissecting complex data and presenting actionable insights.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, scrape_tool]
)

Tasks

Tasks define specific objectives that agents need to accomplish.

Task Types

{
  task_types: {
    research_task: {
      description: 'Information gathering and analysis',
      inputs: ['topic', 'scope', 'sources'],
      outputs: ['research_report', 'key_findings', 'recommendations']
    },
    writing_task: {
      description: 'Content creation and documentation',
      inputs: ['requirements', 'style_guide', 'source_material'],
      outputs: ['written_content', 'formatted_document', 'metadata']
    },
    analysis_task: {
      description: 'Data processing and insights',
      inputs: ['dataset', 'analysis_type', 'parameters'],
      outputs: ['analysis_results', 'visualizations', 'summary']
    },
    review_task: {
      description: 'Quality assurance and validation',
      inputs: ['content_to_review', 'criteria', 'standards'],
      outputs: ['review_report', 'approval_status', 'feedback']
    }
  }
}

Task Configuration

from crewai import Task

research_task = Task(
    description="""Conduct a comprehensive analysis of the latest advancements 
    in AI LLMs in 2024. Identify key trends, breakthrough technologies, and 
    potential industry impacts.""",
    expected_output="A comprehensive 3 paragraphs long report on the latest AI advancements.",
    agent=researcher
)

Crews

Crews are teams of agents working together to accomplish complex objectives.

Crew Structure

{
  crew_structure: {
    agents: 'Collection of specialized agents',
    tasks: 'Sequence of tasks to be executed',
    process: 'Execution methodology (sequential, hierarchical, consensus)',
    memory: 'Shared knowledge and context',
    tools: 'Shared tools and resources',
    manager: 'Optional manager agent for coordination'
  }
}

Crew Configuration

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer, analyst],
    tasks=[research_task, writing_task, analysis_task],
    process=Process.sequential,
    memory=True,
    cache=True,
    max_rpm=100,
    share_crew=True
)

Process Types

Sequential Process

Agents work in a predefined sequence, with each agent building on the previous agent's work.

# Sequential execution
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential
)

Hierarchical Process

A manager agent coordinates and delegates tasks to worker agents.

# Hierarchical execution with manager
crew = Crew(
    agents=[manager, researcher, writer, analyst],
    tasks=[research_task, writing_task, analysis_task],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4")
)

Consensus Process

Agents collaborate and reach consensus on decisions and outputs.

# Consensus-based execution
crew = Crew(
    agents=[expert1, expert2, expert3],
    tasks=[evaluation_task],
    process=Process.consensus,
    consensus_threshold=0.7
)

Advanced Features

Memory System

Short-term Memory

Maintains context within a single crew execution.

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    memory=True,  # Enable short-term memory
    cache=True    # Enable result caching
)

Long-term Memory

Persists knowledge across multiple crew executions.

from crewai.memory import LongTermMemory

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    memory=LongTermMemory(
        storage_path="./crew_memory",
        embeddings_model="text-embedding-ada-002"
    )
)

Shared Memory

Enables agents to share knowledge and context.

from crewai.memory import SharedMemory

shared_memory = SharedMemory(
    storage_type="vector",
    embedding_model="sentence-transformers/all-MiniLM-L6-v2"
)

crew = Crew(
    agents=[agent1, agent2, agent3],
    tasks=[task1, task2, task3],
    memory=shared_memory
)

Tool Integration

Built-in Tools

from crewai_tools import (
    SerperDevTool,
    WebsiteSearchTool,
    FileReadTool,
    DirectoryReadTool,
    CodeInterpreterTool
)

# Search tools
search_tool = SerperDevTool()
website_tool = WebsiteSearchTool()

# File tools
file_tool = FileReadTool()
directory_tool = DirectoryReadTool()

# Code execution
code_tool = CodeInterpreterTool()

Custom Tools

from crewai_tools import BaseTool

class CustomAnalysisTool(BaseTool):
    name: str = "Data Analysis Tool"
    description: str = "Performs custom data analysis"
    
    def _run(self, data: str) -> str:
        # Custom analysis logic
        return analysis_result

Callbacks and Monitoring

Execution Callbacks

from crewai.callbacks import CrewCallback

class CustomCallback(CrewCallback):
    def on_task_start(self, task, agent):
        print(f"Task {task.description} started by {agent.role}")
    
    def on_task_complete(self, task, agent, output):
        print(f"Task completed: {output}")
    
    def on_crew_complete(self, crew, output):
        print(f"Crew execution completed: {output}")

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    callbacks=[CustomCallback()]
)

Performance Monitoring

import time
from crewai.telemetry import CrewTelemetry

telemetry = CrewTelemetry()

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    telemetry=telemetry
)

# Execute with monitoring
start_time = time.time()
result = crew.kickoff()
execution_time = time.time() - start_time

print(f"Execution time: {execution_time}s")
print(f"Telemetry data: {telemetry.get_metrics()}")

Configuration Examples

Research Team Configuration

from crewai import agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool

# Define tools
search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

# Define agents
researcher = agent(
    role='Senior Research Analyst',
    goal='Conduct thorough research on given topics',
    backstory='Expert researcher with 10+ years experience',
    tools=[search_tool, web_tool],
    verbose=True
)

analyst = agent(
    role='Data Analyst',
    goal='Analyze research data and extract insights',
    backstory='Skilled data analyst with expertise in statistical analysis',
    tools=[],
    verbose=True
)

writer = agent(
    role='Technical Writer',
    goal='Create comprehensive reports from research findings',
    backstory='Experienced technical writer specializing in research reports',
    tools=[],
    verbose=True
)

# Define tasks
research_task = Task(
    description='Research the latest trends in artificial intelligence',
    expected_output='Detailed research findings with sources',
    agent=researcher
)

analysis_task = Task(
    description='Analyze the research findings and identify key trends',
    expected_output='Analysis report with key insights and trends',
    agent=analyst
)

writing_task = Task(
    description='Write a comprehensive report based on research and analysis',
    expected_output='Well-structured report with executive summary',
    agent=writer
)

# Create crew
research_crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,
    memory=True,
    cache=True
)

Software Development Team

from crewai import agent, Task, Crew, Process
from crewai_tools import CodeInterpreterTool, FileReadTool

# Tools
code_tool = CodeInterpreterTool()
file_tool = FileReadTool()

# Agents
architect = agent(
    role='Software Architect',
    goal='Design software architecture and technical specifications',
    backstory='Senior architect with expertise in system design',
    tools=[file_tool],
    verbose=True
)

developer = agent(
    role='Senior Developer',
    goal='Implement software solutions based on specifications',
    backstory='Experienced developer proficient in multiple languages',
    tools=[code_tool, file_tool],
    verbose=True
)

tester = agent(
    role='QA Engineer',
    goal='Test software and ensure quality standards',
    backstory='Quality assurance expert with testing automation skills',
    tools=[code_tool],
    verbose=True
)

# Tasks
design_task = Task(
    description='Design the architecture for a web application',
    expected_output='Technical specification document',
    agent=architect
)

development_task = Task(
    description='Implement the web application based on specifications',
    expected_output='Working application code',
    agent=developer
)

testing_task = Task(
    description='Test the application and create test reports',
    expected_output='Test results and quality assessment',
    agent=tester
)

# Development crew
dev_crew = Crew(
    agents=[architect, developer, tester],
    tasks=[design_task, development_task, testing_task],
    process=Process.sequential,
    memory=True
)

Marketing Team Configuration

# Marketing-focused crew
market_researcher = agent(
    role='Market Research Specialist',
    goal='Analyze market trends and customer behavior',
    backstory='Marketing research expert with consumer insights expertise',
    tools=[search_tool, web_tool]
)

content_creator = agent(
    role='Content Marketing Manager',
    goal='Create engaging marketing content',
    backstory='Creative content specialist with brand marketing experience',
    tools=[]
)

campaign_manager = agent(
    role='Campaign Manager',
    goal='Plan and execute marketing campaigns',
    backstory='Strategic campaign manager with digital marketing expertise',
    tools=[]
)

# Marketing tasks
market_research_task = Task(
    description='Research target market and competitor analysis',
    expected_output='Market research report with insights',
    agent=market_researcher
)

content_creation_task = Task(
    description='Create marketing content based on research insights',
    expected_output='Marketing content package',
    agent=content_creator
)

campaign_planning_task = Task(
    description='Develop comprehensive marketing campaign strategy',
    expected_output='Campaign strategy document',
    agent=campaign_manager
)

marketing_crew = Crew(
    agents=[market_researcher, content_creator, campaign_manager],
    tasks=[market_research_task, content_creation_task, campaign_planning_task],
    process=Process.sequential,
    memory=True
)

Integration with Forjinn

Workflow Integration

CrewAI integrates seamlessly with Forjinn workflows:

{
  integration_features: {
    visual_crew_builder: 'Drag-and-drop crew creation',
    agent_templates: 'Pre-built agent role templates',
    task_orchestration: 'Visual task flow design',
    monitoring_dashboard: 'Real-time crew execution monitoring',
    result_handling: 'Automated result processing and routing'
  }
}

Deployment Options

{
  deployment_modes: {
    cloud_native: 'Scalable cloud deployment with auto-scaling',
    on_premise: 'Local deployment for data security',
    hybrid: 'Mix of cloud and on-premise agents',
    edge_computing: 'Edge deployment for low-latency requirements'
  }
}

Performance Optimization

# Optimized crew configuration
crew = Crew(
    agents=[agent1, agent2, agent3],
    tasks=[task1, task2, task3],
    process=Process.sequential,
    memory=True,
    cache=True,
    max_rpm=100,  # Rate limiting
    max_execution_time=3600,  # Timeout
    step_callback=lambda step: print(f"Step completed: {step}")
)

Best Practices

agent Design

1. Clear Roles: Define specific, non-overlapping roles for each agent
2. Appropriate Tools: Equip agents with relevant tools for their tasks
3. Realistic Backstories: Create believable agent personas
4. Goal Alignment: Ensure agent goals support overall crew objectives

Task Definition

1. Specific Descriptions: Provide clear, detailed task descriptions
2. Expected Outputs: Define exact output requirements
3. Dependencies: Clearly specify task dependencies
4. Success Criteria: Establish measurable success metrics

Crew Optimization

1. Right-sized Teams: Use appropriate number of agents for complexity
2. Process Selection: Choose the right process type for the use case
3. Memory Management: Enable memory for context-dependent tasks
4. Performance Monitoring: Implement comprehensive monitoring

Error Handling

try:
    result = crew.kickoff()
except Exception as e:
    print(f"Crew execution failed: {e}")
    # Implement fallback logic
    fallback_result = handle_crew_failure(e)

Monitoring and Analytics

Execution Metrics

from crewai.analytics import CrewAnalytics

analytics = CrewAnalytics()

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    analytics=analytics
)

result = crew.kickoff()

# Get execution metrics
metrics = analytics.get_metrics()
print(f"Total execution time: {metrics['execution_time']}")
print(f"Task completion rate: {metrics['completion_rate']}")
print(f"agent utilization: {metrics['agent_utilization']}")

Performance Dashboard

{
  dashboard_features: {
    real_time_monitoring: 'Live crew execution tracking',
    performance_metrics: 'agent and task performance analytics',
    resource_utilization: 'CPU, memory, and API usage tracking',
    cost_analysis: 'Execution cost breakdown and optimization',
    success_rates: 'Task and crew success rate tracking'
  }
}

Troubleshooting

Common Issues

1. agent Conflicts: Resolve role overlaps and conflicting goals
2. Task Dependencies: Fix circular or missing dependencies
3. Memory Issues: Optimize memory usage for large crews
4. Tool Failures: Implement tool fallbacks and error handling
5. Performance Bottlenecks: Identify and resolve slow agents/tasks

Debug Techniques

# Enable verbose logging
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    verbose=2,  # Maximum verbosity
    debug=True  # Enable debug mode
)

# Custom logging
import logging
logging.basicConfig(level=logging.DEBUG)

result = crew.kickoff()

CrewAI integration in Forjinn enables powerful multi-agent collaboration, allowing teams of AI agents to work together on complex tasks with role-based specialization and coordinated execution.