Skip to content

Commit 1572bac

Browse files
committed
Ensure @start methods emit MethodExecutionStartedEvent
Previously, `@start` methods triggered a `FlowStartedEvent` but did not emit a `MethodExecutionStartedEvent`. This was fine for a single entry point but caused ambiguity when multiple `@start` methods existed. This commit (1) emits events for starting points, (2) adds tests ensuring ordering, (3) adds more fields to events.
1 parent 2fd7506 commit 1572bac

File tree

3 files changed

+277
-22
lines changed

3 files changed

+277
-22
lines changed

src/crewai/flow/flow.py

Lines changed: 47 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -394,7 +394,6 @@ def __new__(mcs, name, bases, dct):
394394
or hasattr(attr_value, "__trigger_methods__")
395395
or hasattr(attr_value, "__is_router__")
396396
):
397-
398397
# Register start methods
399398
if hasattr(attr_value, "__is_start_method__"):
400399
start_methods.append(attr_name)
@@ -569,6 +568,28 @@ class StateWithId(state_type, FlowState): # type: ignore
569568
f"Initial state must be dict or BaseModel, got {type(self.initial_state)}"
570569
)
571570

571+
def _dump_state(self) -> Optional[Dict[str, Any]]:
572+
"""
573+
Dumps the current flow state as a dictionary.
574+
575+
This method converts the internal state into a serializable dictionary format,
576+
ensuring compatibility with both dictionary and Pydantic BaseModel states.
577+
578+
Returns:
579+
Optional[Dict[str, Any]]: The serialized state dictionary, or None if state is not available.
580+
"""
581+
if self._state is None:
582+
return None
583+
584+
if isinstance(self._state, dict):
585+
return self._state.copy()
586+
587+
if isinstance(self._state, BaseModel):
588+
return self._state.model_dump()
589+
590+
logger.warning("Unsupported flow state type for dumping.")
591+
return None
592+
572593
@property
573594
def state(self) -> T:
574595
return self._state
@@ -740,6 +761,7 @@ def kickoff(self, inputs: Optional[Dict[str, Any]] = None) -> Any:
740761
event=FlowStartedEvent(
741762
type="flow_started",
742763
flow_name=self.__class__.__name__,
764+
inputs=inputs,
743765
),
744766
)
745767
self._log_flow_event(
@@ -803,6 +825,18 @@ async def _execute_start_method(self, start_method_name: str) -> None:
803825
async def _execute_method(
804826
self, method_name: str, method: Callable, *args: Any, **kwargs: Any
805827
) -> Any:
828+
dumped_params = {f"_{i}": arg for i, arg in enumerate(args)} | (kwargs or {})
829+
self.event_emitter.send(
830+
self,
831+
event=MethodExecutionStartedEvent(
832+
type="method_execution_started",
833+
method_name=method_name,
834+
flow_name=self.__class__.__name__,
835+
params=dumped_params,
836+
state=self._dump_state(),
837+
),
838+
)
839+
806840
result = (
807841
await method(*args, **kwargs)
808842
if asyncio.iscoroutinefunction(method)
@@ -812,6 +846,18 @@ async def _execute_method(
812846
self._method_execution_counts[method_name] = (
813847
self._method_execution_counts.get(method_name, 0) + 1
814848
)
849+
850+
self.event_emitter.send(
851+
self,
852+
event=MethodExecutionFinishedEvent(
853+
type="method_execution_finished",
854+
method_name=method_name,
855+
flow_name=self.__class__.__name__,
856+
state=self._dump_state(),
857+
result=result,
858+
),
859+
)
860+
815861
return result
816862

817863
async def _execute_listeners(self, trigger_method: str, result: Any) -> None:
@@ -950,16 +996,6 @@ async def _execute_single_listener(self, listener_name: str, result: Any) -> Non
950996
"""
951997
try:
952998
method = self._methods[listener_name]
953-
954-
self.event_emitter.send(
955-
self,
956-
event=MethodExecutionStartedEvent(
957-
type="method_execution_started",
958-
method_name=listener_name,
959-
flow_name=self.__class__.__name__,
960-
),
961-
)
962-
963999
sig = inspect.signature(method)
9641000
params = list(sig.parameters.values())
9651001
method_params = [p for p in params if p.name != "self"]
@@ -971,15 +1007,6 @@ async def _execute_single_listener(self, listener_name: str, result: Any) -> Non
9711007
else:
9721008
listener_result = await self._execute_method(listener_name, method)
9731009

974-
self.event_emitter.send(
975-
self,
976-
event=MethodExecutionFinishedEvent(
977-
type="method_execution_finished",
978-
method_name=listener_name,
979-
flow_name=self.__class__.__name__,
980-
),
981-
)
982-
9831010
# Execute listeners (and possibly routers) of this listener
9841011
await self._execute_listeners(listener_name, listener_result)
9851012

src/crewai/flow/flow_events.py

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
from dataclasses import dataclass, field
22
from datetime import datetime
3-
from typing import Any, Optional
3+
from typing import Any, Dict, Optional
44

55

66
@dataclass
@@ -15,17 +15,21 @@ def __post_init__(self):
1515

1616
@dataclass
1717
class FlowStartedEvent(Event):
18-
pass
18+
inputs: Optional[Dict[str, Any]] = None
1919

2020

2121
@dataclass
2222
class MethodExecutionStartedEvent(Event):
2323
method_name: str
24+
params: Optional[Dict[str, Any]] = None
25+
state: Optional[Dict[str, Any]] = None
2426

2527

2628
@dataclass
2729
class MethodExecutionFinishedEvent(Event):
2830
method_name: str
31+
result: Any = None
32+
state: Optional[Dict[str, Any]] = None
2933

3034

3135
@dataclass

tests/flow_test.py

Lines changed: 224 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,18 @@
11
"""Test Flow creation and execution basic functionality."""
22

33
import asyncio
4+
from datetime import datetime
45

56
import pytest
67
from pydantic import BaseModel
78

89
from crewai.flow.flow import Flow, and_, listen, or_, router, start
10+
from crewai.flow.flow_events import (
11+
FlowFinishedEvent,
12+
FlowStartedEvent,
13+
MethodExecutionFinishedEvent,
14+
MethodExecutionStartedEvent,
15+
)
916

1017

1118
def test_simple_sequential_flow():
@@ -398,3 +405,220 @@ def log_final_step(self):
398405

399406
# final_step should run after router_and
400407
assert execution_order.index("log_final_step") > execution_order.index("router_and")
408+
409+
410+
def test_unstructured_flow_event_emission():
411+
"""Test that the correct events are emitted during unstructured flow
412+
execution with all fields validated."""
413+
414+
class PoemFlow(Flow):
415+
@start()
416+
def prepare_flower(self):
417+
self.state["flower"] = "roses"
418+
return "foo"
419+
420+
@start()
421+
def prepare_color(self):
422+
self.state["color"] = "red"
423+
return "bar"
424+
425+
@listen(prepare_color)
426+
def write_first_sentence(self):
427+
return f"{self.state["flower"]} are {self.state["color"]}"
428+
429+
@listen(write_first_sentence)
430+
def finish_poem(self, first_sentence):
431+
separator = self.state.get("separator", "\n")
432+
return separator.join([first_sentence, "violets are blue"])
433+
434+
@listen(finish_poem)
435+
def save_poem_to_database(self):
436+
# A method without args/kwargs to ensure events are sent correctly
437+
pass
438+
439+
event_log = []
440+
441+
def handle_event(_, event):
442+
event_log.append(event)
443+
444+
flow = PoemFlow()
445+
flow.event_emitter.connect(handle_event)
446+
flow.kickoff(inputs={"separator": ", "})
447+
448+
assert isinstance(event_log[0], FlowStartedEvent)
449+
assert event_log[0].flow_name == "PoemFlow"
450+
assert event_log[0].inputs == {"separator": ", "}
451+
assert isinstance(event_log[0].timestamp, datetime)
452+
453+
# Asserting for concurrent start method executions in a for loop as you
454+
# can't guarantee ordering in asynchronous executions
455+
for i in range(1, 5):
456+
event = event_log[i]
457+
assert isinstance(event.state, dict)
458+
assert isinstance(event.state["id"], str)
459+
460+
if event.method_name == "prepare_flower":
461+
if isinstance(event, MethodExecutionStartedEvent):
462+
assert event.params == {}
463+
assert event.state.get("separator") == ", "
464+
elif isinstance(event, MethodExecutionFinishedEvent):
465+
assert event.result == "foo"
466+
assert event.state.get("flower") == "roses"
467+
assert event.state.get("separator") == ", "
468+
else:
469+
assert False, "Unexpected event type for prepare_flower"
470+
elif event.method_name == "prepare_color":
471+
if isinstance(event, MethodExecutionStartedEvent):
472+
assert event.params == {}
473+
assert event.state.get("separator") == ", "
474+
elif isinstance(event, MethodExecutionFinishedEvent):
475+
assert event.result == "bar"
476+
assert event.state.get("color") == "red"
477+
assert event.state.get("separator") == ", "
478+
else:
479+
assert False, "Unexpected event type for prepare_color"
480+
else:
481+
assert False, f"Unexpected method {event.method_name} in prepare events"
482+
483+
assert isinstance(event_log[5], MethodExecutionStartedEvent)
484+
assert event_log[5].method_name == "write_first_sentence"
485+
assert event_log[5].params == {}
486+
assert isinstance(event_log[5].state, dict)
487+
assert event_log[5].state.get("flower") == "roses"
488+
assert event_log[5].state.get("color") == "red"
489+
assert event_log[5].state.get("separator") == ", "
490+
491+
assert isinstance(event_log[6], MethodExecutionFinishedEvent)
492+
assert event_log[6].method_name == "write_first_sentence"
493+
assert event_log[6].result == "roses are red"
494+
495+
assert isinstance(event_log[7], MethodExecutionStartedEvent)
496+
assert event_log[7].method_name == "finish_poem"
497+
assert event_log[7].params == {"_0": "roses are red"}
498+
assert isinstance(event_log[7].state, dict)
499+
assert event_log[7].state.get("flower") == "roses"
500+
assert event_log[7].state.get("color") == "red"
501+
502+
assert isinstance(event_log[8], MethodExecutionFinishedEvent)
503+
assert event_log[8].method_name == "finish_poem"
504+
assert event_log[8].result == "roses are red, violets are blue"
505+
506+
assert isinstance(event_log[9], MethodExecutionStartedEvent)
507+
assert event_log[9].method_name == "save_poem_to_database"
508+
assert event_log[9].params == {}
509+
assert isinstance(event_log[9].state, dict)
510+
assert event_log[9].state.get("flower") == "roses"
511+
assert event_log[9].state.get("color") == "red"
512+
513+
assert isinstance(event_log[10], MethodExecutionFinishedEvent)
514+
assert event_log[10].method_name == "save_poem_to_database"
515+
assert event_log[10].result is None
516+
517+
assert isinstance(event_log[11], FlowFinishedEvent)
518+
assert event_log[11].flow_name == "PoemFlow"
519+
assert event_log[11].result is None
520+
assert isinstance(event_log[11].timestamp, datetime)
521+
522+
523+
def test_structured_flow_event_emission():
524+
"""Test that the correct events are emitted during structured flow
525+
execution with all fields validated."""
526+
527+
class OnboardingState(BaseModel):
528+
name: str = ""
529+
sent: bool = False
530+
531+
class OnboardingFlow(Flow[OnboardingState]):
532+
@start()
533+
def user_signs_up(self):
534+
self.state.sent = False
535+
536+
@listen(user_signs_up)
537+
def send_welcome_message(self):
538+
self.state.sent = True
539+
return f"Welcome, {self.state.name}!"
540+
541+
event_log = []
542+
543+
def handle_event(_, event):
544+
event_log.append(event)
545+
546+
flow = OnboardingFlow()
547+
flow.event_emitter.connect(handle_event)
548+
flow.kickoff(inputs={"name": "Anakin"})
549+
550+
assert isinstance(event_log[0], FlowStartedEvent)
551+
assert event_log[0].flow_name == "OnboardingFlow"
552+
assert event_log[0].inputs == {"name": "Anakin"}
553+
assert isinstance(event_log[0].timestamp, datetime)
554+
555+
assert isinstance(event_log[1], MethodExecutionStartedEvent)
556+
assert event_log[1].method_name == "user_signs_up"
557+
558+
assert isinstance(event_log[2], MethodExecutionFinishedEvent)
559+
assert event_log[2].method_name == "user_signs_up"
560+
561+
assert isinstance(event_log[3], MethodExecutionStartedEvent)
562+
assert event_log[3].method_name == "send_welcome_message"
563+
assert event_log[3].params == {}
564+
assert isinstance(event_log[3].state, dict)
565+
assert event_log[3].state.get("sent") == False
566+
567+
assert isinstance(event_log[4], MethodExecutionFinishedEvent)
568+
assert event_log[4].method_name == "send_welcome_message"
569+
assert isinstance(event_log[4].state, dict)
570+
assert event_log[4].state.get("sent") == True
571+
assert event_log[4].result == "Welcome, Anakin!"
572+
573+
assert isinstance(event_log[5], FlowFinishedEvent)
574+
assert event_log[5].flow_name == "OnboardingFlow"
575+
assert event_log[5].result == "Welcome, Anakin!"
576+
assert isinstance(event_log[5].timestamp, datetime)
577+
578+
579+
def test_stateless_flow_event_emission():
580+
"""Test that the correct events are emitted stateless during flow execution
581+
with all fields validated."""
582+
583+
class StatelessFlow(Flow):
584+
@start()
585+
def init(self):
586+
pass
587+
588+
@listen(init)
589+
def process(self):
590+
return "Deeds will not be less valiant because they are unpraised."
591+
592+
event_log = []
593+
594+
def handle_event(_, event):
595+
event_log.append(event)
596+
597+
flow = StatelessFlow()
598+
flow.event_emitter.connect(handle_event)
599+
flow.kickoff()
600+
601+
assert isinstance(event_log[0], FlowStartedEvent)
602+
assert event_log[0].flow_name == "StatelessFlow"
603+
assert event_log[0].inputs is None
604+
assert isinstance(event_log[0].timestamp, datetime)
605+
606+
assert isinstance(event_log[1], MethodExecutionStartedEvent)
607+
assert event_log[1].method_name == "init"
608+
609+
assert isinstance(event_log[2], MethodExecutionFinishedEvent)
610+
assert event_log[2].method_name == "init"
611+
612+
assert isinstance(event_log[3], MethodExecutionStartedEvent)
613+
assert event_log[3].method_name == "process"
614+
615+
assert isinstance(event_log[4], MethodExecutionFinishedEvent)
616+
assert event_log[4].method_name == "process"
617+
618+
assert isinstance(event_log[5], FlowFinishedEvent)
619+
assert event_log[5].flow_name == "StatelessFlow"
620+
assert (
621+
event_log[5].result
622+
== "Deeds will not be less valiant because they are unpraised."
623+
)
624+
assert isinstance(event_log[5].timestamp, datetime)

0 commit comments

Comments
 (0)