andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-01-09 08:08:32 +00:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Merge branch 'crewAIInc:main' into bug_fix

Browse Source
This commit is contained in:
sakunkun
2025-03-26 10:50:15 +08:00
committed by GitHub
parent 448d31cad9 807c13e144
commit e4f5c7cdf2
44 changed files with 2741 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
tests/cassettes/test_custom_llm_implementation.yaml Normal file
View File

@@ -0,0 +1,107 @@
interactions:
- request:
body: '{"messages": [{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "What is the answer to life, the universe, and everything?"}],
"model": "gpt-4o-mini", "tools": null}'
headers:
accept:
- application/json
accept-encoding:
- gzip, deflate
connection:
- keep-alive
content-length:
- '206'
content-type:
- application/json
host:
- api.openai.com
user-agent:
- OpenAI/Python 1.61.0
x-stainless-arch:
- arm64
x-stainless-async:
- 'false'
x-stainless-lang:
- python
x-stainless-os:
- MacOS
x-stainless-package-version:
- 1.61.0
x-stainless-retry-count:
- '0'
x-stainless-runtime:
- CPython
x-stainless-runtime-version:
- 3.12.8
method: POST
uri: https://api.openai.com/v1/chat/completions
response:
content: "{\n \"id\": \"chatcmpl-B7W6FS0wpfndLdg12G3H6ZAXcYhJi\",\n \"object\":
\"chat.completion\",\n \"created\": 1741131387,\n \"model\": \"gpt-4o-mini-2024-07-18\",\n
\ \"choices\": [\n {\n \"index\": 0,\n \"message\": {\n \"role\":
\"assistant\",\n \"content\": \"The answer to life, the universe, and
everything, famously found in Douglas Adams' \\\"The Hitchhiker's Guide to the
Galaxy,\\\" is the number 42. However, the question itself is left ambiguous,
leading to much speculation and humor in the story.\",\n \"refusal\":
null\n },\n \"logprobs\": null,\n \"finish_reason\": \"stop\"\n
\ }\n ],\n \"usage\": {\n \"prompt_tokens\": 30,\n \"completion_tokens\":
54,\n \"total_tokens\": 84,\n \"prompt_tokens_details\": {\n \"cached_tokens\":
0,\n \"audio_tokens\": 0\n },\n \"completion_tokens_details\": {\n
\ \"reasoning_tokens\": 0,\n \"audio_tokens\": 0,\n \"accepted_prediction_tokens\":
0,\n \"rejected_prediction_tokens\": 0\n }\n },\n \"service_tier\":
\"default\",\n \"system_fingerprint\": \"fp_06737a9306\"\n}\n"
headers:
CF-RAY:
- 91b532234c18cf1f-SJC
Connection:
- keep-alive
Content-Encoding:
- gzip
Content-Type:
- application/json
Date:
- Tue, 04 Mar 2025 23:36:28 GMT
Server:
- cloudflare
Set-Cookie:
- __cf_bm=DgLb6UAE6W4Oeto1Bi2RiKXQVV5TTzkXdXWFdmAEwQQ-1741131388-1.0.1.1-jWQtsT95wOeQbmIxAK7cv8gJWxYi1tQ.IupuJzBDnZr7iEChwVUQBRfnYUBJPDsNly3bakCDArjD_S.FLKwH6xUfvlxgfd4YSBhBPy7bcgw;
path=/; expires=Wed, 05-Mar-25 00:06:28 GMT; domain=.api.openai.com; HttpOnly;
Secure; SameSite=None
- _cfuvid=Oa59XCmqjKLKwU34la1hkTunN57JW20E.ZHojvRBfow-1741131388236-0.0.1.1-604800000;
path=/; domain=.api.openai.com; HttpOnly; Secure; SameSite=None
Transfer-Encoding:
- chunked
X-Content-Type-Options:
- nosniff
access-control-expose-headers:
- X-Request-ID
alt-svc:
- h3=":443"; ma=86400
cf-cache-status:
- DYNAMIC
openai-organization:
- crewai-iuxna1
openai-processing-ms:
- '776'
openai-version:
- '2020-10-01'
strict-transport-security:
- max-age=31536000; includeSubDomains; preload
x-ratelimit-limit-requests:
- '30000'
x-ratelimit-limit-tokens:
- '150000000'
x-ratelimit-remaining-requests:
- '29999'
x-ratelimit-remaining-tokens:
- '149999960'
x-ratelimit-reset-requests:
- 2ms
x-ratelimit-reset-tokens:
- 0s
x-request-id:
- req_97824e8fe7c1aca3fbcba7c925388b39
http_version: HTTP/1.1
status_code: 200
version: 1

