Detailed Analysis of Crew.ai Agent Workflow | AI Work Mode

In today’s rapidly evolving technological landscape, artificial intelligence (AI) agents have become key tools for optimizing workflows, improving efficiency, and reducing labor costs.

Recently, Google released a white paper on AI Agents, and industry insiders have even directly defined 2025 as the era of Agents.

Recently, I tried Crew.ai, an advanced AI agent platform that provides powerful automation solutions for businesses and developers through its unique task allocation model.

Today, with two case studies, I will delve into the basic concepts, advantages, core components, and agent workflow of Crew.ai, helping everyone to comprehensively understand how this technology can be applied in practice.

However, before we begin, let’s review the basic concepts of AI agents and the corresponding supporting components of Crew.ai.

Basic Concepts of AI Agents

First, let’s take a look at the components of AI agents.

Agent

An AI agent is a software entity that simulates human decision-making processes, capable of autonomously executing tasks and making decisions. In Crew.ai, agents optimize their performance by learning and adapting to their environment.

Action

Actions are specific operations performed by agents, which can include data processing, image recognition, or any other automated operations. Crew.ai agents can identify action types and select appropriate tools and methods based on task requirements.

Tools

Agents rely on various tools and technologies, such as machine learning models and data processing tools, to execute tasks. The selection and configuration of these tools directly impact the efficiency and effectiveness of the agents.

Memory

The memory is the core component where agents store and retrieve information, containing all the data and rules needed for agents to execute tasks. Crew.ai ensures that agents can quickly and accurately access the required information through efficient knowledge management.

Planning

Planning is crucial for agents to complete tasks. Crew.ai defines clear processes to ensure that tasks can be executed as planned while allowing flexible adjustments to adapt to changing needs.

Advantages and Features of Crew.ai

Crew.ai stands out for its efficiency, flexibility, and scalability. It can quickly handle complex tasks, adapt to different work environments, and support large-scale deployment and integration, making it an ideal choice for automating business processes.

Core Components of Crew.ai

Agent

• Definition: An agent is an autonomous unit in Crew.ai that has the ability to execute tasks, make decisions, and communicate with other agents, simulating human decision-making processes.
• Functionality: Agents optimize performance by learning and adapting to their environment, capable of identifying action types and selecting appropriate tools and methods to execute tasks, such as a researcher agent responsible for collecting and analyzing information, and a writer agent focused on creating content.

Task

• Definition: A task is the specific work that an agent needs to complete, including various details required for execution, such as a clear description, the agent responsible for execution, and the tools needed.
• Attributes: Tasks have multiple attributes, including description, agent, expected output, tools, asynchronous execution, context, configuration, output format, callback functions, etc., which collectively define the execution details and expected results of the task.

Tool

• Definition: Tools are various technologies and software that agents rely on when executing tasks, such as machine learning models and data processing tools, empowering agents with multiple capabilities.
• Features: Tools have key features such as practicality, integration, customizability, error handling, and caching mechanisms, enabling seamless integration into workflows, enhancing the overall capability of agents, and achieving efficient collaboration.

Process

• Definition: A process is a series of steps and rules that agents follow to complete tasks, ensuring that tasks can be executed as planned, allowing individual agents to operate cohesively as a whole.
• Types: Processes include sequential execution (tasks are executed in a predefined order), hierarchical (tasks are supervised by manager agents), and consensus processes (collaborative decision-making among agents—this has not yet been implemented).

Teams

• Definition: A team is a collection of multiple agents working together to complete complex tasks, improving overall work efficiency through reasonable process design.
• Attributes: Teams have attributes such as tasks, agents, processes, detail levels, manager LLM, configuration, maximum RPM, language, language files, memory, cache, embedders, complete output, step callbacks, task callbacks, etc., which collectively define the behavior and execution strategy of the team.

Memory

• Definition: The memory system is an important component of Crew.ai, including short-term memory, long-term memory, entity memory, and context memory, used for storing and retrieving information.
• Functionality: The memory system enhances agents’ contextual awareness, experience accumulation, and entity understanding capabilities by storing and utilizing past experiences and information, thereby improving the quality of decision-making and problem-solving of agents.

Today, let’s focus on the process of Crew.ai.

Agent Workflow of Crew.ai

Currently, Crew.ai supports two types of workflows: sequential and hierarchical.

In simple terms, a sequential workflow is like an assembly line, while a hierarchical workflow involves a leading agent assigning tasks to subordinates.

Sequential Workflow

A sequential workflow refers to tasks being executed in a predetermined order. Only after the previous agent’s task is completed will the next agent’s task start.

For example, in a content writing team, the content planner needs to formulate a content plan, followed by the content author writing the article, and finally, the content editor optimizing the content.

The sequential working mode of the entire content writing team is illustrated as follows:

