​
What is ControlFlow?

ControlFlow is a Python framework for building agentic AI workflows.

An agentic workflow is a process that delegates at least some of its work to an LLM agent. An agent is an autonomous entity that is invoked repeatedly to make decisions and perform complex tasks. To learn more, see the AI glossary.

ControlFlow provides a structured, developer-focused framework for defining workflows and delegating work to LLMs, without sacrificing control or transparency:

  • Create discrete, observable tasks for an AI to solve.
  • Assign one or more specialized AI agents to each task.
  • Combine tasks into a flow to orchestrate more complex behaviors.

This task-centric approach allows you to harness the power of AI for complex workflows while maintaining fine-grained control. By defining clear objectives and constraints for each task, you can balance AI autonomy with precise oversight, letting you build sophisticated AI-powered applications with confidence.

import controlflow as cf
from pydantic import BaseModel


# create an agent to write a research report
author = cf.Agent(
    name="Deep Thought",
    instructions="Use a formal tone and clear language",
)


class ResearchTopic(BaseModel):
    title: str
    keywords: list[str]


@cf.flow
def research_workflow() -> str:
    # Task 1: the default agent will work with the user to choose a topic
    topic = cf.Task(
        "Work with the user to come up with a research topic",
        result_type=ResearchTopic,
        user_access=True,
    )

    # Task 2: the default agent will create an outline based on the topic
    outline = cf.Task("Create an outline", context=dict(topic=topic))
    
    # Task 3: the author agent will write a first draft 
    draft = cf.Task(
        "Write a first draft", 
        context=dict(outline=outline),
        agents=[author]
    )
    
    return draft


# run the workflow
result = research_workflow()
print(result)

​
Why ControlFlow?

ControlFlow is designed to address the challenges of building AI-powered applications that are both powerful and predictable.

​
🧩 Task-Centric Architecture

Break complex AI workflows into manageable, observable steps:

topic = cf.Task("Generate a research topic", result_type=ResearchTopic)
outline = cf.Task("Create an outline", context=dict(topic=topic))
draft = cf.Task("Write a first draft", context=dict(outline=outline))

​
🔒 Structured Results

Bridge the gap between AI and traditional software with type-safe outputs:

class ResearchTopic(BaseModel):
    title: str
    keywords: list[str]

topic_task = cf.Task("Generate a topic", result_type=ResearchTopic)

​
🤖 Specialized Agents

Deploy task-specific AI agents for efficient problem-solving. Agents can have their own instructions, tools, and even be backed by different LLM models:

researcher = cf.Agent(name="Researcher", instructions="Conduct thorough research")
writer = cf.Agent(name="Writer", instructions="Write clear, concise content")

topic_task = cf.Task("Research topic", agents=[researcher])
draft_task = cf.Task("Write draft", agents=[writer])

​
🔗 Ecosystem Integration

Seamlessly work with your existing code, tools, and the broader AI ecosystem:

from langchain.tools import WikipediaQueryRun

research_task = cf.Task("Research topic", tools=[WikipediaQueryRun()])

​
🎛ī¸ Flexible Control

Continuously tune the balance of control and autonomy in your agentic workflows by adjusting the scope and oversight of your tasks:

with cf.instructions("Be creative"):
    brainstorm_task.run()

with cf.instructions("Follow APA style strictly"):
    formatting_task.run()

​
🕹ī¸ Multi-Agent Orchestration

Coordinate multiple AI agents within a single workflow - or a single task:

@cf.flow
def research_paper():
    topic = cf.Task("Choose topic", agents=[researcher])
    outline = cf.Task("Create outline", agents=[researcher, writer])
    draft = cf.Task("Write draft", agents=[writer])
    return draft

​
🔍 Native Observability and Debugging

ControlFlow is built on Prefect 3.0, so you can combine agentic and traditional workflows and monitor them all in one place:

@cf.flow(retries=2)
def enhance_data():
    data = etl_pipeline()
    enhanced_data = cf.Task("Add topics to data", context=dict(data=data))
    return enhanced_data

ControlFlow lets you build AI workflows with confidence, maintaining control and visibility throughout the process. It offers a powerful and flexible framework for creating AI-powered applications that are transparent, maintainable, and aligned with software engineering best practices.

Installation & Setup