ControlFlow

A Task is the fundamental building block of a ControlFlow workflow. It represents a discrete objective or goal that an AI agent is assigned to solve.

Key characteristics of a Task include:

• Objective & Instructions: Defines what the task is and how the agent should approach it.
• Expected Result Type: Specifies the format or type of data the task should produce.
• Context & Tools: Provides the necessary information or capabilities required to complete the objective.
• Agents: Tasks are executed by one or more agents assigned to them, allowing developers to match specific tasks with agents that have the appropriate specialized capabilities or model characteristics.
• Dependencies: Tasks can have dependencies to define execution order, ensuring that a task's output is available as input for subsequent tasks or that prerequisites are met before execution.

Flow engineering is an approach to designing and optimizing agentic workflows for Large Language Models (LLMs). Unlike prompt engineering, which focuses on the natural-language content of individual messages, flow engineering focuses on the structural design of the workflow itself.

Key objectives of flow engineering include:

• Guiding decision-making: Using the workflow structure to direct the agent's process.
• Improving output quality: Designing steps, decision points, and feedback loops to ensure better results.
• Task decomposition: Breaking complex tasks into smaller, manageable components.
• Sequence optimization: Defining the optimal order of actions for an agent to follow.
• Control and domain expertise: Incorporating domain-specific knowledge and best practices into the structured sequence of actions to improve agent effectiveness, efficiency, and adaptability.

In ControlFlow, you can assign tools at three distinct levels to control agent capabilities and security. This hierarchical assignment allows you to restrict specific capabilities to specific parts of your workflow:

1. Flow Level: Tools assigned to the @cf.flow decorator are available to any agent participating in that flow. Use this for global utilities like logging or user notifications.
2. Agent Level: Tools assigned to a cf.Agent instance are available to that specific agent throughout the workflow.
3. Task Level: Tools passed to cf.run() (or a cf.Task definition) are only available while that specific task is executing. Use this for sensitive or specialized operations like reading specific files.

Tools can be any Python function. For best results, ensure they have clear type annotations and descriptive docstrings, as agents use these to understand how and when to call them.

import controlflow as cf

# 1. Agent-level tool
agent = cf.Agent(name="FileSearcher", tools=[list_files])

# 2. Flow-level tool
@cf.flow(tools=[update_user], default_agent=agent)  
def file_search_flow(query: str, directory: str):
    
    # 3. Task-level tool
    cf.run(
        "Analyze the contents...",
        tools=[read_file],
        depends_on=[found_files]
    )

A flow provides shared context and history for multiple agents across multiple tasks. You can create a flow by applying the @cf.flow decorator to a Python function. Inside a flow, you can use standard Python logic (like if/else statements) to dynamically adjust the workflow based on the results returned by cf.run() calls.

import controlflow as cf

# Create agents
classifier = cf.Agent(
    name="Email Classifier",
    model="openai/gpt-4o-mini",
    instructions="You are an expert at quickly classifying emails. Always "
                 "respond with exactly one word: either 'important' or 'spam'."
)

responder = cf.Agent(
    name="Email Responder",
    model="openai/gpt-4o",
    instructions="You are an expert at crafting professional email responses. "
                 "Your replies should be concise but friendly."
)

# Create the flow
@cf.flow
def process_email(email_content: str):

    # Classify the email
    category = cf.run(
        f"Classify this email",
        result_type=["important", "spam"],
        agents=[classifier],
        context=dict(email=email_content),
    )

    # If the email is important, write a response
    if category == "important":
        response = cf.run(
            f"Write a response to this important email",
            result_type=str,
            agents=[responder],
            context=dict(email=email_content),
        )
        return response

    # Otherwise, no response is needed for spam email
    else:
        print("No response needed for spam email.")

# Run the flow on each email
for email in emails:
    response = process_email(email)
    print(response)

In ControlFlow, you should follow an incremental approach to building workflows:

1. Start with Tasks

Always begin by defining your workflow using cf.Task. Tasks are the primary way to:

• Define your workflow: Create a roadmap of steps.
• Set clear objectives: Each task has a specific goal and an expected result_type.
• Maintain control: Tasks provide fine-grained control over the application logic.

Note: If you create a cf.Task without providing an agents list, ControlFlow automatically assigns a default agent.

2. Add Agents to Steer Behavior

Introduce cf.Agent instances when you need to fine-tune how tasks are performed. Use agents for:

• Specialized Expertise: When a task requires specific knowledge or skills.
• Consistent Personality: To maintain a specific tone or approach.
• Access to Specific Tools: When tasks require specialized functions or APIs.