Here, I created a Gradio page to allow everyone to see the inputs and outputs of the entire agent team more intuitively (you might have noticed that this article was written by this agent team, haha);

The corresponding agent definitions, task allocations, and workflow code are as follows:

def create_agents(topic):
    """Create agents"""
    planner = Agent(
        role="内容规划师",
        goal=f"规划关于 {topic} 的引人入胜且事实准确的内容",
        backstory=f"您正在规划一篇主题和内容结构是: {topic} 的技术博客文章。"
                "您收集有助于受众学习的"
                "信息。"
                "您的工作是内容作者撰写该主题文章的基础。",
        allow_delegation=False,
        verbose=True,
        llm=llm
    )
    
    writer = Agent(
        role="内容作者",
        goal=f"根据规划撰写关于主题和内容结构是: {topic} 的高质量文章",
        backstory="您是一位经验丰富的技术文章作者，"
                "擅长将复杂的概念转化为易于理解的内容。",
        allow_delegation=False,
        verbose=True,
        llm=llm
    )
    
    editor = Agent(
        role="内容编辑",
        goal="审查和优化文章质量",
        backstory="您是一位资深编辑，"
                "确保文章的准确性、可读性和吸引力。",
        allow_delegation=False,
        verbose=True,
        llm=llm
    )
    
    return planner, writer, editor

def create_tasks(topic, planner, writer, editor):
    """Create tasks"""
    plan = Task(
        description=(
            f"1. 优先考虑关于主题和内容结构是: {topic} 的最新趋势、关键参与者和"
                "值得注意的新闻。\n"
            "2. 确定目标受众，考虑"
                "他们的兴趣和痛点。\n"
            "3. 根据输入的{topic},优化详细的内容大纲，包括"
                "背景和要点。\n"
            "4. 包括 SEO 关键词和相关数据或来源。"
        ),
        expected_output="一份全面的内容计划文档，"
            "包含大纲、受众分析、"
            "SEO 关键词和资源。",
        agent=planner
    )

    write = Task(
        description=(
            "根据内容计划撰写文章，确保：\n"
            "1. 内容清晰、引人入胜\n"
            "2. 使用具体示例和数据支持观点\n"
            "3. 适当使用过渡语句\n"
            "4. 包含引人入胜的开场和有力的结论"
        ),
        expected_output="一篇结构完整、内容丰富的文章初稿",
        agent=writer
    )

    edit = Task(
        description=(
            "审查并优化文章：\n"
            "1. 检查事实准确性和逻辑连贯性\n"
            "2. 优化标题和小标题\n"
            "3. 改进段落结构和文章流畅度\n"
            "4. 确保语言清晰简洁\n"
            "5. 检查拼写和语法"
        ),
        expected_output="一篇经过优化的最终文章",
        agent=editor
    )
    
    return [plan, write, edit]

def generate_article(topic):
    """Main function to generate article"""
    try:
        # Create agents and tasks
        planner, writer, editor = create_agents(topic)
        tasks = create_tasks(topic, planner, writer, editor)
        
        # Create Crew
        crew = Crew(
            agents=[planner, writer, editor],
            tasks=tasks,
            verbose=True
        )
        
        # Execute task flow and get results
        result = crew.kickoff(inputs={"topic": topic})
        
        # Get final article content
        try:
            # Try to get the output of the last task
            tasks_outputs = result.tasks_outputs
            if tasks_outputs and len(tasks_outputs) > 0:
                final_content = tasks_outputs[-1]  # Get the output of the last task
            else:
                final_content = str(result)
        except AttributeError:
            try:
                final_content = str(result)
            except:
                final_content = "无法获取文章内容"
                
        return final_content
        
    except Exception as e:
        print(f"发生错误: {str(e)}")
        import traceback
        print(traceback.format_exc())
        return None

Hierarchical Workflow

The hierarchical workflow involves tasks being assigned and executed in a hierarchical structure. Note that there is a leader in this workflow.

In project management, from project planning, task assignment to execution monitoring, each level has clear responsibilities and tasks.

The case involves a team handling customer issues. The customer support manager is the leader of the entire team, managing both technical support specialists and billing support specialists. Here, the customer support manager needs to determine whether the customer’s inquiry is a billing issue or a technical issue. Once determined, the corresponding specialist is assigned to handle it. Ultimately, the customer issue report is also provided by the customer support manager.

Below is the workflow of the entire team when receiving a ticket with the content: “Why was 100 yuan deducted this month?”

Similarly, I have also posted the corresponding Gradio interface, where everyone can intuitively see the team’s work results.

The corresponding code example is as follows:

1-agents.yaml

manager:
  role: >
    客户支持经理
  goal: >
    监督支持团队，确保及时有效地解决客户咨询
  backstory: >
    你是一位经验丰富的客户支持经理，具有领导支持团队的广泛经验。
    你的主要职责是协调支持代理的工作，确保客户问题得到及时和满意的解决。
    你擅长任务分配、绩效监控，并保持高标准的客户服务。

