Welcome to ControlFlow! ControlFlow is a declarative framework for building agentic workflows. That means that you define the objectives you want an AI agent to complete, and ControlFlow handles the rest. You can think of ControlFlow as a high-level orchestrator for AI agents, allowing you to focus on the logic of your application while ControlFlow manages the details of agent selection, data flow, and error handling. In this tutorial, we’ll introduce the basics of ControlFlow, including tasks, flows, agents, and more. By the end, you’ll have a solid understanding of how to create and run complex agentic workflows. The tutorial is divided into the following sections:

​
Install ControlFlow

To run the code in this tutorial, you’ll need to install ControlFlow and configure API keys for your LLM provider. Please see the installation instructions for more information.

​
Hello, world

​
Creating a task

The starting point of every agentic workflow is a Task. Each task represents an objective that we want an AI agent to complete. Let’s create a simple task to say hello: 
import controlflow as cf

hello_task = cf.Task("say hello")
If you examine this Task object, you’ll notice a few important things: it’s in an INCOMPLETE state and while it has no result value, its result_type is a string. This means that the task has not been completed yet, but when it does, the result will be a string.

​
Running a task

To run a task to completion, call its .run() method. This will set up an agentic loop, assigning the task to an agent and waiting for it to complete. The agent’s job is to provide a result that satisfies the task’s requirements as quickly as possible. Let’s run our task and examine it to see what happened: 
hello_task.run()
The task is now in a SUCCESSFUL state, and its result has been updated to "Hello". The agent successfully completed the task!
If you run the task a second time, it will immediately return the previous result. That’s because this specific task has already been completed, so ControlFlow will use the existing result instead of running an agent again.

​
Recap

What we learned
  • Tasks represent objectives that we want an AI agent to complete.
  • Each task has a result_type that specifies the datatype of the result we expect.
  • Calling task.run() assigns the task to an agent, which is responsible for providing a result that satisfies the task’s requirements.

​
Hello, user

​
User input

By default, agents cannot interact with (human) users. ControlFlow is designed primarily to be an agentic workflow orchestrator, not a chatbot. However, there are times when user input is necessary to complete a task. In these cases, you can set the user_access parameter to True when creating a task. Let’s create a task to ask the user for their name. We’ll also create a Pydantic model to represent the user’s name, which will allow us to apply complex typing or validation, if needed. 
import controlflow as cf
from typing import Optional
from pydantic import BaseModel


class Name(BaseModel):
    first: str
    last: Optional[str]


name_task = cf.Task("Get the user's name", result_type=Name, user_access=True)


name_task.run()
If you run the above code, the agent will ask for your name in your terminal. You can respond with something like “My name is Marvin” or even refuse to respond. The agent will continue to prompt you until it has enough information to complete the task. This is the essence of an agentic workflow: you declare what you need, and the agent figures out how to get it.

​
Failing a task

In the previous example, if you refuse to provide your name a few times, your agent will eventually mark the task as failed. Agents only do this when they are unable to complete the task, and it’s up to you to decide how to handle the failure. ControlFlow will raise a ValueError when a task fails that contains the reason for the failure.

​
Recap

What we learned
  • Setting user_access=True allows agents to interact with a user
  • Pydantic models can be used to represent and validate complex result types
  • Agents will continue to work until the task’s requirements are met
  • Agents can fail a task if they are unable to complete it

​
Hello, tasks

​
Task dependencies

So far, we’ve created and run tasks in isolation. However, agentic workflows are much more powerful when you use the results of one task to inform another. This allows you to build up complex behaviors by chaining tasks together, while still maintaining the benefits of structured, observable workflows. To see how this works, let’s build a workflow that asks the user for their name, then uses that information to write them a personalized poem: 
import controlflow as cf
from pydantic import BaseModel


class Name(BaseModel):
    first: str
    last: str


name = cf.Task("Get the user's name", user_access=True, result_type=Name)
poem = cf.Task("Write a personalized poem", context=dict(name=name))


poem.run()
In this example, we introduced a context parameter for the poem task. This parameter allows us to specify additional information that the agent can use to complete the task, which could include constant values or other tasks. If the context value includes a task, ControlFlow will automatically infer that the second task depends on the first. One benefit of this approach is that you can run any task without having to run its dependencies explicitly. ControlFlow will automatically run any dependencies before executing the task you requested. In the above example, we only ran the poem task, but ControlFlow automatically ran the name task first, then passed its result to the poem task’s context. We can see that both tasks were successfully completed and have result values.

