mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-09 16:18:30 +00:00
Improve code based on PR feedback: add type hints, error handling, docs, and tests
Co-Authored-By: Joe Moura <joao@crewai.com>
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
import inspect
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Dict, TypeVar, cast
|
||||
from typing import Any, Callable, Dict, List, TypeVar, Union, cast
|
||||
|
||||
import yaml
|
||||
from dotenv import load_dotenv
|
||||
@@ -116,15 +116,31 @@ def CrewBase(cls: T) -> T:
|
||||
def _map_agent_variables(
|
||||
self,
|
||||
agent_name: str,
|
||||
agent_info: Dict[str, Any],
|
||||
agent_info: Union[Dict[str, Any], List[Dict[str, Any]]],
|
||||
agents: Dict[str, Callable],
|
||||
llms: Dict[str, Callable],
|
||||
tool_functions: Dict[str, Callable],
|
||||
cache_handler_functions: Dict[str, Callable],
|
||||
callbacks: Dict[str, Callable],
|
||||
) -> None:
|
||||
"""Maps agent variables from configuration to internal state.
|
||||
|
||||
Args:
|
||||
agent_name: Name of the agent.
|
||||
agent_info: Configuration as a dictionary or list of configurations.
|
||||
agents: Dictionary of agent functions.
|
||||
llms: Dictionary of LLM functions.
|
||||
tool_functions: Dictionary of tool functions.
|
||||
cache_handler_functions: Dictionary of cache handler functions.
|
||||
callbacks: Dictionary of callback functions.
|
||||
|
||||
Raises:
|
||||
ValueError: When an empty list is provided as agent_info.
|
||||
"""
|
||||
# If agent_info is a list, use the first item as the configuration
|
||||
if isinstance(agent_info, list) and len(agent_info) > 0:
|
||||
if isinstance(agent_info, list):
|
||||
if not agent_info:
|
||||
raise ValueError(f"Empty agent configuration list for agent {agent_name}")
|
||||
agent_info = agent_info[0]
|
||||
|
||||
if llm := agent_info.get("llm"):
|
||||
|
||||
@@ -6,8 +6,11 @@ from pathlib import Path
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
# Add a simple test to verify the fix works
|
||||
def test_list_format_in_yaml():
|
||||
|
||||
class TestYamlConfig:
|
||||
"""Tests for YAML configuration handling."""
|
||||
|
||||
def test_list_format_in_yaml(self):
|
||||
"""Test that list format in YAML is handled correctly."""
|
||||
# Create a test YAML content with list format
|
||||
yaml_content = """
|
||||
@@ -31,7 +34,9 @@ def test_list_format_in_yaml():
|
||||
# with our fix applied
|
||||
def map_agent_variables(agent_name, agent_info):
|
||||
# This is the fix we implemented
|
||||
if isinstance(agent_info, list) and len(agent_info) > 0:
|
||||
if isinstance(agent_info, list):
|
||||
if not agent_info:
|
||||
raise ValueError(f"Empty agent configuration list for agent {agent_name}")
|
||||
agent_info = agent_info[0]
|
||||
|
||||
# Try to access a dictionary method on agent_info
|
||||
@@ -42,5 +47,69 @@ def test_list_format_in_yaml():
|
||||
# Call the function - this would raise AttributeError before the fix
|
||||
result = map_agent_variables(agent_name, agent_info)
|
||||
|
||||
# Verify the result
|
||||
assert result == "test_agent"
|
||||
def test_empty_list_in_yaml(self):
|
||||
"""Test that empty list in YAML raises appropriate error."""
|
||||
# Create a test YAML content with empty list
|
||||
yaml_content = """
|
||||
test_agent: []
|
||||
"""
|
||||
|
||||
# Parse the YAML content
|
||||
data = yaml.safe_load(yaml_content)
|
||||
|
||||
# Get the agent_info which should be an empty list
|
||||
agent_name = "test_agent"
|
||||
agent_info = data[agent_name]
|
||||
|
||||
# Verify it's a list
|
||||
assert isinstance(agent_info, list)
|
||||
assert len(agent_info) == 0
|
||||
|
||||
# Create a function that simulates the behavior of _map_agent_variables
|
||||
def map_agent_variables(agent_name, agent_info):
|
||||
if isinstance(agent_info, list):
|
||||
if not agent_info:
|
||||
raise ValueError(f"Empty agent configuration list for agent {agent_name}")
|
||||
agent_info = agent_info[0]
|
||||
return agent_info
|
||||
|
||||
# Call the function - should raise ValueError
|
||||
with pytest.raises(ValueError, match=f"Empty agent configuration list for agent {agent_name}"):
|
||||
map_agent_variables(agent_name, agent_info)
|
||||
def test_multiple_items_in_list(self):
|
||||
"""Test that when multiple items are in the list, the first one is used."""
|
||||
# Create a test YAML content with multiple items in the list
|
||||
yaml_content = """
|
||||
test_agent:
|
||||
- name: first_agent
|
||||
role: First Agent
|
||||
goal: First Goal
|
||||
- name: second_agent
|
||||
role: Second Agent
|
||||
goal: Second Goal
|
||||
"""
|
||||
|
||||
# Parse the YAML content
|
||||
data = yaml.safe_load(yaml_content)
|
||||
|
||||
# Get the agent_info which should be a list
|
||||
agent_name = "test_agent"
|
||||
agent_info = data[agent_name]
|
||||
|
||||
# Verify it's a list with multiple items
|
||||
assert isinstance(agent_info, list)
|
||||
assert len(agent_info) > 1
|
||||
|
||||
# Create a function that simulates the behavior of _map_agent_variables
|
||||
def map_agent_variables(agent_name, agent_info):
|
||||
if isinstance(agent_info, list):
|
||||
if not agent_info:
|
||||
raise ValueError(f"Empty agent configuration list for agent {agent_name}")
|
||||
agent_info = agent_info[0]
|
||||
return agent_info.get("name")
|
||||
|
||||
# Call the function - should return name from the first item
|
||||
result = map_agent_variables(agent_name, agent_info)
|
||||
|
||||
# Verify only the first item was used
|
||||
assert result == "first_agent"
|
||||
|
||||
Reference in New Issue
Block a user