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="CurrentTime",
...     description="Tool that returns time",
...     parameters={},
...     output={"type": "string"},
...     func=current_time,
... )
>>>
>>> context_provider = ToolContextProvider(tool=tool, output_name="time_output_io")
>>> display_first_step_time = OutputMessageStep(message_template="{{ time_output_io }}")
>>>
>>> flow = Flow(
...     begin_step=display_first_step_time,
...     steps={
...         "display_first_step_time": display_first_step_time,
...         "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]