Support set max_execution_time to Agent (#2610)
Some checks are pending
Notify Downstream / notify-downstream (push) Waiting to run

* Fixed fake max_execution_time paramenter
---------

Co-authored-by: Lucas Gomide <lucaslg200@gmail.com>
This commit is contained in:
Vidit Ostwal
2025-04-18 01:33:00 +05:30
committed by GitHub
parent 870dffbb89
commit 371f19f3cd
4 changed files with 898 additions and 9 deletions

View File

@@ -3,6 +3,7 @@
import hashlib
import json
import os
import time
from functools import partial
from typing import Tuple, Union
from unittest.mock import MagicMock, patch
@@ -1368,3 +1369,90 @@ def test_interpolate_valid_types():
assert parsed["optional"] is None
assert parsed["nested"]["flag"] is True
assert parsed["nested"]["empty"] is None
def test_task_with_no_max_execution_time():
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,
max_execution_time=None
)
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_without_timeout", return_value = "ok") as execute:
result = task.execute_sync(agent=researcher)
assert result.raw == "ok"
execute.assert_called_once()
@pytest.mark.vcr(filter_headers=["authorization"])
def test_task_with_max_execution_time():
from crewai.tools import tool
"""Test that execution raises TimeoutError when max_execution_time is exceeded."""
@tool("what amazing tool", result_as_answer=True)
def my_tool() -> str:
"My tool"
time.sleep(1)
return "okay"
researcher = Agent(
role="Researcher",
goal="Make the best research and analysis on content about AI and AI agents. Use the tool provided to you.",
backstory=(
"You're an expert researcher, specialized in technology, software engineering, AI and startups. "
"You work as a freelancer and are now working on doing research and analysis for a new customer."
),
allow_delegation=False,
tools=[my_tool],
max_execution_time=4
)
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,
)
result = task.execute_sync(agent=researcher)
assert result.raw == "okay"
@pytest.mark.vcr(filter_headers=["authorization"])
def test_task_with_max_execution_time_exceeded():
from crewai.tools import tool
"""Test that execution raises TimeoutError when max_execution_time is exceeded."""
@tool("what amazing tool", result_as_answer=True)
def my_tool() -> str:
"My tool"
time.sleep(10)
return "okay"
researcher = Agent(
role="Researcher",
goal="Make the best research and analysis on content about AI and AI agents. Use the tool provided to you.",
backstory=(
"You're an expert researcher, specialized in technology, software engineering, AI and startups. "
"You work as a freelancer and are now working on doing research and analysis for a new customer."
),
allow_delegation=False,
tools=[my_tool],
max_execution_time=1
)
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,
)
with pytest.raises(TimeoutError):
task.execute_sync(agent=researcher)