305
tests/cassettes/test_custom_llm_within_crew.yaml Normal file
View File

@@ -0,0 +1,305 @@
interactions:
- request:
body: '{"messages": [{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": [{"role": "system", "content": "You are Say Hi.
You just say hi to the user\nYour personal goal is: Say hi to the user\nTo give
my best complete final answer to the task respond using the exact following
format:\n\nThought: I now can give a great answer\nFinal Answer: Your final
answer must be the great and the most complete as possible, it must be outcome
described.\n\nI MUST use these formats, my job depends on it!"}, {"role": "user",
"content": "\nCurrent Task: Say hi to the user\n\nThis is the expected criteria
for your final answer: A greeting to the user\nyou MUST return the actual complete
content as the final answer, not a summary.\n\nBegin! This is VERY important
to you, use the tools available and give your best Final Answer, your job depends
on it!\n\nThought:"}]}], "model": "gpt-4o-mini", "tools": null}'
headers:
accept:
- application/json
accept-encoding:
- gzip, deflate
connection:
- keep-alive
content-length:
- '931'
content-type:
- application/json
host:
- api.openai.com
user-agent:
- OpenAI/Python 1.61.0
x-stainless-arch:
- arm64
x-stainless-async:
- 'false'
x-stainless-lang:
- python
x-stainless-os:
- MacOS
x-stainless-package-version:
- 1.61.0
x-stainless-retry-count:
- '0'
x-stainless-runtime:
- CPython
x-stainless-runtime-version:
- 3.12.8
method: POST
uri: https://api.openai.com/v1/chat/completions
response:
content: "{\n \"error\": {\n \"message\": \"Missing required parameter: 'messages[1].content[0].type'.\",\n
\ \"type\": \"invalid_request_error\",\n \"param\": \"messages[1].content[0].type\",\n
\ \"code\": \"missing_required_parameter\"\n }\n}"
headers:
CF-RAY:
- 91b54660799a15b4-SJC
Connection:
- keep-alive
Content-Length:
- '219'
Content-Type:
- application/json
Date:
- Tue, 04 Mar 2025 23:50:16 GMT
Server:
- cloudflare
Set-Cookie:
- __cf_bm=OwS.6cyfDpbxxx8vPp4THv5eNoDMQK0qSVN.wSUyOYk-1741132216-1.0.1.1-QBVd08CjfmDBpNnYQM5ILGbTUWKh6SDM9E4ARG4SV2Z9Q4ltFSFLXoo38OGJApUNZmzn4PtRsyAPsHt_dsrHPF6MD17FPcGtrnAHqCjJrfU;
path=/; expires=Wed, 05-Mar-25 00:20:16 GMT; domain=.api.openai.com; HttpOnly;
Secure; SameSite=None
- _cfuvid=n_ebDsAOhJm5Mc7OMx8JDiOaZq5qzHCnVxyS3KN0BwA-1741132216951-0.0.1.1-604800000;
path=/; domain=.api.openai.com; HttpOnly; Secure; SameSite=None
X-Content-Type-Options:
- nosniff
access-control-expose-headers:
- X-Request-ID
alt-svc:
- h3=":443"; ma=86400
cf-cache-status:
- DYNAMIC
openai-organization:
- crewai-iuxna1
openai-processing-ms:
- '19'
openai-version:
- '2020-10-01'
strict-transport-security:
- max-age=31536000; includeSubDomains; preload
x-ratelimit-limit-requests:
- '30000'
x-ratelimit-limit-tokens:
- '150000000'
x-ratelimit-remaining-requests:
- '29999'
x-ratelimit-remaining-tokens:
- '149999974'
x-ratelimit-reset-requests:
- 2ms
x-ratelimit-reset-tokens:
- 0s
x-request-id:
- req_042a4e8f9432f6fde7a02037bb6caafa
http_version: HTTP/1.1
status_code: 400
- request:
body: '{"messages": [{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": [{"role": "system", "content": "You are Say Hi.
You just say hi to the user\nYour personal goal is: Say hi to the user\nTo give
my best complete final answer to the task respond using the exact following
format:\n\nThought: I now can give a great answer\nFinal Answer: Your final
answer must be the great and the most complete as possible, it must be outcome
described.\n\nI MUST use these formats, my job depends on it!"}, {"role": "user",
"content": "\nCurrent Task: Say hi to the user\n\nThis is the expected criteria
for your final answer: A greeting to the user\nyou MUST return the actual complete
content as the final answer, not a summary.\n\nBegin! This is VERY important
to you, use the tools available and give your best Final Answer, your job depends
on it!\n\nThought:"}]}], "model": "gpt-4o-mini", "tools": null}'
headers:
accept:
- application/json
accept-encoding:
- gzip, deflate
connection:
- keep-alive
content-length:
- '931'
content-type:
- application/json
host:
- api.openai.com
user-agent:
- OpenAI/Python 1.61.0
x-stainless-arch:
- arm64
x-stainless-async:
- 'false'
x-stainless-lang:
- python
x-stainless-os:
- MacOS
x-stainless-package-version:
- 1.61.0
x-stainless-retry-count:
- '0'
x-stainless-runtime:
- CPython
x-stainless-runtime-version:
- 3.12.8
method: POST
uri: https://api.openai.com/v1/chat/completions
response:
content: "{\n \"error\": {\n \"message\": \"Missing required parameter: 'messages[1].content[0].type'.\",\n
\ \"type\": \"invalid_request_error\",\n \"param\": \"messages[1].content[0].type\",\n
\ \"code\": \"missing_required_parameter\"\n }\n}"
headers:
CF-RAY:
- 91b54664bb1acef1-SJC
Connection:
- keep-alive
Content-Length:
- '219'
Content-Type:
- application/json
Date:
- Tue, 04 Mar 2025 23:50:17 GMT
Server:
- cloudflare
Set-Cookie:
- __cf_bm=.wGU4pJEajaSzFWjp05TBQwWbCNA2CgpYNu7UYOzbbM-1741132217-1.0.1.1-NoLiAx4qkplllldYYxZCOSQGsX6hsPUJIEyqmt84B3g7hjW1s7.jk9C9PYzXagHWjT0sQ9Ny4LZBA94lDJTfDBZpty8NJQha7ZKW0P_msH8;
path=/; expires=Wed, 05-Mar-25 00:20:17 GMT; domain=.api.openai.com; HttpOnly;
Secure; SameSite=None
- _cfuvid=GAjgJjVLtN49bMeWdWZDYLLkEkK51z5kxK4nKqhAzxY-1741132217161-0.0.1.1-604800000;
path=/; domain=.api.openai.com; HttpOnly; Secure; SameSite=None
X-Content-Type-Options:
- nosniff
access-control-expose-headers:
- X-Request-ID
alt-svc:
- h3=":443"; ma=86400
cf-cache-status:
- DYNAMIC
openai-organization:
- crewai-iuxna1
openai-processing-ms:
- '25'
openai-version:
- '2020-10-01'
strict-transport-security:
- max-age=31536000; includeSubDomains; preload
x-ratelimit-limit-requests:
- '30000'
x-ratelimit-limit-tokens:
- '150000000'
x-ratelimit-remaining-requests:
- '29999'
x-ratelimit-remaining-tokens:
- '149999974'
x-ratelimit-reset-requests:
- 2ms
x-ratelimit-reset-tokens:
- 0s
x-request-id:
- req_7a1d027da1ef4468e861e570c72e98fb
http_version: HTTP/1.1
status_code: 400
- request:
body: '{"messages": [{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": [{"role": "system", "content": "You are Say Hi.
You just say hi to the user\nYour personal goal is: Say hi to the user\nTo give
my best complete final answer to the task respond using the exact following
format:\n\nThought: I now can give a great answer\nFinal Answer: Your final
answer must be the great and the most complete as possible, it must be outcome
described.\n\nI MUST use these formats, my job depends on it!"}, {"role": "user",
"content": "\nCurrent Task: Say hi to the user\n\nThis is the expected criteria
for your final answer: A greeting to the user\nyou MUST return the actual complete
content as the final answer, not a summary.\n\nBegin! This is VERY important
to you, use the tools available and give your best Final Answer, your job depends
on it!\n\nThought:"}]}], "model": "gpt-4o-mini", "tools": null}'
headers:
accept:
- application/json
accept-encoding:
- gzip, deflate
connection:
- keep-alive
content-length:
- '931'
content-type:
- application/json
host:
- api.openai.com
user-agent:
- OpenAI/Python 1.61.0
x-stainless-arch:
- arm64
x-stainless-async:
- 'false'
x-stainless-lang:
- python
x-stainless-os:
- MacOS
x-stainless-package-version:
- 1.61.0
x-stainless-retry-count:
- '0'
x-stainless-runtime:
- CPython
x-stainless-runtime-version:
- 3.12.8
method: POST
uri: https://api.openai.com/v1/chat/completions
response:
content: "{\n \"error\": {\n \"message\": \"Missing required parameter: 'messages[1].content[0].type'.\",\n
\ \"type\": \"invalid_request_error\",\n \"param\": \"messages[1].content[0].type\",\n
\ \"code\": \"missing_required_parameter\"\n }\n}"
headers:
CF-RAY:
- 91b54666183beb22-SJC
Connection:
- keep-alive
Content-Length:
- '219'
Content-Type:
- application/json
Date:
- Tue, 04 Mar 2025 23:50:17 GMT
Server:
- cloudflare
Set-Cookie:
- __cf_bm=VwjWHHpkZMJlosI9RbMqxYDBS1t0JK4tWpAy4lST2QM-1741132217-1.0.1.1-u7PU.ZvVBTXNB5R8vaYfWdPXAjWZ3ZcTAy656VaGDZmKIckk5od._eQdn0W0EGVtEMm3TuF60z4GZAPDwMYvb3_3cw1RuEMmQbp4IIrl7VY;
path=/; expires=Wed, 05-Mar-25 00:20:17 GMT; domain=.api.openai.com; HttpOnly;
Secure; SameSite=None
- _cfuvid=NglAAsQBoiabMuuHFgilRjflSPFqS38VGKnGyweuCuw-1741132217438-0.0.1.1-604800000;
path=/; domain=.api.openai.com; HttpOnly; Secure; SameSite=None
X-Content-Type-Options:
- nosniff
access-control-expose-headers:
- X-Request-ID
alt-svc:
- h3=":443"; ma=86400
cf-cache-status:
- DYNAMIC
openai-organization:
- crewai-iuxna1
openai-processing-ms:
- '56'
openai-version:
- '2020-10-01'
strict-transport-security:
- max-age=31536000; includeSubDomains; preload
x-ratelimit-limit-requests:
- '30000'
x-ratelimit-limit-tokens:
- '150000000'
x-ratelimit-remaining-requests:
- '29999'
x-ratelimit-remaining-tokens:
- '149999974'
x-ratelimit-reset-requests:
- 2ms
x-ratelimit-reset-tokens:
- 0s
x-request-id:
- req_3c335b308b82cc2214783a4bf2fc0fd4
http_version: HTTP/1.1
status_code: 400
version: 1

