mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-08-10 08:21:54 +00:00
7fd1997e8e
The `kickoff_async` try block starts before `FlowStartedEvent` is emitted, so an abort in the execution-start hooks, in input handling or in state restore emitted a `flow_failed` with no opener, which pops an unrelated scope and warns about an empty scope stack. The failure event is now gated on the flow scope actually being open, either from this kickoff's `flow_started` or from a restored deferred session scope.
1850 lines
58 KiB
Python
1850 lines
58 KiB
Python
import threading
|
|
from datetime import datetime
|
|
import os
|
|
from unittest.mock import Mock, patch
|
|
|
|
from crewai.agent import Agent
|
|
from crewai.agents.crew_agent_executor import CrewAgentExecutor
|
|
from crewai.crew import Crew
|
|
from crewai.events.event_bus import crewai_event_bus
|
|
from crewai.events.event_listener import EventListener
|
|
from crewai.events.types.agent_events import (
|
|
AgentExecutionCompletedEvent,
|
|
AgentExecutionErrorEvent,
|
|
AgentExecutionStartedEvent,
|
|
)
|
|
from crewai.events.types.crew_events import (
|
|
CrewKickoffCompletedEvent,
|
|
CrewKickoffFailedEvent,
|
|
CrewKickoffStartedEvent,
|
|
CrewTestCompletedEvent,
|
|
CrewTestResultEvent,
|
|
CrewTestStartedEvent,
|
|
)
|
|
from crewai.events.types.flow_events import (
|
|
FlowCreatedEvent,
|
|
FlowFailedEvent,
|
|
FlowFinishedEvent,
|
|
FlowStartedEvent,
|
|
HumanFeedbackReceivedEvent,
|
|
HumanFeedbackRequestedEvent,
|
|
MethodExecutionFailedEvent,
|
|
MethodExecutionFinishedEvent,
|
|
MethodExecutionStartedEvent,
|
|
)
|
|
from crewai.events.types.llm_events import (
|
|
LLMCallCompletedEvent,
|
|
LLMCallFailedEvent,
|
|
LLMCallStartedEvent,
|
|
LLMStreamChunkEvent,
|
|
)
|
|
from crewai.events.types.task_events import (
|
|
TaskCompletedEvent,
|
|
TaskFailedEvent,
|
|
TaskStartedEvent,
|
|
)
|
|
from crewai.events.types.tool_usage_events import (
|
|
ToolUsageErrorEvent,
|
|
ToolUsageFinishedEvent,
|
|
)
|
|
from crewai.flow.async_feedback.types import PendingFeedbackContext
|
|
from crewai.flow.flow import Flow, listen, start
|
|
from crewai.flow.human_feedback import human_feedback
|
|
from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
|
|
from crewai.hooks.dispatch import HookAborted, InterceptionPoint, clear_all, on
|
|
from crewai.llm import LLM
|
|
from crewai.task import Task
|
|
from crewai.tools.base_tool import BaseTool
|
|
from pydantic import BaseModel, Field
|
|
import pytest
|
|
|
|
from ..utils import wait_for_event_handlers
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def base_agent():
|
|
return Agent(
|
|
role="base_agent",
|
|
llm="gpt-4o-mini",
|
|
goal="Just say hi",
|
|
backstory="You are a helpful assistant that just says hi",
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def base_task(base_agent):
|
|
return Task(
|
|
description="Just say hi",
|
|
expected_output="hi",
|
|
agent=base_agent,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def reset_event_listener_singleton():
|
|
"""Reset EventListener singleton for clean test state."""
|
|
original_instance = EventListener._instance
|
|
original_initialized = (
|
|
getattr(EventListener._instance, "_initialized", False)
|
|
if EventListener._instance
|
|
else False
|
|
)
|
|
|
|
EventListener._instance = None
|
|
|
|
yield
|
|
|
|
EventListener._instance = original_instance
|
|
if original_instance and original_initialized:
|
|
EventListener._instance._initialized = original_initialized
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_crew_emits_start_kickoff_event(
|
|
base_agent, base_task, reset_event_listener_singleton
|
|
):
|
|
received_events = []
|
|
mock_span = Mock()
|
|
|
|
@crewai_event_bus.on(CrewKickoffStartedEvent)
|
|
def handle_crew_start(source, event):
|
|
received_events.append(event)
|
|
|
|
mock_telemetry = Mock()
|
|
mock_telemetry.crew_execution_span = Mock(return_value=mock_span)
|
|
mock_telemetry.end_crew = Mock(return_value=mock_span)
|
|
mock_telemetry.set_tracer = Mock()
|
|
mock_telemetry.task_started = Mock(return_value=mock_span)
|
|
mock_telemetry.task_ended = Mock(return_value=mock_span)
|
|
|
|
with patch("crewai.events.event_listener.Telemetry", return_value=mock_telemetry):
|
|
crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
|
|
crew.kickoff()
|
|
wait_for_event_handlers()
|
|
|
|
mock_telemetry.crew_execution_span.assert_called_once_with(crew, None)
|
|
mock_telemetry.end_crew.assert_called_once_with(crew, "hi")
|
|
|
|
assert len(received_events) == 1
|
|
assert received_events[0].crew_name == "TestCrew"
|
|
assert isinstance(received_events[0].timestamp, datetime)
|
|
assert received_events[0].type == "crew_kickoff_started"
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_crew_emits_end_kickoff_event(base_agent, base_task):
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(CrewKickoffCompletedEvent)
|
|
def handle_crew_end(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
|
|
|
|
crew.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), (
|
|
"Timeout waiting for crew kickoff completed event"
|
|
)
|
|
assert len(received_events) == 1
|
|
assert received_events[0].crew_name == "TestCrew"
|
|
assert isinstance(received_events[0].timestamp, datetime)
|
|
assert received_events[0].type == "crew_kickoff_completed"
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_crew_emits_test_kickoff_type_event(base_agent, base_task):
|
|
received_events = []
|
|
|
|
@crewai_event_bus.on(CrewTestStartedEvent)
|
|
def handle_crew_end(source, event):
|
|
received_events.append(event)
|
|
|
|
@crewai_event_bus.on(CrewTestCompletedEvent)
|
|
def handle_crew_test_end(source, event):
|
|
received_events.append(event)
|
|
|
|
@crewai_event_bus.on(CrewTestResultEvent)
|
|
def handle_crew_test_result(source, event):
|
|
received_events.append(event)
|
|
|
|
eval_llm = LLM(model="gpt-4o-mini")
|
|
crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
|
|
crew.test(n_iterations=1, eval_llm=eval_llm)
|
|
wait_for_event_handlers()
|
|
|
|
assert len(received_events) == 3
|
|
assert received_events[0].crew_name == "TestCrew"
|
|
assert isinstance(received_events[0].timestamp, datetime)
|
|
assert received_events[0].type == "crew_test_started"
|
|
assert received_events[1].crew_name == "TestCrew"
|
|
assert isinstance(received_events[1].timestamp, datetime)
|
|
assert received_events[1].type == "crew_test_result"
|
|
assert received_events[2].crew_name == "TestCrew"
|
|
assert isinstance(received_events[2].timestamp, datetime)
|
|
assert received_events[2].type == "crew_test_completed"
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_crew_emits_kickoff_failed_event(base_agent, base_task):
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(CrewKickoffFailedEvent)
|
|
def handle_crew_failed(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
|
|
|
|
with patch.object(Crew, "_execute_tasks") as mock_execute:
|
|
error_message = "Simulated crew kickoff failure"
|
|
mock_execute.side_effect = Exception(error_message)
|
|
|
|
with pytest.raises(Exception): # noqa: B017
|
|
crew.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for failed event"
|
|
assert len(received_events) == 1
|
|
assert received_events[0].error == error_message
|
|
assert isinstance(received_events[0].timestamp, datetime)
|
|
assert received_events[0].type == "crew_kickoff_failed"
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_crew_emits_start_task_event(base_agent, base_task):
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(TaskStartedEvent)
|
|
def handle_task_start(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
|
|
|
|
crew.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for task started event"
|
|
assert len(received_events) == 1
|
|
assert isinstance(received_events[0].timestamp, datetime)
|
|
assert received_events[0].type == "task_started"
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_crew_emits_end_task_event(base_agent, base_task):
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(TaskCompletedEvent)
|
|
def handle_task_end(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
|
|
crew.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for task completed event"
|
|
assert len(received_events) == 1
|
|
assert isinstance(received_events[0].timestamp, datetime)
|
|
assert received_events[0].type == "task_completed"
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_task_emits_failed_event_on_execution_error(base_agent, base_task):
|
|
received_events = []
|
|
received_sources = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(TaskFailedEvent)
|
|
def handle_task_failed(source, event):
|
|
received_events.append(event)
|
|
received_sources.append(source)
|
|
event_received.set()
|
|
|
|
with patch.object(
|
|
Task,
|
|
"_execute_core",
|
|
) as mock_execute:
|
|
error_message = "Simulated task failure"
|
|
mock_execute.side_effect = Exception(error_message)
|
|
agent = Agent(
|
|
role="base_agent",
|
|
goal="Just say hi",
|
|
backstory="You are a helpful assistant that just says hi",
|
|
)
|
|
task = Task(
|
|
description="Just say hi",
|
|
expected_output="hi",
|
|
agent=agent,
|
|
)
|
|
|
|
with pytest.raises(Exception): # noqa: B017
|
|
agent.execute_task(task=task)
|
|
|
|
assert event_received.wait(timeout=5), (
|
|
"Timeout waiting for task failed event"
|
|
)
|
|
assert len(received_events) == 1
|
|
assert received_sources[0] == task
|
|
assert received_events[0].error == error_message
|
|
assert isinstance(received_events[0].timestamp, datetime)
|
|
assert received_events[0].type == "task_failed"
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_agent_emits_execution_started_and_completed_events(base_agent, base_task):
|
|
started_events: list[AgentExecutionStartedEvent] = []
|
|
completed_events: list[AgentExecutionCompletedEvent] = []
|
|
condition = threading.Condition()
|
|
|
|
@crewai_event_bus.on(AgentExecutionStartedEvent)
|
|
def handle_agent_start(source, event):
|
|
with condition:
|
|
started_events.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(AgentExecutionCompletedEvent)
|
|
def handle_agent_completed(source, event):
|
|
with condition:
|
|
completed_events.append(event)
|
|
condition.notify()
|
|
|
|
crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
|
|
crew.kickoff()
|
|
|
|
with condition:
|
|
success = condition.wait_for(
|
|
lambda: len(started_events) >= 1 and len(completed_events) >= 1,
|
|
timeout=10,
|
|
)
|
|
assert success, "Timeout waiting for agent execution events"
|
|
|
|
assert len(started_events) == 1
|
|
assert len(completed_events) == 1
|
|
assert started_events[0].agent == base_agent
|
|
assert started_events[0].task == base_task
|
|
assert started_events[0].tools == []
|
|
assert isinstance(started_events[0].task_prompt, str)
|
|
assert (
|
|
started_events[0].task_prompt
|
|
== "Just say hi\n\nThis is the expected criteria for your final answer: hi\nyou MUST return the actual complete content as the final answer, not a summary."
|
|
)
|
|
assert isinstance(started_events[0].timestamp, datetime)
|
|
assert started_events[0].type == "agent_execution_started"
|
|
assert isinstance(completed_events[0].timestamp, datetime)
|
|
assert completed_events[0].type == "agent_execution_completed"
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_agent_emits_execution_error_event(base_agent, base_task):
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(AgentExecutionErrorEvent)
|
|
def handle_agent_start(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
from crewai.experimental.agent_executor import AgentExecutor
|
|
|
|
error_message = "Error happening while sending prompt to model."
|
|
base_agent.max_retry_limit = 0
|
|
|
|
# Patch at the class level since agent_executor is created lazily
|
|
with patch.object(
|
|
AgentExecutor, "invoke", side_effect=Exception(error_message)
|
|
):
|
|
with pytest.raises(Exception): # noqa: B017
|
|
base_agent.execute_task(
|
|
task=base_task,
|
|
)
|
|
|
|
assert event_received.wait(timeout=5), (
|
|
"Timeout waiting for agent execution error event"
|
|
)
|
|
assert len(received_events) == 1
|
|
assert received_events[0].agent == base_agent
|
|
assert received_events[0].task == base_task
|
|
assert received_events[0].error == error_message
|
|
assert isinstance(received_events[0].timestamp, datetime)
|
|
assert received_events[0].type == "agent_execution_error"
|
|
|
|
|
|
class SayHiTool(BaseTool):
|
|
name: str = Field(default="say_hi", description="The name of the tool")
|
|
description: str = Field(
|
|
default="Say hi", description="The description of the tool"
|
|
)
|
|
|
|
def _run(self) -> str:
|
|
return "hi"
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_tools_emits_finished_events():
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(ToolUsageFinishedEvent)
|
|
def handle_tool_end(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
agent = Agent(
|
|
role="base_agent",
|
|
goal="Just say hi",
|
|
backstory="You are a helpful assistant that just says hi",
|
|
tools=[SayHiTool()],
|
|
)
|
|
|
|
task = Task(
|
|
description="Just say hi",
|
|
expected_output="hi",
|
|
agent=agent,
|
|
)
|
|
crew = Crew(agents=[agent], tasks=[task], name="TestCrew")
|
|
crew.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), (
|
|
"Timeout waiting for tool usage finished event"
|
|
)
|
|
assert len(received_events) == 1
|
|
assert received_events[0].agent_key == agent.key
|
|
assert received_events[0].agent_role == agent.role
|
|
assert received_events[0].tool_name == SayHiTool().name
|
|
assert received_events[0].tool_args == "{}" or received_events[0].tool_args == {}
|
|
assert received_events[0].type == "tool_usage_finished"
|
|
assert isinstance(received_events[0].timestamp, datetime)
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_tools_emits_error_events():
|
|
received_events = []
|
|
lock = threading.Lock()
|
|
all_events_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(ToolUsageErrorEvent)
|
|
def handle_tool_end(source, event):
|
|
with lock:
|
|
received_events.append(event)
|
|
# Set event when we receive at least 1 error event
|
|
if len(received_events) >= 1:
|
|
all_events_received.set()
|
|
|
|
class ErrorTool(BaseTool):
|
|
name: str = Field(
|
|
default="error_tool", description="A tool that raises an error"
|
|
)
|
|
description: str = Field(
|
|
default="This tool always raises an error",
|
|
description="The description of the tool",
|
|
)
|
|
|
|
def _run(self) -> str:
|
|
raise Exception("Simulated tool error")
|
|
|
|
agent = Agent(
|
|
role="base_agent",
|
|
goal="Try to use the error tool",
|
|
backstory="You are an assistant that tests error handling",
|
|
tools=[ErrorTool()],
|
|
llm=LLM(model="gpt-4o-mini"),
|
|
)
|
|
|
|
task = Task(
|
|
description="Use the error tool",
|
|
expected_output="This should error",
|
|
agent=agent,
|
|
)
|
|
|
|
crew = Crew(agents=[agent], tasks=[task], name="TestCrew")
|
|
crew.kickoff()
|
|
|
|
assert all_events_received.wait(timeout=10), (
|
|
"Timeout waiting for tool usage error events"
|
|
)
|
|
# At least one error event should be received (number varies by execution path)
|
|
assert len(received_events) >= 1
|
|
assert received_events[0].agent_key == agent.key
|
|
assert received_events[0].agent_role == agent.role
|
|
assert received_events[0].tool_name == "error_tool"
|
|
assert received_events[0].tool_args == "{}" or received_events[0].tool_args == {}
|
|
assert str(received_events[0].error) == "Simulated tool error"
|
|
assert received_events[0].type == "tool_usage_error"
|
|
assert isinstance(received_events[0].timestamp, datetime)
|
|
|
|
|
|
def test_flow_emits_start_event(reset_event_listener_singleton):
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
mock_span = Mock()
|
|
|
|
@crewai_event_bus.on(FlowStartedEvent)
|
|
def handle_flow_start(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
class TestFlow(Flow[dict]):
|
|
@start()
|
|
def begin(self):
|
|
return "started"
|
|
|
|
mock_telemetry = Mock()
|
|
mock_telemetry.flow_execution_span = Mock(return_value=mock_span)
|
|
mock_telemetry.flow_creation_span = Mock()
|
|
mock_telemetry.set_tracer = Mock()
|
|
|
|
with patch("crewai.events.event_listener.Telemetry", return_value=mock_telemetry):
|
|
# Force creation of EventListener singleton with mocked telemetry
|
|
_ = EventListener()
|
|
|
|
flow = TestFlow()
|
|
flow.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for flow started event"
|
|
mock_telemetry.flow_execution_span.assert_called_once_with("TestFlow", ["begin"])
|
|
assert len(received_events) == 1
|
|
assert received_events[0].flow_name == "TestFlow"
|
|
assert received_events[0].type == "flow_started"
|
|
|
|
|
|
def test_flow_name_emitted_to_event_bus():
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
class MyFlowClass(Flow):
|
|
name = "PRODUCTION_FLOW"
|
|
|
|
@start()
|
|
def start(self):
|
|
return "Hello, world!"
|
|
|
|
@crewai_event_bus.on(FlowStartedEvent)
|
|
def handle_flow_start(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
flow = MyFlowClass()
|
|
flow.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for flow started event"
|
|
assert len(received_events) == 1
|
|
assert received_events[0].flow_name == "PRODUCTION_FLOW"
|
|
|
|
|
|
def test_flow_emits_finish_event():
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(FlowFinishedEvent)
|
|
def handle_flow_finish(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
class TestFlow(Flow[dict]):
|
|
@start()
|
|
def begin(self):
|
|
return "completed"
|
|
|
|
flow = TestFlow()
|
|
result = flow.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for finish event"
|
|
assert len(received_events) == 1
|
|
assert received_events[0].flow_name == "TestFlow"
|
|
assert received_events[0].type == "flow_finished"
|
|
assert received_events[0].result == "completed"
|
|
assert result == "completed"
|
|
|
|
|
|
def test_flow_emits_failed_event_paired_with_started_event():
|
|
started: list[FlowStartedEvent] = []
|
|
failed: list[FlowFailedEvent] = []
|
|
|
|
class BoomFlow(Flow[dict]):
|
|
@start()
|
|
def begin(self):
|
|
raise RuntimeError("boom")
|
|
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowStartedEvent)
|
|
def handle_flow_started(source, event):
|
|
started.append(event)
|
|
|
|
@crewai_event_bus.on(FlowFailedEvent)
|
|
def handle_flow_failed(source, event):
|
|
failed.append(event)
|
|
|
|
with pytest.raises(RuntimeError, match="boom"):
|
|
BoomFlow().kickoff()
|
|
wait_for_event_handlers()
|
|
|
|
assert len(failed) == 1
|
|
assert failed[0].type == "flow_failed"
|
|
assert failed[0].flow_name == "BoomFlow"
|
|
assert isinstance(failed[0].error, RuntimeError)
|
|
assert str(failed[0].error) == "boom"
|
|
assert failed[0].started_event_id == started[0].event_id
|
|
|
|
|
|
def test_suppressed_flow_failure_matches_finished_event_emission():
|
|
finished: list[FlowFinishedEvent] = []
|
|
failed: list[FlowFailedEvent] = []
|
|
|
|
class SuppressedFlow(Flow):
|
|
suppress_flow_events: bool = True
|
|
|
|
@start()
|
|
def begin(self):
|
|
return "ok"
|
|
|
|
class SuppressedBoomFlow(Flow):
|
|
suppress_flow_events: bool = True
|
|
|
|
@start()
|
|
def begin(self):
|
|
raise RuntimeError("boom")
|
|
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowFinishedEvent)
|
|
def handle_flow_finished(source, event):
|
|
finished.append(event)
|
|
|
|
@crewai_event_bus.on(FlowFailedEvent)
|
|
def handle_flow_failed(source, event):
|
|
failed.append(event)
|
|
|
|
SuppressedFlow().kickoff()
|
|
with pytest.raises(RuntimeError, match="boom"):
|
|
SuppressedBoomFlow().kickoff()
|
|
wait_for_event_handlers()
|
|
|
|
assert len(finished) == 1
|
|
assert len(failed) == 1
|
|
|
|
|
|
def test_abort_before_flow_started_emits_no_failed_event():
|
|
started: list[FlowStartedEvent] = []
|
|
failed: list[FlowFailedEvent] = []
|
|
|
|
class BlockedFlow(Flow):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "never runs"
|
|
|
|
clear_all()
|
|
try:
|
|
|
|
@on(InterceptionPoint.EXECUTION_START)
|
|
def block(_ctx):
|
|
raise HookAborted(reason="blocked by policy")
|
|
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowStartedEvent)
|
|
def handle_flow_started(source, event):
|
|
started.append(event)
|
|
|
|
@crewai_event_bus.on(FlowFailedEvent)
|
|
def handle_flow_failed(source, event):
|
|
failed.append(event)
|
|
|
|
with pytest.raises(HookAborted):
|
|
BlockedFlow().kickoff()
|
|
wait_for_event_handlers()
|
|
finally:
|
|
clear_all()
|
|
|
|
assert started == []
|
|
assert failed == []
|
|
|
|
|
|
def test_resume_emits_failed_event_paired_with_resume_started_event(tmp_path):
|
|
started: list[FlowStartedEvent] = []
|
|
failed: list[FlowFailedEvent] = []
|
|
|
|
class ResumeBoomFlow(Flow):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "content"
|
|
|
|
@listen(begin)
|
|
def after_feedback(self, _feedback):
|
|
raise RuntimeError("boom on resume")
|
|
|
|
persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db"))
|
|
flow_id = "resume-failure-test"
|
|
persistence.save_pending_feedback(
|
|
flow_uuid=flow_id,
|
|
context=PendingFeedbackContext(
|
|
flow_id=flow_id,
|
|
flow_class="ResumeBoomFlow",
|
|
method_name="begin",
|
|
method_output="content",
|
|
message="Review:",
|
|
),
|
|
state_data={"id": flow_id},
|
|
)
|
|
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowStartedEvent)
|
|
def handle_flow_started(source, event):
|
|
started.append(event)
|
|
|
|
@crewai_event_bus.on(FlowFailedEvent)
|
|
def handle_flow_failed(source, event):
|
|
failed.append(event)
|
|
|
|
flow = ResumeBoomFlow.from_pending(flow_id, persistence)
|
|
with pytest.raises(RuntimeError, match="boom on resume"):
|
|
flow.resume("ok")
|
|
wait_for_event_handlers()
|
|
|
|
assert len(started) == 1
|
|
assert len(failed) == 1
|
|
assert failed[0].flow_name == "ResumeBoomFlow"
|
|
assert str(failed[0].error) == "boom on resume"
|
|
assert failed[0].started_event_id == started[0].event_id
|
|
|
|
|
|
def test_resume_pairs_resumed_method_events_with_their_own_scope(tmp_path):
|
|
started: list[FlowStartedEvent] = []
|
|
finished: list[FlowFinishedEvent] = []
|
|
method_started: list[MethodExecutionStartedEvent] = []
|
|
method_finished: list[MethodExecutionFinishedEvent] = []
|
|
|
|
class ResumeFlow(Flow):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "content"
|
|
|
|
@listen(begin)
|
|
def after_feedback(self, _feedback):
|
|
return "done"
|
|
|
|
persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db"))
|
|
flow_id = "resume-pairing-test"
|
|
persistence.save_pending_feedback(
|
|
flow_uuid=flow_id,
|
|
context=PendingFeedbackContext(
|
|
flow_id=flow_id,
|
|
flow_class="ResumeFlow",
|
|
method_name="begin",
|
|
method_output="content",
|
|
message="Review:",
|
|
),
|
|
state_data={"id": flow_id},
|
|
)
|
|
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowStartedEvent)
|
|
def handle_flow_started(source, event):
|
|
started.append(event)
|
|
|
|
@crewai_event_bus.on(FlowFinishedEvent)
|
|
def handle_flow_finished(source, event):
|
|
finished.append(event)
|
|
|
|
@crewai_event_bus.on(MethodExecutionStartedEvent)
|
|
def handle_method_started(source, event):
|
|
method_started.append(event)
|
|
|
|
@crewai_event_bus.on(MethodExecutionFinishedEvent)
|
|
def handle_method_finished(source, event):
|
|
method_finished.append(event)
|
|
|
|
ResumeFlow.from_pending(flow_id, persistence).resume("ok")
|
|
wait_for_event_handlers()
|
|
|
|
resumed_started = next(e for e in method_started if e.method_name == "begin")
|
|
resumed_finished = next(e for e in method_finished if e.method_name == "begin")
|
|
|
|
assert resumed_finished.started_event_id == resumed_started.event_id
|
|
assert finished[0].started_event_id == started[0].event_id
|
|
|
|
|
|
def test_resume_failing_before_method_finishes_keeps_flow_pairing(tmp_path):
|
|
started: list[FlowStartedEvent] = []
|
|
failed: list[FlowFailedEvent] = []
|
|
method_failed: list[MethodExecutionFailedEvent] = []
|
|
|
|
class ResumeFlow(Flow):
|
|
@start()
|
|
def begin(self) -> str:
|
|
return "content"
|
|
|
|
@listen(begin)
|
|
def after_feedback(self, _feedback):
|
|
return "done"
|
|
|
|
persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db"))
|
|
flow_id = "resume-finalize-failure-test"
|
|
persistence.save_pending_feedback(
|
|
flow_uuid=flow_id,
|
|
context=PendingFeedbackContext(
|
|
flow_id=flow_id,
|
|
flow_class="ResumeFlow",
|
|
method_name="begin",
|
|
method_output="content",
|
|
message="Review:",
|
|
),
|
|
state_data={"id": flow_id},
|
|
)
|
|
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(FlowStartedEvent)
|
|
def handle_flow_started(source, event):
|
|
started.append(event)
|
|
|
|
@crewai_event_bus.on(FlowFailedEvent)
|
|
def handle_flow_failed(source, event):
|
|
failed.append(event)
|
|
|
|
@crewai_event_bus.on(MethodExecutionFailedEvent)
|
|
def handle_method_failed(source, event):
|
|
method_failed.append(event)
|
|
|
|
flow = ResumeFlow.from_pending(flow_id, persistence)
|
|
with patch.object(
|
|
Flow,
|
|
"_finalize_human_feedback",
|
|
side_effect=RuntimeError("feedback collapse failed"),
|
|
):
|
|
with pytest.raises(RuntimeError, match="feedback collapse failed"):
|
|
flow.resume("ok")
|
|
wait_for_event_handlers()
|
|
|
|
assert len(method_failed) == 1
|
|
assert method_failed[0].method_name == "begin"
|
|
assert len(failed) == 1
|
|
assert failed[0].started_event_id == started[0].event_id
|
|
|
|
|
|
def test_flow_emits_method_execution_started_event():
|
|
received_events = []
|
|
lock = threading.Lock()
|
|
second_event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(MethodExecutionStartedEvent)
|
|
async def handle_method_start(source, event):
|
|
with lock:
|
|
received_events.append(event)
|
|
if event.method_name == "second_method":
|
|
second_event_received.set()
|
|
|
|
class TestFlow(Flow[dict]):
|
|
@start()
|
|
def begin(self):
|
|
return "started"
|
|
|
|
@listen("begin")
|
|
def second_method(self):
|
|
return "executed"
|
|
|
|
flow = TestFlow()
|
|
flow.kickoff()
|
|
|
|
assert second_event_received.wait(timeout=5), (
|
|
"Timeout waiting for second_method event"
|
|
)
|
|
assert len(received_events) == 2
|
|
|
|
# Events may arrive in any order due to async handlers, so check both are present
|
|
method_names = {event.method_name for event in received_events}
|
|
assert method_names == {"begin", "second_method"}
|
|
|
|
for event in received_events:
|
|
assert event.flow_name == "TestFlow"
|
|
assert event.type == "method_execution_started"
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_register_handler_adds_new_handler(base_agent, base_task):
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
def custom_handler(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
crewai_event_bus.register_handler(CrewKickoffStartedEvent, custom_handler)
|
|
|
|
crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
|
|
crew.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for handler event"
|
|
assert len(received_events) == 1
|
|
assert isinstance(received_events[0].timestamp, datetime)
|
|
assert received_events[0].type == "crew_kickoff_started"
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_multiple_handlers_for_same_event(base_agent, base_task):
|
|
received_events_1 = []
|
|
received_events_2 = []
|
|
event_received = threading.Event()
|
|
|
|
def handler_1(source, event):
|
|
received_events_1.append(event)
|
|
|
|
def handler_2(source, event):
|
|
received_events_2.append(event)
|
|
event_received.set()
|
|
|
|
crewai_event_bus.register_handler(CrewKickoffStartedEvent, handler_1)
|
|
crewai_event_bus.register_handler(CrewKickoffStartedEvent, handler_2)
|
|
|
|
crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
|
|
crew.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for handler events"
|
|
assert len(received_events_1) == 1
|
|
assert len(received_events_2) == 1
|
|
assert received_events_1[0].type == "crew_kickoff_started"
|
|
assert received_events_2[0].type == "crew_kickoff_started"
|
|
|
|
|
|
def test_flow_emits_created_event():
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(FlowCreatedEvent)
|
|
def handle_flow_created(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
class TestFlow(Flow[dict]):
|
|
@start()
|
|
def begin(self):
|
|
return "started"
|
|
|
|
flow = TestFlow()
|
|
flow.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for flow created event"
|
|
assert len(received_events) == 1
|
|
assert received_events[0].flow_name == "TestFlow"
|
|
assert received_events[0].type == "flow_created"
|
|
|
|
|
|
def test_flow_emits_method_execution_failed_event():
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
error = Exception("Simulated method failure")
|
|
|
|
@crewai_event_bus.on(MethodExecutionFailedEvent)
|
|
def handle_method_failed(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
class TestFlow(Flow[dict]):
|
|
@start()
|
|
def begin(self):
|
|
raise error
|
|
|
|
flow = TestFlow()
|
|
with pytest.raises(Exception): # noqa: B017
|
|
flow.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), (
|
|
"Timeout waiting for method execution failed event"
|
|
)
|
|
assert len(received_events) == 1
|
|
assert received_events[0].method_name == "begin"
|
|
assert received_events[0].flow_name == "TestFlow"
|
|
assert received_events[0].type == "method_execution_failed"
|
|
assert received_events[0].error == error
|
|
|
|
|
|
def test_flow_method_execution_started_includes_unstructured_state():
|
|
"""Test that MethodExecutionStartedEvent includes unstructured (dict) state."""
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(MethodExecutionStartedEvent)
|
|
def handle_method_started(source, event):
|
|
received_events.append(event)
|
|
if event.method_name == "process":
|
|
event_received.set()
|
|
|
|
class TestFlow(Flow[dict]):
|
|
@start()
|
|
def begin(self):
|
|
self.state["counter"] = 1
|
|
self.state["message"] = "test"
|
|
return "started"
|
|
|
|
@listen("begin")
|
|
def process(self):
|
|
self.state["counter"] = 2
|
|
return "processed"
|
|
|
|
flow = TestFlow()
|
|
flow.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), (
|
|
"Timeout waiting for method execution started event"
|
|
)
|
|
|
|
begin_event = next(e for e in received_events if e.method_name == "begin")
|
|
process_event = next(e for e in received_events if e.method_name == "process")
|
|
|
|
assert begin_event.state is not None
|
|
assert isinstance(begin_event.state, dict)
|
|
assert "id" in begin_event.state
|
|
|
|
assert process_event.state is not None
|
|
assert isinstance(process_event.state, dict)
|
|
assert process_event.state["counter"] == 1
|
|
assert process_event.state["message"] == "test"
|
|
|
|
|
|
def test_flow_method_execution_started_includes_structured_state():
|
|
"""Test that MethodExecutionStartedEvent includes structured (BaseModel) state and serializes it properly."""
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
class FlowState(BaseModel):
|
|
counter: int = 0
|
|
message: str = ""
|
|
items: list[str] = []
|
|
|
|
@crewai_event_bus.on(MethodExecutionStartedEvent)
|
|
def handle_method_started(source, event):
|
|
received_events.append(event)
|
|
if event.method_name == "process":
|
|
event_received.set()
|
|
|
|
class TestFlow(Flow[FlowState]):
|
|
@start()
|
|
def begin(self):
|
|
self.state.counter = 1
|
|
self.state.message = "initial"
|
|
self.state.items = ["a", "b"]
|
|
return "started"
|
|
|
|
@listen("begin")
|
|
def process(self):
|
|
self.state.counter += 1
|
|
return "processed"
|
|
|
|
flow = TestFlow()
|
|
flow.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), (
|
|
"Timeout waiting for method execution started event"
|
|
)
|
|
|
|
begin_event = next(e for e in received_events if e.method_name == "begin")
|
|
process_event = next(e for e in received_events if e.method_name == "process")
|
|
|
|
assert begin_event.state is not None
|
|
assert isinstance(begin_event.state, dict)
|
|
assert begin_event.state["counter"] == 0
|
|
assert begin_event.state["message"] == ""
|
|
assert begin_event.state["items"] == []
|
|
|
|
assert process_event.state is not None
|
|
assert isinstance(process_event.state, dict)
|
|
assert process_event.state["counter"] == 1
|
|
assert process_event.state["message"] == "initial"
|
|
assert process_event.state["items"] == ["a", "b"]
|
|
|
|
|
|
def test_flow_method_execution_finished_includes_serialized_state():
|
|
"""Test that MethodExecutionFinishedEvent includes properly serialized state."""
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
class FlowState(BaseModel):
|
|
result: str = ""
|
|
completed: bool = False
|
|
|
|
@crewai_event_bus.on(MethodExecutionFinishedEvent)
|
|
def handle_method_finished(source, event):
|
|
received_events.append(event)
|
|
if event.method_name == "process":
|
|
event_received.set()
|
|
|
|
class TestFlow(Flow[FlowState]):
|
|
@start()
|
|
def begin(self):
|
|
self.state.result = "begin done"
|
|
return "started"
|
|
|
|
@listen("begin")
|
|
def process(self):
|
|
self.state.result = "process done"
|
|
self.state.completed = True
|
|
return "final_result"
|
|
|
|
flow = TestFlow()
|
|
final_output = flow.kickoff()
|
|
|
|
assert event_received.wait(timeout=5), (
|
|
"Timeout waiting for method execution finished event"
|
|
)
|
|
|
|
begin_finished = next(e for e in received_events if e.method_name == "begin")
|
|
process_finished = next(e for e in received_events if e.method_name == "process")
|
|
|
|
assert begin_finished.state is not None
|
|
assert isinstance(begin_finished.state, dict)
|
|
assert begin_finished.state["result"] == "begin done"
|
|
assert begin_finished.state["completed"] is False
|
|
assert begin_finished.result == "started"
|
|
|
|
assert process_finished.state is not None
|
|
assert isinstance(process_finished.state, dict)
|
|
assert process_finished.state["result"] == "process done"
|
|
assert process_finished.state["completed"] is True
|
|
assert process_finished.result == "final_result"
|
|
assert final_output == "final_result"
|
|
|
|
|
|
def test_suppress_flow_events_silences_method_lifecycle_events():
|
|
"""``suppress_flow_events=True`` emits no MethodExecution* events on the
|
|
bus (used by infrastructure flows like AgentExecutor so their control-flow
|
|
methods don't pollute traces), while default flows still emit them."""
|
|
captured: list[tuple[str, str]] = []
|
|
|
|
class SuppressedFlow(Flow):
|
|
suppress_flow_events: bool = True
|
|
|
|
@start()
|
|
def begin(self):
|
|
return "started"
|
|
|
|
@listen("begin")
|
|
def process(self):
|
|
return "done"
|
|
|
|
class ControlFlow(Flow):
|
|
@start()
|
|
def begin(self):
|
|
return "started"
|
|
|
|
@listen("begin")
|
|
def process(self):
|
|
return "done"
|
|
|
|
with crewai_event_bus.scoped_handlers():
|
|
|
|
@crewai_event_bus.on(MethodExecutionStartedEvent)
|
|
def _on_started(source, event):
|
|
captured.append(("started", type(source).__name__))
|
|
|
|
@crewai_event_bus.on(MethodExecutionFinishedEvent)
|
|
def _on_finished(source, event):
|
|
captured.append(("finished", type(source).__name__))
|
|
|
|
SuppressedFlow().kickoff()
|
|
wait_for_event_handlers()
|
|
assert [e for e in captured if e[1] == "SuppressedFlow"] == [], (
|
|
"suppress_flow_events=True must emit no MethodExecution* events"
|
|
)
|
|
|
|
captured.clear()
|
|
ControlFlow().kickoff()
|
|
wait_for_event_handlers()
|
|
control = [e for e in captured if e[1] == "ControlFlow"]
|
|
assert ("started", "ControlFlow") in control
|
|
assert ("finished", "ControlFlow") in control
|
|
|
|
|
|
def test_infrastructure_flows_suppress_flow_events_by_default():
|
|
"""Pin the infra flows that must stay silent in traces.
|
|
|
|
The gating in ``_execute_method`` only helps if these flows actually set
|
|
``suppress_flow_events=True``; without this guard, removing the flag from
|
|
AgentExecutor would silently bring back the verbose per-method trace spans.
|
|
"""
|
|
from crewai.experimental.agent_executor import AgentExecutor
|
|
from crewai.memory.encoding_flow import EncodingFlow
|
|
from crewai.memory.recall_flow import RecallFlow
|
|
|
|
assert AgentExecutor.model_fields["suppress_flow_events"].default is True
|
|
|
|
for flow_cls in (EncodingFlow, RecallFlow):
|
|
flow = flow_cls(storage=None, llm=None, embedder=None)
|
|
assert flow.suppress_flow_events is True
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_llm_emits_call_started_event():
|
|
started_events: list[LLMCallStartedEvent] = []
|
|
completed_events: list[LLMCallCompletedEvent] = []
|
|
condition = threading.Condition()
|
|
|
|
@crewai_event_bus.on(LLMCallStartedEvent)
|
|
def handle_llm_call_started(source, event):
|
|
with condition:
|
|
started_events.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMCallCompletedEvent)
|
|
def handle_llm_call_completed(source, event):
|
|
with condition:
|
|
completed_events.append(event)
|
|
condition.notify()
|
|
|
|
llm = LLM(model="gpt-4o-mini")
|
|
llm.call("Hello, how are you?")
|
|
|
|
with condition:
|
|
success = condition.wait_for(
|
|
lambda: len(started_events) >= 1 and len(completed_events) >= 1,
|
|
timeout=10,
|
|
)
|
|
assert success, "Timeout waiting for LLM events"
|
|
|
|
assert started_events[0].type == "llm_call_started"
|
|
assert completed_events[0].type == "llm_call_completed"
|
|
|
|
assert started_events[0].task_name is None
|
|
assert started_events[0].agent_role is None
|
|
assert started_events[0].agent_id is None
|
|
assert started_events[0].task_id is None
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_llm_completed_event_includes_usage():
|
|
completed_events: list[LLMCallCompletedEvent] = []
|
|
condition = threading.Condition()
|
|
|
|
@crewai_event_bus.on(LLMCallCompletedEvent)
|
|
def handle_llm_call_completed(source, event):
|
|
with condition:
|
|
completed_events.append(event)
|
|
condition.notify()
|
|
|
|
llm = LLM(model="gpt-4o-mini")
|
|
llm.call("Say hello")
|
|
|
|
with condition:
|
|
success = condition.wait_for(
|
|
lambda: len(completed_events) >= 1,
|
|
timeout=10,
|
|
)
|
|
assert success, "Timeout waiting for LLMCallCompletedEvent"
|
|
|
|
event = completed_events[0]
|
|
assert event.usage is not None
|
|
assert isinstance(event.usage, dict)
|
|
assert event.usage.get("prompt_tokens", 0) > 0
|
|
assert event.usage.get("completion_tokens", 0) > 0
|
|
assert event.usage.get("total_tokens", 0) > 0
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_llm_emits_call_failed_event():
|
|
received_events = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(LLMCallFailedEvent)
|
|
def handle_llm_call_failed(source, event):
|
|
received_events.append(event)
|
|
event_received.set()
|
|
|
|
error_message = "OpenAI API call failed: Simulated API failure"
|
|
|
|
with patch(
|
|
"crewai.llms.providers.openai.completion.OpenAICompletion._handle_completion"
|
|
) as mock_handle_completion:
|
|
mock_handle_completion.side_effect = Exception("Simulated API failure")
|
|
|
|
llm = LLM(model="gpt-4o-mini")
|
|
with pytest.raises(Exception) as exc_info:
|
|
llm.call("Hello, how are you?")
|
|
|
|
assert str(exc_info.value) == "Simulated API failure"
|
|
assert event_received.wait(timeout=5), "Timeout waiting for failed event"
|
|
assert len(received_events) == 1
|
|
assert received_events[0].type == "llm_call_failed"
|
|
assert received_events[0].error == error_message
|
|
assert received_events[0].task_name is None
|
|
assert received_events[0].agent_role is None
|
|
assert received_events[0].agent_id is None
|
|
assert received_events[0].task_id is None
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_llm_emits_stream_chunk_events():
|
|
"""Test that LLM emits stream chunk events when streaming is enabled."""
|
|
received_chunks = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(LLMStreamChunkEvent)
|
|
def handle_stream_chunk(source, event):
|
|
received_chunks.append(event.chunk)
|
|
if len(received_chunks) >= 1:
|
|
event_received.set()
|
|
|
|
llm = LLM(model="gpt-4o", stream=True)
|
|
|
|
response = llm.call("Tell me a short joke")
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for stream chunks"
|
|
|
|
assert len(received_chunks) > 0
|
|
|
|
assert "".join(received_chunks) == response
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_llm_no_stream_chunks_when_streaming_disabled():
|
|
"""Test that LLM doesn't emit stream chunk events when streaming is disabled."""
|
|
received_chunks = []
|
|
|
|
@crewai_event_bus.on(LLMStreamChunkEvent)
|
|
def handle_stream_chunk(source, event):
|
|
received_chunks.append(event.chunk)
|
|
|
|
llm = LLM(model="gpt-4o", stream=False)
|
|
|
|
response = llm.call("Tell me a short joke")
|
|
|
|
assert len(received_chunks) == 0
|
|
|
|
assert response and isinstance(response, str)
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_streaming_fallback_to_non_streaming():
|
|
"""Test that streaming falls back to non-streaming when there's an error."""
|
|
received_chunks = []
|
|
fallback_called = False
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(LLMStreamChunkEvent)
|
|
def handle_stream_chunk(source, event):
|
|
received_chunks.append(event.chunk)
|
|
if len(received_chunks) >= 2:
|
|
event_received.set()
|
|
|
|
llm = LLM(model="gpt-4o", stream=True)
|
|
|
|
original_call = llm.call
|
|
|
|
def mock_call(messages, tools=None, callbacks=None, available_functions=None):
|
|
nonlocal fallback_called
|
|
# Emit a couple of chunks to simulate partial streaming
|
|
crewai_event_bus.emit(llm, event=LLMStreamChunkEvent(chunk="Test chunk 1", response_id="Id", call_id="test-call-id"))
|
|
crewai_event_bus.emit(llm, event=LLMStreamChunkEvent(chunk="Test chunk 2", response_id="Id", call_id="test-call-id"))
|
|
|
|
# Mark that fallback would be called
|
|
fallback_called = True
|
|
|
|
# Return a response as if fallback succeeded
|
|
return "Fallback response after streaming error"
|
|
|
|
llm.call = mock_call
|
|
|
|
try:
|
|
response = llm.call("Tell me a short joke")
|
|
wait_for_event_handlers()
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for stream chunks"
|
|
|
|
assert len(received_chunks) == 2
|
|
assert received_chunks[0] == "Test chunk 1"
|
|
assert received_chunks[1] == "Test chunk 2"
|
|
|
|
# Verify fallback was triggered
|
|
assert fallback_called
|
|
|
|
# Verify we got the fallback response
|
|
assert response == "Fallback response after streaming error"
|
|
|
|
finally:
|
|
llm.call = original_call
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_streaming_empty_response_handling():
|
|
"""Test that streaming handles empty responses correctly."""
|
|
received_chunks = []
|
|
event_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(LLMStreamChunkEvent)
|
|
def handle_stream_chunk(source, event):
|
|
received_chunks.append(event.chunk)
|
|
if len(received_chunks) >= 3:
|
|
event_received.set()
|
|
|
|
llm = LLM(model="gpt-3.5-turbo", stream=True)
|
|
|
|
original_call = llm.call
|
|
|
|
# Create a mock call method that simulates empty chunks
|
|
def mock_call(messages, tools=None, callbacks=None, available_functions=None):
|
|
for _ in range(3):
|
|
crewai_event_bus.emit(llm, event=LLMStreamChunkEvent(chunk="", response_id="id", call_id="test-call-id"))
|
|
|
|
return "I apologize, but I couldn't generate a proper response. Please try again or rephrase your request."
|
|
|
|
llm.call = mock_call
|
|
|
|
try:
|
|
response = llm.call("Tell me a short joke")
|
|
|
|
assert event_received.wait(timeout=5), "Timeout waiting for empty chunks"
|
|
|
|
assert len(received_chunks) == 3
|
|
assert all(chunk == "" for chunk in received_chunks)
|
|
|
|
assert "I apologize" in response and "couldn't generate" in response
|
|
|
|
finally:
|
|
llm.call = original_call
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_stream_llm_emits_event_with_task_and_agent_info():
|
|
completed_event = []
|
|
failed_event = []
|
|
started_event = []
|
|
stream_event = []
|
|
condition = threading.Condition()
|
|
|
|
@crewai_event_bus.on(LLMCallFailedEvent)
|
|
def handle_llm_failed(source, event):
|
|
with condition:
|
|
failed_event.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMCallStartedEvent)
|
|
def handle_llm_started(source, event):
|
|
with condition:
|
|
started_event.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMCallCompletedEvent)
|
|
def handle_llm_completed(source, event):
|
|
with condition:
|
|
completed_event.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMStreamChunkEvent)
|
|
def handle_llm_stream_chunk(source, event):
|
|
with condition:
|
|
stream_event.append(event)
|
|
condition.notify()
|
|
|
|
agent = Agent(
|
|
role="TestAgent",
|
|
llm=LLM(model="gpt-4o-mini", stream=True),
|
|
goal="Just say hi",
|
|
backstory="You are a helpful assistant that just says hi",
|
|
)
|
|
task = Task(
|
|
description="Just say hi",
|
|
expected_output="hi",
|
|
llm=LLM(model="gpt-4o-mini", stream=True),
|
|
agent=agent,
|
|
)
|
|
|
|
crew = Crew(agents=[agent], tasks=[task])
|
|
crew.kickoff()
|
|
|
|
with condition:
|
|
success = condition.wait_for(
|
|
lambda: len(completed_event) >= 1
|
|
and len(started_event) >= 1
|
|
and len(stream_event) >= 12,
|
|
timeout=10,
|
|
)
|
|
assert success, "Timeout waiting for LLM events"
|
|
assert len(completed_event) == 1
|
|
assert len(failed_event) == 0
|
|
assert len(started_event) == 1
|
|
assert len(stream_event) == 12
|
|
|
|
all_events = completed_event + failed_event + started_event + stream_event
|
|
all_agent_roles = [event.agent_role for event in all_events]
|
|
all_agent_id = [event.agent_id for event in all_events]
|
|
all_task_id = [event.task_id for event in all_events]
|
|
all_task_name = [event.task_name for event in all_events]
|
|
|
|
# ensure all events have the agent + task props set
|
|
assert len(all_agent_roles) == 14
|
|
assert len(all_agent_id) == 14
|
|
assert len(all_task_id) == 14
|
|
assert len(all_task_name) == 14
|
|
|
|
assert set(all_agent_roles) == {agent.role}
|
|
assert set(all_agent_id) == {str(agent.id)}
|
|
assert set(all_task_id) == {str(task.id)}
|
|
assert set(all_task_name) == {task.name or task.description}
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_llm_emits_event_with_task_and_agent_info(base_agent, base_task):
|
|
completed_event: list[LLMCallCompletedEvent] = []
|
|
failed_event: list[LLMCallFailedEvent] = []
|
|
started_event: list[LLMCallStartedEvent] = []
|
|
stream_event: list[LLMStreamChunkEvent] = []
|
|
condition = threading.Condition()
|
|
|
|
@crewai_event_bus.on(LLMCallFailedEvent)
|
|
def handle_llm_failed(source, event):
|
|
with condition:
|
|
failed_event.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMCallStartedEvent)
|
|
def handle_llm_started(source, event):
|
|
with condition:
|
|
started_event.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMCallCompletedEvent)
|
|
def handle_llm_completed(source, event):
|
|
with condition:
|
|
completed_event.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMStreamChunkEvent)
|
|
def handle_llm_stream_chunk(source, event):
|
|
with condition:
|
|
stream_event.append(event)
|
|
condition.notify()
|
|
|
|
crew = Crew(agents=[base_agent], tasks=[base_task])
|
|
crew.kickoff()
|
|
|
|
with condition:
|
|
success = condition.wait_for(
|
|
lambda: len(completed_event) >= 1 and len(started_event) >= 1,
|
|
timeout=10,
|
|
)
|
|
assert success, "Timeout waiting for LLM events"
|
|
assert len(completed_event) == 1
|
|
assert len(failed_event) == 0
|
|
assert len(started_event) == 1
|
|
assert len(stream_event) == 0
|
|
|
|
all_events = completed_event + failed_event + started_event + stream_event
|
|
all_agent_roles = [event.agent_role for event in all_events]
|
|
all_agent_id = [event.agent_id for event in all_events]
|
|
all_task_id = [event.task_id for event in all_events]
|
|
all_task_name = [event.task_name for event in all_events]
|
|
|
|
# ensure all events have the agent + task props set
|
|
assert len(all_agent_roles) == 2
|
|
assert len(all_agent_id) == 2
|
|
assert len(all_task_id) == 2
|
|
assert len(all_task_name) == 2
|
|
|
|
assert set(all_agent_roles) == {base_agent.role}
|
|
assert set(all_agent_id) == {str(base_agent.id)}
|
|
assert set(all_task_id) == {str(base_task.id)}
|
|
assert set(all_task_name) == {base_task.name or base_task.description}
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_llm_emits_event_with_lite_agent():
|
|
completed_event = []
|
|
failed_event = []
|
|
started_event = []
|
|
stream_event = []
|
|
condition = threading.Condition()
|
|
|
|
@crewai_event_bus.on(LLMCallFailedEvent)
|
|
def handle_llm_failed(source, event):
|
|
with condition:
|
|
failed_event.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMCallStartedEvent)
|
|
def handle_llm_started(source, event):
|
|
with condition:
|
|
started_event.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMCallCompletedEvent)
|
|
def handle_llm_completed(source, event):
|
|
with condition:
|
|
completed_event.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMStreamChunkEvent)
|
|
def handle_llm_stream_chunk(source, event):
|
|
with condition:
|
|
stream_event.append(event)
|
|
condition.notify()
|
|
|
|
agent = Agent(
|
|
role="Speaker",
|
|
llm=LLM(model="gpt-4o-mini", stream=True),
|
|
goal="Just say hi",
|
|
backstory="You are a helpful assistant that just says hi",
|
|
)
|
|
agent.kickoff(messages=[{"role": "user", "content": "say hi!"}])
|
|
|
|
with condition:
|
|
success = condition.wait_for(
|
|
lambda: len(completed_event) >= 1
|
|
and len(started_event) >= 1
|
|
and len(stream_event) >= 1,
|
|
timeout=10,
|
|
)
|
|
assert success, "Timeout waiting for all events"
|
|
|
|
assert len(completed_event) == 1
|
|
assert len(failed_event) == 0
|
|
assert len(started_event) == 1
|
|
assert len(stream_event) >= 1
|
|
|
|
all_events = completed_event + failed_event + started_event + stream_event
|
|
all_agent_roles = [event.agent_role for event in all_events]
|
|
all_agent_id = [event.agent_id for event in all_events]
|
|
all_task_id = [event.task_id for event in all_events if event.task_id]
|
|
all_task_name = [event.task_name for event in all_events if event.task_name]
|
|
|
|
# ensure all events have the agent + task props set
|
|
expected_total = 1 + 1 + len(stream_event) # completed + started + stream
|
|
assert len(all_agent_roles) == expected_total
|
|
assert len(all_agent_id) == expected_total
|
|
assert len(all_task_id) == 0
|
|
assert len(all_task_name) == 0
|
|
|
|
assert set(all_agent_roles) == {agent.role}
|
|
assert set(all_agent_id) == {str(agent.id)}
|
|
|
|
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_llm_call_events_share_call_id():
|
|
"""All events from a single LLM call should share the same call_id."""
|
|
import uuid
|
|
|
|
events = []
|
|
condition = threading.Condition()
|
|
|
|
@crewai_event_bus.on(LLMCallStartedEvent)
|
|
def on_start(source, event):
|
|
with condition:
|
|
events.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMCallCompletedEvent)
|
|
def on_complete(source, event):
|
|
with condition:
|
|
events.append(event)
|
|
condition.notify()
|
|
|
|
llm = LLM(model="gpt-4o-mini")
|
|
llm.call("Say hi")
|
|
|
|
with condition:
|
|
success = condition.wait_for(lambda: len(events) >= 2, timeout=10)
|
|
assert success, "Timeout waiting for LLM events"
|
|
|
|
# Behavior: all events from the call share the same call_id
|
|
assert len(events) == 2
|
|
assert events[0].call_id == events[1].call_id
|
|
# call_id should be a valid UUID
|
|
uuid.UUID(events[0].call_id)
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_streaming_chunks_share_call_id_with_call():
|
|
"""Streaming chunks should share call_id with started/completed events."""
|
|
events = []
|
|
condition = threading.Condition()
|
|
|
|
@crewai_event_bus.on(LLMCallStartedEvent)
|
|
def on_start(source, event):
|
|
with condition:
|
|
events.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMStreamChunkEvent)
|
|
def on_chunk(source, event):
|
|
with condition:
|
|
events.append(event)
|
|
condition.notify()
|
|
|
|
@crewai_event_bus.on(LLMCallCompletedEvent)
|
|
def on_complete(source, event):
|
|
with condition:
|
|
events.append(event)
|
|
condition.notify()
|
|
|
|
llm = LLM(model="gpt-4o-mini", stream=True)
|
|
llm.call("Say hi")
|
|
|
|
with condition:
|
|
success = condition.wait_for(lambda: len(events) >= 3, timeout=10)
|
|
assert success, "Timeout waiting for streaming events"
|
|
|
|
# Behavior: all events (started, chunks, completed) share the same call_id
|
|
call_ids = {e.call_id for e in events}
|
|
assert len(call_ids) == 1
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_separate_llm_calls_have_different_call_ids():
|
|
"""Different LLM calls should have different call_ids."""
|
|
call_ids = []
|
|
condition = threading.Condition()
|
|
|
|
@crewai_event_bus.on(LLMCallStartedEvent)
|
|
def on_start(source, event):
|
|
with condition:
|
|
call_ids.append(event.call_id)
|
|
condition.notify()
|
|
|
|
llm = LLM(model="gpt-4o-mini")
|
|
llm.call("Say hi")
|
|
llm.call("Say bye")
|
|
|
|
with condition:
|
|
success = condition.wait_for(lambda: len(call_ids) >= 2, timeout=10)
|
|
assert success, "Timeout waiting for LLM call events"
|
|
|
|
# Behavior: each call has its own call_id
|
|
assert len(call_ids) == 2
|
|
assert call_ids[0] != call_ids[1]
|
|
|
|
|
|
|
|
|
|
@patch("builtins.input", return_value="looks good")
|
|
@patch("builtins.print")
|
|
def test_human_feedback_emits_requested_and_received_events(mock_print, mock_input):
|
|
"""Test that @human_feedback decorator emits HumanFeedbackRequested and Received events."""
|
|
requested_events = []
|
|
received_events = []
|
|
events_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(HumanFeedbackRequestedEvent)
|
|
def handle_requested(source, event):
|
|
requested_events.append(event)
|
|
|
|
@crewai_event_bus.on(HumanFeedbackReceivedEvent)
|
|
def handle_received(source, event):
|
|
received_events.append(event)
|
|
events_received.set()
|
|
|
|
class TestFlow(Flow):
|
|
@start()
|
|
@human_feedback(
|
|
message="Review:",
|
|
emit=["approved", "rejected"],
|
|
llm="gpt-4o-mini",
|
|
)
|
|
def review(self):
|
|
return "test content"
|
|
|
|
flow = TestFlow()
|
|
|
|
with patch.object(flow, "_collapse_to_outcome", return_value="approved"):
|
|
flow.kickoff()
|
|
|
|
assert events_received.wait(timeout=5), (
|
|
"Timeout waiting for human feedback events"
|
|
)
|
|
|
|
assert len(requested_events) == 1
|
|
assert requested_events[0].type == "human_feedback_requested"
|
|
assert requested_events[0].emit == ["approved", "rejected"]
|
|
assert requested_events[0].message == "Review:"
|
|
assert requested_events[0].output == "test content"
|
|
|
|
assert len(received_events) == 1
|
|
assert received_events[0].type == "human_feedback_received"
|
|
assert received_events[0].feedback == "looks good"
|
|
assert received_events[0].outcome is None
|
|
|
|
assert flow.last_human_feedback is not None
|
|
assert flow.last_human_feedback.outcome == "approved"
|
|
|
|
|
|
@patch("builtins.input", return_value="feedback text")
|
|
@patch("builtins.print")
|
|
def test_human_feedback_without_routing_emits_events(mock_print, mock_input):
|
|
"""Test that @human_feedback without emit still emits events."""
|
|
requested_events = []
|
|
received_events = []
|
|
events_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(HumanFeedbackRequestedEvent)
|
|
def handle_requested(source, event):
|
|
requested_events.append(event)
|
|
|
|
@crewai_event_bus.on(HumanFeedbackReceivedEvent)
|
|
def handle_received(source, event):
|
|
received_events.append(event)
|
|
events_received.set()
|
|
|
|
class SimpleFlow(Flow):
|
|
@start()
|
|
@human_feedback(message="Please review:")
|
|
def review(self):
|
|
return "content to review"
|
|
|
|
flow = SimpleFlow()
|
|
flow.kickoff()
|
|
|
|
assert events_received.wait(timeout=5), (
|
|
"Timeout waiting for human feedback events"
|
|
)
|
|
|
|
assert len(requested_events) == 1
|
|
assert requested_events[0].emit is None
|
|
|
|
assert len(received_events) == 1
|
|
assert received_events[0].feedback == "feedback text"
|
|
assert received_events[0].outcome is None
|
|
|
|
|
|
@patch("builtins.input", return_value="")
|
|
@patch("builtins.print")
|
|
def test_human_feedback_empty_feedback_emits_events(mock_print, mock_input):
|
|
"""Test that empty feedback (skipped) still emits events correctly."""
|
|
received_events = []
|
|
events_received = threading.Event()
|
|
|
|
@crewai_event_bus.on(HumanFeedbackReceivedEvent)
|
|
def handle_received(source, event):
|
|
received_events.append(event)
|
|
events_received.set()
|
|
|
|
class SkipFlow(Flow):
|
|
@start()
|
|
@human_feedback(
|
|
message="Review:",
|
|
emit=["approved", "rejected"],
|
|
llm="gpt-4o-mini",
|
|
default_outcome="rejected",
|
|
)
|
|
def review(self):
|
|
return "content"
|
|
|
|
flow = SkipFlow()
|
|
flow.kickoff()
|
|
|
|
assert events_received.wait(timeout=5), (
|
|
"Timeout waiting for human feedback events"
|
|
)
|
|
|
|
|
|
assert len(received_events) == 1
|
|
assert received_events[0].feedback == ""
|
|
assert received_events[0].outcome is None
|
|
|
|
|
|
assert flow.last_human_feedback is not None
|
|
assert flow.last_human_feedback.outcome == "rejected"
|