Conversations#

Conversation & Messages#

class wayflowcore.conversation.Conversation(component: wayflowcore.conversationalcomponent.ConversationalComponent, state: 'ConversationExecutionState', inputs: Dict[str, Any], message_list: wayflowcore.messagelist.MessageList, status: Optional[wayflowcore.executors.executionstatus.ExecutionStatus], conversation_id: str = '', *, id: str = <factory>, __metadata_info__: Dict[str, Any] = <factory>, name: str = '', description: Optional[str] = None)#

Parameters:

component (ConversationalComponent) –
state (ConversationExecutionState) –
inputs (Dict[str, Any]) –
message_list (MessageList) –
status (ExecutionStatus | None) –
conversation_id (str) –
id (str) –
__metadata_info__ (Dict[str, Any]) –
name (str) –
description (str | None) –

append_agent_message(agent_input, is_error=False)#

Append a new message object of type MessageType.AGENT to the messages list.

Parameters:

agent_input (str) – message to append.
is_error (bool) –

Return type:

None

append_message(message)#

Append a message to the messages list of this Conversation object.

Parameters:: message (Message) – message to append.
Return type:: None

append_tool_result(tool_result)#

Append a new message object of type MessageType.TOOL_RESULT to the messages list.

Parameters:: tool_result (ToolResult) – message to append.
Return type:: None

append_user_message(user_input)#

Append a new message object of type MessageType.USER to the messages list.

Parameters:: user_input (str) – message to append.
Return type:: None

component: ConversationalComponent#

conversation_id: str = ''#

abstract property current_step_name: str#

execute(execution_interrupts=None)#

Execute the conversation and get its ExecutionStatus based on the outcome.

The Execution status is returned by the Assistant and indicates if the assistant yielded, finished the conversation.

Parameters:: execution_interrupts (Sequence[ExecutionInterrupt] | None) –
Return type:: ExecutionStatus

async execute_async(execution_interrupts=None)#

Execute the conversation and get its ExecutionStatus based on the outcome.

The Execution status is returned by the Assistant and indicates if the assistant yielded, finished the conversation.

Parameters:: execution_interrupts (Sequence[ExecutionInterrupt] | None) –
Return type:: ExecutionStatus

get_last_message()#

Get the last message from the messages List.

Return type:: Message | None

get_messages()#

Return all Message objects of the messages list in a python list.

Return type:: List[Message]

inputs: Dict[str, Any]#

message_list: MessageList#

property plan: ExecutionPlan | None#

state: ConversationExecutionState#

status: ExecutionStatus | None#

token_usage: TokenUsage#

class wayflowcore.messagelist.Message(content='', message_type=None, tool_requests=None, tool_result=None, is_error=False, display_only=False, sender=None, recipients=None, time_created=None, time_updated=None, contents=None, role=None, __metadata_info__=None)#

Messages are an exchange medium between the user, LLM agent, and controller logic. This helps determining who provided what information.

Parameters:

content (str) – Content of the message.
message_type (MessageType | None) – A message type out of the following ones: SYSTEM, AGENT, USER, THOUGHT, INTERNAL, TOOL_REQUEST, TOOL_RESULT.
tool_requests (List[wayflowcore.tools.tools.ToolRequest] | None) – A list of ToolRequest objects representing the tools invoked as part of this message. Each request includes the tool’s name, arguments, and a unique identifier.
tool_result (wayflowcore.tools.tools.ToolResult | None) – A ToolResult object representing the outcome of a tool invocation. It includes the returned content and a reference to the related tool request ID.
display_only (bool) – If True, the message is excluded from any context. Its only purpose is to be displayed in the chat UI (e.g debugging message)
sender (str | None) – Sender of the message in str format.
recipients (Set[str]) – Recipients of the message in str format.
time_created (datetime.datetime) – Creation timestamp of the message.
time_updated (datetime.datetime) – Update timestamp of the message.
contents (List[wayflowcore.messagelist.MessageContent]) – Message content. Is a list of chunks with potentially different types
role (Literal['user', 'assistant', 'system']) – Role of the sender of the message. Can be user, system or assistant
is_error (bool) –
__metadata_info__ (Dict[str, Any]) –

property content: str#: Text content getter

contents: List[MessageContent]#

copy(**kwargs)#

Create a copy of the given message.

Parameters:: kwargs (Any) –
Return type:: Message

display_only: bool = False#

property is_error: bool#

property message_type: MessageType#: Getter to guarantee backward compatibility

recipients: Set[str]#

role: Literal['user', 'assistant', 'system']#

sender: str | None = None#

time_created: datetime#

time_updated: datetime#

tool_requests: List[ToolRequest] | None = None#

tool_result: ToolResult | None = None#

class wayflowcore.messagelist.MessageType(value)#

Type of messages

AGENT = 'AGENT'#

DISPLAY_ONLY = 'DISPLAY_ONLY'#

ERROR = 'ERROR'#

INTERNAL = 'INTERNAL'#

SYSTEM = 'SYSTEM'#

THOUGHT = 'THOUGHT'#

TOOL_REQUEST = 'TOOL_REQUEST'#

TOOL_RESULT = 'TOOL_RESULT'#

USER = 'USER'#

