Events

This page presents all APIs and classes related to events in WayFlow.

Events

class wayflowcore.events.event.Event(name=None, event_id=<factory>, timestamp=<factory>)

Base Event class. It contains information relevant to all events.

Parameters:

• name (str | None) –

• event_id (str) –

• timestamp (int) –

event_id: str

A unique identifier for the event

name: Optional[str] = None

The optional name of the event

timestamp: int

The timestamp of when the event occurred

to_tracing_info(mask_sensitive_information=True)

Return a serialized version of the event’s information to be used for tracing.

Parameters:

mask_sensitive_information (bool) – Whether to mask potentially sensitive information from the span and its events

Return type:

Dict[str, Any]

Returns:

A dictionary containing the serialized information of this event

class wayflowcore.events.event.LlmGenerationRequestEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, llm=<factory>, prompt=<factory>)

This event is recorded when the llm receives a generation request.

Parameters:

• name (str | None) –

• event_id (str) –

• timestamp (int) –

• span (Span | None) –

• llm (LlmModel) –

• prompt (Prompt) –

llm: LlmModel

The model that receives the generation request

prompt: Prompt

The prompt for the generation request

to_tracing_info(mask_sensitive_information=True)

Return a serialized version of the event’s information to be used for tracing.

Parameters:

mask_sensitive_information (bool) – Whether to mask potentially sensitive information from the span and its events

Return type:

Dict[str, Any]

Returns:

A dictionary containing the serialized information of this event

class wayflowcore.events.event.LlmGenerationResponseEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, llm=<factory>, completion=<factory>)

This event is recorded when the llm generates a response.

Parameters:

• name (str | None) –

• event_id (str) –

• timestamp (int) –

• span (Span | None) –

• llm (LlmModel) –

• completion (LlmCompletion) –

completion: LlmCompletion

The completion generated by the llm model

llm: LlmModel

The model that was used for generating the response

to_tracing_info(mask_sensitive_information=True)

Return a serialized version of the event’s information to be used for tracing.

Parameters:

mask_sensitive_information (bool) – Whether to mask potentially sensitive information from the span and its events

Return type:

Dict[str, Any]

Returns:

A dictionary containing the serialized information of this event

class wayflowcore.events.event.ConversationalComponentExecutionStartedEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, conversational_component=<factory>)

This event is recorded when the agent/flow execution has started.

Parameters:

• name (str | None) –

• event_id (str) –

• timestamp (int) –

• span (Span | None) –

• conversational_component (ConversationalComponent) –

conversational_component: ConversationalComponent

Agent/flow that started the execution of the conversation

class wayflowcore.events.event.ConversationalComponentExecutionFinishedEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, conversational_component=<factory>, execution_status=<factory>)

This event is recorded when the agent/flow execution has ended.

Parameters:

• name (str | None) –

• event_id (str) –

• timestamp (int) –

• span (Span | None) –

• conversational_component (ConversationalComponent) –

• execution_status (ExecutionStatus) –

conversational_component: ConversationalComponent

Agent/flow that started the execution of the conversation

execution_status: ExecutionStatus

Indicates the status of the conversation (finished, yielding, etc.)

to_tracing_info(mask_sensitive_information=True)

Return a serialized version of the event’s information to be used for tracing.

Parameters:

mask_sensitive_information (bool) – Whether to mask potentially sensitive information from the span and its events

Return type:

Dict[str, Any]

Returns:

A dictionary containing the serialized information of this event

class wayflowcore.events.event.ConversationCreatedEvent(name=None, event_id=<factory>, timestamp=<factory>, conversational_component=<factory>, inputs=<factory>, messages=<factory>, conversation_id=<factory>, nesting_level=<factory>)

This event is recorded whenever a new conversation with an agent or a flow was created.

Parameters:

• name (str | None) –

• event_id (str) –

• timestamp (int) –

• conversational_component (ConversationalComponent) –

• inputs (Dict[str, Any] | None) –

• messages (MessageList | List[Message] | None) –

• conversation_id (str | None) –

• nesting_level (int | None) –

conversation_id: Optional[str]

Unique id of the conversation

