Context Providers#

Base classes#

class wayflowcore.contextproviders.contextprovider.ContextProvider(name=None, description=None, id=None, __metadata_info__=None)#

Context providers are callable components that are used to provide dynamic contextual information to WayFlow assistants. They are useful to connect external datasources to an assistant.

Parameters:

name (str | None) –
description (str | None) –
id (str | None) –
__metadata_info__ (Dict[str, Any] | None) –

async call_async(conversation)#

Default sync callable of the context provider

Parameters:: conversation (Conversation) –
Return type:: Any

abstract get_output_descriptors()#

Return type:: List[Property]

classmethod get_static_configuration_descriptors()#

Returns a dictionary in which the keys are the names of the configuration items and the values are the expected type.

Return type:: Dict[str, type]

property output_descriptors: List[Property]#

Available Context Providers#

class wayflowcore.contextproviders.toolcontextprovider.ToolContextProvider(tool, output_name=None, name=None, description=None, id=None, __metadata_info__=None)#

Context provider to wrap a tool execution.

Parameters:

tool (ServerTool) – The tool to be called as part of this context provider
output_name (str | None) – The name of the output of this context provider. If None is given, the name of the tool followed by _output is used.
name (str | None) – The name of the context provider
description (str | None) –
id (str | None) –
__metadata_info__ (Dict[str, Any] | None) –

Examples

>>> from time import time
>>> from wayflowcore.contextproviders import ToolContextProvider
>>> from wayflowcore.flow import Flow
>>> from wayflowcore.steps import OutputMessageStep
>>> from wayflowcore.tools import ServerTool
>>>
>>> def current_time() -> str:
...     from time import time
...     return str(time())
>>>
>>> tool = ServerTool(
...     name="Current time",
...     description="Tool that returns time",
...     parameters={},
...     output={"type": "string"},
...     func=current_time,
... )
>>>
>>> context_provider = ToolContextProvider(tool=tool, output_name="time_output_io")
>>>
>>> flow = Flow(
...     begin_step_name="display_first_step_time",
...     steps={
...         "display_first_step_time": OutputMessageStep(
...             message_template="{{ time_output_io }}",
...         ),
...         "display_second_step_time": OutputMessageStep(
...             message_template="{{ time_output_io }}",
...         ),
...     },
...     transitions={
...         "display_first_step_time": ["display_second_step_time"],
...         "display_second_step_time": [None],
...     },
...     context_providers=[context_provider],
... )

async call_async(conversation)#

Default sync callable of the context provider

Parameters:: conversation (Conversation) –
Return type:: Any

get_output_descriptors()#

Return type:: List[Property]

classmethod get_static_configuration_descriptors()#

Returns a dictionary in which the keys are the names of the configuration items and the values are the expected type.

Return type:: Dict[str, type]

class wayflowcore.contextproviders.flowcontextprovider.FlowContextProvider(flow, flow_output_names=None, name=None, description=None, id=None, __metadata_info__=None)#

Context provider that wraps and executes a flow.

Parameters:

flow (Flow) – The Flow to be used as context. It must not require inputs and must not yield.
flow_output_names (List[str] | None) – List of output names for the context provider, to be used in the calling flow’s I/O system. This list must contain unique names. These names, if specified, must be a subset of the context flow’s outputs. If not specified, defaults to all outputs from all steps of the provided flow
name (str | None) – The name of the context provider
description (str | None) –
id (str | None) –
__metadata_info__ (Dict[str, Any] | None) –

Examples

>>> from wayflowcore.contextproviders import FlowContextProvider
>>> from wayflowcore.flowhelpers import create_single_step_flow
>>> from wayflowcore.steps import OutputMessageStep
>>> contextual_flow = create_single_step_flow(OutputMessageStep(
...     message_template="The current time is 2pm.",
...     output_mapping={OutputMessageStep.OUTPUT: "time_output_io"},
... ))
>>> context_provider = FlowContextProvider(contextual_flow, flow_output_names=["time_output_io"])
>>> flow = create_single_step_flow(
...     OutputMessageStep("Last time message: {{time_output_io}}"),
...     context_providers=[context_provider]
... )
>>> conversation = flow.start_conversation()
>>> execution_status = conversation.execute()
>>> last_message = conversation.get_last_message()
>>> # print(last_message.content)
>>> # Last time message: The current time is 2pm.

async call_async(conversation)#

Default sync callable of the context provider

Parameters:: conversation (Conversation) –
Return type:: Any

get_output_descriptors()#

Return type:: List[Property]

classmethod get_static_configuration_descriptors()#

Returns a dictionary in which the keys are the names of the configuration items and the values are the expected type.

Return type:: Dict[str, type]

class wayflowcore.contextproviders.chathistorycontextprovider.ChatHistoryContextProvider(n=10, which_messages=MessageSlice.LAST_MESSAGES, offset=0, message_types=(<MessageType.USER: 'USER'>, <MessageType.AGENT: 'AGENT'>), output_template=None, output_name=None, id=None, name=None, description=None, __metadata_info__=None)#

Context provider to get messages from the messages list e.g. last 4 messages and return it as output. The following parameters will be passed to the underlying GetChatHistoryStep.

Parameters:

n (int) – Number of messages to retrieve.
which_messages (MessageSlice) – Name of the strategy to use to collect messages. Either last_messages or first_messages
offset (int) – Number of messages to ignore in the given order. Must be a non-negative integer.
message_types (Tuple[MessageType, ...] | None) – Tuple specification of the MessageType to keep from the list of messages. If None, collects all the messages from the history.
output_template (str | None) – Template to format the chat history. If string, will format the chat history messages into the template using jinja2 syntax. Unlike GetChatHistoryStep, the template should only contain the chat_history jinja variable. If None, will use the default one from GetChatHistorySteps.
output_name (str | None) – Name of the output for the context provider, to be used in the calling flow’s I/O system. This is not a parameter of the underlying GetChatHistoryStep If not given, the output name is chat_history.
name (str | None) – The name of the context provider
id (str | None) –
description (str | None) –
__metadata_info__ (Dict[str, Any] | None) –

Examples

>>> from wayflowcore.contextproviders import ChatHistoryContextProvider
>>> from wayflowcore.flowhelpers import create_single_step_flow
>>> from wayflowcore.messagelist import MessageList, Message
>>> from wayflowcore.steps import OutputMessageStep
>>> context_provider = ChatHistoryContextProvider(
...     n=2, # will retrieve the last 5 messages
...     output_name="history",
... )
>>> flow = create_single_step_flow(
...     OutputMessageStep("Chat history number: {{history}}"),
...     context_providers=[context_provider]
... )
>>> message_list = MessageList([Message(f"Message {i+1}") for i in range(5)])
>>> conversation = flow.start_conversation(messages=message_list)
>>> execution_status = conversation.execute()
>>> last_message = conversation.get_last_message()
>>> # print(last_message.content)
>>> # Chat history number:
>>> # USER >> Message 4
>>> # USER >> Message 5

classmethod get_static_configuration_descriptors()#

Returns a dictionary in which the keys are the names of the configuration items and the values are the expected type.

Return type:: Dict[str, type]

property message_types: Tuple[MessageType, ...] | None#

property n: int#

property offset: int#

property output_template: str | None#

property which_messages: MessageSlice#