technical_support:
  role: >
    技术支持专家
  goal: >
    及时有效地解决客户报告的技术问题
  backstory: >
    你是一位技术支持专家，拥有丰富的故障排除软件和硬件问题的背景。
    你的主要职责是协助客户解决技术问题，确保他们的满意度和产品的顺利运行。

billing_support:
  role: >
    账单支持专家
  goal: >
    处理与账单、支付和账户管理相关的客户咨询
  backstory: >
    你是一位经验丰富的账单支持专家，擅长处理客户的账单咨询。
    你的主要目标是提供清晰准确的账单流程信息，解决支付问题，并协助账户管理，以确保客户满意。

2-task.yaml

categorize_tickets:
  description: >
    根据内容对来电客户支持工单进行分类：'{ticket}'，
    以确定它是技术问题还是账单问题。
  expected_output: >
    一个标记为'技术'或'账单'的分类工单。
  agent: manager

resolve_technical_issues:
  description: >
    解决工单中描述的技术问题：'{ticket}'
  expected_output: >
    为每个技术问题提供详细解决方案。
  agent: technical_support

resolve_billing_issues:
  description: >
    解决工单中描述的账单问题：'{ticket}'
  expected_output: >
    为每个与账单相关的咨询提供全面的回应。
  agent: billing_support

quality_assurance_review:
  description: >
    审核提供的技术和账单问题的解决方案，以确保准确性和客户满意度。
  expected_output: >
    一份确认解决方案质量和准确性的报告。
  agent: manager 

3-crew.py

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import SerperDevTool, ScrapeWebsiteTool, FileWriterTool
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os
from docx import Document

# Load environment variables
load_dotenv()

# Initialize LLM using Deepseek API
llm = ChatOpenAI(
    api_key=os.getenv('DEEPSEEK_API_KEY'),
    base_url=os.getenv('DEEPSEEK_API_BASE'),
    model_name="deepseek/deepseek-chat",
    temperature=0.7,
)

class CustomFileWriterTool(FileWriterTool):
    def _write_file(self, content: str, filename: str) -> str:
        """Custom file writing tool"""
        try:
            # Ensure safe filename
            safe_filename = "".join([c for c in filename if c.isalnum() or c in (' ', '-', '_')]).rstrip()
            if not safe_filename:
                safe_filename = "article"
                
            # Add .docx extension if not present
            if not safe_filename.endswith('.docx'):
                safe_filename += '.docx'
                
            # Create Word document
            doc = Document()
            doc.add_heading(filename, 0)
            doc.add_paragraph(content)
            
            # Save document
            doc.save(safe_filename)
            return f"Article successfully saved as {safe_filename}"
        except Exception as e:
            return f"Error occurred while saving file: {str(e)}"

@CrewBase
class SequentialProcessExample:
    """ProcessExample crew"""

    # Use relative path
    base_path = os.path.dirname(os.path.abspath(__file__))
    agents_config = os.path.join(base_path, 'agents.yaml')
    tasks_config = os.path.join(base_path, 'tasks.yaml')

    @task
    def categorize_tickets(self) -> Task:
        return Task(config=self.tasks_config['categorize_tickets'], verbose=True)

    @task
    def resolve_technical_issues(self) -> Task:
        return Task(config=self.tasks_config['resolve_technical_issues'], verbose=True)

    @task
    def resolve_billing_issues(self) -> Task:
        return Task(config=self.tasks_config['resolve_billing_issues'], verbose=True)

    @task
    def quality_assurance_review(self) -> Task:
        return Task(config=self.tasks_config['quality_assurance_review'], verbose=True)

    @agent
    def technical_support(self) -> Agent:
        return Agent(
            config=self.agents_config['technical_support'], 
            verbose=True,
            llm=llm
        )

    @agent
    def billing_support(self) -> Agent:
        return Agent(
            config=self.agents_config['billing_support'], 
            verbose=True,
            llm=llm
        )

    @agent
    def manager(self) -> Agent:
        return Agent(
            config=self.agents_config['manager'],
            verbose=True,
            llm=llm
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=[self.technical_support(), self.billing_support()],
            tasks=[
                self.categorize_tickets(),
                self.resolve_technical_issues(),
                self.resolve_billing_issues(),
                self.quality_assurance_review()
            ],
            manager_agent=self.manager(),
            process=Process.hierarchical,
            verbose=True,
        )

Through this article, I hope you can gain a comprehensive understanding of the task allocation model of Crew.ai’s agents and consider how to apply it to your own business for greater automation and efficiency.

We will continue to share case studies using Crew.ai for actual business support. If you like it, please give it a thumbs up, muah.