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.ConversationMessageStreamStartedEvent(name=None, event_id=<factory>, timestamp=<factory>, message=<factory>)#

This event is recorded whenever aa new message start being streamed to the conversation

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
message (Message) –

message: Message#: The message that is being appended to the 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.ConversationMessageStreamChunkEvent(name=None, event_id=<factory>, timestamp=<factory>, chunk=<factory>)#

This event is recorded whenever a message is being streamed and a delta is added to the conversation.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
chunk (str) –

chunk: str#: The chunk that is being appended to the 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.ConversationMessageStreamEndedEvent(name=None, event_id=<factory>, timestamp=<factory>, message=<factory>)#

This event is recorded whenever aa new message start being streamed to the conversation

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
message (Message) –

message: Message#: The message that is being appended to the 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.ConversationExecutionStartedEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, conversation=<factory>)#

This event is recorded whenever a conversation is started.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
span (Span | None) –
conversation (Conversation) –

conversation: 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.ConversationExecutionFinishedEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, conversation=<factory>, execution_status=<factory>)#

This event is recorded whenever a conversation is started.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
span (Span | None) –
conversation (Conversation) –
execution_status (ExecutionStatus) –

conversation: Conversation#

execution_status: ExecutionStatus#

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.ToolExecutionStartEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, tool=<factory>, tool_request=<factory>)#

This event is recorded whenever a tool is executed.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
span (Span | None) –
tool (Tool) –
tool_request (ToolRequest) –

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

tool: Tool#: Tool that triggered this event

tool_request: ToolRequest#: ToolRequest object containing the id of the tool request made as well as the tool call’s inputs

class wayflowcore.events.event.ToolExecutionResultEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, tool=<factory>, tool_result=<factory>)#

This event is recorded whenever a tool has finished execution.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
span (Span | None) –
tool (Tool) –
tool_result (ToolResult) –

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

tool: Tool#: Tool that triggered this event

tool_result: ToolResult#: ToolResult object containing the id of the tool request made and the tool call’s output

class wayflowcore.events.event.ToolConfirmationRequestStartEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, tool=<factory>, tool_request=<factory>)#

This event is recorded whenever a tool confirmation is required.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
span (Span | None) –
tool (Tool) –
tool_request (ToolRequest) –

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

tool: Tool#: Tool that triggered this event

tool_request: ToolRequest#: ToolRequest object containing the id of the tool request made as well as the tool call’s inputs

class wayflowcore.events.event.ToolConfirmationRequestEndEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, tool=<factory>, tool_request=<factory>)#

This event is recorded whenever a tool confirmation has been handled.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
span (Span | None) –
tool (Tool) –
tool_request (ToolRequest) –

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

tool: Tool#: Tool that triggered this event

tool_request: ToolRequest#: ToolRequest object containing the id of the tool request made as well as the tool call’s inputs

class wayflowcore.events.event.StepInvocationStartEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, step=<factory>, inputs=<factory>)#

This event is recorded whenever a step is invoked.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
span (Span | None) –
step (Step) –
inputs (Dict[str, Any]) –

inputs: Dict[str, Any]#: Inputs to the step invocation

step: Step#: Step that triggered the event

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.StepInvocationResultEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, step=<factory>, step_result=<factory>)#

This event is recorded whenever a step invocation has finished.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
span (Span | None) –
step (Step) –
step_result (StepResult) –

step: Step#: Step that triggered the event

step_result: StepResult#: Result of the step invocation

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.ContextProviderExecutionRequestEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, context_provider=<factory>)#

This event is recorded whenever a context provider is called.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
span (Span | None) –
context_provider (ContextProvider) –

context_provider: ContextProvider#: Used to pass contextual information to assistants

class wayflowcore.events.event.ContextProviderExecutionResultEvent(name=None, event_id=<factory>, timestamp=<factory>, span=<factory>, context_provider=<factory>, output=<factory>)#

This event is recorded whenever a context provider has returned a result.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
span (Span | None) –
context_provider (ContextProvider) –
output (Any) –

context_provider: ContextProvider#: Used to pass contextual information to assistants

output: Any#: Result of the context provider call

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.FlowExecutionIterationStartedEvent(name=None, event_id=<factory>, timestamp=<factory>, execution_state=<factory>)#

This event is recorded whenever an iteration of a flow has started executing.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
execution_state (FlowConversationExecutionState) –

execution_state: FlowConversationExecutionState#: State of a flow (doesn’t contain messages)

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.FlowExecutionIterationFinishedEvent(name=None, event_id=<factory>, timestamp=<factory>, execution_state=<factory>)#

This event is recorded whenever an iteration of a flow has finished executing.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
execution_state (FlowConversationExecutionState) –

execution_state: FlowConversationExecutionState#: State of a flow (doesn’t contain messages)

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.AgentExecutionIterationStartedEvent(name=None, event_id=<factory>, timestamp=<factory>, execution_state=<factory>)#

This event is recorded whenever an iteration of an agent has started executing.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
execution_state (AgentConversationExecutionState) –

execution_state: AgentConversationExecutionState#: State of an agent (doesn’t contain messages)

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.AgentExecutionIterationFinishedEvent(name=None, event_id=<factory>, timestamp=<factory>, execution_state=<factory>)#

This event is recorded whenever an iteration of an agent has finished executing.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
execution_state (AgentConversationExecutionState) –

execution_state: AgentConversationExecutionState#: State of an agent (doesn’t contain messages)

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.ExceptionRaisedEvent(name=None, event_id=<factory>, timestamp=<factory>, exception=<factory>)#

This event is recorded whenever an exception occurs.

Parameters:

name (str | None) –
event_id (str) –
timestamp (int) –
exception (Exception) –

exception: Exception#: Exception that was thrown during execution

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