ControlFlow

ControlFlow is a Python framework for building agentic AI workflows.

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 work on.
• Assign one or more specialized AI agents to each task.
• Combine tasks into a flow to orchestrate more complex behaviors.

Example

The simplest ControlFlow workflow has one task, a default agent, and automatic thread management:

import controlflow as cf

result = cf.run("Write a short poem about artificial intelligence")

print(result)

Result:

In circuits and code, a mind does bloom,
With algorithms weaving through the gloom.
A spark of thought in silicon's embrace,
Artificial intelligence finds its place.

Why ControlFlow?

ControlFlow addresses the challenges of building AI-powered applications that are both powerful and predictable:

• 🧩 Task-Centric Architecture: Break complex AI workflows into manageable, observable steps.
• 🔒 Structured Results: Bridge the gap between AI and traditional software with type-safe, validated outputs.
• 🤖 Specialized Agents: Deploy task-specific AI agents for efficient problem-solving.
• 🎛️ Flexible Control: Continuously tune the balance of control and autonomy in your workflows.
• 🕹️ Multi-Agent Orchestration: Coordinate multiple AI agents within a single workflow or task.
• 🔍 Native Observability: Monitor and debug your AI workflows with full Prefect 3.0 support.
• 🔗 Ecosystem Integration: Seamlessly work with your existing code, tools, and the broader AI ecosystem.

Installation

Install ControlFlow with pip:

pip install controlflow

Next, configure your LLM provider. ControlFlow's default provider is OpenAI, which requires the OPENAI_API_KEY environment variable:

export OPENAI_API_KEY=your-api-key

To use a different LLM provider, see the LLM configuration docs.

Workflow Example

Here's a more involved example that showcases user interaction, a multi-step workflow, and structured outputs:

import controlflow as cf
from pydantic import BaseModel


class ResearchProposal(BaseModel):
    title: str
    abstract: str
    key_points: list[str]


@cf.flow
def research_proposal_flow():

    # Task 1: Get the research topic from the user
    user_input = cf.Task(
        "Work with the user to choose a research topic",
        interactive=True,
    )
    
    # Task 2: Generate a structured research proposal
    proposal = cf.run(
        "Generate a structured research proposal",
        result_type=ResearchProposal,
        depends_on=[user_input]
    )
    
    return proposal


result = research_proposal_flow()

print(result.model_dump_json(indent=2))

In this example, ControlFlow is automatically managing a flow, or a shared context for a series of tasks. You can switch between standard Python functions and agentic tasks at any time, making it easy to incrementally build out complex workflows.

Learn More

To dive deeper into ControlFlow:

• Read the full documentation
• Explore example projects
• Join our community on Slack