Introduction to CrewAI
CrewAI is a cutting-edge framework for coordinating autonomous AI agents.
CrewAI allows you to create AI teams, where each agent has specific roles, tools, and goals, working together to accomplish complex tasks.
Think of it as assembling your dream team—each member (agent) brings unique skills and expertise, collaborating seamlessly to achieve your objectives.
Recently, I used the CrewAI framework and found it to be a very user-friendly AI agent framework, so I recommend it to everyone.
CrewAI covers core concepts such as Agents, Tasks, Crews, Flows, Knowledge, LLMs, and Tools.
Next, I will introduce how to use CrewAI with a specific example.
The GitHub address for CrewAI is: https://github.com/crewAIInc/crewAI
Building a Translation Agent with CrewAI
Create a Python virtual environment, and install CrewAI and CrewAI-tools.
Run the command:
crewai create crew translation_agent
A template project will appear.
In the config directory, use YAML to configure the agents and tasks:
First, let’s set up the agents:
file_reader:
role: >
File reading agent
goal: >
Read file contents based on file path
backstory: >
You are a file reading agent, and your task is to read file contents based on the file path.
translation_agent:
role: >
Translation agent
goal: >
Translate text based on user requirements
backstory: >
You are a translation agent, and your task is to translate text based on user requirements.
file_saver:
role: >
File saving agent
goal: >
Save files based on user requirements
backstory: >
You are a file saving agent, and your task is to save files based on user requirements.
Here, we have set up three agents: the file reading agent, the translation agent, and the file saving agent.
Next, let’s configure the tasks:
file_read_task:
description: >
Based on user requirements: {question}
Get the file path to read
Use tools to read file contents
expected_output: >
Return file contents
agent: file_reader
translation_task:
description: >
Translate the text obtained from file_reader into English
expected_output: >
Return the translated text
agent: translation_agent
file_save_task:
description: >
Based on user requirements: {question} extract the file path and related information to save
Save the translation content from translation_agent to the specified file
expected_output: >
Return save result
agent: file_saver
Three tasks have been set up: file_read_task, translation_task, and file_save_task.
To complete these tasks, the agents need to use tools for reading and saving files.
You can write tool code in the tools directory:
File read tool code:
from typing import Any, Optional, Type
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
class FileReadToolSchema(BaseModel):
"""Input for FileReadTool."""
# Mandatory file full path to read the file
# 必须的文件全路径，以读取文件
file_path: str = Field(..., description="Mandatory file full path to read the file")
class FileReadTool(BaseTool):
"""A tool for reading file contents.
This tool inherits its schema handling from BaseTool to avoid recursive schema
definition issues. The args_schema is set to FileReadToolSchema which defines
the required file_path parameter. The schema should not be overridden in the
constructor as it would break the inheritance chain and cause infinite loops.
The tool supports two ways of specifying the file path:
1. At construction time via the file_path parameter
2. At runtime via the file_path parameter in the tool's input
Args:
file_path (Optional[str]): Path to the file to be read. If provided,
this becomes the default file path for the tool.
**kwargs: Additional keyword arguments passed to BaseTool.
Example:
>>> tool = FileReadTool(file_path="/path/to/file.txt")
>>> content = tool.run() # Reads /path/to/file.txt
>>> content = tool.run(file_path="/path/to/other.txt") # Reads other.txt
用于读取文件内容的工具。
该工具继承自 BaseTool 的 schema 处理，以避免递归 schema 定义问题。args_schema 设置为 FileReadToolSchema，定义了必需的 file_path 参数。构造函数中不应该覆盖 schema，否则会破坏继承链并导致无限循环。
该工具支持两种指定文件路径的方法：
在构造时通过 file_path 参数
在运行时通过工具的输入参数 file_path
参数：
file_path (可选[str]): 要读取的文件路径。如果提供，则成为工具的默认文件路径。
**kwargs: 传递给 BaseTool 的其他关键字参数。
示例：
>>> tool = FileReadTool(file_path="/path/to/file.txt")
>>> content = tool.run() # 读取 /path/to/file.txt
>>> content = tool.run(file_path="/path/to/other.txt") # 读取 other.txt
"""
name: str = "Read a file's content"
description: str = "A tool that reads the content of a file. To use this tool, provide a 'file_path' parameter with the path to the file you want to read."
args_schema: Type[BaseModel] = FileReadToolSchema
file_path: Optional[str] = None
def __init__(self, file_path: Optional[str] = None, **kwargs: Any) -> None:
"""Initialize the FileReadTool."""
Args:
file_path (Optional[str]): Path to the file to be read. If provided,
this becomes the default file path for the tool.
**kwargs: Additional keyword arguments passed to BaseTool.
初始化 FileReadTool。
参数：
file_path（可选[str]）：要读取的文件路径。如果提供，则此路径成为工具的默认文件路径。
**kwargs：传递给 BaseTool 的其他关键字参数。
"""
if file_path is not None:
kwargs['description'] = f"A tool that reads file content. The default file is {file_path}, but you can provide a different 'file_path' parameter to read another file."
super().__init__(**kwargs)
self.file_path = file_path
def _run(
self,
**kwargs: Any,
) -> str:
file_path = kwargs.get("file_path", self.file_path)
if file_path is None:
return "Error: No file path provided. Please provide a file path either in the constructor or as an argument."
try:
with open(file_path, "r", encoding='utf-8') as file:
return file.read()
except FileNotFoundError:
return f"Error: File not found at path: {file_path}"
except PermissionError:
return f"Error: Permission denied when trying to read file: {file_path}"
except Exception as e:
return f"Error: Failed to read file {file_path}. {str(e)}"
File writer tool code:
import os
from ast import literal_eval
from typing import Any, Optional, Type
from crewai.tools import BaseTool
from pydantic import BaseModel
class FileWriterToolInput(BaseModel):
filename: str
directory: Optional[str] = "./"
overwrite: str = "False"
content: str
class FileWriterTool(BaseTool):
name: str = "File Writer Tool"
description: str = "A tool to write content to a specified file. Accepts filename, content, and optionally a directory path and overwrite flag as input, overwrite flag is True or False."
args_schema: Type[BaseModel] = FileWriterToolInput
def _run(self, **kwargs: Any) -> str:
try:
# Create the directory if it doesn't exist
if kwargs.get("directory") and not os.path.exists(kwargs["directory"]):
os.makedirs(kwargs["directory"])
# Construct the full path
filepath = os.path.join(kwargs.get("directory") or "", kwargs["filename"])
# Convert overwrite to boolean
kwargs["overwrite"] = bool(literal_eval(kwargs["overwrite"]))
# Check if file exists and overwrite is not allowed
if os.path.exists(filepath) and not kwargs["overwrite"]:
return f"File {filepath} already exists and overwrite option was not passed."
# Write content to the file
mode = "w" if kwargs["overwrite"] else "x"
with open(filepath, mode) as file:
content = kwargs["content"]
file.write(content)
return f"Content successfully written to {filepath}"
except FileExistsError:
return (
f"File {filepath} already exists and overwrite option was not passed."
)
except KeyError as e:
return f"An error occurred while accessing key: {str(e)}"
except Exception as e:
return f"An error occurred while writing to the file: {str(e)}"
Now, we need to build a team.
Code to build the team:
from crewai import Agent, Crew, Process, Task, LLM
from crewai.project import CrewBase, agent, crew, task
from translation_agent.tools.file_read_tool import FileReadTool
from translation_agent.tools.file_writer_tool import FileWriterTool
import os
from dotenv import load_dotenv
load_dotenv()
file_read_tool = FileReadTool()
file_writer_tool = FileWriterTool()
api_key = os.getenv('OPENAI_API_KEY')
base_url = os.getenv('OPENAI_API_BASE')
model = os.getenv('OPENAI_MODEL_NAME', 'Qwen/Qwen2.5-72B-Instruct') # Provide a default model if not set
agent_llm = LLM(
model=model,
base_url=base_url,
api_key=api_key
)
# If you want to run a snippet of code before or after the crew starts,
# you can use the @before_kickoff and @after_kickoff decorators
# https://docs.crewai.com/concepts/crews#example-crew-class-with-decorators
@CrewBase
class TranslationAgent():
"""TranslationAgent crew"""
# Learn more about YAML configuration files here:
# Agents: https://docs.crewai.com/concepts/agents#yaml-configuration-recommended
# Tasks: https://docs.crewai.com/concepts/tasks#yaml-configuration-recommended
agents_config = 'config/agents.yaml'
tasks_config = 'config/tasks.yaml'
# If you would like to add tools to your agents, you can learn more about it here:
# https://docs.crewai.com/concepts/agents#agent-tools
# @agent
# def researcher(self) -> Agent:
# return Agent(
# config=self.agents_config['researcher'],
# verbose=True
# )
# @agent
# def reporting_analyst(self) -> Agent:
# return Agent(
# config=self.agents_config['reporting_analyst'],
# verbose=True
# )
@agent
def file_reader(self) -> Agent:
return Agent(
config=self.agents_config['file_reader'],
verbose=True,
llm=agent_llm,
tools=[file_read_tool],
)
@agent
def translation_agent(self) -> Agent:
return Agent(
config=self.agents_config['translation_agent'],
verbose=True,
llm=agent_llm,
)
@agent
def file_saver(self) -> Agent:
return Agent(
config=self.agents_config['file_saver'],
verbose=True,
llm=agent_llm,
tools=[file_writer_tool],
)
# To learn more about structured task outputs,
# task dependencies, and task callbacks, check out the documentation:
# https://docs.crewai.com/concepts/tasks#overview-of-a-task
@task
def file_read_task(self) -> Task:
return Task(
config=self.tasks_config['file_read_task'],
)
@task
def translation_task(self) -> Task:
return Task(
config=self.tasks_config['translation_task'],
)
@task
def file_save_task(self) -> Task:
return Task(
config=self.tasks_config['file_save_task'],
)
@crew
def crew(self) -> Crew:
"""Creates the TranslationAgent crew"""
# To learn how to add knowledge sources to your crew, check out the documentation:
# https://docs.crewai.com/concepts/knowledge#what-is-knowledge
return Crew(
agents=self.agents, # Automatically created by the @agent decorator
tasks=self.tasks, # Automatically created by the @task decorator
process=Process.sequential,
verbose=True,
# process=Process.hierarchical, # In case you wanna use that instead https://docs.crewai.com/how-to/Hierarchical/
)
To allow agents to use the Silicon Flow model, you can write it like this:
You need to prefix the model name with openai; otherwise, an error will occur.
If you haven’t registered yet, you can click the invitation link to register: https://cloud.siliconflow.cn/i/Ia3zOSCU.
Here, I will use the meta-llama/Llama-3.3-70B-Instruct with tool-calling capability as an example.
Then you can use it like this:
import os
from dotenv import load_dotenv
load_dotenv()
file_read_tool = FileReadTool()
file_writer_tool = FileWriterTool()
api_key = os.getenv('OPENAI_API_KEY')
base_url = os.getenv('OPENAI_API_BASE')
model = os.getenv('OPENAI_MODEL_NAME', 'Qwen/Qwen2.5-72B-Instruct') # Provide a default model if not set
agent_llm = LLM(
model=model,
base_url=base_url,
api_key=api_key
)
When creating agents, remember to use this large model and ensure to use tools:
@agent
def file_reader(self) -> Agent:
return Agent(
config=self.agents_config['file_reader'],
verbose=True,
llm=agent_llm,
tools=[file_read_tool],
)
Thus, the team is successfully built.
In main.py, write it like this:
#!/usr/bin/env python
import sys
import warnings
from datetime import datetime
from translation_agent.crew import TranslationAgent
warnings.filterwarnings("ignore", category=SyntaxWarning, module="pysbd")
# This main file is intended to be a way for you to run your
# crew locally, so refrain from adding unnecessary logic into this file.
# Replace with inputs you want to test with, it will automatically
# interpolate any tasks and agents information
def run():
"""Run the crew."""
inputs = {
'question': 'Read the contents of test.txt file, translate it into English, and then write it into test4.txt file',
}
try:
TranslationAgent().crew().kickoff(inputs=inputs)
except Exception as e:
raise Exception(f"An error occurred while running the crew: {e}")
def train():
"""Train the crew for a given number of iterations."""
inputs = {
"topic": "AI LLMs"
}
try:
TranslationAgent().crew().train(n_iterations=int(sys.argv[1]), filename=sys.argv[2], inputs=inputs)
except Exception as e:
raise Exception(f"An error occurred while training the crew: {e}")
def replay():
"""Replay the crew execution from a specific task."""
try:
TranslationAgent().crew().replay(task_id=sys.argv[1])
except Exception as e:
raise Exception(f"An error occurred while replaying the crew: {e}")
def test():
"""Test the crew execution and returns the results."""
inputs = {
"topic": "AI LLMs"
}
try:
TranslationAgent().crew().test(n_iterations=int(sys.argv[1]), openai_model_name=sys.argv[2], inputs=inputs)
except Exception as e:
raise Exception(f"An error occurred while testing the crew: {e}")
Focus mainly here:
def run():
"""Run the crew."""
inputs = {
'question': 'Read the contents of test.txt file, translate it into English, and then write it into test4.txt file',
}
try:
TranslationAgent().crew().kickoff(inputs=inputs)
except Exception as e:
raise Exception(f"An error occurred while running the crew: {e}")
Input the content for the question placeholder in the task in inputs.
Now create a test.txt file with the content:
CrewAI: A production-level framework for orchestrating complex AI agent systems. From simple automation to complex real-world applications, CrewAI provides precise control and deep customization. Facilitating collaborative intelligence through a flexible, production-ready architecture, CrewAI enables agents to collaborate seamlessly to solve complex business challenges with predictable and consistent results.
Now run crewai run and see the effect of this translation agent.
You can find that the file reading agent did not perform well as it included extra content.
Improvements are needed.
Change it like this and try again:
file_reader:
role: >
File reading agent
goal: >
Read file contents based on file path
backstory: >
You are a file reading agent, and your task is to read file contents based on the file path; just return the file contents.
Now the effect is very good, as shown below:
The translation agent translated very well, as shown below:
The file saving agent saved the translation result, as shown below:
Conclusion
This is the process and effect of building a translation agent using CrewAI. There are many interesting tools worth exploring in CrewAI, and next time I will introduce the use of the code interpreter tool.