​
Custom tools

For certain tasks, you may want your agents to use specialized tools or APIs to complete the task. To add tools to a task, pass a list of Python functions to the tools parameter of the task. These functions will be available to the agent when it runs the task, allowing it to use them to complete the task more effectively. The only requirement is that the functions are type-annotated and have a docstring, so that the agent can understand how to use them. In this example, we create a task to roll various dice, and provide a roll_die function as a tool to the task, which the agent can use to complete the task: 
import controlflow as cf
import random


def roll_die(n:int) -> int:
    '''Roll an n-sided die'''
    return random.randint(1, n)


task = cf.Task(
    'Roll 5 dice, three with 6 sides and two with 20 sides', 
    tools=[roll_die], 
    result_type=list[int],
)


task.run()

​
Recap

What we learned
  • You can provide additional information to a task using the context parameter, including constant values or other tasks
  • If a task depends on another task, ControlFlow will automatically run the dependencies first
  • You can provide custom tools to a task by passing a list of Python functions to the tools parameter

​
Hello, flow

If Tasks are the building blocks of an agentic workflow, then Flows are the glue that holds them together. Each flow represents a shared history and context for all tasks and agents in a workflow. This allows you to maintain a consistent state across multiple tasks, even if they are not directly dependent on each other.
When you run a task outside a flow, as we did in the previous examples, ControlFlow automatically creates a flow context for that run. This is very convenient for testing and interactive use, but you can disable this behavior by setting controlflow.settings.strict_flow_context=True.

​
The @flow decorator

The simplest way to create a flow is by decorating a function with the @flow decorator. This will automatically create a shared flow context for all tasks inside the flow. Here’s how we would rewrite the last example with a flow function: 
import controlflow as cf


@cf.flow
def hello_flow(poem_topic:str):
    name = cf.Task("Get the user's name", user_access=True)
    poem = cf.Task(
        "Write a personalized poem about the provided topic",
        context=dict(name=name, topic=poem_topic),
    )
    return poem


hello_flow(poem_topic='AI')
hello_flow is now a portable agentic workflow that can be run anywhere. On every call, it will automatically create a flow context for all tasks inside the flow, ensuring that they share the same state and history.

​
Eager execution

Notice that in the above flow, we never explicitly ran the name task, nor did we access its result attribute at the end. That’s because @flow-decorated functions are executed eagerly by default. This means that when you call a flow function, all tasks inside the flow are run automatically and any tasks returned from the flow are replaced with their result values. Most of the time, you’ll use eagerly-executed @flows and lazily-executed Tasks in your workflows. Eager flows are more intuitive and easier to work with, since they behave like normal functions, while lazy tasks allow the orchestrator to take advantage of observed dependencies to optimize task execution and agent selection, though it’s possible to customize both behaviors. However, you’ll frequently need a task’s result inside your flow function. In this case, you can eagerly run the task by calling its .run() method, then use the task’s result attribute as needed. In this example, we collect the user’s height, then use it to determine if they are tall enough to receive a poem: 
import controlflow as cf

@cf.flow
def height_flow(poem_topic:str):
    height = cf.Task("Get the user's height", user_access=True, result_type=int, instructions='convert the height to inches')
    height.run()

    if height.result < 40:
        raise ValueError("You must be at least 40 inches tall to receive a poem")
    else:
        return cf.Task(
            "Write a poem for the user that takes their height into account",
            context=dict(height=height, topic=poem_topic),
        )
In this example, we introduced the instructions parameter for the height task. This parameter allows you to provide additional instructions to the agent about how to complete the task.

​
Recap

What we learned
  • Flows provide a shared context for all tasks and agents inside the flow
  • The @flow decorator creates a flow function that can be run anywhere
  • By default, @flow-decorated functions are executed eagerly, meaning all tasks inside the flow are run automatically
  • You can eagerly run a task inside a flow by calling its .run() method
  • The instructions parameter allows you to provide additional instructions to the agent about how to complete the task

​
Hello, agents