359
tests/custom_llm_test.py Normal file
View File

@@ -0,0 +1,359 @@
from typing import Any, Dict, List, Optional, Union
from unittest.mock import Mock
import pytest
from crewai import Agent, Crew, Process, Task
from crewai.llms.base_llm import BaseLLM
from crewai.utilities.llm_utils import create_llm
class CustomLLM(BaseLLM):
"""Custom LLM implementation for testing.
This is a simple implementation of the BaseLLM abstract base class
that returns a predefined response for testing purposes.
"""
def __init__(self, response="Default response", model="test-model"):
"""Initialize the CustomLLM with a predefined response.
Args:
response: The predefined response to return from call().
"""
super().__init__(model=model)
self.response = response
self.call_count = 0
def call(
self,
messages,
tools=None,
callbacks=None,
available_functions=None,
):
"""
Mock LLM call that returns a predefined response.
Properly formats messages to match OpenAI's expected structure.
"""
self.call_count += 1
# If input is a string, convert to proper message format
if isinstance(messages, str):
messages = [{"role": "user", "content": messages}]
# Ensure each message has properly formatted content
for message in messages:
if isinstance(message["content"], str):
message["content"] = [{"type": "text", "text": message["content"]}]
# Return predefined response in expected format
if "Thought:" in str(messages):
return f"Thought: I will say hi\nFinal Answer: {self.response}"
return self.response
def supports_function_calling(self) -> bool:
"""Return False to indicate that function calling is not supported.
Returns:
False, indicating that this LLM does not support function calling.
"""
return False
def supports_stop_words(self) -> bool:
"""Return False to indicate that stop words are not supported.
Returns:
False, indicating that this LLM does not support stop words.
"""
return False
def get_context_window_size(self) -> int:
"""Return a default context window size.
Returns:
4096, a typical context window size for modern LLMs.
"""
return 4096
@pytest.mark.vcr(filter_headers=["authorization"])
def test_custom_llm_implementation():
"""Test that a custom LLM implementation works with create_llm."""
custom_llm = CustomLLM(response="The answer is 42")
# Test that create_llm returns the custom LLM instance directly
result_llm = create_llm(custom_llm)
assert result_llm is custom_llm
# Test calling the custom LLM
response = result_llm.call(
"What is the answer to life, the universe, and everything?"
)
# Verify that the response from the custom LLM was used
assert "42" in response
@pytest.mark.vcr(filter_headers=["authorization"])
def test_custom_llm_within_crew():
"""Test that a custom LLM implementation works with create_llm."""
custom_llm = CustomLLM(response="Hello! Nice to meet you!", model="test-model")
agent = Agent(
role="Say Hi",
goal="Say hi to the user",
backstory="""You just say hi to the user""",
llm=custom_llm,
)
task = Task(
description="Say hi to the user",
expected_output="A greeting to the user",
agent=agent,
)
crew = Crew(
agents=[agent],
tasks=[task],
process=Process.sequential,
)
result = crew.kickoff()
# Assert the LLM was called
assert custom_llm.call_count > 0
# Assert we got a response
assert "Hello!" in result.raw
def test_custom_llm_message_formatting():
"""Test that the custom LLM properly formats messages"""
custom_llm = CustomLLM(response="Test response", model="test-model")
# Test with string input
result = custom_llm.call("Test message")
assert result == "Test response"
# Test with message list
messages = [
{"role": "system", "content": "System message"},
{"role": "user", "content": "User message"},
]
result = custom_llm.call(messages)
assert result == "Test response"
class JWTAuthLLM(BaseLLM):
"""Custom LLM implementation with JWT authentication."""
def __init__(self, jwt_token: str):
super().__init__(model="test-model")
if not jwt_token or not isinstance(jwt_token, str):
raise ValueError("Invalid JWT token")
self.jwt_token = jwt_token
self.calls = []
self.stop = []
def call(
self,
messages: Union[str, List[Dict[str, str]]],
tools: Optional[List[dict]] = None,
callbacks: Optional[List[Any]] = None,
available_functions: Optional[Dict[str, Any]] = None,
) -> Union[str, Any]:
"""Record the call and return a predefined response."""
self.calls.append(
{
"messages": messages,
"tools": tools,
"callbacks": callbacks,
"available_functions": available_functions,
}
)
# In a real implementation, this would use the JWT token to authenticate
# with an external service
return "Response from JWT-authenticated LLM"
def supports_function_calling(self) -> bool:
"""Return True to indicate that function calling is supported."""
return True
def supports_stop_words(self) -> bool:
"""Return True to indicate that stop words are supported."""
return True
def get_context_window_size(self) -> int:
"""Return a default context window size."""
return 8192
def test_custom_llm_with_jwt_auth():
"""Test a custom LLM implementation with JWT authentication."""
jwt_llm = JWTAuthLLM(jwt_token="example.jwt.token")
# Test that create_llm returns the JWT-authenticated LLM instance directly
result_llm = create_llm(jwt_llm)
assert result_llm is jwt_llm
# Test calling the JWT-authenticated LLM
response = result_llm.call("Test message")
# Verify that the JWT-authenticated LLM was called
assert len(jwt_llm.calls) > 0
# Verify that the response from the JWT-authenticated LLM was used
assert response == "Response from JWT-authenticated LLM"
def test_jwt_auth_llm_validation():
"""Test that JWT token validation works correctly."""
# Test with invalid JWT token (empty string)
with pytest.raises(ValueError, match="Invalid JWT token"):
JWTAuthLLM(jwt_token="")
# Test with invalid JWT token (non-string)
with pytest.raises(ValueError, match="Invalid JWT token"):
JWTAuthLLM(jwt_token=None)
class TimeoutHandlingLLM(BaseLLM):
"""Custom LLM implementation with timeout handling and retry logic."""
def __init__(self, max_retries: int = 3, timeout: int = 30):
"""Initialize the TimeoutHandlingLLM with retry and timeout settings.
Args:
max_retries: Maximum number of retry attempts.
timeout: Timeout in seconds for each API call.
"""
super().__init__(model="test-model")
self.max_retries = max_retries
self.timeout = timeout
self.calls = []
self.stop = []
self.fail_count = 0 # Number of times to simulate failure
def call(
self,
messages: Union[str, List[Dict[str, str]]],
tools: Optional[List[dict]] = None,
callbacks: Optional[List[Any]] = None,
available_functions: Optional[Dict[str, Any]] = None,
) -> Union[str, Any]:
"""Simulate API calls with timeout handling and retry logic.
Args:
messages: Input messages for the LLM.
tools: Optional list of tool schemas for function calling.
callbacks: Optional list of callback functions.
available_functions: Optional dict mapping function names to callables.
Returns:
A response string based on whether this is the first attempt or a retry.
Raises:
TimeoutError: If all retry attempts fail.
"""
# Record the initial call
self.calls.append(
{
"messages": messages,
"tools": tools,
"callbacks": callbacks,
"available_functions": available_functions,
"attempt": 0,
}
)
# Simulate retry logic
for attempt in range(self.max_retries):
# Skip the first attempt recording since we already did that above
if attempt == 0:
# Simulate a failure if fail_count > 0
if self.fail_count > 0:
self.fail_count -= 1
# If we've used all retries, raise an error
if attempt == self.max_retries - 1:
raise TimeoutError(
f"LLM request failed after {self.max_retries} attempts"
)
# Otherwise, continue to the next attempt (simulating backoff)
continue
else:
# Success on first attempt
return "First attempt response"
else:
# This is a retry attempt (attempt > 0)
# Always record retry attempts
self.calls.append(
{
"retry_attempt": attempt,
"messages": messages,
"tools": tools,
"callbacks": callbacks,
"available_functions": available_functions,
}
)
# Simulate a failure if fail_count > 0
if self.fail_count > 0:
self.fail_count -= 1
# If we've used all retries, raise an error
if attempt == self.max_retries - 1:
raise TimeoutError(
f"LLM request failed after {self.max_retries} attempts"
)
# Otherwise, continue to the next attempt (simulating backoff)
continue
else:
# Success on retry
return "Response after retry"
def supports_function_calling(self) -> bool:
"""Return True to indicate that function calling is supported.
Returns:
True, indicating that this LLM supports function calling.
"""
return True
def supports_stop_words(self) -> bool:
"""Return True to indicate that stop words are supported.
Returns:
True, indicating that this LLM supports stop words.
"""
return True
def get_context_window_size(self) -> int:
"""Return a default context window size.
Returns:
8192, a typical context window size for modern LLMs.
"""
return 8192
def test_timeout_handling_llm():
"""Test a custom LLM implementation with timeout handling and retry logic."""
# Test successful first attempt
llm = TimeoutHandlingLLM()
response = llm.call("Test message")
assert response == "First attempt response"
assert len(llm.calls) == 1
# Test successful retry
llm = TimeoutHandlingLLM()
llm.fail_count = 1 # Fail once, then succeed
response = llm.call("Test message")
assert response == "Response after retry"
assert len(llm.calls) == 2 # Initial call + successful retry call
# Test failure after all retries
llm = TimeoutHandlingLLM(max_retries=2)
llm.fail_count = 2 # Fail twice, which is all retries
with pytest.raises(TimeoutError, match="LLM request failed after 2 attempts"):
llm.call("Test message")
assert len(llm.calls) == 2 # Initial call + failed retry attempt