conversational_component: ConversationalComponent

Agent/flow from which the conversation was created

inputs: Optional[Dict[str, Any]]

A dictionary of inputs where the keys are the input names and the values are the actual inputs of the conversation.

messages: Union[MessageList, List[Message], None]

List of messages to populate the conversation

nesting_level: Optional[int]

The level of nested sub-conversations or tasks within the main conversation.

to_tracing_info(mask_sensitive_information=True)

Return a serialized version of the event’s information to be used for tracing.

Parameters:

mask_sensitive_information (bool) – Whether to mask potentially sensitive information from the span and its events

Return type:

Dict[str, Any]

Returns:

A dictionary containing the serialized information of this event

class wayflowcore.events.event.ConversationMessageAddedEvent(name=None, event_id=<factory>, timestamp=<factory>, message=<factory>, streamed=<factory>)

This event is recorded whenever a new message was added to the conversation.

Parameters:

• name (str | None) –

• event_id (str) –

• timestamp (int) –

• message (Message) –

• streamed (bool) –

message: Message

The message that is being appended to the conversation

streamed: bool

Whether the message was streamed or not.

to_tracing_info(mask_sensitive_information=True)

Return a serialized version of the event’s information to be used for tracing.

Parameters:

mask_sensitive_information (bool) – Whether to mask potentially sensitive information from the span and its events

Return type:

Dict[str, Any]

Returns:

A dictionary containing the serialized information of this event

class wayflowcore.events.event.AgentNextActionDecisionStartEvent(name=None, event_id=<factory>, timestamp=<factory>, execution_state=<factory>)

This event is recorded at the start of the agent taking a decision on what to do next.

Parameters:

• name (str | None) –

• event_id (str) –

• timestamp (int) –

• execution_state (AgentConversationExecutionState) –

execution_state: AgentConversationExecutionState

to_tracing_info(mask_sensitive_information=True)

Return a serialized version of the event’s information to be used for tracing.

Parameters:

mask_sensitive_information (bool) – Whether to mask potentially sensitive information from the span and its events

Return type:

Dict[str, Any]

Returns:

A dictionary containing the serialized information of this event

class wayflowcore.events.event.AgentDecidedNextActionEvent(name=None, event_id=<factory>, timestamp=<factory>, should_yield=<factory>, execution_state=<factory>)

This event is recorded whenever the agent decided what to do next.

Parameters:

• name (str | None) –

• event_id (str) –

• timestamp (int) –

• should_yield (bool) –

• execution_state (AgentConversationExecutionState) –

execution_state: AgentConversationExecutionState

should_yield: bool

to_tracing_info(mask_sensitive_information=True)

Return a serialized version of the event’s information to be used for tracing.

Parameters:

mask_sensitive_information (bool) – Whether to mask potentially sensitive information from the span and its events

Return type:

Dict[str, Any]

Returns:

A dictionary containing the serialized information of this event

Event Listeners

class wayflowcore.events.eventlistener.EventListener

Base class for EventListeners.

An EventListener is a Callable that gets called every time an Event is recorded.

class wayflowcore.events.eventlistener.GenericEventListener(event_classes, function)

Special implementation of EventListener that calls a function every time an event whose type is among the given event classes is recorded.

Parameters:

• event_classes (List[Type[Event]]) – The list of Event classes that will trigger the function call when recorded

• function (Callable[[Event], None]) – The function to be called. It must have a single argument of type Event, that will contain the event being recorded, it is not supposed to have any return value.

wayflowcore.events.eventlistener.register_event_listeners(event_listeners)

Register event listeners to be called for all events recorded inside the context manager’s scope

Parameters:

event_listeners (List[EventListener]) – The list of EventListeners to be called

Return type:

Iterator[None]

wayflowcore.events.eventlistener.get_event_listeners()

Get all the event listeners that are registered in the current context

Return type:

List[EventListener]

Returns:

The list of EventListeners registered in the current context

wayflowcore.events.eventlistener.record_event(event)

Record the event and execute all the event listeners that are registered in the current context

Parameters:

event (Event) – The Event being recorded

Return type:

None