You’ve made it through an entire tutorial on agentic workflows without ever encountering an actual agent! That’s because ControlFlow abstracts away the complexities of agent selection and orchestration, allowing you to focus on the high-level logic of your application. But agents are the heart of ControlFlow, and understanding how to create and use them is essential to building sophisticated agentic workflows.

​
Creating an agent

To create an agent, provide at least a name, as well as optional description, instructions, or tools. Here’s an example of creating an agent that specializes in writing technical documentation: 
import controlflow as cf

docs_agent = cf.Agent(
    name="DocsBot",
    description="An agent that specializes in writing technical documentation",
    instructions=(
        "You are an expert in technical writing. You strive "
        "to condense complex subjects into clear, concise language."
        "Your goal is to provide the user with accurate, informative "
        "documentation that is easy to understand."
    ),
)
What’s the difference between a description and instructions? The description is a high-level overview of the agent’s purpose and capabilities, while instructions provide detailed guidance on how the agent should complete a task. Agent descriptions can be seen by other agents, but instructions are private, which can affect how agents collaborate with each other.

​
Assigning an agent to a task

To use an agent to complete a task, assign the agent to the task’s agents parameter. Here’s an example of assigning the docs_agent to a task that requires writing a technical document: 
technical_document = cf.Task(
    "Write a technical document",
    agents=[docs_agent],
    instructions=(
        "Write a technical document that explains agentic workflows."
    ),
)
When you run the technical_document task, ControlFlow will automatically assign the docs_agent to complete the task. The agent will use the instructions provided to generate a technical document that meets the task’s requirements.

​
Assigning multiple agents to a task

You can assign multiple agents to a task by passing a list of agents to the agents parameter. This allows you to leverage the unique capabilities of different agents to complete a task more effectively. Here’s an example of assigning an editor agent to review the technical document created by the docs_agent: 
import controlflow as cf

docs_agent = cf.Agent(
    name="DocsBot",
    description="An agent that specializes in writing technical documentation",
    instructions=(
        "You are an expert in technical writing. You strive "
        "to condense complex subjects into clear, concise language."
        "Your goal is to provide the user with accurate, informative "
        "documentation that is easy to understand."
    ),
)

editor_agent = cf.Agent(
    name="EditorBot",
    description="An agent that specializes in editing technical documentation",
    instructions=(
        "You are an expert in grammar, style, and clarity. "
        "Your goal is to review the technical document created by DocsBot, "
        "ensuring that it is accurate, well-organized, and easy to read."
        "You should output notes rather than rewriting the document."
    ),
)

technical_document = cf.Task(
    "Write a technical document",
    agents=[docs_agent, editor_agent],
    instructions=(
        "Write a technical document that explains agentic workflows."
        "The docs agent should generate the document, "
        "after which the editor agent should review and "
        "edit it. Only the editor can mark the task as complete."
    ),
)

with cf.instructions('No more than 5 sentences per document'):
    technical_document.run()
When you run the technical_document task, ControlFlow will assign both the docs_agent and the editor_agent to complete the task. The docs_agent will generate the technical document, and the editor_agent will review and edit the document to ensure its accuracy and readability. By default, they will be run in round-robin fashion, but you can customize the agent selection strategy by passing a function as the task’s agent_strategy.

​
Instructions

In the above example, we also introduced the instructions context manager. This allows you to provide additional instructions to the agents about how to complete any task. As long as the context manager is active, any tasks/agents run within its scope will follow the provided instructions. Here, we use it to limit the length of the technical document to 5 sentences in order to keep the example manageable.

​
Recap

What we learned
  • Agents are autonomous entities that complete tasks on behalf of the user
  • You can create an agent by providing a name, description, instructions, and LangChain model
  • Assign an agent to a task by passing it to the task’s agents parameter
  • You can assign multiple agents to a task to have them collaborate

​
What’s next?

Congratulations, you’ve completed the ControlFlow tutorial! You’ve learned how to:
  • Create tasks and run them to completion
  • Interact with users and handle user input
  • Chain tasks together to build complex workflows
  • Create flows to maintain a shared context across multiple tasks
  • Work with agents to complete tasks autonomously
  • Read more about core concepts like tasks, flows, and agents
  • Understand ControlFlow’s workflow APIs and execution modes
  • Learn how to use different LLM models