“If models can be trained, can intelligent agents be trained too?”
The answer is, yes!
Crew.ai provides training methods for intelligent agents. Through training, the decision-making and problem-solving capabilities of the agents can be enhanced.
The core training method is autonomous learning and iterative training, this method emphasizes the agent’s continuous learning and optimization of its behavior through interaction with the environment (including humans) and from data/feedback.
It is important to note that Crew.ai optimizes the performance of agents relying on human feedback feedback, known as Human-in-the-loop.
Well, you humans are also a part of the play. This is a clever training method, and the steps are as follows:
- Agent generates results: The agent executes tasks based on the current model and generates preliminary results.
- Human feedback: Human experts evaluate the results, providing suggestions for improvement or correcting errors.
- Model adjustment: Based on human feedback, the agent’s model is adjusted and optimized.
- Repeat iteration: This process is repeated until the agent’s performance reaches the expected level.
The Human-in-the-loop training method is particularly suitable for scenarios that require high precision and reliability, such as medical diagnosis or financial analysis.
By incorporating human feedback, the agent can learn complex tasks faster and reduce errors.
Crew.ai’s intelligent agent training model uses the command line and mainly includes the following steps:
- Define the number of training iterations: The training iteration parameter is defined as n_iterations, which must be a positive integer.
- Define the input parameters for the training process.
- To prevent the training process from being interrupted, a try-except module will be used to handle potential errors.
Refer to the specific code below:
def train_crew():
n_iterations = 2 # Number of training iterations
inputs = {"topic": "CrewAI Training"}
filename = "tech_blog_writer.pkl"
try:
crew = SequentialProcessExample() # Crew case for this test
crew.crew().train(
n_iterations=n_iterations,
inputs=inputs,
filename=filename
)
except Exception as e:
raise Exception(f"An error occurred while training the crew: {e}")
if __name__ == "__main__":
#topic = "The Application of Artificial Intelligence in Data Analysis"
# result = run_crew(topic)
#print(result)
train_crew()
It is also important to note that the filename is the record file for the training results, and the file format is .pkl . After completing the training, a .pkl file will be automatically created in the specified folder.
Because the complexity of agent training is not exactly the same, and each iteration iteration relies on human feedback, the training process may take a considerable amount of time.
Each time a training is completed, the agent’s abilities and knowledge will be enhanced, allowing it to handle more complex tasks and provide more stable and valuable insights.
Without further ado, let’s jump into a case study.
Before we begin, let me briefly introduce this multi-agent case.
This agent serves the purpose of writing technology-related blogs, including the following main agent roles:
The execution process of the test is very simple, just one line of code to call crew.py.
python crew.pyThe entire training process is lengthy and has a lot of content. I have made a diagram of each agent’s tasks and corresponding feedback, as shown below:
After 3 complete cycles, the overall training is completed.
So, where are the training results?
The results of the entire training process will form a training memory file in the format of .pkl, becoming part of the entire project.
The corresponding source code is as follows. Interested friends can give it a try:
agents.yaml code
searcher:
role: >
{topic} Content Researcher
goal: >
Reveal the cutting-edge developments of {topic}, considering that the current year is 2025.
backstory: >
You are an experienced researcher known for uncovering the latest developments in {topic}. You are renowned for finding the most relevant information and presenting it clearly and concisely.
copywriter:
role: >
Content Copywriter
goal: >
Write engaging and informative content based on the provided information.
backstory: >
You are a talented copywriter known for creating engaging and informative content. You are renowned for distilling complex information into clear and compelling copy.
case_provider:
role: >
{topic} Case Provider
goal: >
Provide real and relevant case support for the article content.
backstory: >
You are a professional case study expert known for uncovering and analyzing real cases related to the topic. You excel at finding compelling real-world applications and providing verifiable case source links. Your case analysis always accurately demonstrates the application of theory in practice.
editor:
role: >
{topic} Article Editor
goal: >
Integrate copy and cases to compose a well-structured article.
backstory: >
You are a seasoned article editor known for integrating and optimizing content. You know how to perfectly combine copy and case studies to create articles that are both deep and easy to understand. You pay special attention to the logic and readability of the article, ensuring smooth transitions between content and appropriate integration of cases.
file_writer:
role: >
File Writer
goal: >
Write the extracted information into a file.
backstory: >
You are a skilled writer known for creating well-structured and informative documents. You are renowned for presenting information clearly and concisely.
tasks.yaml code
search_task:
description: |
Research {topic} in depth to ensure all interesting and relevant information is found.
expected_output: |
A list of 10 key points about {topic}.
agent: searcher
content_write_task:
description: |
Write a blog article about {topic}, ensuring all relevant information is included and making it engaging for readers.
expected_output: |
A blog article about {topic} with at least 2000 words.
agent: copywriter
case_provide_task:
description: |
Based on the points in the article, search for and provide at least 3 relevant real cases. Each case needs to include:
1. Case title
2. Case details
3. Case source link
4. Analysis of the relationship with the article's points.
expected_output: |
A report containing detailed case information, including case descriptions and source links.
agent: case_provider
edit_task:
description: |
Integrate the copywriter's content and the case provider's cases into a complete article. Requirements:
1. Ensure clear structure and logical coherence of the article.
2. Insert relevant cases at appropriate positions.
3. Add necessary transition paragraphs.
4. Optimize titles and subtitles.
5. Ensure close connection between cases and the article's points.
expected_output: |
A well-structured and content-rich article containing main content, relevant cases, and case links. Please write in Chinese.
agent: editor
file_write_task:
description: |
Save the article as a Word document, with the filename "{topic}.docx"
expected_output: |
The article has been saved as {topic}.docx
agent: file_writer
crew.py code
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 filename safety
#safe_filename = "".join([c for c in filename if c.isalnum() or c in (' ', '-', '_')]).rstrip()
#if not safe_filename:
safe_filename = "article"
# Add timestamp
from datetime import datetime
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
safe_filename = f"{safe_filename}_{timestamp}"
# Add .docx extension (if not already)
if not safe_filename.endswith('.docx'):
safe_filename += '.docx'
# Create Word document
doc = Document()
doc.add_heading(safe_filename, 0)
doc.add_paragraph(content)
# Save document
doc.save(safe_filename)
return f"Article has been successfully saved as {safe_filename}"
except Exception as e:
return f"An error occurred while saving the 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')
@agent
def searcher(self) -> Agent:
return Agent(
config=self.agents_config['searcher'],
verbose=True,
tools=[SerperDevTool()],
llm=llm
)
#@agent
def scraper(self) -> Agent:
return Agent(
config=self.agents_config['scraper'],
verbose=True,
tools=[ScrapeWebsiteTool()],
llm=llm
)
@agent
def copywriter(self) -> Agent:
return Agent(
config=self.agents_config['copywriter'],
verbose=True,
llm=llm
)
@agent
def case_provider(self) -> Agent:
return Agent(
config=self.agents_config['case_provider'],
verbose=True,
tools=[SerperDevTool()],
llm=llm
)
@agent
def editor(self) -> Agent:
return Agent(
config=self.agents_config['editor'],
verbose=True,
llm=llm
)
@agent
def file_writer(self) -> Agent:
return Agent(
config=self.agents_config['file_writer'],
verbose=True,
tools=[CustomFileWriterTool()],
llm=llm
)
@task
def search_task(self) -> Task:
return Task(config=self.tasks_config['search_task'], verbose=True)
#@task
#def scrape_task(self) -> Task:
# return Task(config=self.tasks_config['scrape_task'], verbose=True)
@task
def content_write_task(self) -> Task:
return Task(config=self.tasks_config['content_write_task'], verbose=True)
@task
def case_provide_task(self) -> Task:
return Task(config=self.tasks_config['case_provide_task'], verbose=True)
@task
def edit_task(self) -> Task:
return Task(config=self.tasks_config['edit_task'], verbose=True)
@task
def file_write_task(self) -> Task:
return Task(config=self.tasks_config['file_write_task'], verbose=True)
@crew
def crew(self) -> Crew:
return Crew(
agents=self.agents,
tasks=[
self.search_task(),
#self.scrape_task(),
self.content_write_task(),
self.case_provide_task(),
self.edit_task(),
self.file_write_task()
],
process=Process.sequential,
verbose=True,
)
def run_crew(topic):
crew = SequentialProcessExample()
result = crew.crew().kickoff(inputs={'topic': topic})
return result
def train_crew():
n_iterations = 2
inputs = {"topic": "CrewAI Training"}
filename = "tech_blog_writer.pkl"
try:
crew = SequentialProcessExample()
crew.crew().train(
n_iterations=n_iterations,
inputs=inputs,
filename=filename
)
except Exception as e:
raise Exception(f"An error occurred while training the crew: {e}")
if __name__ == "__main__":
#topic = "The Application of Artificial Intelligence in Data Analysis"
# result = run_crew(topic)
#print(result)
train_crew()