mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-20 13:28:13 +00:00
feat: restructure project as UV workspace with crewai in lib/
This commit is contained in:
Greyson LaLonde
@@ -1 +0,0 @@
|
||||
"""Tests for agent adapters."""
|
||||
@@ -1,116 +0,0 @@
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from pydantic import BaseModel
|
||||
|
||||
from crewai.agent import BaseAgent
|
||||
from crewai.agents.agent_adapters.base_agent_adapter import BaseAgentAdapter
|
||||
from crewai.tools import BaseTool
|
||||
from crewai.utilities.token_counter_callback import TokenProcess
|
||||
|
||||
|
||||
# Concrete implementation for testing
|
||||
class ConcreteAgentAdapter(BaseAgentAdapter):
|
||||
def configure_tools(
|
||||
self, tools: list[BaseTool] | None = None, **kwargs: Any
|
||||
) -> None:
|
||||
# Simple implementation for testing
|
||||
self.tools = tools or []
|
||||
|
||||
def execute_task(
|
||||
self,
|
||||
task: Any,
|
||||
context: str | None = None,
|
||||
tools: list[Any] | None = None,
|
||||
) -> str:
|
||||
# Dummy implementation needed due to BaseAgent inheritance
|
||||
return "Task executed"
|
||||
|
||||
def create_agent_executor(self, tools: list[BaseTool] | None = None) -> Any:
|
||||
# Dummy implementation
|
||||
return None
|
||||
|
||||
def get_delegation_tools(
|
||||
self, tools: list[BaseTool], tool_map: dict[str, BaseTool] | None
|
||||
) -> list[BaseTool]:
|
||||
# Dummy implementation
|
||||
return []
|
||||
|
||||
def _parse_output(self, agent_output: Any, token_process: TokenProcess):
|
||||
# Dummy implementation
|
||||
pass
|
||||
|
||||
def get_output_converter(self, tools: list[BaseTool] | None = None) -> Any:
|
||||
# Dummy implementation
|
||||
return None
|
||||
|
||||
def get_platform_tools(self, apps: Any) -> list[BaseTool]:
|
||||
# Dummy implementation
|
||||
return []
|
||||
|
||||
|
||||
def test_base_agent_adapter_initialization():
|
||||
"""Test initialization of the concrete agent adapter."""
|
||||
adapter = ConcreteAgentAdapter(
|
||||
role="test role", goal="test goal", backstory="test backstory"
|
||||
)
|
||||
assert isinstance(adapter, BaseAgent)
|
||||
assert isinstance(adapter, BaseAgentAdapter)
|
||||
assert adapter.role == "test role"
|
||||
assert adapter._agent_config is None
|
||||
assert adapter.adapted_structured_output is False
|
||||
|
||||
|
||||
def test_base_agent_adapter_initialization_with_config():
|
||||
"""Test initialization with agent_config."""
|
||||
config = {"model": "gpt-4"}
|
||||
adapter = ConcreteAgentAdapter(
|
||||
agent_config=config,
|
||||
role="test role",
|
||||
goal="test goal",
|
||||
backstory="test backstory",
|
||||
)
|
||||
assert adapter._agent_config == config
|
||||
|
||||
|
||||
def test_configure_tools_method_exists():
|
||||
"""Test that configure_tools method exists and can be called."""
|
||||
adapter = ConcreteAgentAdapter(
|
||||
role="test role", goal="test goal", backstory="test backstory"
|
||||
)
|
||||
# Create dummy tools if needed, or pass None
|
||||
tools = []
|
||||
adapter.configure_tools(tools)
|
||||
assert hasattr(adapter, "tools")
|
||||
assert adapter.tools == tools
|
||||
|
||||
|
||||
def test_configure_structured_output_method_exists():
|
||||
"""Test that configure_structured_output method exists and can be called."""
|
||||
adapter = ConcreteAgentAdapter(
|
||||
role="test role", goal="test goal", backstory="test backstory"
|
||||
)
|
||||
|
||||
# Define a dummy structure or pass None/Any
|
||||
class DummyOutput(BaseModel):
|
||||
data: str
|
||||
|
||||
structured_output = DummyOutput
|
||||
adapter.configure_structured_output(structured_output)
|
||||
# Add assertions here if configure_structured_output modifies state
|
||||
# For now, just ensuring it runs without error is sufficient
|
||||
|
||||
|
||||
def test_base_agent_adapter_inherits_base_agent():
|
||||
"""Test that BaseAgentAdapter inherits from BaseAgent."""
|
||||
assert issubclass(BaseAgentAdapter, BaseAgent)
|
||||
|
||||
|
||||
class ConcreteAgentAdapterWithoutRequiredMethods(BaseAgentAdapter):
|
||||
pass
|
||||
|
||||
|
||||
def test_base_agent_adapter_fails_without_required_methods():
|
||||
"""Test that BaseAgentAdapter fails without required methods."""
|
||||
with pytest.raises(TypeError):
|
||||
ConcreteAgentAdapterWithoutRequiredMethods() # type: ignore
|
||||
@@ -1,94 +0,0 @@
|
||||
from typing import Any, List
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from crewai.agents.agent_adapters.base_tool_adapter import BaseToolAdapter
|
||||
from crewai.tools.base_tool import BaseTool
|
||||
|
||||
|
||||
class ConcreteToolAdapter(BaseToolAdapter):
|
||||
def configure_tools(self, tools: List[BaseTool]) -> None:
|
||||
self.converted_tools = [f"converted_{tool.name}" for tool in tools]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_tool_1():
|
||||
tool = Mock(spec=BaseTool)
|
||||
tool.name = "Mock Tool 1"
|
||||
return tool
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_tool_2():
|
||||
tool = Mock(spec=BaseTool)
|
||||
tool.name = "MockTool2"
|
||||
return tool
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tools_list(mock_tool_1, mock_tool_2):
|
||||
return [mock_tool_1, mock_tool_2]
|
||||
|
||||
|
||||
def test_initialization_with_tools(tools_list):
|
||||
adapter = ConcreteToolAdapter(tools=tools_list)
|
||||
assert adapter.original_tools == tools_list
|
||||
assert adapter.converted_tools == [] # Conversion happens in configure_tools
|
||||
|
||||
|
||||
def test_initialization_without_tools():
|
||||
adapter = ConcreteToolAdapter()
|
||||
assert adapter.original_tools == []
|
||||
assert adapter.converted_tools == []
|
||||
|
||||
|
||||
def test_configure_tools(tools_list):
|
||||
adapter = ConcreteToolAdapter()
|
||||
adapter.configure_tools(tools_list)
|
||||
assert adapter.converted_tools == ["converted_Mock Tool 1", "converted_MockTool2"]
|
||||
assert adapter.original_tools == [] # original_tools is only set in init
|
||||
|
||||
adapter_with_init_tools = ConcreteToolAdapter(tools=tools_list)
|
||||
adapter_with_init_tools.configure_tools(tools_list)
|
||||
assert adapter_with_init_tools.converted_tools == [
|
||||
"converted_Mock Tool 1",
|
||||
"converted_MockTool2",
|
||||
]
|
||||
assert adapter_with_init_tools.original_tools == tools_list
|
||||
|
||||
|
||||
def test_tools_method(tools_list):
|
||||
adapter = ConcreteToolAdapter()
|
||||
adapter.configure_tools(tools_list)
|
||||
assert adapter.tools() == ["converted_Mock Tool 1", "converted_MockTool2"]
|
||||
|
||||
|
||||
def test_tools_method_empty():
|
||||
adapter = ConcreteToolAdapter()
|
||||
assert adapter.tools() == []
|
||||
|
||||
|
||||
def test_sanitize_tool_name_with_spaces():
|
||||
adapter = ConcreteToolAdapter()
|
||||
assert adapter.sanitize_tool_name("Tool With Spaces") == "Tool_With_Spaces"
|
||||
|
||||
|
||||
def test_sanitize_tool_name_without_spaces():
|
||||
adapter = ConcreteToolAdapter()
|
||||
assert adapter.sanitize_tool_name("ToolWithoutSpaces") == "ToolWithoutSpaces"
|
||||
|
||||
|
||||
def test_sanitize_tool_name_empty():
|
||||
adapter = ConcreteToolAdapter()
|
||||
assert adapter.sanitize_tool_name("") == ""
|
||||
|
||||
|
||||
class ConcreteToolAdapterWithoutRequiredMethods(BaseToolAdapter):
|
||||
pass
|
||||
|
||||
|
||||
def test_tool_adapted_fails_without_required_methods():
|
||||
"""Test that BaseToolAdapter fails without required methods."""
|
||||
with pytest.raises(TypeError):
|
||||
ConcreteToolAdapterWithoutRequiredMethods() # type: ignore
|
||||
@@ -1 +0,0 @@
|
||||
"""Tests for agent builder."""
|
||||
@@ -1,37 +0,0 @@
|
||||
import hashlib
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from crewai.agents.agent_builder.base_agent import BaseAgent
|
||||
from crewai.tools.base_tool import BaseTool
|
||||
|
||||
|
||||
class MockAgent(BaseAgent):
|
||||
def execute_task(
|
||||
self,
|
||||
task: Any,
|
||||
context: str | None = None,
|
||||
tools: list[BaseTool] | None = None,
|
||||
) -> str:
|
||||
return ""
|
||||
|
||||
def create_agent_executor(self, tools=None) -> None: ...
|
||||
|
||||
def get_delegation_tools(self, agents: list["BaseAgent"]): ...
|
||||
|
||||
def get_platform_tools(self, apps: list[Any]): ...
|
||||
|
||||
def get_output_converter(
|
||||
self, llm: Any, text: str, model: type[BaseModel] | None, instructions: str
|
||||
): ...
|
||||
|
||||
|
||||
def test_key():
|
||||
agent = MockAgent(
|
||||
role="test role",
|
||||
goal="test goal",
|
||||
backstory="test backstory",
|
||||
)
|
||||
hash = hashlib.md5("test role|test goal|test backstory".encode(), usedforsecurity=False).hexdigest()
|
||||
assert agent.key == hash
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,117 +0,0 @@
|
||||
from datetime import datetime
|
||||
from unittest.mock import patch
|
||||
|
||||
from crewai.agent import Agent
|
||||
from crewai.task import Task
|
||||
|
||||
|
||||
def test_agent_inject_date():
|
||||
"""Test that the inject_date flag injects the current date into the task.
|
||||
|
||||
Tests that when inject_date=True, the current date is added to the task description.
|
||||
"""
|
||||
with patch("datetime.datetime") as mock_datetime:
|
||||
mock_datetime.now.return_value = datetime(2025, 1, 1)
|
||||
|
||||
agent = Agent(
|
||||
role="test_agent",
|
||||
goal="test_goal",
|
||||
backstory="test_backstory",
|
||||
inject_date=True,
|
||||
)
|
||||
|
||||
task = Task(
|
||||
description="Test task",
|
||||
expected_output="Test output",
|
||||
agent=agent,
|
||||
)
|
||||
|
||||
# Store original description
|
||||
original_description = task.description
|
||||
|
||||
agent._inject_date_to_task(task)
|
||||
|
||||
assert "Current Date: 2025-01-01" in task.description
|
||||
assert task.description != original_description
|
||||
|
||||
|
||||
def test_agent_without_inject_date():
|
||||
"""Test that without inject_date flag, no date is injected.
|
||||
|
||||
Tests that when inject_date=False (default), no date is added to the task description.
|
||||
"""
|
||||
agent = Agent(
|
||||
role="test_agent",
|
||||
goal="test_goal",
|
||||
backstory="test_backstory",
|
||||
# inject_date is False by default
|
||||
)
|
||||
|
||||
task = Task(
|
||||
description="Test task",
|
||||
expected_output="Test output",
|
||||
agent=agent,
|
||||
)
|
||||
|
||||
original_description = task.description
|
||||
|
||||
agent._inject_date_to_task(task)
|
||||
|
||||
assert task.description == original_description
|
||||
|
||||
|
||||
def test_agent_inject_date_custom_format():
|
||||
"""Test that the inject_date flag with custom date_format works correctly.
|
||||
|
||||
Tests that when inject_date=True with a custom date_format, the date is formatted correctly.
|
||||
"""
|
||||
with patch("datetime.datetime") as mock_datetime:
|
||||
mock_datetime.now.return_value = datetime(2025, 1, 1)
|
||||
|
||||
agent = Agent(
|
||||
role="test_agent",
|
||||
goal="test_goal",
|
||||
backstory="test_backstory",
|
||||
inject_date=True,
|
||||
date_format="%d/%m/%Y",
|
||||
)
|
||||
|
||||
task = Task(
|
||||
description="Test task",
|
||||
expected_output="Test output",
|
||||
agent=agent,
|
||||
)
|
||||
|
||||
# Store original description
|
||||
original_description = task.description
|
||||
|
||||
agent._inject_date_to_task(task)
|
||||
|
||||
assert "Current Date: 01/01/2025" in task.description
|
||||
assert task.description != original_description
|
||||
|
||||
|
||||
def test_agent_inject_date_invalid_format():
|
||||
"""Test error handling with invalid date format.
|
||||
|
||||
Tests that when an invalid date_format is provided, the task description remains unchanged.
|
||||
"""
|
||||
agent = Agent(
|
||||
role="test_agent",
|
||||
goal="test_goal",
|
||||
backstory="test_backstory",
|
||||
inject_date=True,
|
||||
date_format="invalid",
|
||||
)
|
||||
|
||||
task = Task(
|
||||
description="Test task",
|
||||
expected_output="Test output",
|
||||
agent=agent,
|
||||
)
|
||||
|
||||
original_description = task.description
|
||||
|
||||
agent._inject_date_to_task(task)
|
||||
|
||||
assert task.description == original_description
|
||||
@@ -1,238 +0,0 @@
|
||||
"""Tests for reasoning in agents."""
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from crewai import Agent, Task
|
||||
from crewai.llm import LLM
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_llm_responses():
|
||||
"""Fixture for mock LLM responses."""
|
||||
return {
|
||||
"ready": "I'll solve this simple math problem.\n\nREADY: I am ready to execute the task.\n\n",
|
||||
"not_ready": "I need to think about derivatives.\n\nNOT READY: I need to refine my plan because I'm not sure about the derivative rules.",
|
||||
"ready_after_refine": "I'll use the power rule for derivatives where d/dx(x^n) = n*x^(n-1).\n\nREADY: I am ready to execute the task.",
|
||||
"execution": "4",
|
||||
}
|
||||
|
||||
|
||||
def test_agent_with_reasoning(mock_llm_responses):
|
||||
"""Test agent with reasoning."""
|
||||
llm = LLM("gpt-3.5-turbo")
|
||||
|
||||
agent = Agent(
|
||||
role="Test Agent",
|
||||
goal="To test the reasoning feature",
|
||||
backstory="I am a test agent created to verify the reasoning feature works correctly.",
|
||||
llm=llm,
|
||||
reasoning=True,
|
||||
verbose=True,
|
||||
)
|
||||
|
||||
task = Task(
|
||||
description="Simple math task: What's 2+2?",
|
||||
expected_output="The answer should be a number.",
|
||||
agent=agent,
|
||||
)
|
||||
|
||||
agent.llm.call = lambda messages, *args, **kwargs: (
|
||||
mock_llm_responses["ready"]
|
||||
if any("create a detailed plan" in msg.get("content", "") for msg in messages)
|
||||
else mock_llm_responses["execution"]
|
||||
)
|
||||
|
||||
result = agent.execute_task(task)
|
||||
|
||||
assert result == mock_llm_responses["execution"]
|
||||
assert "Reasoning Plan:" in task.description
|
||||
|
||||
|
||||
def test_agent_with_reasoning_not_ready_initially(mock_llm_responses):
|
||||
"""Test agent with reasoning that requires refinement."""
|
||||
llm = LLM("gpt-3.5-turbo")
|
||||
|
||||
agent = Agent(
|
||||
role="Test Agent",
|
||||
goal="To test the reasoning feature",
|
||||
backstory="I am a test agent created to verify the reasoning feature works correctly.",
|
||||
llm=llm,
|
||||
reasoning=True,
|
||||
max_reasoning_attempts=2,
|
||||
verbose=True,
|
||||
)
|
||||
|
||||
task = Task(
|
||||
description="Complex math task: What's the derivative of x²?",
|
||||
expected_output="The answer should be a mathematical expression.",
|
||||
agent=agent,
|
||||
)
|
||||
|
||||
call_count = [0]
|
||||
|
||||
def mock_llm_call(messages, *args, **kwargs):
|
||||
if any(
|
||||
"create a detailed plan" in msg.get("content", "") for msg in messages
|
||||
) or any("refine your plan" in msg.get("content", "") for msg in messages):
|
||||
call_count[0] += 1
|
||||
if call_count[0] == 1:
|
||||
return mock_llm_responses["not_ready"]
|
||||
return mock_llm_responses["ready_after_refine"]
|
||||
return "2x"
|
||||
|
||||
agent.llm.call = mock_llm_call
|
||||
|
||||
result = agent.execute_task(task)
|
||||
|
||||
assert result == "2x"
|
||||
assert call_count[0] == 2 # Should have made 2 reasoning calls
|
||||
assert "Reasoning Plan:" in task.description
|
||||
|
||||
|
||||
def test_agent_with_reasoning_max_attempts_reached():
|
||||
"""Test agent with reasoning that reaches max attempts without being ready."""
|
||||
llm = LLM("gpt-3.5-turbo")
|
||||
|
||||
agent = Agent(
|
||||
role="Test Agent",
|
||||
goal="To test the reasoning feature",
|
||||
backstory="I am a test agent created to verify the reasoning feature works correctly.",
|
||||
llm=llm,
|
||||
reasoning=True,
|
||||
max_reasoning_attempts=2,
|
||||
verbose=True,
|
||||
)
|
||||
|
||||
task = Task(
|
||||
description="Complex math task: Solve the Riemann hypothesis.",
|
||||
expected_output="A proof or disproof of the hypothesis.",
|
||||
agent=agent,
|
||||
)
|
||||
|
||||
call_count = [0]
|
||||
|
||||
def mock_llm_call(messages, *args, **kwargs):
|
||||
if any(
|
||||
"create a detailed plan" in msg.get("content", "") for msg in messages
|
||||
) or any("refine your plan" in msg.get("content", "") for msg in messages):
|
||||
call_count[0] += 1
|
||||
return f"Attempt {call_count[0]}: I need more time to think.\n\nNOT READY: I need to refine my plan further."
|
||||
return "This is an unsolved problem in mathematics."
|
||||
|
||||
agent.llm.call = mock_llm_call
|
||||
|
||||
result = agent.execute_task(task)
|
||||
|
||||
assert result == "This is an unsolved problem in mathematics."
|
||||
assert (
|
||||
call_count[0] == 2
|
||||
) # Should have made exactly 2 reasoning calls (max_attempts)
|
||||
assert "Reasoning Plan:" in task.description
|
||||
|
||||
|
||||
def test_agent_reasoning_error_handling():
|
||||
"""Test error handling during the reasoning process."""
|
||||
llm = LLM("gpt-3.5-turbo")
|
||||
|
||||
agent = Agent(
|
||||
role="Test Agent",
|
||||
goal="To test the reasoning feature",
|
||||
backstory="I am a test agent created to verify the reasoning feature works correctly.",
|
||||
llm=llm,
|
||||
reasoning=True,
|
||||
)
|
||||
|
||||
task = Task(
|
||||
description="Task that will cause an error",
|
||||
expected_output="Output that will never be generated",
|
||||
agent=agent,
|
||||
)
|
||||
|
||||
call_count = [0]
|
||||
|
||||
def mock_llm_call_error(*args, **kwargs):
|
||||
call_count[0] += 1
|
||||
if call_count[0] <= 2: # First calls are for reasoning
|
||||
raise Exception("LLM error during reasoning")
|
||||
return "Fallback execution result" # Return a value for task execution
|
||||
|
||||
agent.llm.call = mock_llm_call_error
|
||||
|
||||
result = agent.execute_task(task)
|
||||
|
||||
assert result == "Fallback execution result"
|
||||
assert call_count[0] > 2 # Ensure we called the mock multiple times
|
||||
|
||||
|
||||
def test_agent_with_function_calling():
|
||||
"""Test agent with reasoning using function calling."""
|
||||
llm = LLM("gpt-3.5-turbo")
|
||||
|
||||
agent = Agent(
|
||||
role="Test Agent",
|
||||
goal="To test the reasoning feature",
|
||||
backstory="I am a test agent created to verify the reasoning feature works correctly.",
|
||||
llm=llm,
|
||||
reasoning=True,
|
||||
verbose=True,
|
||||
)
|
||||
|
||||
task = Task(
|
||||
description="Simple math task: What's 2+2?",
|
||||
expected_output="The answer should be a number.",
|
||||
agent=agent,
|
||||
)
|
||||
|
||||
agent.llm.supports_function_calling = lambda: True
|
||||
|
||||
def mock_function_call(messages, *args, **kwargs):
|
||||
if "tools" in kwargs:
|
||||
return json.dumps(
|
||||
{"plan": "I'll solve this simple math problem: 2+2=4.", "ready": True}
|
||||
)
|
||||
return "4"
|
||||
|
||||
agent.llm.call = mock_function_call
|
||||
|
||||
result = agent.execute_task(task)
|
||||
|
||||
assert result == "4"
|
||||
assert "Reasoning Plan:" in task.description
|
||||
assert "I'll solve this simple math problem: 2+2=4." in task.description
|
||||
|
||||
|
||||
def test_agent_with_function_calling_fallback():
|
||||
"""Test agent with reasoning using function calling that falls back to text parsing."""
|
||||
llm = LLM("gpt-3.5-turbo")
|
||||
|
||||
agent = Agent(
|
||||
role="Test Agent",
|
||||
goal="To test the reasoning feature",
|
||||
backstory="I am a test agent created to verify the reasoning feature works correctly.",
|
||||
llm=llm,
|
||||
reasoning=True,
|
||||
verbose=True,
|
||||
)
|
||||
|
||||
task = Task(
|
||||
description="Simple math task: What's 2+2?",
|
||||
expected_output="The answer should be a number.",
|
||||
agent=agent,
|
||||
)
|
||||
|
||||
agent.llm.supports_function_calling = lambda: True
|
||||
|
||||
def mock_function_call(messages, *args, **kwargs):
|
||||
if "tools" in kwargs:
|
||||
return "Invalid JSON that will trigger fallback. READY: I am ready to execute the task."
|
||||
return "4"
|
||||
|
||||
agent.llm.call = mock_function_call
|
||||
|
||||
result = agent.execute_task(task)
|
||||
|
||||
assert result == "4"
|
||||
assert "Reasoning Plan:" in task.description
|
||||
assert "Invalid JSON that will trigger fallback" in task.description
|
||||
@@ -1,363 +0,0 @@
|
||||
import pytest
|
||||
|
||||
from crewai.agents import parser
|
||||
from crewai.agents.parser import (
|
||||
AgentAction,
|
||||
AgentFinish,
|
||||
)
|
||||
from crewai.agents.parser import (
|
||||
OutputParserError as OutputParserException,
|
||||
)
|
||||
|
||||
|
||||
def test_valid_action_parsing_special_characters():
|
||||
text = "Thought: Let's find the temperature\nAction: search\nAction Input: what's the temperature in SF?"
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "what's the temperature in SF?"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_json_tool_input():
|
||||
text = """
|
||||
Thought: Let's find the information
|
||||
Action: query
|
||||
Action Input: ** {"task": "What are some common challenges or barriers that you have observed or experienced when implementing AI-powered solutions in healthcare settings?", "context": "As we've discussed recent advancements in AI applications in healthcare, it's crucial to acknowledge the potential hurdles. Some possible obstacles include...", "coworker": "Senior Researcher"}
|
||||
"""
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
expected_tool_input = '{"task": "What are some common challenges or barriers that you have observed or experienced when implementing AI-powered solutions in healthcare settings?", "context": "As we\'ve discussed recent advancements in AI applications in healthcare, it\'s crucial to acknowledge the potential hurdles. Some possible obstacles include...", "coworker": "Senior Researcher"}'
|
||||
assert result.tool == "query"
|
||||
assert result.tool_input == expected_tool_input
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_quotes():
|
||||
text = 'Thought: Let\'s find the temperature\nAction: search\nAction Input: "temperature in SF"'
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "temperature in SF"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_curly_braces():
|
||||
text = "Thought: Let's find the temperature\nAction: search\nAction Input: {temperature in SF}"
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "{temperature in SF}"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_angle_brackets():
|
||||
text = "Thought: Let's find the temperature\nAction: search\nAction Input: <temperature in SF>"
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "<temperature in SF>"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_parentheses():
|
||||
text = "Thought: Let's find the temperature\nAction: search\nAction Input: (temperature in SF)"
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "(temperature in SF)"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_mixed_brackets():
|
||||
text = "Thought: Let's find the temperature\nAction: search\nAction Input: [temperature in {SF}]"
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "[temperature in {SF}]"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_nested_quotes():
|
||||
text = "Thought: Let's find the temperature\nAction: search\nAction Input: \"what's the temperature in 'SF'?\""
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "what's the temperature in 'SF'?"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_incomplete_json():
|
||||
text = 'Thought: Let\'s find the temperature\nAction: search\nAction Input: {"query": "temperature in SF"'
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == '{"query": "temperature in SF"}'
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_special_characters():
|
||||
text = "Thought: Let's find the temperature\nAction: search\nAction Input: what is the temperature in SF? @$%^&*"
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "what is the temperature in SF? @$%^&*"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_combination():
|
||||
text = 'Thought: Let\'s find the temperature\nAction: search\nAction Input: "[what is the temperature in SF?]"'
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "[what is the temperature in SF?]"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_mixed_quotes():
|
||||
text = "Thought: Let's find the temperature\nAction: search\nAction Input: \"what's the temperature in SF?\""
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "what's the temperature in SF?"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_newlines():
|
||||
text = "Thought: Let's find the temperature\nAction: search\nAction Input: what is\nthe temperature in SF?"
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "what is\nthe temperature in SF?"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_escaped_characters():
|
||||
text = "Thought: Let's find the temperature\nAction: search\nAction Input: what is the temperature in SF? \\n"
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "what is the temperature in SF? \\n"
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_json_string():
|
||||
text = 'Thought: Let\'s find the temperature\nAction: search\nAction Input: {"query": "temperature in SF"}'
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == '{"query": "temperature in SF"}'
|
||||
|
||||
|
||||
def test_valid_action_parsing_with_unbalanced_quotes():
|
||||
text = "Thought: Let's find the temperature\nAction: search\nAction Input: \"what is the temperature in SF?"
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "what is the temperature in SF?"
|
||||
|
||||
|
||||
def test_clean_action_no_formatting():
|
||||
action = "Ask question to senior researcher"
|
||||
cleaned_action = parser._clean_action(action)
|
||||
assert cleaned_action == "Ask question to senior researcher"
|
||||
|
||||
|
||||
def test_clean_action_with_leading_asterisks():
|
||||
action = "** Ask question to senior researcher"
|
||||
cleaned_action = parser._clean_action(action)
|
||||
assert cleaned_action == "Ask question to senior researcher"
|
||||
|
||||
|
||||
def test_clean_action_with_trailing_asterisks():
|
||||
action = "Ask question to senior researcher **"
|
||||
cleaned_action = parser._clean_action(action)
|
||||
assert cleaned_action == "Ask question to senior researcher"
|
||||
|
||||
|
||||
def test_clean_action_with_leading_and_trailing_asterisks():
|
||||
action = "** Ask question to senior researcher **"
|
||||
cleaned_action = parser._clean_action(action)
|
||||
assert cleaned_action == "Ask question to senior researcher"
|
||||
|
||||
|
||||
def test_clean_action_with_multiple_leading_asterisks():
|
||||
action = "**** Ask question to senior researcher"
|
||||
cleaned_action = parser._clean_action(action)
|
||||
assert cleaned_action == "Ask question to senior researcher"
|
||||
|
||||
|
||||
def test_clean_action_with_multiple_trailing_asterisks():
|
||||
action = "Ask question to senior researcher ****"
|
||||
cleaned_action = parser._clean_action(action)
|
||||
assert cleaned_action == "Ask question to senior researcher"
|
||||
|
||||
|
||||
def test_clean_action_with_spaces_and_asterisks():
|
||||
action = " ** Ask question to senior researcher ** "
|
||||
cleaned_action = parser._clean_action(action)
|
||||
assert cleaned_action == "Ask question to senior researcher"
|
||||
|
||||
|
||||
def test_clean_action_with_only_asterisks():
|
||||
action = "****"
|
||||
cleaned_action = parser._clean_action(action)
|
||||
assert cleaned_action == ""
|
||||
|
||||
|
||||
def test_clean_action_with_empty_string():
|
||||
action = ""
|
||||
cleaned_action = parser._clean_action(action)
|
||||
assert cleaned_action == ""
|
||||
|
||||
|
||||
def test_valid_final_answer_parsing():
|
||||
text = (
|
||||
"Thought: I found the information\nFinal Answer: The temperature is 100 degrees"
|
||||
)
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentFinish)
|
||||
assert result.output == "The temperature is 100 degrees"
|
||||
|
||||
|
||||
def test_missing_action_error():
|
||||
text = "Thought: Let's find the temperature\nAction Input: what is the temperature in SF?"
|
||||
with pytest.raises(OutputParserException) as exc_info:
|
||||
parser.parse(text)
|
||||
assert "Invalid Format: I missed the 'Action:' after 'Thought:'." in str(
|
||||
exc_info.value
|
||||
)
|
||||
|
||||
|
||||
def test_missing_action_input_error():
|
||||
text = "Thought: Let's find the temperature\nAction: search"
|
||||
with pytest.raises(OutputParserException) as exc_info:
|
||||
parser.parse(text)
|
||||
assert "I missed the 'Action Input:' after 'Action:'." in str(exc_info.value)
|
||||
|
||||
|
||||
def test_safe_repair_json():
|
||||
invalid_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": Senior Researcher'
|
||||
expected_repaired_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_safe_repair_json_unrepairable():
|
||||
invalid_json = "{invalid_json"
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == invalid_json # Should return the original if unrepairable
|
||||
|
||||
|
||||
def test_safe_repair_json_missing_quotes():
|
||||
invalid_json = (
|
||||
'{task: "Research XAI", context: "Explainable AI", coworker: Senior Researcher}'
|
||||
)
|
||||
expected_repaired_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_safe_repair_json_unclosed_brackets():
|
||||
invalid_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"'
|
||||
expected_repaired_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_safe_repair_json_extra_commas():
|
||||
invalid_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher",}'
|
||||
expected_repaired_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_safe_repair_json_trailing_commas():
|
||||
invalid_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher",}'
|
||||
expected_repaired_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_safe_repair_json_single_quotes():
|
||||
invalid_json = "{'task': 'Research XAI', 'context': 'Explainable AI', 'coworker': 'Senior Researcher'}"
|
||||
expected_repaired_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_safe_repair_json_mixed_quotes():
|
||||
invalid_json = "{'task': \"Research XAI\", 'context': \"Explainable AI\", 'coworker': 'Senior Researcher'}"
|
||||
expected_repaired_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_safe_repair_json_unescaped_characters():
|
||||
invalid_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher\n"}'
|
||||
expected_repaired_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_safe_repair_json_missing_colon():
|
||||
invalid_json = '{"task" "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
expected_repaired_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_safe_repair_json_missing_comma():
|
||||
invalid_json = '{"task": "Research XAI" "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
expected_repaired_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_safe_repair_json_unexpected_trailing_characters():
|
||||
invalid_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"} random text'
|
||||
expected_repaired_json = '{"task": "Research XAI", "context": "Explainable AI", "coworker": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_safe_repair_json_special_characters_key():
|
||||
invalid_json = '{"task!@#": "Research XAI", "context$%^": "Explainable AI", "coworker&*()": "Senior Researcher"}'
|
||||
expected_repaired_json = '{"task!@#": "Research XAI", "context$%^": "Explainable AI", "coworker&*()": "Senior Researcher"}'
|
||||
result = parser._safe_repair_json(invalid_json)
|
||||
assert result == expected_repaired_json
|
||||
|
||||
|
||||
def test_parsing_with_whitespace():
|
||||
text = " Thought: Let's find the temperature \n Action: search \n Action Input: what is the temperature in SF? "
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "what is the temperature in SF?"
|
||||
|
||||
|
||||
def test_parsing_with_special_characters():
|
||||
text = 'Thought: Let\'s find the temperature\nAction: search\nAction Input: "what is the temperature in SF?"'
|
||||
result = parser.parse(text)
|
||||
assert isinstance(result, AgentAction)
|
||||
assert result.tool == "search"
|
||||
assert result.tool_input == "what is the temperature in SF?"
|
||||
|
||||
|
||||
def test_integration_valid_and_invalid():
|
||||
text = """
|
||||
Thought: Let's find the temperature
|
||||
Action: search
|
||||
Action Input: what is the temperature in SF?
|
||||
|
||||
Thought: I found the information
|
||||
Final Answer: The temperature is 100 degrees
|
||||
|
||||
Thought: Missing action
|
||||
Action Input: invalid
|
||||
|
||||
Thought: Missing action input
|
||||
Action: invalid
|
||||
"""
|
||||
parts = text.strip().split("\n\n")
|
||||
results = []
|
||||
for part in parts:
|
||||
try:
|
||||
result = parser.parse(part.strip())
|
||||
except OutputParserException as e:
|
||||
result = e
|
||||
results.append(result)
|
||||
|
||||
assert isinstance(results[0], AgentAction)
|
||||
assert isinstance(results[1], AgentFinish)
|
||||
assert isinstance(results[2], OutputParserException)
|
||||
assert isinstance(results[3], OutputParserException)
|
||||
|
||||
|
||||
# TODO: ADD TEST TO MAKE SURE ** REMOVAL DOESN'T MESS UP ANYTHING
|
||||
@@ -1,539 +0,0 @@
|
||||
# mypy: ignore-errors
|
||||
from collections import defaultdict
|
||||
from typing import cast
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from crewai import LLM, Agent
|
||||
from crewai.events.event_bus import crewai_event_bus
|
||||
from crewai.events.types.agent_events import LiteAgentExecutionStartedEvent
|
||||
from crewai.events.types.tool_usage_events import ToolUsageStartedEvent
|
||||
from crewai.flow import Flow, start
|
||||
from crewai.lite_agent import LiteAgent, LiteAgentOutput
|
||||
from crewai.llms.base_llm import BaseLLM
|
||||
from crewai.tools import BaseTool
|
||||
|
||||
|
||||
# A simple test tool
|
||||
class SecretLookupTool(BaseTool):
|
||||
name: str = "secret_lookup"
|
||||
description: str = "A tool to lookup secrets"
|
||||
|
||||
def _run(self) -> str:
|
||||
return "SUPERSECRETPASSWORD123"
|
||||
|
||||
|
||||
# Define Mock Search Tool
|
||||
class WebSearchTool(BaseTool):
|
||||
"""Tool for searching the web for information."""
|
||||
|
||||
name: str = "search_web"
|
||||
description: str = "Search the web for information about a topic."
|
||||
|
||||
def _run(self, query: str) -> str:
|
||||
"""Search the web for information about a topic."""
|
||||
# This is a mock implementation
|
||||
if "tokyo" in query.lower():
|
||||
return "Tokyo's population in 2023 was approximately 21 million people in the city proper, and 37 million in the greater metropolitan area."
|
||||
if "climate change" in query.lower() and "coral" in query.lower():
|
||||
return "Climate change severely impacts coral reefs through: 1) Ocean warming causing coral bleaching, 2) Ocean acidification reducing calcification, 3) Sea level rise affecting light availability, 4) Increased storm frequency damaging reef structures. Sources: NOAA Coral Reef Conservation Program, Global Coral Reef Alliance."
|
||||
return f"Found information about {query}: This is a simulated search result for demonstration purposes."
|
||||
|
||||
|
||||
# Define Mock Calculator Tool
|
||||
class CalculatorTool(BaseTool):
|
||||
"""Tool for performing calculations."""
|
||||
|
||||
name: str = "calculate"
|
||||
description: str = "Calculate the result of a mathematical expression."
|
||||
|
||||
def _run(self, expression: str) -> str:
|
||||
"""Calculate the result of a mathematical expression."""
|
||||
try:
|
||||
# Using eval with restricted builtins for test purposes only
|
||||
result = eval(expression, {"__builtins__": {}}) # noqa: S307
|
||||
return f"The result of {expression} is {result}"
|
||||
except Exception as e:
|
||||
return f"Error calculating {expression}: {e!s}"
|
||||
|
||||
|
||||
# Define a custom response format using Pydantic
|
||||
class ResearchResult(BaseModel):
|
||||
"""Structure for research results."""
|
||||
|
||||
main_findings: str = Field(description="The main findings from the research")
|
||||
key_points: list[str] = Field(description="List of key points")
|
||||
sources: list[str] = Field(description="List of sources used")
|
||||
|
||||
|
||||
@pytest.mark.vcr(filter_headers=["authorization"])
|
||||
@pytest.mark.parametrize("verbose", [True, False])
|
||||
def test_lite_agent_created_with_correct_parameters(monkeypatch, verbose):
|
||||
"""Test that LiteAgent is created with the correct parameters when Agent.kickoff() is called."""
|
||||
# Create a test agent with specific parameters
|
||||
llm = LLM(model="gpt-4o-mini")
|
||||
custom_tools = [WebSearchTool(), CalculatorTool()]
|
||||
max_iter = 10
|
||||
max_execution_time = 300
|
||||
|
||||
agent = Agent(
|
||||
role="Test Agent",
|
||||
goal="Test Goal",
|
||||
backstory="Test Backstory",
|
||||
llm=llm,
|
||||
tools=custom_tools,
|
||||
max_iter=max_iter,
|
||||
max_execution_time=max_execution_time,
|
||||
verbose=verbose,
|
||||
)
|
||||
|
||||
# Create a mock to capture the created LiteAgent
|
||||
created_lite_agent = None
|
||||
original_lite_agent = LiteAgent
|
||||
|
||||
# Define a mock LiteAgent class that captures its arguments
|
||||
class MockLiteAgent(original_lite_agent):
|
||||
def __init__(self, **kwargs):
|
||||
nonlocal created_lite_agent
|
||||
created_lite_agent = kwargs
|
||||
super().__init__(**kwargs)
|
||||
|
||||
# Patch the LiteAgent class
|
||||
monkeypatch.setattr("crewai.agent.LiteAgent", MockLiteAgent)
|
||||
|
||||
# Call kickoff to create the LiteAgent
|
||||
agent.kickoff("Test query")
|
||||
|
||||
# Verify all parameters were passed correctly
|
||||
assert created_lite_agent is not None
|
||||
assert created_lite_agent["role"] == "Test Agent"
|
||||
assert created_lite_agent["goal"] == "Test Goal"
|
||||
assert created_lite_agent["backstory"] == "Test Backstory"
|
||||
assert created_lite_agent["llm"] == llm
|
||||
assert len(created_lite_agent["tools"]) == 2
|
||||
assert isinstance(created_lite_agent["tools"][0], WebSearchTool)
|
||||
assert isinstance(created_lite_agent["tools"][1], CalculatorTool)
|
||||
assert created_lite_agent["max_iterations"] == max_iter
|
||||
assert created_lite_agent["max_execution_time"] == max_execution_time
|
||||
assert created_lite_agent["verbose"] == verbose
|
||||
assert created_lite_agent["response_format"] is None
|
||||
|
||||
# Test with a response_format
|
||||
monkeypatch.setattr("crewai.agent.LiteAgent", MockLiteAgent)
|
||||
|
||||
class TestResponse(BaseModel):
|
||||
test_field: str
|
||||
|
||||
agent.kickoff("Test query", response_format=TestResponse)
|
||||
assert created_lite_agent["response_format"] == TestResponse
|
||||
|
||||
|
||||
@pytest.mark.vcr(filter_headers=["authorization"])
|
||||
def test_lite_agent_with_tools():
|
||||
"""Test that Agent can use tools."""
|
||||
# Create a LiteAgent with tools
|
||||
llm = LLM(model="gpt-4o-mini")
|
||||
agent = Agent(
|
||||
role="Research Assistant",
|
||||
goal="Find information about the population of Tokyo",
|
||||
backstory="You are a helpful research assistant who can search for information about the population of Tokyo.",
|
||||
llm=llm,
|
||||
tools=[WebSearchTool()],
|
||||
verbose=True,
|
||||
)
|
||||
|
||||
result = agent.kickoff(
|
||||
"What is the population of Tokyo and how many people would that be per square kilometer if Tokyo's area is 2,194 square kilometers?"
|
||||
)
|
||||
|
||||
assert "21 million" in result.raw or "37 million" in result.raw, (
|
||||
"Agent should find Tokyo's population"
|
||||
)
|
||||
assert "per square kilometer" in result.raw, (
|
||||
"Agent should calculate population density"
|
||||
)
|
||||
|
||||
received_events = []
|
||||
|
||||
@crewai_event_bus.on(ToolUsageStartedEvent)
|
||||
def event_handler(source, event):
|
||||
received_events.append(event)
|
||||
|
||||
agent.kickoff("What are the effects of climate change on coral reefs?")
|
||||
|
||||
# Verify tool usage events were emitted
|
||||
assert len(received_events) > 0, "Tool usage events should be emitted"
|
||||
event = received_events[0]
|
||||
assert isinstance(event, ToolUsageStartedEvent)
|
||||
assert event.agent_role == "Research Assistant"
|
||||
assert event.tool_name == "search_web"
|
||||
|
||||
|
||||
@pytest.mark.vcr(filter_headers=["authorization"])
|
||||
def test_lite_agent_structured_output():
|
||||
"""Test that Agent can return a simple structured output."""
|
||||
|
||||
class SimpleOutput(BaseModel):
|
||||
"""Simple structure for agent outputs."""
|
||||
|
||||
summary: str = Field(description="A brief summary of findings")
|
||||
confidence: int = Field(description="Confidence level from 1-100")
|
||||
|
||||
web_search_tool = WebSearchTool()
|
||||
|
||||
llm = LLM(model="gpt-4o-mini")
|
||||
agent = Agent(
|
||||
role="Info Gatherer",
|
||||
goal="Provide brief information",
|
||||
backstory="You gather and summarize information quickly.",
|
||||
llm=llm,
|
||||
tools=[web_search_tool],
|
||||
verbose=True,
|
||||
)
|
||||
|
||||
result = agent.kickoff(
|
||||
"What is the population of Tokyo? Return your structured output in JSON format with the following fields: summary, confidence",
|
||||
response_format=SimpleOutput,
|
||||
)
|
||||
|
||||
print(f"\n=== Agent Result Type: {type(result)}")
|
||||
print(f"=== Agent Result: {result}")
|
||||
print(f"=== Pydantic: {result.pydantic}")
|
||||
|
||||
assert result.pydantic is not None, "Should return a Pydantic model"
|
||||
|
||||
output = cast(SimpleOutput, result.pydantic)
|
||||
|
||||
assert isinstance(output.summary, str), "Summary should be a string"
|
||||
assert len(output.summary) > 0, "Summary should not be empty"
|
||||
assert isinstance(output.confidence, int), "Confidence should be an integer"
|
||||
assert 1 <= output.confidence <= 100, "Confidence should be between 1 and 100"
|
||||
|
||||
assert "tokyo" in output.summary.lower() or "population" in output.summary.lower()
|
||||
|
||||
assert result.usage_metrics is not None
|
||||
|
||||
return result
|
||||
|
||||
|
||||
@pytest.mark.vcr(filter_headers=["authorization"])
|
||||
def test_lite_agent_returns_usage_metrics():
|
||||
"""Test that LiteAgent returns usage metrics."""
|
||||
llm = LLM(model="gpt-4o-mini")
|
||||
agent = Agent(
|
||||
role="Research Assistant",
|
||||
goal="Find information about the population of Tokyo",
|
||||
backstory="You are a helpful research assistant who can search for information about the population of Tokyo.",
|
||||
llm=llm,
|
||||
tools=[WebSearchTool()],
|
||||
verbose=True,
|
||||
)
|
||||
|
||||
result = agent.kickoff(
|
||||
"What is the population of Tokyo? Return your structured output in JSON format with the following fields: summary, confidence"
|
||||
)
|
||||
|
||||
assert result.usage_metrics is not None
|
||||
assert result.usage_metrics["total_tokens"] > 0
|
||||
|
||||
|
||||
@pytest.mark.vcr(filter_headers=["authorization"])
|
||||
@pytest.mark.asyncio
|
||||
async def test_lite_agent_returns_usage_metrics_async():
|
||||
"""Test that LiteAgent returns usage metrics when run asynchronously."""
|
||||
llm = LLM(model="gpt-4o-mini")
|
||||
agent = Agent(
|
||||
role="Research Assistant",
|
||||
goal="Find information about the population of Tokyo",
|
||||
backstory="You are a helpful research assistant who can search for information about the population of Tokyo.",
|
||||
llm=llm,
|
||||
tools=[WebSearchTool()],
|
||||
verbose=True,
|
||||
)
|
||||
|
||||
result = await agent.kickoff_async(
|
||||
"What is the population of Tokyo? Return your structured output in JSON format with the following fields: summary, confidence"
|
||||
)
|
||||
assert isinstance(result, LiteAgentOutput)
|
||||
assert "21 million" in result.raw or "37 million" in result.raw
|
||||
assert result.usage_metrics is not None
|
||||
assert result.usage_metrics["total_tokens"] > 0
|
||||
|
||||
|
||||
class TestFlow(Flow):
|
||||
"""A test flow that creates and runs an agent."""
|
||||
|
||||
def __init__(self, llm, tools):
|
||||
self.llm = llm
|
||||
self.tools = tools
|
||||
super().__init__()
|
||||
|
||||
@start()
|
||||
def start(self):
|
||||
agent = Agent(
|
||||
role="Test Agent",
|
||||
goal="Test Goal",
|
||||
backstory="Test Backstory",
|
||||
llm=self.llm,
|
||||
tools=self.tools,
|
||||
)
|
||||
return agent.kickoff("Test query")
|
||||
|
||||
|
||||
def verify_agent_parent_flow(result, agent, flow):
|
||||
"""Verify that both the result and agent have the correct parent flow."""
|
||||
assert result.parent_flow is flow
|
||||
assert agent is not None
|
||||
assert agent.parent_flow is flow
|
||||
|
||||
|
||||
def test_sets_parent_flow_when_inside_flow():
|
||||
captured_agent = None
|
||||
|
||||
mock_llm = Mock(spec=LLM)
|
||||
mock_llm.call.return_value = "Test response"
|
||||
mock_llm.stop = []
|
||||
|
||||
class MyFlow(Flow):
|
||||
@start()
|
||||
def start(self):
|
||||
agent = Agent(
|
||||
role="Test Agent",
|
||||
goal="Test Goal",
|
||||
backstory="Test Backstory",
|
||||
llm=mock_llm,
|
||||
tools=[WebSearchTool()],
|
||||
)
|
||||
return agent.kickoff("Test query")
|
||||
|
||||
flow = MyFlow()
|
||||
with crewai_event_bus.scoped_handlers():
|
||||
|
||||
@crewai_event_bus.on(LiteAgentExecutionStartedEvent)
|
||||
def capture_agent(source, event):
|
||||
nonlocal captured_agent
|
||||
captured_agent = source
|
||||
|
||||
flow.kickoff()
|
||||
assert captured_agent.parent_flow is flow
|
||||
|
||||
|
||||
@pytest.mark.vcr(filter_headers=["authorization"])
|
||||
def test_guardrail_is_called_using_string():
|
||||
guardrail_events = defaultdict(list)
|
||||
from crewai.events.event_types import (
|
||||
LLMGuardrailCompletedEvent,
|
||||
LLMGuardrailStartedEvent,
|
||||
)
|
||||
|
||||
agent = Agent(
|
||||
role="Sports Analyst",
|
||||
goal="Gather information about the best soccer players",
|
||||
backstory="""You are an expert at gathering and organizing information. You carefully collect details and present them in a structured way.""",
|
||||
guardrail="""Only include Brazilian players, both women and men""",
|
||||
)
|
||||
|
||||
with crewai_event_bus.scoped_handlers():
|
||||
|
||||
@crewai_event_bus.on(LLMGuardrailStartedEvent)
|
||||
def capture_guardrail_started(source, event):
|
||||
assert isinstance(source, LiteAgent)
|
||||
assert source.original_agent == agent
|
||||
guardrail_events["started"].append(event)
|
||||
|
||||
@crewai_event_bus.on(LLMGuardrailCompletedEvent)
|
||||
def capture_guardrail_completed(source, event):
|
||||
assert isinstance(source, LiteAgent)
|
||||
assert source.original_agent == agent
|
||||
guardrail_events["completed"].append(event)
|
||||
|
||||
result = agent.kickoff(messages="Top 10 best players in the world?")
|
||||
|
||||
assert len(guardrail_events["started"]) == 2
|
||||
assert len(guardrail_events["completed"]) == 2
|
||||
assert not guardrail_events["completed"][0].success
|
||||
assert guardrail_events["completed"][1].success
|
||||
assert (
|
||||
"Here are the top 10 best soccer players in the world, focusing exclusively on Brazilian players"
|
||||
in result.raw
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.vcr(filter_headers=["authorization"])
|
||||
def test_guardrail_is_called_using_callable():
|
||||
guardrail_events = defaultdict(list)
|
||||
from crewai.events.event_types import (
|
||||
LLMGuardrailCompletedEvent,
|
||||
LLMGuardrailStartedEvent,
|
||||
)
|
||||
|
||||
with crewai_event_bus.scoped_handlers():
|
||||
|
||||
@crewai_event_bus.on(LLMGuardrailStartedEvent)
|
||||
def capture_guardrail_started(source, event):
|
||||
guardrail_events["started"].append(event)
|
||||
|
||||
@crewai_event_bus.on(LLMGuardrailCompletedEvent)
|
||||
def capture_guardrail_completed(source, event):
|
||||
guardrail_events["completed"].append(event)
|
||||
|
||||
agent = Agent(
|
||||
role="Sports Analyst",
|
||||
goal="Gather information about the best soccer players",
|
||||
backstory="""You are an expert at gathering and organizing information. You carefully collect details and present them in a structured way.""",
|
||||
guardrail=lambda output: (True, "Pelé - Santos, 1958"),
|
||||
)
|
||||
|
||||
result = agent.kickoff(messages="Top 1 best players in the world?")
|
||||
|
||||
assert len(guardrail_events["started"]) == 1
|
||||
assert len(guardrail_events["completed"]) == 1
|
||||
assert guardrail_events["completed"][0].success
|
||||
assert "Pelé - Santos, 1958" in result.raw
|
||||
|
||||
|
||||
@pytest.mark.vcr(filter_headers=["authorization"])
|
||||
def test_guardrail_reached_attempt_limit():
|
||||
guardrail_events = defaultdict(list)
|
||||
from crewai.events.event_types import (
|
||||
LLMGuardrailCompletedEvent,
|
||||
LLMGuardrailStartedEvent,
|
||||
)
|
||||
|
||||
with crewai_event_bus.scoped_handlers():
|
||||
|
||||
@crewai_event_bus.on(LLMGuardrailStartedEvent)
|
||||
def capture_guardrail_started(source, event):
|
||||
guardrail_events["started"].append(event)
|
||||
|
||||
@crewai_event_bus.on(LLMGuardrailCompletedEvent)
|
||||
def capture_guardrail_completed(source, event):
|
||||
guardrail_events["completed"].append(event)
|
||||
|
||||
agent = Agent(
|
||||
role="Sports Analyst",
|
||||
goal="Gather information about the best soccer players",
|
||||
backstory="""You are an expert at gathering and organizing information. You carefully collect details and present them in a structured way.""",
|
||||
guardrail=lambda output: (
|
||||
False,
|
||||
"You are not allowed to include Brazilian players",
|
||||
),
|
||||
guardrail_max_retries=2,
|
||||
)
|
||||
|
||||
with pytest.raises(
|
||||
Exception, match="Agent's guardrail failed validation after 2 retries"
|
||||
):
|
||||
agent.kickoff(messages="Top 10 best players in the world?")
|
||||
|
||||
assert len(guardrail_events["started"]) == 3 # 2 retries + 1 initial call
|
||||
assert len(guardrail_events["completed"]) == 3 # 2 retries + 1 initial call
|
||||
assert not guardrail_events["completed"][0].success
|
||||
assert not guardrail_events["completed"][1].success
|
||||
assert not guardrail_events["completed"][2].success
|
||||
|
||||
|
||||
@pytest.mark.vcr(filter_headers=["authorization"])
|
||||
def test_agent_output_when_guardrail_returns_base_model():
|
||||
class Player(BaseModel):
|
||||
name: str
|
||||
country: str
|
||||
|
||||
agent = Agent(
|
||||
role="Sports Analyst",
|
||||
goal="Gather information about the best soccer players",
|
||||
backstory="""You are an expert at gathering and organizing information. You carefully collect details and present them in a structured way.""",
|
||||
guardrail=lambda output: (
|
||||
True,
|
||||
Player(name="Lionel Messi", country="Argentina"),
|
||||
),
|
||||
)
|
||||
|
||||
result = agent.kickoff(messages="Top 10 best players in the world?")
|
||||
|
||||
assert result.pydantic == Player(name="Lionel Messi", country="Argentina")
|
||||
|
||||
|
||||
def test_lite_agent_with_custom_llm_and_guardrails():
|
||||
"""Test that CustomLLM (inheriting from BaseLLM) works with guardrails."""
|
||||
|
||||
class CustomLLM(BaseLLM):
|
||||
def __init__(self, response: str = "Custom response"):
|
||||
super().__init__(model="custom-model")
|
||||
self.response = response
|
||||
self.call_count = 0
|
||||
|
||||
def call(
|
||||
self,
|
||||
messages,
|
||||
tools=None,
|
||||
callbacks=None,
|
||||
available_functions=None,
|
||||
from_task=None,
|
||||
from_agent=None,
|
||||
) -> str:
|
||||
self.call_count += 1
|
||||
|
||||
if "valid" in str(messages) and "feedback" in str(messages):
|
||||
return '{"valid": true, "feedback": null}'
|
||||
|
||||
if "Thought:" in str(messages):
|
||||
return f"Thought: I will analyze soccer players\nFinal Answer: {self.response}"
|
||||
|
||||
return self.response
|
||||
|
||||
def supports_function_calling(self) -> bool:
|
||||
return False
|
||||
|
||||
def supports_stop_words(self) -> bool:
|
||||
return False
|
||||
|
||||
def get_context_window_size(self) -> int:
|
||||
return 4096
|
||||
|
||||
custom_llm = CustomLLM(response="Brazilian soccer players are the best!")
|
||||
|
||||
agent = LiteAgent(
|
||||
role="Sports Analyst",
|
||||
goal="Analyze soccer players",
|
||||
backstory="You analyze soccer players and their performance.",
|
||||
llm=custom_llm,
|
||||
guardrail="Only include Brazilian players",
|
||||
)
|
||||
|
||||
result = agent.kickoff("Tell me about the best soccer players")
|
||||
|
||||
assert custom_llm.call_count > 0
|
||||
assert "Brazilian" in result.raw
|
||||
|
||||
custom_llm2 = CustomLLM(response="Original response")
|
||||
|
||||
def test_guardrail(output):
|
||||
return (True, "Modified by guardrail")
|
||||
|
||||
agent2 = LiteAgent(
|
||||
role="Test Agent",
|
||||
goal="Test goal",
|
||||
backstory="Test backstory",
|
||||
llm=custom_llm2,
|
||||
guardrail=test_guardrail,
|
||||
)
|
||||
|
||||
result2 = agent2.kickoff("Test message")
|
||||
assert result2.raw == "Modified by guardrail"
|
||||
|
||||
|
||||
@pytest.mark.vcr(filter_headers=["authorization"])
|
||||
def test_lite_agent_with_invalid_llm():
|
||||
"""Test that LiteAgent raises proper error when create_llm returns None."""
|
||||
with patch("crewai.lite_agent.create_llm", return_value=None):
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
LiteAgent(
|
||||
role="Test Agent",
|
||||
goal="Test goal",
|
||||
backstory="Test backstory",
|
||||
llm="invalid-model",
|
||||
)
|
||||
assert "Expected LLM instance of type BaseLLM" in str(exc_info.value)
|
||||
Reference in New Issue
Block a user