65
tests/task_test.py
View File

@@ -15,6 +15,7 @@ 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
from crewai.utilities.string_utils import interpolate_only
def test_task_tool_reflect_agent_tools():
@@ -822,7 +823,7 @@ def test_interpolate_only():
# Test JSON structure preservation
json_string = '{"info": "Look at {placeholder}", "nested": {"val": "{nestedVal}"}}'
result = task.interpolate_only(
result = interpolate_only(
input_string=json_string,
inputs={"placeholder": "the data", "nestedVal": "something else"},
)
@@ -833,20 +834,18 @@ def test_interpolate_only():
# Test normal string interpolation
normal_string = "Hello {name}, welcome to {place}!"
result = task.interpolate_only(
result = 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"})
result = 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"}
)
result = interpolate_only(input_string=no_placeholders, inputs={"unused": "value"})
assert result == no_placeholders
@@ -858,7 +857,7 @@ def test_interpolate_only_with_dict_inside_expected_output():
)
json_string = '{"questions": {"main_question": "What is the user\'s name?", "secondary_question": "What is the user\'s age?"}}'
result = task.interpolate_only(
result = interpolate_only(
input_string=json_string,
inputs={
"questions": {
@@ -872,18 +871,16 @@ def test_interpolate_only_with_dict_inside_expected_output():
assert result == json_string
normal_string = "Hello {name}, welcome to {place}!"
result = task.interpolate_only(
result = 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"})
result = 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"}
)
result = interpolate_only(input_string=no_placeholders, inputs={"unused": "value"})
assert result == no_placeholders
@@ -1085,12 +1082,12 @@ def test_interpolate_with_list_of_strings():
# Test simple list of strings
input_str = "Available items: {items}"
inputs = {"items": ["apple", "banana", "cherry"]}
result = task.interpolate_only(input_str, inputs)
result = 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)
result = interpolate_only(input_str, empty_list_input)
assert result == "Available items: []"
@@ -1106,7 +1103,7 @@ def test_interpolate_with_list_of_dicts():
{"name": "Bob", "age": 25, "skills": ["Java", "Cloud"]},
]
}
result = task.interpolate_only("{people}", input_data)
result = interpolate_only("{people}", input_data)
parsed_result = eval(result)
assert isinstance(parsed_result, list)
@@ -1138,7 +1135,7 @@ def test_interpolate_with_nested_structures():
],
}
}
result = task.interpolate_only("{company}", input_data)
result = interpolate_only("{company}", input_data)
parsed = eval(result)
assert parsed["name"] == "TechCorp"
@@ -1161,7 +1158,7 @@ def test_interpolate_with_special_characters():
"empty": "",
}
}
result = task.interpolate_only("{special_data}", input_data)
result = interpolate_only("{special_data}", input_data)
parsed = eval(result)
assert parsed["quotes"] == """This has "double" and 'single' quotes"""
@@ -1188,7 +1185,7 @@ def test_interpolate_mixed_types():
},
}
}
result = task.interpolate_only("{data}", input_data)
result = interpolate_only("{data}", input_data)
parsed = eval(result)
assert parsed["name"] == "Test Dataset"
@@ -1216,7 +1213,7 @@ def test_interpolate_complex_combination():
},
]
}
result = task.interpolate_only("{report}", input_data)
result = interpolate_only("{report}", input_data)
parsed = eval(result)
assert len(parsed) == 2
@@ -1233,7 +1230,7 @@ def test_interpolate_invalid_type_validation():
# 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
interpolate_only("{data}", {"data": set()}) # type: ignore we are purposely testing this failure
assert "Unsupported type set" in str(excinfo.value)
@@ -1246,7 +1243,7 @@ def test_interpolate_invalid_type_validation():
}
}
with pytest.raises(ValueError) as excinfo:
task.interpolate_only("{data}", {"data": invalid_nested})
interpolate_only("{data}", {"data": invalid_nested})
assert "Unsupported type set" in str(excinfo.value)
@@ -1265,24 +1262,22 @@ def test_interpolate_custom_object_validation():
# 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
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)}}
)
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)]})
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(
interpolate_only(
"{data}", {"data": {"level1": {"level2": [{"level3": CustomObject(5)}]}}}
)
assert "Unsupported type CustomObject" in str(excinfo.value)
@@ -1306,7 +1301,7 @@ def test_interpolate_valid_complex_types():
}
# Should not raise any errors
result = task.interpolate_only("{data}", {"data": valid_data})
result = interpolate_only("{data}", {"data": valid_data})
parsed = eval(result)
assert parsed["name"] == "Valid Dataset"
assert parsed["stats"]["nested"]["deeper"]["b"] == 2.5
@@ -1319,16 +1314,16 @@ def test_interpolate_edge_cases():
)
# Test empty dict and list
assert task.interpolate_only("{}", {"data": {}}) == "{}"
assert task.interpolate_only("[]", {"data": []}) == "[]"
assert interpolate_only("{}", {"data": {}}) == "{}"
assert interpolate_only("[]", {"data": []}) == "[]"
# Test numeric types
assert task.interpolate_only("{num}", {"num": 42}) == "42"
assert task.interpolate_only("{num}", {"num": 3.14}) == "3.14"
assert interpolate_only("{num}", {"num": 42}) == "42"
assert 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"
assert interpolate_only("{flag}", {"flag": True}) == "True"
assert interpolate_only("{flag}", {"flag": False}) == "False"
def test_interpolate_valid_types():
@@ -1346,7 +1341,7 @@ def test_interpolate_valid_types():
"nested": {"flag": True, "empty": None},
}
result = task.interpolate_only("{data}", {"data": valid_data})
result = interpolate_only("{data}", {"data": valid_data})
parsed = eval(result)
assert parsed["active"] is True