class wayflowcore.messagelist.TextContent(content='')

Represents the content of a text message.

Parameters:: content (str) –

content

The textual content of the message.

Type:: str

type

Identifier for the text content type.

Type:: Literal[“text”]

content: str = ''

type: ClassVar[Literal['text']] = 'text'

class wayflowcore.messagelist.ImageContent(base64_content)

Represents the content of an image message, storing image data as a base64-encoded string.

Parameters:: base64_content (str) –

base64_content

A base64-encoded string representing the image data.

Type:: str

type

Identifier for the image content type.

Type:: str

Examples

>>> import requests
>>> from wayflowcore.messagelist import Message, TextContent, ImageContent
>>> from wayflowcore.models import Prompt
>>> # Download the Oracle logo as bytes
>>> url = "https://www.oracle.com/a/ocom/img/oracle-logo.png"
>>> response = requests.get(url)
>>> img_content = ImageContent.from_bytes(response.content, format="png")
>>> prompt = Prompt(messages=[Message(contents = [TextContent("Which company's logo is this?") , img_content])])
>>> completion = multimodal_llm.generate(prompt)
>>> # LlmCompletion(message=Message(content="That is the logo for **Oracle Corporation**."))

base64_content: str

classmethod from_bytes(bytes_content, format)

Parameters:

bytes_content (bytes) –
format (str) –

Return type:

ImageContent

type: ClassVar[str] = 'image'

class wayflowcore.messagelist.MessageList(messages=<factory>, *, id=<factory>, __metadata_info__=<factory>)#

Object that carries a list of messages. We may only append to this object, not remove

Parameters:

messages (List[wayflowcore.messagelist.Message]) – list of messages to start from.
id (str) –
__metadata_info__ (Dict[str, Any]) –

append_agent_message(agent_input, is_error=False)#

Append a new message object of type MessageType.AGENT to the messages list.

Parameters:

agent_input (str) – message to append.
is_error (bool) –

Return type:

None

append_message(message)#

Add a message to a message list.

Parameters:: message (Message) – Message to append to the message list.
Return type:: None

append_tool_result(tool_result)#

Append a new message object of type MessageType.TOOL_RESULT to the messages list.

Parameters:: tool_result (ToolResult) – message to append.
Return type:: None

append_user_message(user_input)#

Append a new message object of type MessageType.USER to the messages list.

Parameters:: user_input (str) – message to append.
Return type:: None

copy()#

Create a copy of the given message list.

Return type:: MessageList

get_last_message()#

Returns the last message from the conversation

Return type:: Message | None

get_messages()#

Returns a copy of the messages list

Return type:: List[Message]

messages: List[Message]#

ExecutionStatus#

class wayflowcore.executors.executionstatus.ExecutionStatus(*, id=<factory>, __metadata_info__=<factory>)#

Execution status returned by the Assistant. This indicates if the assistant yielded, finished the conversation, …

Parameters:

id (str) –
__metadata_info__ (Dict[str, Any]) –

class wayflowcore.executors.executionstatus.UserMessageRequestStatus(*, id=<factory>, __metadata_info__=<factory>)#

Execution status for when the assistant answered and will be waiting for the next user input

Parameters:

id (str) –
__metadata_info__ (Dict[str, Any]) –

class wayflowcore.executors.executionstatus.FinishedStatus(output_values, complete_step_name=None, *, id=<factory>, __metadata_info__=<factory>)#

Execution status for when the conversation is finished. Contains the outputs of the conversation

Parameters:

output_values (Dict[str, Any]) –
complete_step_name (str | None) –
id (str) –
__metadata_info__ (Dict[str, Any]) –

complete_step_name: str | None = None#: The name of the last step reached if the flow returning this execution status transitioned to a CompleteStep, otherwise None.

output_values: Dict[str, Any]#: The outputs produced by the agent or flow returning this execution status.

class wayflowcore.executors.executionstatus.ToolRequestStatus(tool_requests, *, id=<factory>, __metadata_info__=<factory>)#

Execution status for when the assistant is asking the user to call a tool and send back its result

Parameters:

tool_requests (List[ToolRequest]) –
id (str) –
__metadata_info__ (Dict[str, Any]) –

tool_requests: List[ToolRequest]#

Conversation#

Base class for conversations. Can manipulate a conversation object, and can be serialized/deserialized.

class wayflowcore.conversationalcomponent.ConversationalComponent(name, description, input_descriptors, output_descriptors, runner, conversation_class, id=None, __metadata_info__=None)#

Parameters:

name (str) –
description (str | None) –
input_descriptors (List[Property]) –
output_descriptors (List[Property]) –
runner (type['ConversationExecutor']) –
conversation_class (Any) –
id (str | None) –
__metadata_info__ (Dict[str, Any] | None) –

execute(conversation, execution_interrupts=None, _validate_same_component=True)#

Parameters:

conversation (Conversation) –
execution_interrupts (List[ExecutionInterrupt] | None) –
_validate_same_component (bool) –

Return type:

ExecutionStatus

property llms: List[LlmModel]#

abstract start_conversation(inputs=None, messages=None)#

Parameters:

inputs (Dict[str, Any] | None) –
messages (MessageList | List[Message] | None) –

Return type:

Conversation