mirror of
https://github.com/crewAIInc/crewAI.git
synced 2025-12-16 04:18:35 +00:00
89c4b3fe88
Co-authored-by: Brandon Hancock (bhancock_ai) <109994880+bhancockio@users.noreply.github.com>
1286 lines
42 KiB
Python
1286 lines
42 KiB
Python
"""Test Agent creation and execution basic functionality."""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from pydantic import BaseModel
|
|
from pydantic_core import ValidationError
|
|
|
|
from crewai import Agent, Crew, Process, Task
|
|
from crewai.tasks.conditional_task import ConditionalTask
|
|
from crewai.tasks.task_output import TaskOutput
|
|
from crewai.utilities.converter import Converter
|
|
|
|
|
|
def test_task_tool_reflect_agent_tools():
|
|
from crewai.tools import tool
|
|
|
|
@tool
|
|
def fake_tool() -> None:
|
|
"Fake tool"
|
|
|
|
researcher = Agent(
|
|
role="Researcher",
|
|
goal="Make the best research and analysis on content about AI and AI agents",
|
|
backstory="You're an expert researcher, specialized in technology, software engineering, AI and startups. You work as a freelancer and is now working on doing research and analysis for a new customer.",
|
|
tools=[fake_tool],
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me a list of 5 interesting ideas to explore for na article, what makes them unique and interesting.",
|
|
expected_output="Bullet point list of 5 ideas.",
|
|
agent=researcher,
|
|
)
|
|
|
|
assert task.tools == [fake_tool]
|
|
|
|
|
|
def test_task_tool_takes_precedence_over_agent_tools():
|
|
from crewai.tools import tool
|
|
|
|
@tool
|
|
def fake_tool() -> None:
|
|
"Fake tool"
|
|
|
|
@tool
|
|
def fake_task_tool() -> None:
|
|
"Fake tool"
|
|
|
|
researcher = Agent(
|
|
role="Researcher",
|
|
goal="Make the best research and analysis on content about AI and AI agents",
|
|
backstory="You're an expert researcher, specialized in technology, software engineering, AI and startups. You work as a freelancer and is now working on doing research and analysis for a new customer.",
|
|
tools=[fake_tool],
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me a list of 5 interesting ideas to explore for an article, what makes them unique and interesting.",
|
|
expected_output="Bullet point list of 5 ideas.",
|
|
agent=researcher,
|
|
tools=[fake_task_tool],
|
|
)
|
|
|
|
assert task.tools == [fake_task_tool]
|
|
|
|
|
|
def test_task_prompt_includes_expected_output():
|
|
researcher = Agent(
|
|
role="Researcher",
|
|
goal="Make the best research and analysis on content about AI and AI agents",
|
|
backstory="You're an expert researcher, specialized in technology, software engineering, AI and startups. You work as a freelancer and is now working on doing research and analysis for a new customer.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me a list of 5 interesting ideas to explore for na article, what makes them unique and interesting.",
|
|
expected_output="Bullet point list of 5 interesting ideas.",
|
|
agent=researcher,
|
|
)
|
|
|
|
with patch.object(Agent, "execute_task") as execute:
|
|
execute.return_value = "ok"
|
|
task.execute_sync(agent=researcher)
|
|
execute.assert_called_once_with(task=task, context=None, tools=[])
|
|
|
|
|
|
def test_task_callback():
|
|
researcher = Agent(
|
|
role="Researcher",
|
|
goal="Make the best research and analysis on content about AI and AI agents",
|
|
backstory="You're an expert researcher, specialized in technology, software engineering, AI and startups. You work as a freelancer and is now working on doing research and analysis for a new customer.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task_completed = MagicMock(return_value="done")
|
|
|
|
task = Task(
|
|
name="Brainstorm",
|
|
description="Give me a list of 5 interesting ideas to explore for na article, what makes them unique and interesting.",
|
|
expected_output="Bullet point list of 5 interesting ideas.",
|
|
agent=researcher,
|
|
callback=task_completed,
|
|
)
|
|
|
|
with patch.object(Agent, "execute_task") as execute:
|
|
execute.return_value = "ok"
|
|
task.execute_sync(agent=researcher)
|
|
task_completed.assert_called_once_with(task.output)
|
|
|
|
assert task.output.description == task.description
|
|
assert task.output.expected_output == task.expected_output
|
|
assert task.output.name == task.name
|
|
|
|
|
|
def test_task_callback_returns_task_output():
|
|
from crewai.tasks.output_format import OutputFormat
|
|
|
|
researcher = Agent(
|
|
role="Researcher",
|
|
goal="Make the best research and analysis on content about AI and AI agents",
|
|
backstory="You're an expert researcher, specialized in technology, software engineering, AI and startups. You work as a freelancer and is now working on doing research and analysis for a new customer.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task_completed = MagicMock(return_value="done")
|
|
|
|
task = Task(
|
|
description="Give me a list of 5 interesting ideas to explore for an article, what makes them unique and interesting.",
|
|
expected_output="Bullet point list of 5 interesting ideas.",
|
|
agent=researcher,
|
|
callback=task_completed,
|
|
)
|
|
|
|
with patch.object(Agent, "execute_task") as execute:
|
|
execute.return_value = "exported_ok"
|
|
task.execute_sync(agent=researcher)
|
|
# Ensure the callback is called with a TaskOutput object serialized to JSON
|
|
task_completed.assert_called_once()
|
|
callback_data = task_completed.call_args[0][0]
|
|
|
|
# Check if callback_data is TaskOutput object or JSON string
|
|
if isinstance(callback_data, TaskOutput):
|
|
callback_data = json.dumps(callback_data.model_dump())
|
|
|
|
assert isinstance(callback_data, str)
|
|
output_dict = json.loads(callback_data)
|
|
expected_output = {
|
|
"description": task.description,
|
|
"raw": "exported_ok",
|
|
"pydantic": None,
|
|
"json_dict": None,
|
|
"agent": researcher.role,
|
|
"summary": "Give me a list of 5 interesting ideas to explore...",
|
|
"name": None,
|
|
"expected_output": "Bullet point list of 5 interesting ideas.",
|
|
"output_format": OutputFormat.RAW,
|
|
}
|
|
assert output_dict == expected_output
|
|
|
|
|
|
def test_execute_with_agent():
|
|
researcher = Agent(
|
|
role="Researcher",
|
|
goal="Make the best research and analysis on content about AI and AI agents",
|
|
backstory="You're an expert researcher, specialized in technology, software engineering, AI and startups. You work as a freelancer and is now working on doing research and analysis for a new customer.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me a list of 5 interesting ideas to explore for na article, what makes them unique and interesting.",
|
|
expected_output="Bullet point list of 5 interesting ideas.",
|
|
)
|
|
|
|
with patch.object(Agent, "execute_task", return_value="ok") as execute:
|
|
task.execute_sync(agent=researcher)
|
|
execute.assert_called_once_with(task=task, context=None, tools=[])
|
|
|
|
|
|
def test_async_execution():
|
|
researcher = Agent(
|
|
role="Researcher",
|
|
goal="Make the best research and analysis on content about AI and AI agents",
|
|
backstory="You're an expert researcher, specialized in technology, software engineering, AI and startups. You work as a freelancer and is now working on doing research and analysis for a new customer.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me a list of 5 interesting ideas to explore for na article, what makes them unique and interesting.",
|
|
expected_output="Bullet point list of 5 interesting ideas.",
|
|
async_execution=True,
|
|
agent=researcher,
|
|
)
|
|
|
|
with patch.object(Agent, "execute_task", return_value="ok") as execute:
|
|
execution = task.execute_async(agent=researcher)
|
|
result = execution.result()
|
|
assert result.raw == "ok"
|
|
execute.assert_called_once_with(task=task, context=None, tools=[])
|
|
|
|
|
|
def test_multiple_output_type_error():
|
|
class Output(BaseModel):
|
|
field: str
|
|
|
|
with pytest.raises(ValidationError):
|
|
Task(
|
|
description="Give me a list of 5 interesting ideas to explore for na article, what makes them unique and interesting.",
|
|
expected_output="Bullet point list of 5 interesting ideas.",
|
|
output_json=Output,
|
|
output_pydantic=Output,
|
|
)
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_output_pydantic_sequential():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_pydantic=ScoreOutput,
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(agents=[scorer], tasks=[task], process=Process.sequential)
|
|
result = crew.kickoff()
|
|
assert isinstance(result.pydantic, ScoreOutput)
|
|
assert result.to_dict() == {"score": 4}
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_output_pydantic_hierarchical():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_pydantic=ScoreOutput,
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(
|
|
agents=[scorer],
|
|
tasks=[task],
|
|
process=Process.hierarchical,
|
|
manager_llm="gpt-4o",
|
|
)
|
|
result = crew.kickoff()
|
|
assert isinstance(result.pydantic, ScoreOutput)
|
|
assert result.to_dict() == {"score": 4}
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_output_json_sequential():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_json=ScoreOutput,
|
|
output_file="score.json",
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(agents=[scorer], tasks=[task], process=Process.sequential)
|
|
result = crew.kickoff()
|
|
assert '{"score": 4}' == result.json
|
|
assert result.to_dict() == {"score": 4}
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_output_json_hierarchical():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_json=ScoreOutput,
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(
|
|
agents=[scorer],
|
|
tasks=[task],
|
|
process=Process.hierarchical,
|
|
manager_llm="gpt-4o",
|
|
)
|
|
result = crew.kickoff()
|
|
assert result.json == '{"score": 4}'
|
|
assert result.to_dict() == {"score": 4}
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_json_property_without_output_json():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_pydantic=ScoreOutput, # Using output_pydantic instead of output_json
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(agents=[scorer], tasks=[task], process=Process.sequential)
|
|
result = crew.kickoff()
|
|
|
|
with pytest.raises(ValueError) as excinfo:
|
|
_ = result.json # Attempt to access the json property
|
|
|
|
assert "No JSON output found in the final task." in str(excinfo.value)
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_output_json_dict_sequential():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_json=ScoreOutput,
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(agents=[scorer], tasks=[task], process=Process.sequential)
|
|
result = crew.kickoff()
|
|
assert {"score": 4} == result.json_dict
|
|
assert result.to_dict() == {"score": 4}
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_output_json_dict_hierarchical():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_json=ScoreOutput,
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(
|
|
agents=[scorer],
|
|
tasks=[task],
|
|
process=Process.hierarchical,
|
|
manager_llm="gpt-4o",
|
|
)
|
|
result = crew.kickoff()
|
|
assert {"score": 4} == result.json_dict
|
|
assert result.to_dict() == {"score": 4}
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_output_pydantic_to_another_task():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
llm="gpt-4-0125-preview",
|
|
function_calling_llm="gpt-3.5-turbo-0125",
|
|
verbose=True,
|
|
)
|
|
|
|
task1 = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_pydantic=ScoreOutput,
|
|
agent=scorer,
|
|
)
|
|
|
|
task2 = Task(
|
|
description="Given the score the title 'The impact of AI in the future of work' got, give me an integer score between 1-5 for the following title: 'Return of the Jedi', you MUST give it a score, use your best judgment",
|
|
expected_output="The score of the title.",
|
|
output_pydantic=ScoreOutput,
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(agents=[scorer], tasks=[task1, task2], verbose=True)
|
|
result = crew.kickoff()
|
|
pydantic_result = result.pydantic
|
|
assert isinstance(
|
|
pydantic_result, ScoreOutput
|
|
), "Expected pydantic result to be of type ScoreOutput"
|
|
assert pydantic_result.score == 5
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_output_json_to_another_task():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task1 = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_json=ScoreOutput,
|
|
agent=scorer,
|
|
)
|
|
|
|
task2 = Task(
|
|
description="Given the score the title 'The impact of AI in the future of work' got, give me an integer score between 1-5 for the following title: 'Return of the Jedi'",
|
|
expected_output="The score of the title.",
|
|
output_json=ScoreOutput,
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(agents=[scorer], tasks=[task1, task2])
|
|
result = crew.kickoff()
|
|
assert '{"score": 4}' == result.json
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_save_task_output():
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_file="score.json",
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(agents=[scorer], tasks=[task])
|
|
|
|
with patch.object(Task, "_save_file") as save_file:
|
|
save_file.return_value = None
|
|
crew.kickoff()
|
|
save_file.assert_called_once()
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_save_task_json_output():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_file="score.json",
|
|
output_json=ScoreOutput,
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(agents=[scorer], tasks=[task])
|
|
crew.kickoff()
|
|
|
|
output_file_exists = os.path.exists("score.json")
|
|
assert output_file_exists
|
|
assert {"score": 4} == json.loads(open("score.json").read())
|
|
if output_file_exists:
|
|
os.remove("score.json")
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_save_task_pydantic_output():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_file="score.json",
|
|
output_pydantic=ScoreOutput,
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(agents=[scorer], tasks=[task])
|
|
crew.kickoff()
|
|
|
|
output_file_exists = os.path.exists("score.json")
|
|
assert output_file_exists
|
|
assert {"score": 4} == json.loads(open("score.json").read())
|
|
if output_file_exists:
|
|
os.remove("score.json")
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_custom_converter_cls():
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
class ScoreConverter(Converter):
|
|
pass
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
output_pydantic=ScoreOutput,
|
|
converter_cls=ScoreConverter,
|
|
agent=scorer,
|
|
)
|
|
|
|
crew = Crew(agents=[scorer], tasks=[task])
|
|
|
|
with patch.object(
|
|
ScoreConverter, "to_pydantic", return_value=ScoreOutput(score=5)
|
|
) as mock_to_pydantic:
|
|
crew.kickoff()
|
|
mock_to_pydantic.assert_called_once()
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_increment_delegations_for_hierarchical_process():
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
)
|
|
|
|
crew = Crew(
|
|
agents=[scorer],
|
|
tasks=[task],
|
|
process=Process.hierarchical,
|
|
manager_llm="gpt-4o",
|
|
)
|
|
|
|
with patch.object(Task, "increment_delegations") as increment_delegations:
|
|
increment_delegations.return_value = None
|
|
crew.kickoff()
|
|
increment_delegations.assert_called_once()
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_increment_delegations_for_sequential_process():
|
|
manager = Agent(
|
|
role="Manager",
|
|
goal="Coordinate scoring processes",
|
|
backstory="You're great at delegating work about scoring.",
|
|
allow_delegation=True,
|
|
)
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
allow_delegation=True,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work'",
|
|
expected_output="The score of the title.",
|
|
agent=manager,
|
|
)
|
|
|
|
crew = Crew(
|
|
agents=[manager, scorer],
|
|
tasks=[task],
|
|
process=Process.sequential,
|
|
)
|
|
|
|
with patch.object(Task, "increment_delegations") as increment_delegations:
|
|
increment_delegations.return_value = None
|
|
crew.kickoff()
|
|
increment_delegations.assert_called_once()
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_increment_tool_errors():
|
|
from crewai.tools import tool
|
|
|
|
@tool
|
|
def scoring_examples() -> None:
|
|
"Useful examples for scoring titles."
|
|
raise Exception("Error")
|
|
|
|
scorer = Agent(
|
|
role="Scorer",
|
|
goal="Score the title",
|
|
backstory="You're an expert scorer, specialized in scoring titles.",
|
|
tools=[scoring_examples],
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work', check examples to based your evaluation.",
|
|
expected_output="The score of the title.",
|
|
)
|
|
|
|
crew = Crew(
|
|
agents=[scorer],
|
|
tasks=[task],
|
|
process=Process.hierarchical,
|
|
manager_llm="gpt-4-0125-preview",
|
|
)
|
|
|
|
with patch.object(Task, "increment_tools_errors") as increment_tools_errors:
|
|
increment_tools_errors.return_value = None
|
|
crew.kickoff()
|
|
assert len(increment_tools_errors.mock_calls) > 0
|
|
|
|
|
|
def test_task_definition_based_on_dict():
|
|
config = {
|
|
"description": "Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work', check examples to based your evaluation.",
|
|
"expected_output": "The score of the title.",
|
|
}
|
|
|
|
task = Task(**config)
|
|
|
|
assert task.description == config["description"]
|
|
assert task.expected_output == config["expected_output"]
|
|
assert task.agent is None
|
|
|
|
|
|
def test_conditional_task_definition_based_on_dict():
|
|
config = {
|
|
"description": "Give me an integer score between 1-5 for the following title: 'The impact of AI in the future of work', check examples to based your evaluation.",
|
|
"expected_output": "The score of the title.",
|
|
}
|
|
|
|
task = ConditionalTask(**config, condition=lambda x: True)
|
|
|
|
assert task.description == config["description"]
|
|
assert task.expected_output == config["expected_output"]
|
|
assert task.agent is None
|
|
|
|
|
|
def test_interpolate_inputs():
|
|
task = Task(
|
|
description="Give me a list of 5 interesting ideas about {topic} to explore for an article, what makes them unique and interesting.",
|
|
expected_output="Bullet point list of 5 interesting ideas about {topic}.",
|
|
output_file="/tmp/{topic}/output_{date}.txt",
|
|
)
|
|
|
|
task.interpolate_inputs_and_add_conversation_history(
|
|
inputs={"topic": "AI", "date": "2025"}
|
|
)
|
|
assert (
|
|
task.description
|
|
== "Give me a list of 5 interesting ideas about AI to explore for an article, what makes them unique and interesting."
|
|
)
|
|
assert task.expected_output == "Bullet point list of 5 interesting ideas about AI."
|
|
assert task.output_file == "/tmp/AI/output_2025.txt"
|
|
|
|
task.interpolate_inputs_and_add_conversation_history(
|
|
inputs={"topic": "ML", "date": "2025"}
|
|
)
|
|
assert (
|
|
task.description
|
|
== "Give me a list of 5 interesting ideas about ML to explore for an article, what makes them unique and interesting."
|
|
)
|
|
assert task.expected_output == "Bullet point list of 5 interesting ideas about ML."
|
|
assert task.output_file == "/tmp/ML/output_2025.txt"
|
|
|
|
|
|
def test_interpolate_only():
|
|
"""Test the interpolate_only method for various scenarios including JSON structure preservation."""
|
|
task = Task(
|
|
description="Unused in this test", expected_output="Unused in this test"
|
|
)
|
|
|
|
# Test JSON structure preservation
|
|
json_string = '{"info": "Look at {placeholder}", "nested": {"val": "{nestedVal}"}}'
|
|
result = task.interpolate_only(
|
|
input_string=json_string,
|
|
inputs={"placeholder": "the data", "nestedVal": "something else"},
|
|
)
|
|
assert '"info": "Look at the data"' in result
|
|
assert '"val": "something else"' in result
|
|
assert "{placeholder}" not in result
|
|
assert "{nestedVal}" not in result
|
|
|
|
# Test normal string interpolation
|
|
normal_string = "Hello {name}, welcome to {place}!"
|
|
result = task.interpolate_only(
|
|
input_string=normal_string, inputs={"name": "John", "place": "CrewAI"}
|
|
)
|
|
assert result == "Hello John, welcome to CrewAI!"
|
|
|
|
# Test empty string
|
|
result = task.interpolate_only(input_string="", inputs={"unused": "value"})
|
|
assert result == ""
|
|
|
|
# Test string with no placeholders
|
|
no_placeholders = "Hello, this is a test"
|
|
result = task.interpolate_only(
|
|
input_string=no_placeholders, inputs={"unused": "value"}
|
|
)
|
|
assert result == no_placeholders
|
|
|
|
|
|
def test_interpolate_only_with_dict_inside_expected_output():
|
|
"""Test the interpolate_only method for various scenarios including JSON structure preservation."""
|
|
task = Task(
|
|
description="Unused in this test",
|
|
expected_output="Unused in this test: {questions}",
|
|
)
|
|
|
|
json_string = '{"questions": {"main_question": "What is the user\'s name?", "secondary_question": "What is the user\'s age?"}}'
|
|
result = task.interpolate_only(
|
|
input_string=json_string,
|
|
inputs={
|
|
"questions": {
|
|
"main_question": "What is the user's name?",
|
|
"secondary_question": "What is the user's age?",
|
|
}
|
|
},
|
|
)
|
|
assert '"main_question": "What is the user\'s name?"' in result
|
|
assert '"secondary_question": "What is the user\'s age?"' in result
|
|
assert result == json_string
|
|
|
|
normal_string = "Hello {name}, welcome to {place}!"
|
|
result = task.interpolate_only(
|
|
input_string=normal_string, inputs={"name": "John", "place": "CrewAI"}
|
|
)
|
|
assert result == "Hello John, welcome to CrewAI!"
|
|
|
|
result = task.interpolate_only(input_string="", inputs={"unused": "value"})
|
|
assert result == ""
|
|
|
|
no_placeholders = "Hello, this is a test"
|
|
result = task.interpolate_only(
|
|
input_string=no_placeholders, inputs={"unused": "value"}
|
|
)
|
|
assert result == no_placeholders
|
|
|
|
|
|
def test_task_output_str_with_pydantic():
|
|
from crewai.tasks.output_format import OutputFormat
|
|
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
score_output = ScoreOutput(score=4)
|
|
task_output = TaskOutput(
|
|
description="Test task",
|
|
agent="Test Agent",
|
|
pydantic=score_output,
|
|
output_format=OutputFormat.PYDANTIC,
|
|
)
|
|
|
|
assert str(task_output) == str(score_output)
|
|
|
|
|
|
def test_task_output_str_with_json_dict():
|
|
from crewai.tasks.output_format import OutputFormat
|
|
|
|
json_dict = {"score": 4}
|
|
task_output = TaskOutput(
|
|
description="Test task",
|
|
agent="Test Agent",
|
|
json_dict=json_dict,
|
|
output_format=OutputFormat.JSON,
|
|
)
|
|
|
|
assert str(task_output) == str(json_dict)
|
|
|
|
|
|
def test_task_output_str_with_raw():
|
|
from crewai.tasks.output_format import OutputFormat
|
|
|
|
raw_output = "Raw task output"
|
|
task_output = TaskOutput(
|
|
description="Test task",
|
|
agent="Test Agent",
|
|
raw=raw_output,
|
|
output_format=OutputFormat.RAW,
|
|
)
|
|
|
|
assert str(task_output) == raw_output
|
|
|
|
|
|
def test_task_output_str_with_pydantic_and_json_dict():
|
|
from crewai.tasks.output_format import OutputFormat
|
|
|
|
class ScoreOutput(BaseModel):
|
|
score: int
|
|
|
|
score_output = ScoreOutput(score=4)
|
|
json_dict = {"score": 4}
|
|
task_output = TaskOutput(
|
|
description="Test task",
|
|
agent="Test Agent",
|
|
pydantic=score_output,
|
|
json_dict=json_dict,
|
|
output_format=OutputFormat.PYDANTIC,
|
|
)
|
|
|
|
# When both pydantic and json_dict are present, pydantic should take precedence
|
|
assert str(task_output) == str(score_output)
|
|
|
|
|
|
def test_task_output_str_with_none():
|
|
from crewai.tasks.output_format import OutputFormat
|
|
|
|
task_output = TaskOutput(
|
|
description="Test task",
|
|
agent="Test Agent",
|
|
output_format=OutputFormat.RAW,
|
|
)
|
|
|
|
assert str(task_output) == ""
|
|
|
|
|
|
def test_key():
|
|
original_description = "Give me a list of 5 interesting ideas about {topic} to explore for an article, what makes them unique and interesting."
|
|
original_expected_output = "Bullet point list of 5 interesting ideas about {topic}."
|
|
task = Task(
|
|
description=original_description,
|
|
expected_output=original_expected_output,
|
|
)
|
|
hash = hashlib.md5(
|
|
f"{original_description}|{original_expected_output}".encode()
|
|
).hexdigest()
|
|
|
|
assert task.key == hash, "The key should be the hash of the description."
|
|
|
|
task.interpolate_inputs_and_add_conversation_history(inputs={"topic": "AI"})
|
|
assert (
|
|
task.key == hash
|
|
), "The key should be the hash of the non-interpolated description."
|
|
|
|
|
|
def test_output_file_validation():
|
|
"""Test output file path validation."""
|
|
# Valid paths
|
|
assert (
|
|
Task(
|
|
description="Test task",
|
|
expected_output="Test output",
|
|
output_file="output.txt",
|
|
).output_file
|
|
== "output.txt"
|
|
)
|
|
assert (
|
|
Task(
|
|
description="Test task",
|
|
expected_output="Test output",
|
|
output_file="/tmp/output.txt",
|
|
).output_file
|
|
== "tmp/output.txt"
|
|
)
|
|
assert (
|
|
Task(
|
|
description="Test task",
|
|
expected_output="Test output",
|
|
output_file="{dir}/output_{date}.txt",
|
|
).output_file
|
|
== "{dir}/output_{date}.txt"
|
|
)
|
|
|
|
# Invalid paths
|
|
with pytest.raises(ValueError, match="Path traversal"):
|
|
Task(
|
|
description="Test task",
|
|
expected_output="Test output",
|
|
output_file="../output.txt",
|
|
)
|
|
with pytest.raises(ValueError, match="Path traversal"):
|
|
Task(
|
|
description="Test task",
|
|
expected_output="Test output",
|
|
output_file="folder/../output.txt",
|
|
)
|
|
with pytest.raises(ValueError, match="Shell special characters"):
|
|
Task(
|
|
description="Test task",
|
|
expected_output="Test output",
|
|
output_file="output.txt | rm -rf /",
|
|
)
|
|
with pytest.raises(ValueError, match="Shell expansion"):
|
|
Task(
|
|
description="Test task",
|
|
expected_output="Test output",
|
|
output_file="~/output.txt",
|
|
)
|
|
with pytest.raises(ValueError, match="Shell expansion"):
|
|
Task(
|
|
description="Test task",
|
|
expected_output="Test output",
|
|
output_file="$HOME/output.txt",
|
|
)
|
|
with pytest.raises(ValueError, match="Invalid template variable"):
|
|
Task(
|
|
description="Test task",
|
|
expected_output="Test output",
|
|
output_file="{invalid-name}/output.txt",
|
|
)
|
|
|
|
|
|
@pytest.mark.vcr(filter_headers=["authorization"])
|
|
def test_task_execution_times():
|
|
researcher = Agent(
|
|
role="Researcher",
|
|
goal="Make the best research and analysis on content about AI and AI agents",
|
|
backstory="You're an expert researcher, specialized in technology, software engineering, AI and startups. You work as a freelancer and is now working on doing research and analysis for a new customer.",
|
|
allow_delegation=False,
|
|
)
|
|
|
|
task = Task(
|
|
description="Give me a list of 5 interesting ideas to explore for na article, what makes them unique and interesting.",
|
|
expected_output="Bullet point list of 5 interesting ideas.",
|
|
agent=researcher,
|
|
)
|
|
|
|
assert task.start_time is None
|
|
assert task.end_time is None
|
|
assert task.execution_duration is None
|
|
|
|
task.execute_sync(agent=researcher)
|
|
|
|
assert task.start_time is not None
|
|
assert task.end_time is not None
|
|
assert task.execution_duration == (task.end_time - task.start_time).total_seconds()
|
|
|
|
|
|
def test_interpolate_with_list_of_strings():
|
|
task = Task(
|
|
description="Test list interpolation",
|
|
expected_output="List: {items}",
|
|
)
|
|
|
|
# Test simple list of strings
|
|
input_str = "Available items: {items}"
|
|
inputs = {"items": ["apple", "banana", "cherry"]}
|
|
result = task.interpolate_only(input_str, inputs)
|
|
assert result == f"Available items: {inputs['items']}"
|
|
|
|
# Test empty list
|
|
empty_list_input = {"items": []}
|
|
result = task.interpolate_only(input_str, empty_list_input)
|
|
assert result == "Available items: []"
|
|
|
|
|
|
def test_interpolate_with_list_of_dicts():
|
|
task = Task(
|
|
description="Test list of dicts interpolation",
|
|
expected_output="People: {people}",
|
|
)
|
|
|
|
input_data = {
|
|
"people": [
|
|
{"name": "Alice", "age": 30, "skills": ["Python", "AI"]},
|
|
{"name": "Bob", "age": 25, "skills": ["Java", "Cloud"]},
|
|
]
|
|
}
|
|
result = task.interpolate_only("{people}", input_data)
|
|
|
|
parsed_result = eval(result)
|
|
assert isinstance(parsed_result, list)
|
|
assert len(parsed_result) == 2
|
|
assert parsed_result[0]["name"] == "Alice"
|
|
assert parsed_result[0]["age"] == 30
|
|
assert parsed_result[0]["skills"] == ["Python", "AI"]
|
|
assert parsed_result[1]["name"] == "Bob"
|
|
assert parsed_result[1]["age"] == 25
|
|
assert parsed_result[1]["skills"] == ["Java", "Cloud"]
|
|
|
|
|
|
def test_interpolate_with_nested_structures():
|
|
task = Task(
|
|
description="Test nested structures",
|
|
expected_output="Company: {company}",
|
|
)
|
|
|
|
input_data = {
|
|
"company": {
|
|
"name": "TechCorp",
|
|
"departments": [
|
|
{
|
|
"name": "Engineering",
|
|
"employees": 50,
|
|
"tools": ["Git", "Docker", "Kubernetes"],
|
|
},
|
|
{"name": "Sales", "employees": 20, "regions": {"north": 5, "south": 3}},
|
|
],
|
|
}
|
|
}
|
|
result = task.interpolate_only("{company}", input_data)
|
|
parsed = eval(result)
|
|
|
|
assert parsed["name"] == "TechCorp"
|
|
assert len(parsed["departments"]) == 2
|
|
assert parsed["departments"][0]["tools"] == ["Git", "Docker", "Kubernetes"]
|
|
assert parsed["departments"][1]["regions"]["north"] == 5
|
|
|
|
|
|
def test_interpolate_with_special_characters():
|
|
task = Task(
|
|
description="Test special characters in dicts",
|
|
expected_output="Data: {special_data}",
|
|
)
|
|
|
|
input_data = {
|
|
"special_data": {
|
|
"quotes": """This has "double" and 'single' quotes""",
|
|
"unicode": "文字化けテスト",
|
|
"symbols": "!@#$%^&*()",
|
|
"empty": "",
|
|
}
|
|
}
|
|
result = task.interpolate_only("{special_data}", input_data)
|
|
parsed = eval(result)
|
|
|
|
assert parsed["quotes"] == """This has "double" and 'single' quotes"""
|
|
assert parsed["unicode"] == "文字化けテスト"
|
|
assert parsed["symbols"] == "!@#$%^&*()"
|
|
assert parsed["empty"] == ""
|
|
|
|
|
|
def test_interpolate_mixed_types():
|
|
task = Task(
|
|
description="Test mixed type interpolation",
|
|
expected_output="Mixed: {data}",
|
|
)
|
|
|
|
input_data = {
|
|
"data": {
|
|
"name": "Test Dataset",
|
|
"samples": 1000,
|
|
"features": ["age", "income", "location"],
|
|
"metadata": {
|
|
"source": "public",
|
|
"validated": True,
|
|
"tags": ["demo", "test", "temp"],
|
|
},
|
|
}
|
|
}
|
|
result = task.interpolate_only("{data}", input_data)
|
|
parsed = eval(result)
|
|
|
|
assert parsed["name"] == "Test Dataset"
|
|
assert parsed["samples"] == 1000
|
|
assert parsed["metadata"]["tags"] == ["demo", "test", "temp"]
|
|
|
|
|
|
def test_interpolate_complex_combination():
|
|
task = Task(
|
|
description="Test complex combination",
|
|
expected_output="Report: {report}",
|
|
)
|
|
|
|
input_data = {
|
|
"report": [
|
|
{
|
|
"month": "January",
|
|
"metrics": {"sales": 15000, "expenses": 8000, "profit": 7000},
|
|
"top_products": ["Product A", "Product B"],
|
|
},
|
|
{
|
|
"month": "February",
|
|
"metrics": {"sales": 18000, "expenses": 8500, "profit": 9500},
|
|
"top_products": ["Product C", "Product D"],
|
|
},
|
|
]
|
|
}
|
|
result = task.interpolate_only("{report}", input_data)
|
|
parsed = eval(result)
|
|
|
|
assert len(parsed) == 2
|
|
assert parsed[0]["month"] == "January"
|
|
assert parsed[1]["metrics"]["profit"] == 9500
|
|
assert "Product D" in parsed[1]["top_products"]
|
|
|
|
|
|
def test_interpolate_invalid_type_validation():
|
|
task = Task(
|
|
description="Test invalid type validation",
|
|
expected_output="Should never reach here",
|
|
)
|
|
|
|
# Test with invalid top-level type
|
|
with pytest.raises(ValueError) as excinfo:
|
|
task.interpolate_only("{data}", {"data": set()}) # type: ignore we are purposely testing this failure
|
|
|
|
assert "Unsupported type set" in str(excinfo.value)
|
|
|
|
# Test with invalid nested type
|
|
invalid_nested = {
|
|
"profile": {
|
|
"name": "John",
|
|
"age": 30,
|
|
"tags": {"a", "b", "c"}, # Set is invalid
|
|
}
|
|
}
|
|
with pytest.raises(ValueError) as excinfo:
|
|
task.interpolate_only("{data}", {"data": invalid_nested})
|
|
assert "Unsupported type set" in str(excinfo.value)
|
|
|
|
|
|
def test_interpolate_custom_object_validation():
|
|
task = Task(
|
|
description="Test custom object rejection",
|
|
expected_output="Should never reach here",
|
|
)
|
|
|
|
class CustomObject:
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
def __str__(self):
|
|
return str(self.value)
|
|
|
|
# Test with custom object at top level
|
|
with pytest.raises(ValueError) as excinfo:
|
|
task.interpolate_only("{obj}", {"obj": CustomObject(5)}) # type: ignore we are purposely testing this failure
|
|
assert "Unsupported type CustomObject" in str(excinfo.value)
|
|
|
|
# Test with nested custom object in dictionary
|
|
with pytest.raises(ValueError) as excinfo:
|
|
task.interpolate_only(
|
|
"{data}", {"data": {"valid": 1, "invalid": CustomObject(5)}}
|
|
)
|
|
assert "Unsupported type CustomObject" in str(excinfo.value)
|
|
|
|
# Test with nested custom object in list
|
|
with pytest.raises(ValueError) as excinfo:
|
|
task.interpolate_only("{data}", {"data": [1, "valid", CustomObject(5)]})
|
|
assert "Unsupported type CustomObject" in str(excinfo.value)
|
|
|
|
# Test with deeply nested custom object
|
|
with pytest.raises(ValueError) as excinfo:
|
|
task.interpolate_only(
|
|
"{data}", {"data": {"level1": {"level2": [{"level3": CustomObject(5)}]}}}
|
|
)
|
|
assert "Unsupported type CustomObject" in str(excinfo.value)
|
|
|
|
|
|
def test_interpolate_valid_complex_types():
|
|
task = Task(
|
|
description="Test valid complex types",
|
|
expected_output="Validation should pass",
|
|
)
|
|
|
|
# Valid complex structure
|
|
valid_data = {
|
|
"name": "Valid Dataset",
|
|
"stats": {
|
|
"count": 1000,
|
|
"distribution": [0.2, 0.3, 0.5],
|
|
"features": ["age", "income"],
|
|
"nested": {"deep": [1, 2, 3], "deeper": {"a": 1, "b": 2.5}},
|
|
},
|
|
}
|
|
|
|
# Should not raise any errors
|
|
result = task.interpolate_only("{data}", {"data": valid_data})
|
|
parsed = eval(result)
|
|
assert parsed["name"] == "Valid Dataset"
|
|
assert parsed["stats"]["nested"]["deeper"]["b"] == 2.5
|
|
|
|
|
|
def test_interpolate_edge_cases():
|
|
task = Task(
|
|
description="Test edge cases",
|
|
expected_output="Edge case handling",
|
|
)
|
|
|
|
# Test empty dict and list
|
|
assert task.interpolate_only("{}", {"data": {}}) == "{}"
|
|
assert task.interpolate_only("[]", {"data": []}) == "[]"
|
|
|
|
# Test numeric types
|
|
assert task.interpolate_only("{num}", {"num": 42}) == "42"
|
|
assert task.interpolate_only("{num}", {"num": 3.14}) == "3.14"
|
|
|
|
# Test boolean values (valid JSON types)
|
|
assert task.interpolate_only("{flag}", {"flag": True}) == "True"
|
|
assert task.interpolate_only("{flag}", {"flag": False}) == "False"
|
|
|
|
|
|
def test_interpolate_valid_types():
|
|
task = Task(
|
|
description="Test valid types including null and boolean",
|
|
expected_output="Should pass validation",
|
|
)
|
|
|
|
# Test with boolean and null values (valid JSON types)
|
|
valid_data = {
|
|
"name": "Test",
|
|
"active": True,
|
|
"deleted": False,
|
|
"optional": None,
|
|
"nested": {"flag": True, "empty": None},
|
|
}
|
|
|
|
result = task.interpolate_only("{data}", {"data": valid_data})
|
|
parsed = eval(result)
|
|
|
|
assert parsed["active"] is True
|
|
assert parsed["deleted"] is False
|
|
assert parsed["optional"] is None
|
|
assert parsed["nested"]["flag"] is True
|
|
assert parsed["nested"]["empty"] is None
|