Tasks
Tasks are the fundamental building blocks of ControlFlow workflows, representing specific objectives or goals within an AI-powered application. They act as a bridge between AI agents and application logic, enabling developers to define and structure desired outcomes in a clear and intuitive manner.
Creating Tasks
ControlFlow provides two convenient ways to create tasks: using the Task class or the @task decorator.
Using the Task Class
The Task class offers a flexible and expressive way to define tasks by specifying various properties and requirements.
from controlflow import Task
interests = Task(
objective="Ask user for three interests",
result_type=list[str],
user_access=True,
instructions="Politely ask the user to provide three of their interests or hobbies."
)
The Task class allows you to explicitly define the objective, instructions, agents, context, result type, tools, and other properties of a task. This approach provides full control over the task definition and is particularly useful when you need to specify complex requirements or dependencies.
Using the @task Decorator
The @task decorator provides a concise and intuitive way to define tasks using familiar Python functions. The decorator automatically infers key properties from the function definition, making task creation more streamlined.
from controlflow import task
@task(user_access=True)
def get_user_name() -> str:
"Politely ask the user for their name."
pass
When using the @task decorator, the objective is inferred from the function name, instructions are derived from the docstring, context is inferred from the function arguments, and the result type is inferred from the return annotation. This approach is ideal for simple tasks or when you want to leverage existing functions as tasks.
Defining Task Objectives and Instructions
Clear objectives and instructions are crucial for guiding AI agents and ensuring successful task execution.
Objectives
The objective of a task should be a brief description of the task’s goal or desired outcome. It helps both developers and AI agents understand the purpose of the task and what it aims to achieve.
When defining objectives, aim for clarity and specificity. Use action-oriented language and avoid ambiguity. For example:
summary_task = Task(
objective="Summarize the key points of the customer feedback",
result_type=str,
)
Instructions
Instructions provide detailed guidelines or steps for completing the task. They offer more context and direction to the AI agents, beyond what is conveyed in the objective.
When writing instructions, use concise language and bullet points or numbered steps if applicable. Avoid ambiguity and provide sufficient detail to enable the AI agents to complete the task effectively.
data_analysis_task = Task(
objective="Analyze the sales data and identify top-performing products",
instructions="""
1. Load the sales data from the provided CSV file
2. Calculate the total revenue for each product
3. Sort the products by total revenue in descending order
4. Select the top 5 products based on total revenue
5. Return a list of tuples containing the product name and total revenue
""",
result_type=list[tuple[str, float]],
)
Specifying Result Types
The result_type property allows you to define the expected type of the task’s result. It provides a contract for the task’s output, ensuring consistency and enabling seamless integration with the broader workflow.
By specifying a result type, you make it clear to both the AI agents and the developers what kind of data to expect. The result_type can be any valid Python type, such as str, int, list, dict, or even custom classes.
sentiment_analysis_task = Task(
objective="Analyze the sentiment of the given text",
result_type=float,
)
product_classification_task = Task(
objective="Classify the product based on its description",
result_type=list[str],
)
When using the @task decorator, the result type is inferred from the function’s return annotation:
@task
def count_words(text: str) -> int:
"Count the number of words in the provided text."
pass
Assigning Agents and Tools
ControlFlow allows you to assign specific AI agents and tools to tasks, enabling you to leverage their specialized skills and capabilities.
Assigning Agents
By assigning agents to a task, you can ensure that the most suitable agent is responsible for executing the task. Agents can be assigned using the agents property of the Task class or the agents parameter of the @task decorator.
from controlflow import Agent
data_analyst = Agent(
name="DataAnalyst",
description="Specializes in data analysis and statistical modeling",
)
business_analyst = Agent(
name="BusinessAnalyst",
description="Expert in business strategy and market research",
instructions="Use the DataAnalyst's insights to inform business decisions.",
)
analysis_task = Task(
objective="Analyze the customer data and provide insights",
agents=[data_analyst, business_analyst],
result_type=str,
)
If no agents are explicitly assigned to a task, ControlFlow will use the agents defined in a task’s parent task, flow, or fall back on a global default agent, respectively.
Providing Tools
Tools are Python functions that can be used by agents to perform specific actions or computations. By providing relevant tools to a task, you empower the AI agents to tackle more complex problems and enhance their problem-solving abilities.
Tools can be specified using the tools property of the Task class or the tools parameter of the @task decorator.
def calculate_square_root(number: float) -> float:
return number ** 0.5
calculation_task = Task(
objective="Calculate the square root of the given number",
tools=[calculate_square_root],
result_type=float,
)
Handling User Interaction
ControlFlow provides a built-in mechanism for tasks to interact with human users. By setting the user_access property to True, a task can indicate that it requires human input or feedback to be completed.
feedback_task = Task(
objective="Collect user feedback on the new feature",
user_access=True,
result_type=str,
instructions="Ask the user to provide their thoughts on the new feature.",
)
When a task with user_access=True is executed, the AI agents assigned to the task will be given access to a special talk_to_user tool. This tool allows the agents to send messages to the user and receive their responses, enabling a conversation between the AI and the human.
Creating Task Dependencies and Subtasks
ControlFlow allows you to define dependencies between tasks and create subtasks to break down complex tasks into smaller, more manageable units of work.
Task Dependencies
Dependencies can be specified using the depends_on property of the Task class. By specifying dependencies, you ensure that tasks are executed in the correct order and have access to the necessary data or results from previous tasks.
data_collection_task = Task(
objective="Collect user data from the database",
result_type=pd.DataFrame,
)
data_cleaning_task = Task(
objective="Clean and preprocess the collected user data",
depends_on=[data_collection_task],
result_type=pd.DataFrame,
)
data_analysis_task = Task(
objective="Analyze the cleaned user data and generate insights",
depends_on=[data_cleaning_task],
result_type=dict,
)
Subtasks
Subtasks allow you to break down complex tasks into smaller steps and manage the workflow more effectively. Subtasks can be defined either by creating tasks with the context of another task, or by passing a task as a parent parameter to the subtask.
Running Tasks
Running to Completion
Tasks can be executed using the run() method, which coordinates the execution of the task, its subtasks, and any dependent tasks, ensuring that the necessary steps are performed in the correct order.
from controlflow import Task
title_task = Task('Generate a title for a poem about AI', result_type=str)
poem_task = Task(
'Write a poem about AI using the provided title',
result_type=str,
context=dict(title=title_task),
)
poem_task.run()
print(poem_task.result)
When you run a task, ControlFlow orchestrates the execution of the task and its dependencies in a loop, ensuring that each step is completed successfully before proceeding to the next one. The run() method exits when the task is completed, at which point the task’s result is available (if it succeeded) or an exception is raised (if it failed).
You can limit the number of iterations in the task loop by passing max_iterations=n to the run() method, or set a global limit using controlflow.settings.max_iterations. There is no default limit.
Do I need to create a flow?
Tasks must always be run within the context of a flow in order to manage dependencies, history, and agent interactions effectively. As a convenience, if you call task.run() outside a flow context, a new flow will be automatically created to manage the task’s execution for that run only. In the above example, poem_task.run() implicitly creates a new flow for both tasks.
This is useful for testing tasks in isolation or running them as standalone units of work. However, it can lead to confusing behavior if you try to combine multiple tasks that created their own flows, because they will not have access to each other’s context or history.
Controlling Iteration
The run() method starts a loop and orchestrates the execution of the task, its subtasks, and dependencies until the task is completed. If you need more fine-grained control over task execution, you can use the run_once() method to execute only a single step of the graph.
from controlflow import Flow, Task
title_task = Task('Generate a title for a poem about AI', result_type=str)
poem_task = Task(
'Write a poem about AI using the provided title',
result_type=str,
context=dict(title=title_task),
)
with Flow():
while poem_task.is_incomplete():
poem_task.run_once()
print(poem_task.result)
Note that run_once requires the task to be run within a flow context, as it relies on the flow to manage the task’s execution and history over each invocation.