187
tests/utilities/test_string_utils.py Normal file
View File

@@ -0,0 +1,187 @@
from typing import Any, Dict, List, Union
import pytest
from crewai.utilities.string_utils import interpolate_only
class TestInterpolateOnly:
"""Tests for the interpolate_only function in string_utils.py."""
def test_basic_variable_interpolation(self):
"""Test basic variable interpolation works correctly."""
template = "Hello, {name}! Welcome to {company}."
inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
"name": "Alice",
"company": "CrewAI",
}
result = interpolate_only(template, inputs)
assert result == "Hello, Alice! Welcome to CrewAI."
def test_multiple_occurrences_of_same_variable(self):
"""Test that multiple occurrences of the same variable are replaced."""
template = "{name} is using {name}'s account."
inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
"name": "Bob"
}
result = interpolate_only(template, inputs)
assert result == "Bob is using Bob's account."
def test_json_structure_preservation(self):
"""Test that JSON structures are preserved and not interpolated incorrectly."""
template = """
Instructions for {agent}:
Please return the following object:
{"name": "person's name", "age": 25, "skills": ["coding", "testing"]}
"""
inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
"agent": "DevAgent"
}
result = interpolate_only(template, inputs)
assert "Instructions for DevAgent:" in result
assert (
'{"name": "person\'s name", "age": 25, "skills": ["coding", "testing"]}'
in result
)
def test_complex_nested_json(self):
"""Test with complex JSON structures containing curly braces."""
template = """
{agent} needs to process:
{
"config": {
"nested": {
"value": 42
},
"arrays": [1, 2, {"inner": "value"}]
}
}
"""
inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
"agent": "DataProcessor"
}
result = interpolate_only(template, inputs)
assert "DataProcessor needs to process:" in result
assert '"nested": {' in result
assert '"value": 42' in result
assert '[1, 2, {"inner": "value"}]' in result
def test_missing_variable(self):
"""Test that an error is raised when a required variable is missing."""
template = "Hello, {name}!"
inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
"not_name": "Alice"
}
with pytest.raises(KeyError) as excinfo:
interpolate_only(template, inputs)
assert "template variable" in str(excinfo.value).lower()
assert "name" in str(excinfo.value)
def test_invalid_input_types(self):
"""Test that an error is raised with invalid input types."""
template = "Hello, {name}!"
# Using Any for this test since we're intentionally testing an invalid type
inputs: Dict[str, Any] = {"name": object()} # Object is not a valid input type
with pytest.raises(ValueError) as excinfo:
interpolate_only(template, inputs)
assert "unsupported type" in str(excinfo.value).lower()
def test_empty_input_string(self):
"""Test handling of empty or None input string."""
inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
"name": "Alice"
}
assert interpolate_only("", inputs) == ""
assert interpolate_only(None, inputs) == ""
def test_no_variables_in_template(self):
"""Test a template with no variables to replace."""
template = "This is a static string with no variables."
inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
"name": "Alice"
}
result = interpolate_only(template, inputs)
assert result == template
def test_variable_name_starting_with_underscore(self):
"""Test variables starting with underscore are replaced correctly."""
template = "Variable: {_special_var}"
inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
"_special_var": "Special Value"
}
result = interpolate_only(template, inputs)
assert result == "Variable: Special Value"
def test_preserves_non_matching_braces(self):
"""Test that non-matching braces patterns are preserved."""
template = (
"This {123} and {!var} should not be replaced but {valid_var} should."
)
inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
"valid_var": "works"
}
result = interpolate_only(template, inputs)
assert (
result == "This {123} and {!var} should not be replaced but works should."
)
def test_complex_mixed_scenario(self):
"""Test a complex scenario with both valid variables and JSON structures."""
template = """
{agent_name} is working on task {task_id}.
Instructions:
1. Process the data
2. Return results as:
{
"taskId": "{task_id}",
"results": {
"processed_by": "agent_name",
"status": "complete",
"values": [1, 2, 3]
}
}
"""
inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
"agent_name": "AnalyticsAgent",
"task_id": "T-12345",
}
result = interpolate_only(template, inputs)
assert "AnalyticsAgent is working on task T-12345" in result
assert '"taskId": "T-12345"' in result
assert '"processed_by": "agent_name"' in result # This shouldn't be replaced
assert '"values": [1, 2, 3]' in result
def test_empty_inputs_dictionary(self):
"""Test that an error is raised with empty inputs dictionary."""
template = "Hello, {name}!"
inputs: Dict[str, Any] = {}
with pytest.raises(ValueError) as excinfo:
interpolate_only(template, inputs)
assert "inputs dictionary cannot be empty" in str(excinfo.value).lower()