WayFlow Reference Sheet#
This reference sheet provides a single-page overview of basic code snippets covering the core concepts used in WayFlow.
Each section includes links to additional tutorials and guides for deeper learning.
LLMs#
WayFlow Agents and Flows may require the use of Large Language Models (LLMs). This section shows how to initialize an LLM and perform quick tests.
Loading an LLM instance#
WayFlow supports several LLM API providers. For an overview of supported LLMs, see the guide How to Use LLMs from Different Providers.
Start by selecting an LLM from one of the available providers:
from wayflowcore.models import OCIGenAIModel
if __name__ == "__main__":
llm = OCIGenAIModel(
model_id="provider.model-id",
service_endpoint="https://url-to-service-endpoint.com",
compartment_id="compartment-id",
auth_type="API_KEY",
)
from wayflowcore.models import VllmModel
llm = VllmModel(
model_id="model-id",
host_port="VLLM_HOST_PORT",
)
from wayflowcore.models import OllamaModel
llm = OllamaModel(
model_id="model-id",
)
Read more about the LLMs support in the API reference.
Testing inference with LLMs#
Single prompt generation#
Use a simple PromptExecutionStep to test an LLM.
from wayflowcore.flowhelpers import create_single_step_flow, run_flow_and_return_outputs
from wayflowcore.steps import PromptExecutionStep
flow = create_single_step_flow(PromptExecutionStep(prompt_template="{{prompt}}", llm=llm))
prompt = "Write a simple Python function to sum two numbers"
response = run_flow_and_return_outputs(flow, {"prompt": prompt})[PromptExecutionStep.OUTPUT]
print(response) # Here's a simple Python function...
API Reference: PromptExecutionStep | Flow
Tip
Use the helper methods create_single_step_flow and run_flow_and_return_outputs for quick prototyping.
Parallel generation#
Add a MapStep to perform inference on a batch of inputs (parallel generation).
from wayflowcore.flowhelpers import create_single_step_flow, run_flow_and_return_outputs
from wayflowcore.property import ListProperty
from wayflowcore.steps import MapStep, PromptExecutionStep
flow = create_single_step_flow(
MapStep(
create_single_step_flow(PromptExecutionStep(prompt_template="{{prompt}}", llm=llm)),
parallel_execution=True,
unpack_input={"prompt": "."},
output_descriptors=[ListProperty(PromptExecutionStep.OUTPUT)],
)
)
NUM_RESPONSES = 3
prompt = "Write a simple Python function to sum two numbers"
prompt_batch = [prompt] * NUM_RESPONSES
response = run_flow_and_return_outputs(flow, {MapStep.ITERATED_INPUT: prompt_batch})[
PromptExecutionStep.OUTPUT
]
print(*response, sep=f"\n\n{'-'*30}\n\n")
API Reference: ListProperty | PromptExecutionStep | MapStep | Flow
Note
Note the use of a ListProperty to specify the output of the MapStep.
Structured generation#
WayFlow supports structured generation (such as controlling LLM outputs to conform to specific formats, schemas, or patterns, for example, Json Schema).
Structured generation can be achieved by specifying the output descriptors of the PromptExecutionStep.
from wayflowcore.flowhelpers import create_single_step_flow, run_flow_and_return_outputs
from wayflowcore.property import ListProperty, StringProperty
from wayflowcore.steps import PromptExecutionStep
product_name = StringProperty(
name="product_name",
description="name of the product",
default_value="",
)
product_description = StringProperty(
name="product_description",
description="One sentence description of the product.",
default_value="",
)
functionalities = ListProperty(
name="functionalities",
description="List of functionalities of the product",
item_type=StringProperty("functionality"),
default_value=[],
)
flow = create_single_step_flow(
PromptExecutionStep(
prompt_template="{{prompt}}",
llm=llm,
output_descriptors=[product_name, product_description, functionalities],
)
)
prompt = "Create a simple product for students."
response = run_flow_and_return_outputs(flow, {"prompt": prompt})
print(
f"Name: {response['product_name']}\n"
f"Description: {response['product_description']}\n"
f"Functionalities:\n{response['functionalities']}"
)
API Reference: StringProperty | PromptExecutionStep | Flow
Read the guide on How to Perform Structured Generation for more information.
Tools#
Tools are essential for building powerful Agents and Flows. WayFlow supports the use of ServerTool (which can be simply built with the tool decorator), the RemoteTool as well as the ClientTool.
Figure: The different tools in WayFlow.
Creating a simple tool#
The simplest way to create a tool in WayFlow is by using the tool decorator, which creates a ServerTool (see definition in the glossary).
from datetime import datetime
from typing import Annotated
from wayflowcore.tools import tool
@tool
def days_between_dates(
date1: Annotated[str, "First date in 'dd/mm/yyyy' format."],
date2: Annotated[str, "Second date in 'dd/mm/yyyy' format."],
) -> Annotated[int, "Absolute difference in days between the two dates."]:
"""
Calculate the absolute difference in days between two dates.
"""
return abs((datetime.strptime(date2, "%d/%m/%Y") - datetime.strptime(date1, "%d/%m/%Y")).days)
# days_between_dates is not a callable anymore, it is a `ServerTool`
print(days_between_dates.func("01/01/2020", "31/12/2020")) # 365
API Reference: tool
For more information, read the guide on How to Build Assistants with Tools or read the API reference to learn about the available types of tools in WayFlow.
Creating a stateful tool#
To build stateful tools, simply use the tool helper as a wrapper to the method of an instantiated class.
from wayflowcore.tools import tool
class Counter:
def __init__(self):
self.value = 0
def increment(self) -> str:
"""Increment the counter"""
self.value += 1
return f"The updated count is {self.value}"
counter = Counter()
counter_tool = tool("increment_counter", counter.increment)
print(counter_tool.func()) # The updated count is 1
API Reference: tool
Creating and using a Client tool#
Use the ClientTool to create tools that are meant to be executed on the client side (see definition in the glossary).
from datetime import datetime
from typing import Any
from wayflowcore.executors.executionstatus import ToolRequestStatus
from wayflowcore.flowhelpers import create_single_step_flow
from wayflowcore.steps import ToolExecutionStep
from wayflowcore.tools import ClientTool, ToolRequest, ToolResult
def _days_between_dates(date1: str, date2: str) -> int:
return abs((datetime.strptime(date2, "%d/%m/%Y") - datetime.strptime(date1, "%d/%m/%Y")).days)
def execute_client_tool_from_tool_request(tool_request: ToolRequest) -> Any:
if tool_request.name == "days_between_dates":
return _days_between_dates(**tool_request.args)
else:
raise ValueError(f"Tool name {tool_request.name} is not recognized")
days_client_tools = ClientTool(
name="days_between_dates",
description="Calculate the absolute difference in days between two dates.",
parameters={
"date1": {
"description": "First date in 'dd/mm/yyyy' format.",
"type": "string",
},
"date2": {
"description": "Second date in 'dd/mm/yyyy' format.",
"type": "string",
},
},
output={"type": "string", "description": "Absolute difference in days between the two dates."},
)
flow = create_single_step_flow(ToolExecutionStep(days_client_tools))
conversation = flow.start_conversation({"date1": "01/01/2020", "date2": "31/12/2020"})
status = conversation.execute()
assert isinstance(status, ToolRequestStatus)
tool_request = status.tool_requests[0]
tool_execution_content = execute_client_tool_from_tool_request(tool_request)
conversation.append_tool_result(ToolResult(tool_execution_content, tool_request.tool_request_id))
# conversation.execute() # continue the execution of the Flow
API Reference: ClientTool | ToolRequest | ToolResult | ToolExecutionStep | ToolRequestStatus | Flow
Learn more about tools by reading How to Build Assistants with Tools, and the Tools API reference.
Execution loop and statuses#
This section illustrates a basic execution loop for WayFlow assistants (Agents and Flows).
A new conversation is created.
The assistant is executed on the conversation.
Based on the status returned from the assistant execution: * The loop exits if the status is
FinishedStatus. * The user is prompted for input if the status isUserMessageRequestStatus. * AClientToolis executed if the status isToolRequestStatus.
The loop continues until the assistant returns a FinishedStatus.
from typing import Any
from wayflowcore.messagelist import Message, MessageType
from wayflowcore.executors.executionstatus import (
FinishedStatus, UserMessageRequestStatus, ToolRequestStatus
)
from wayflowcore.tools import ToolRequest, ToolResult
def execute_client_tool_from_tool_request(tool_request: ToolRequest) -> Any:
if tool_request.name == "my_tool_name":
return _my_tool_callable(**tool_request.args)
else:
raise ValueError(f"Tool name {tool_request.name} is not recognized")
conversation_inputs = {}
conversation = assistant.start_conversation(inputs=conversation_inputs)
while True:
status = conversation.execute()
assistant_reply = conversation.get_last_message()
if assistant_reply:
print(f"Assistant>>> {assistant_reply.content}\n")
if isinstance(status, FinishedStatus):
print(f"Finished assistant execution. Output values:\n{status.output_values}",)
break
elif isinstance(status, UserMessageRequestStatus):
user_input = input("User>>> ")
print("\n")
conversation.append_user_message(user_input)
elif isinstance(status, ToolRequestStatus):
tool_request = status.tool_requests[0]
tool_result = execute_client_tool_from_tool_request(tool_request)
print(f"{tool_result!r}")
conversation.append_message(
Message(
tool_result=ToolResult(content=tool_result, tool_request_id=tool_request.tool_request_id),
message_type=MessageType.TOOL_RESULT,
)
)
else:
raise ValueError(f"Unsupported execution status: '{status}'")
Learn more about execution loops by reading the Execution Status API reference.
Agents#
WayFlow Agents are LLM-powered assistants that can interact with users, leverage external tools, and interact with other WayFlow assistants to take specific actions in order to solve user requests through conversational interfaces.
Creating a simple Agent#
Creating an Agent only requires an LLM and optional instructions to guide the agent behavior.
from wayflowcore.agent import Agent
agent = Agent(
llm, custom_instruction="You are a helpful assistant, please answer the user requests."
)
conversation = agent.start_conversation()
conversation.append_user_message(
"Please write a simple Python function to compute the sum of 2 numbers."
)
conversation.execute()
print(conversation.get_last_message().content)
# Here's a simple Python function that...
API Reference: Agent
Learn more about Agents by reading the tutorial Build a Simple Conversational Assistant with Agents and the Agent API reference.
Creating a Agent with tools#
You can simply equip Agents with tools using the tools attribute of the agent.
from datetime import datetime
from typing import Annotated
from wayflowcore.agent import Agent
from wayflowcore.tools import tool
@tool
def days_between_dates(
date1: Annotated[str, "First date in 'dd/mm/yyyy' format."],
date2: Annotated[str, "Second date in 'dd/mm/yyyy' format."],
) -> Annotated[int, "Absolute difference in days between the two dates."]:
"""
Calculate the absolute difference in days between two dates.
"""
return abs((datetime.strptime(date2, "%d/%m/%Y") - datetime.strptime(date1, "%d/%m/%Y")).days)
agent = Agent(llm, tools=[days_between_dates])
conversation = agent.start_conversation()
conversation.append_user_message("How many days are there between 01/01/2020 and 31/12/2020?")
conversation.execute()
print(conversation.get_last_message().content)
# There are 365 days between 01/01/2020 and 31/12/2020.
Flows#
WayFlow Flows are LLM-powered structured assistants composed of individual steps that are connected to form a coherent sequence of actions.
Each step in a Flow is designed to perform a specific function, similar to functions in programming.
Creating a simple Flow#
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.steps import OutputMessageStep
opening_step = OutputMessageStep("Opening session")
closing_step = OutputMessageStep('Closing session"')
flow = Flow(
begin_step_name="open_step",
steps={
"open_step": opening_step,
"close_step": closing_step,
},
control_flow_edges=[
ControlFlowEdge(source_step=opening_step, destination_step=closing_step),
ControlFlowEdge(source_step=closing_step, destination_step=None),
],
)
conversation = flow.start_conversation()
status = conversation.execute()
print(conversation.get_messages())
API Reference: ControlFlowEdge | Flow | OutputMessageStep
Learn more about Flows by reading the tutorial Build a Simple Fixed-Flow Assistant with Flows and the Flow API reference.
Creating Flow with explicit data connection#
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.dataconnection import DataFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.steps import OutputMessageStep
FAKE_PROCESSING_STEP = "processing_step"
OUTPUT_STEP = "output_step"
fake_processing_step = OutputMessageStep("Sucessfully processed username {{username}}")
output_step = OutputMessageStep('{{session_id}}: Received message "{{processing_message}}"')
flow = Flow(
begin_step_name=FAKE_PROCESSING_STEP,
steps={
FAKE_PROCESSING_STEP: fake_processing_step,
OUTPUT_STEP: output_step,
},
control_flow_edges=[
ControlFlowEdge(source_step=fake_processing_step, destination_step=output_step),
ControlFlowEdge(source_step=output_step, destination_step=None),
],
data_flow_edges=[
DataFlowEdge(
fake_processing_step, OutputMessageStep.OUTPUT, output_step, "processing_message"
)
],
)
conversation = flow.start_conversation(
inputs={"username": "Username#123", "session_id": "Session#456"}
)
status = conversation.execute()
last_message = conversation.get_last_message()
# last_message.content
# Session#456: Received message "Sucessfully processed username Username#123"
API Reference: ControlFlowEdge | DataFlowEdge | Flow | OutputMessageStep
Learn more about data flow edges in the Data Flow Edges API reference.
Executing a sub-flow to an iterable with the MapStep#
Applying or executing a sub-flow to an iterable is a common pattern and can be achieved in WayFlow using the MapStep.
from wayflowcore.flowhelpers import create_single_step_flow
from wayflowcore.property import AnyProperty
from wayflowcore.steps import MapStep, OutputMessageStep
sub_flow = create_single_step_flow(
OutputMessageStep(message_template="username={{user}}"), step_name="step"
)
step = MapStep(
flow=sub_flow,
unpack_input={"user": "."},
output_descriptors=[AnyProperty(name=OutputMessageStep.OUTPUT)],
)
iterable = ["a", "b"]
assistant = create_single_step_flow(step, "step")
conversation = assistant.start_conversation(inputs={MapStep.ITERATED_INPUT: iterable})
status = conversation.execute()
status.output_values # {'output_message': ['username=a', 'username=b']}
API Reference: Flow | MapStep | OutputMessageStep | AnyProperty
Learn more about MapSteps by reading How to Do Map and Reduce Operations in Flows and the MapStep API reference.
Adding conditional branching to Flows with the BranchingStep#
It is also frequent to want to transition in a Flow depending on a condition, and this can be achieved in WayFlow with the BranchingStep.
Figure: An example of a Flow using a BranchingStep.
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.steps import BranchingStep, OutputMessageStep
NEXT_STEP_NAME_IO = "$next_step_name"
branching_step = BranchingStep(
branch_name_mapping={"yes": "access_is_granted", "no": "access_is_denied"},
input_mapping={BranchingStep.NEXT_BRANCH_NAME: NEXT_STEP_NAME_IO},
)
access_granted_output_step = OutputMessageStep("Access granted. Press any key to continue...")
access_denied_output_step = OutputMessageStep("Access denied. Please exit the conversation.")
assistant = Flow(
begin_step_name="branching_step",
steps={
"branching_step": branching_step,
"access_granted_output_step": access_granted_output_step,
"access_denied_output_step": access_denied_output_step,
},
control_flow_edges=[
ControlFlowEdge(
branching_step, access_granted_output_step, source_branch="access_is_granted"
),
ControlFlowEdge(
branching_step, access_denied_output_step, source_branch="access_is_denied"
),
ControlFlowEdge(
branching_step, access_denied_output_step, source_branch=branching_step.BRANCH_DEFAULT
),
ControlFlowEdge(access_granted_output_step, None),
ControlFlowEdge(access_denied_output_step, None),
],
)
conversation = assistant.start_conversation(inputs={NEXT_STEP_NAME_IO: "yes"})
status = conversation.execute()
# conversation.get_last_message().content
# Access granted. Press any key to continue...
API Reference: Flow | BranchingStep | OutputMessageStep
Learn more about branching steps by reading How to Create Conditional Transitions in Flows and the BranchingStep API reference.
Adding tools to Flows with the ToolExecutionStep#
To use tools in Flows, use the ToolExecutionStep.
from typing import Annotated
from wayflowcore.flowhelpers import create_single_step_flow
from wayflowcore.steps import ToolExecutionStep
from wayflowcore.tools import tool
@tool
def compute_square_root(x: Annotated[float, "input number to square"]) -> float:
"""Computes the square root of a number"""
return x**0.5
step = ToolExecutionStep(tool=compute_square_root)
assistant = create_single_step_flow(step)
conversation = assistant.start_conversation(inputs={"x": 123456789.0})
status = conversation.execute()
print(status.output_values)
API Reference: Flow | ToolExecutionStep | ServerTool
Learn more about ToolexecutionSteps by reading How to Build Assistants with Tools
and the ToolexecutionSteps API reference.
Agentic composition patterns#
There are four majors agentic composition patterns supported in WayFlow:
Calling Agents in Flows
Calling Agents in Agents
Calling Flows in Agents
Calling Flows in Flows
Using an Agent in a Flow#
To use Agents in Flows, you can use the AgentExecutionStep.
from wayflowcore.agent import Agent
from wayflowcore.flowhelpers import create_single_step_flow
from wayflowcore.steps import AgentExecutionStep
code_agent = Agent(
llm=llm, custom_instruction="Please assist the user by answering their code-related questions"
)
flow = create_single_step_flow(AgentExecutionStep(code_agent))
conversation = flow.start_conversation()
status = conversation.execute()
print(conversation.get_last_message().content) # Hi! How can I help you?
conversation.append_user_message("Write a simple Python function to sum two numbers")
status = conversation.execute()
print(conversation.get_last_message().content) # Here's a simple Python function that ...
API Reference: Flow | Agent | AgentExecutionStep
Learn more about Agents in Flows by reading How to Use Agents in Flows and the Agent Execution Step API reference.
Warning
The AgentExecutionStep is currently in beta and may undergo significant changes.
The API and behaviour are not guaranteed to be stable and may change in future versions.
Multi-Level Agent Workflows#
WayFlow supports hierarchical multi-agent systems, by using expert Agents with a master / manager agent. This can be achieved by using a DescribedAgent.
from wayflowcore.agent import Agent
code_expert_agent = Agent(
llm=llm,
custom_instruction="Please assist the user by answering their code-related questions",
agent_id="code_expert_subagent",
name="code_expert",
description="Expert agent that can assist with code questions",
)
agent = Agent(
llm=llm,
custom_instruction="Please assist the user by answering their questions. Call the expert agents at your disposal when needed.",
agents=[code_expert_agent],
agent_id="main_agent",
)
conversation = agent.start_conversation()
status = conversation.execute()
print(conversation.get_last_message().content) # Hi! How can I help you?
conversation.append_user_message("Write a simple Python function to sum two numbers")
status = conversation.execute()
print(conversation.get_last_message().content) # Here's a simple Python function that ...
API Reference: Agent | DescribedAgent
Warning
The use of expert agents is currently in beta and may undergo significant changes. The API and behaviour are not guaranteed to be stable and may change in future versions.
Using Flows Within Agents#
To use Flows in Agents, use the DescribedFlow.
from wayflowcore.agent import Agent
from wayflowcore.flowhelpers import create_single_step_flow
from wayflowcore.steps import PromptExecutionStep
from wayflowcore.tools import DescribedFlow
code_reviewing_flow = create_single_step_flow(
PromptExecutionStep(
prompt_template="Here is some code:\n{{input_code}}\n---\nPlease check for potential bugs in the code",
llm=llm,
)
)
agent = Agent(
llm=llm,
custom_instruction="Please assist the user by answering their code questions. When creating code, use the code reviewing tool to ensure the code validity before answering the user.",
flows=[
DescribedFlow(
flow=code_reviewing_flow,
name="reviewing_tool",
description="Tool to check for potential bugs in a given code",
)
],
)
conversation = agent.start_conversation()
status = conversation.execute()
print(conversation.get_last_message().content) # Hi! How can I help you?
conversation.append_user_message("Write a simple Python function to sum two numbers")
status = conversation.execute()
print(conversation.get_last_message().content) # Here's a simple Python function that ...
API Reference: ControlFlowEdge | DataFlowEdge | Flow | PromptExecutionStep | DescribedFlow
Learn more about the use of Flows in Agents in the API reference.
Using Sub-Flows Within Flows#
To use sub-flows in Flows, use the FlowExecutionStep.
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.dataconnection import DataFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.flowhelpers import create_single_step_flow
from wayflowcore.steps import FlowExecutionStep, OutputMessageStep, PromptExecutionStep
code_generation_subflow = create_single_step_flow(
PromptExecutionStep(
prompt_template="Please answer the following user question:\n{{user_input}}", llm=llm
)
)
code_reviewing_subflow = create_single_step_flow(
PromptExecutionStep(
prompt_template="Here is some code:\n{{code_input}}\n---\nPlease check for potential bugs in the code and propose an improved version of the code",
llm=llm,
)
)
code_generation_step = FlowExecutionStep(code_generation_subflow)
code_reviewing_step = FlowExecutionStep(code_reviewing_subflow)
output_step = OutputMessageStep("{{output_content}}")
flow = Flow(
begin_step_name="code_generation",
steps={
"code_generation": code_generation_step,
"code_reviewing": code_reviewing_step,
"display_output": output_step,
},
control_flow_edges=[
ControlFlowEdge(code_generation_step, code_reviewing_step),
ControlFlowEdge(code_reviewing_step, output_step),
ControlFlowEdge(output_step, None),
],
data_flow_edges=[
DataFlowEdge(
code_generation_step, PromptExecutionStep.OUTPUT, code_reviewing_step, "code_input"
),
DataFlowEdge(
code_reviewing_step, PromptExecutionStep.OUTPUT, output_step, "output_content"
),
],
)
conversation = flow.start_conversation(
{"user_input": "Write a simple Python function to sum two numbers"}
)
status = conversation.execute()
print(conversation.get_last_message().content)
API Reference: ControlFlowEdge | DataFlowEdge | Flow | FlowExecutionStep | OutputMessageStep | PromptExecutionStep
Learn more about the use of sub-flows in Flows by reading the FlowExecutionStep API reference.
Saving and loading WayFlow assistants#
Figure: How serialization works in WayFlow.
Saving and loading simple assistants#
Save and load WayFlow assistants using the serialize and autodeserialize helper functions.
from wayflowcore.agent import Agent
from wayflowcore.serialization import autodeserialize, serialize
agent = Agent(
llm, custom_instruction="You are a helpful assistant, please answer the user requests."
)
# saving an assistant to its serialized form
serialized_assistant = serialize(agent)
# with open("path/to/agent_config.yaml", "w") as f:
# f.write(serialized_assistant)
# loading an assistant from its serialized form
# with open("path/to/agent_config.yaml") as f:
# serialized_assistant = f.read()
agent = autodeserialize(serialized_assistant)
API Reference: Agent | serialize | autodeserialize
Learn more about Serialisation by reading How to Serialize and Deserialize Flows and Agents and the Serialisation API reference.
Saving and loading assistants with tools#
Register tools to a DeserializationContext to load assistants using tools.
from wayflowcore.agent import Agent
from wayflowcore.serialization import autodeserialize, serialize
from wayflowcore.serialization.context import DeserializationContext
from wayflowcore.tools import register_server_tool, tool
@tool
def say_hello() -> str:
"""Say hello"""
return "hello"
agent = Agent(
llm,
tools=[say_hello],
custom_instruction="You are a helpful assistant, please answer the user requests.",
)
# saving an assistant to its serialized form
serialized_assistant = serialize(agent)
# with open("path/to/agent_config.yaml", "w") as f:
# f.write(serialized_assistant)
# loading an assistant from its serialized form
# with open("path/to/agent_config.yaml") as f:
# serialized_assistant = f.read()
deserialization_context = DeserializationContext()
register_server_tool(say_hello, deserialization_context.registered_tools)
agent = autodeserialize(serialized_assistant, deserialization_context)
API Reference: Agent | serialize | autodeserialize | tool
Learn more about Serialisation by reading How to Serialize and Deserialize Flows and Agents and the Serialisation API reference.
Providing context to assistants#
Passing contextual information to assistants can be done in several ways, including:
By specifying input values when creating the Conversation.
By using ContextProviders.
By using Variables.
Providing context with inputs#
You can pass static inputs when creating a new Conversation.
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.steps import OutputMessageStep
output_step = OutputMessageStep("{{message_content}}")
flow = Flow(
begin_step_name="output_step",
steps={"output_step": output_step},
control_flow_edges=[ControlFlowEdge(output_step, None)],
)
input_context = {"message_content": "Here is my input context"}
conversation = flow.start_conversation(inputs=input_context)
conversation.execute()
print(conversation.get_last_message().content)
API Reference: ControlFlowEdge | Flow | OutputMessageStep
Learn more about passing static inputs in the Conversation API reference.
Providing dynamic inputs with ContextProviders#
ContextProviders can be used to provide dynamic information to WayFlow assistants.
Using the ToolContextProvider#
Use the ToolContextProvider to provide information to an assistant with a tool.
from wayflowcore.contextproviders import ToolContextProvider
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.dataconnection import DataFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.steps import OutputMessageStep
from wayflowcore.tools import tool
@tool
def current_time() -> str:
"""Return the current time"""
from datetime import datetime
return str(datetime.now())
current_time_contextprovider = ToolContextProvider(current_time, "current_time")
output_step = OutputMessageStep("Current time: {{time}}\nMessage content: {{message_content}}")
flow = Flow(
begin_step_name="output_step",
steps={"output_step": output_step},
control_flow_edges=[ControlFlowEdge(output_step, None)],
data_flow_edges=[
DataFlowEdge(current_time_contextprovider, "current_time", output_step, "time")
],
context_providers=[current_time_contextprovider],
)
input_context = {"message_content": "Here is my input context"}
conversation = flow.start_conversation(inputs=input_context)
conversation.execute()
print(conversation.get_last_message().content)
API Reference: ControlFlowEdge | ToolContextProvider | DataFlowEdge | Flow | tool | OutputMessageStep
Learn more by reading the ToolContextProvider API reference.
Using the FlowContextProvider#
Use the FlowContextProvider to provide information to an assistant with a Flow.
from wayflowcore.contextproviders import FlowContextProvider
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.dataconnection import DataFlowEdge
from wayflowcore.flow import Flow
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"},
)
)
context_provider = FlowContextProvider(contextual_flow, flow_output_names=["time_output"])
output_step = OutputMessageStep("Last time message: {{time_output_io}}")
flow = Flow(
begin_step_name="output_step",
steps={"output_step": output_step},
control_flow_edges=[ControlFlowEdge(output_step, None)],
data_flow_edges=[DataFlowEdge(context_provider, "time_output", output_step, "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.
API Reference: FlowContextProvider | ControlFlowEdge | DataFlowEdge | Flow | OutputMessageStep
Learn more by reading the FlowContextProvider API reference.
Using the ChatHistoryContextProvider#
Use the ChatHistoryContextProvider to provide chat history information to an assistant.
from wayflowcore.contextproviders import ChatHistoryContextProvider
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.messagelist import Message, MessageList
from wayflowcore.steps import OutputMessageStep
context_provider = ChatHistoryContextProvider(
n=2, # will retrieve the last 2 messages
output_name="history",
)
output_step = OutputMessageStep("Chat history number: {{history}}")
flow = Flow(
begin_step_name="output_step",
steps={"output_step": output_step},
control_flow_edges=[ControlFlowEdge(output_step, None)],
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()
print(conversation.get_last_message().content)
# Chat history number: USER >> Message 4
# USER >> Message 5
API Reference: ChatHistoryContextProvider | ControlFlowEdge | DataFlowEdge | Flow | OutputMessageStep
Learn more by reading the ChatHistoryContextProvider API reference.
Using Variables to provide context#
You can use Variables as an alternative way to manage context (or shared state) in Flows. They let you store and reuse information across different steps in Flows.
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.dataconnection import DataFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.property import FloatProperty
from wayflowcore.steps import OutputMessageStep, VariableReadStep, ToolExecutionStep, VariableWriteStep
from wayflowcore.variable import Variable
from wayflowcore.tools import tool
float_variable = Variable(
name="float_variable",
type=FloatProperty(),
description="a float variable",
default_value=5.0,
)
read_step_1 = VariableReadStep(variable=float_variable)
@tool(description_mode="only_docstring")
def triple_number(x: float) -> float:
"Tool that triples a number"
return x*3
triple_step = ToolExecutionStep(tool=triple_number)
write_step = VariableWriteStep(variable=float_variable)
read_step_2 = VariableReadStep(variable=float_variable)
output_step = OutputMessageStep("The variable is {{ variable }}")
flow = Flow(
begin_step=read_step_1,
control_flow_edges=[
ControlFlowEdge(read_step_1, triple_step),
ControlFlowEdge(triple_step, write_step),
ControlFlowEdge(write_step, read_step_2),
ControlFlowEdge(read_step_2, output_step),
ControlFlowEdge(output_step, None),
],
data_flow_edges=[
DataFlowEdge(read_step_1, VariableReadStep.VALUE, triple_step, "x"),
DataFlowEdge(triple_step, ToolExecutionStep.TOOL_OUTPUT, write_step, VariableWriteStep.VALUE),
DataFlowEdge(read_step_2, VariableReadStep.VALUE, output_step, "variable")
],
variables=[float_variable]
)
conversation = flow.start_conversation()
status = conversation.execute()
print(conversation.get_last_message().content)
API Reference: ControlFlowEdge | DataFlowEdge | Flow | ListProperty | FloatProperty | VariableReadStep | VariableWriteStep | OutputMessageStep | Variable
Learn more by reading the Variables API reference.