Building 3 Fun AI Applications with ControlFlow: A Step-by-Step Guide

The AI industry is evolving quickly, with a focus on harnessing the power of large language models (LLMs) and optimizing AI solutions. Companies are looking for tools that integrate AI smoothly into their existing codebases without the high costs of hiring experts or investing in extensive resources. This is where ControlFlow comes in. With ControlFlow, you can create advanced AI applications with just a few lines of code.

In this tutorial, we'll dive into ControlFlow and demonstrate how to build three exciting AI applications. These projects range from a simple text classifier to a more complex system with multiple agents, tasks, and flows.

What is ControlFlow?

ControlFlow is a Python framework designed to streamline the creation of AI workflows using large language models (LLMs). It offers a structured approach with three key components:

• Tasks: The building blocks of AI workflows, tasks define clear, specific objectives that need to be completed by one or more AI agents.

• Agents: Intelligent, autonomous entities that power your AI workflows. You can customize these agents by defining models, providing instructions, and integrating tools.

• Flows: Flows orchestrate multiple AI workflows in a specified sequence, allowing you to manage tasks, agents, tools, and shared contexts in an organized manner.

With ControlFlow, you can easily integrate AI into your Python applications, giving you greater control over AI workflows and generating structured outputs instead of just raw text. It simplifies the process of building complex workflows and offers an intuitive, user-friendly experience. One of its key benefits is the ability to track and observe the decision-making process of AI models throughout each task.

In short, ControlFlow enables you to orchestrate LLM workflows and generate structured outputs, providing enhanced control over your AI applications.

Setting up ControlFlow

To set up ControlFlow, simply install it using the following command in your terminal. This will automatically install all the required dependencies:

$ pip install controlflow

Next, generate your OpenAI API key and set it as an environment variable:

$ export OPENAI_API_KEY="your-api-key"

Before using ControlFlow, ensure it's installed correctly. Run the following command to check the version of ControlFlow and its associated packages:

$ controlflow version

You should see output like this:

ControlFlow version: 0.9.4
Prefect version: 3.0.2
LangChain Core version: 0.2.40
Python version: 3.10.12
Platform: Linux-6.1.85+-x86_64-with-glibc2.35
Path: /usr/local/lib/python3.10

Creating an Agent and Running a Task

Creating an agent and running a task in ControlFlow is straightforward. In this example, we create a "Horror Storyteller" agent, providing it with custom instructions. Then, we use the agent to run a simple task—generating a short story.

import controlflow as cf

teller = cf.Agent(name="Horror Storyteller",
                  model="openai/gpt-4",
                  instructions="You are an older man telling horror stories to kids.")

story = cf.run("Write a short story.", agents=[teller])

This will produce the generated short story.

If you encounter a RuntimeError or runtime issue while running the following code in Colab, please rerun the cell once again.

1. Tweet Classification

Creating a tweet classifier is a common project for students, but it can take months to develop a fully functional text classifier. With ControlFlow, you can quickly build an effective tweet classifier in just a few lines of code.

Start by creating a list of 4 short tweets. Then, set up an agent with custom instructions using the GPT-4-mini model.

import controlflow as cf

tweets = [
    "Negativity spreads too easily here. #sigh",
    "Sometimes venting is necessary. #HateTherapy",
    "Love fills the air today! 💖 #Blessed",
    "Thankful for all my Twitter friends! 🌟"
]
# Create a specialized agent
classifier = cf.Agent(
    name="Tweet Classifier",
    model="openai/gpt-4o-mini",
    instructions="You are an expert at quickly classifying tweets.",
)

Next, create a task to classify the tweets as either “hate” or “love.” Provide the list of tweets as context, and set up the result type, agents, and prompt.

from typing import Literal

# Set up a ControlFlow task to classify tweets
classifications = cf.run(
    'Classify the tweets',
    result_type=list[Literal['hate', 'love']],
    agents=[classifier],
    context=dict(tweets=tweets),
)

print(classifications)

The task will classify the tweets and generate a list of labels indicating whether each tweet is categorized as "hate" or "love."

Let’s display them in a proper way using the color code (red for hate and green for love).

# ANSI escape code for green text
GREEN = '\033[92m'
RED = '\033[91m'
RESET = '\033[0m'
 
# Print tweets alongside their classifications with classification in green
for twt, cls in zip(tweets, classifications):
  if cls == 'hate':
    print(f"Tweet: {twt} | Classification: {RED}{cls.upper()}{RESET}\n")
  else:
    print(f"Tweet: {twt} | Classification: {GREEN}{cls.upper()}{RESET}\n")

Two tweets are labeled “hate,” and two tweets are labeled “love.” This is quite accurate and gives output in a structured format that we can integrate into our existing applications.

2. Book Recommender

Two years ago, I built a Goodreads application on Deepnote, which sparked my interest in creating a book recommendation app.

In this project, we’ll define a data model for book recommendations using the Pydantic model. This model ensures that each recommendation includes essential details like title, author, publication year, and genre, all with the correct data types.

The recommend_books function will utilize this Pydantic model to define the result type and generate a list of book recommendations based on a specified genre and the desired number of books.

import controlflow as cf
from pydantic import BaseModel, Field

class BookRecommendation(BaseModel):
    title: str = Field(description='The title of the recommended book')
    author: str = Field(description='The author of the book')
    year_published: int = Field(description='The year the book was published')
    genre: str = Field(description='The genre of the book')

def recommend_books(genre: str, count: int) -> list[BookRecommendation]:
    return cf.run(
        f"Recommend {count} books in the {genre} genre with their details",
        result_type=list[BookRecommendation],
        context={"genre": genre, "count": count}
    )

Now, let’s generate 5 Science Fiction book recommendations:

recommended_books = recommend_books(genre="Science Fiction", count=5)
print(recommended_books)

The agent will first generate the recommended books in text format, convert them into JSON format, and then transform the data into the Pydantic model, ensuring all fields have the correct data types.

[
    BookRecommendation(title='Dune', author='Frank Herbert', year_published=1965, genre='Science Fiction'),
    BookRecommendation(
        title='Neuromancer',
        author='William Gibson',
        year_published=1984,
        genre='Science Fiction'
    ),
    BookRecommendation(
        title='The Left Hand of Darkness',
        author='Ursula K. Le Guin',
        year_published=1969,
        genre='Science Fiction'
    ),
    BookRecommendation(title='Foundation', author='Isaac Asimov', year_published=1951, genre='Science Fiction'),
    BookRecommendation(
        title='Snow Crash',
        author='Neal Stephenson',
        year_published=1992,
        genre='Science Fiction'
    )
]

To convert the output into the JSON format, we only have to use the ‘.model_dump_json(indent=2)` function. 

for book in recommended_books:
  print(book.model_dump_json(indent=2))

We have an output in JSON format that can be easily integrated into a web application or any code base. 

{
  "title": "Dune",
  "author": "Frank Herbert",
  "year_published": 1965,
  "genre": "Science Fiction"
}
 
{
  "title": "Neuromancer",
  "author": "William Gibson",
  "year_published": 1984,
  "genre": "Science Fiction"
}
 
{
  "title": "The Left Hand of Darkness",
  "author": "Ursula K. Le Guin",
  "year_published": 1969,
  "genre": "Science Fiction"
}
 
{
  "title": "Foundation",
  "author": "Isaac Asimov",
  "year_published": 1951,
  "genre": "Science Fiction"
}
 
{
  "title": "Snow Crash",
  "author": "Neal Stephenson",
  "year_published": 1992,
  "genre": "Science Fiction"
}

3. Travel Agent

In this project, we’ll connect two tasks to create a personalized travel itinerary. The first task will generate a destination based on user preferences, and the second task will use that destination to plan a detailed trip itinerary. To build this, we’ll use multiple agents, multiple tasks, and combine them using a flow.

We'll define two model classes: one for user travel preferences and another for the travel itinerary output. The flow function will combine these tasks—one to generate the destination and another to plan the trip based on the user's input.

Here’s how the process works:

1. Generate a destination based on user preferences.

2. Use that destination to plan a trip for a specified number of days.

import controlflow as cf
from pydantic import BaseModel
from typing import List

# Create specialized agents
destination_recommender = cf.Agent(name="DestinationRecommender", model="openai/gpt-4o-mini")
itinerary_planner = cf.Agent(name="ItineraryPlanner", model="openai/gpt-4o")

# Define our data models
class TravelPreferences(BaseModel):
    preferred_activities: List[str]
    budget: str  # e.g., "low", "medium", "high"
    travel_duration: int  # in days
    preferred_region: str  # e.g., "Asia", "Europe", "South America"

class TravelItinerary(BaseModel):
    destination: str
    daily_schedule: List[str]

@cf.flow
def create_travel_itinerary(preferences: TravelPreferences) -> TravelItinerary:
    # Recommend a single destination based on user preferences
    destination = cf.run(
        "Suggest a travel destination based on user preference.",
        agents=[destination_recommender],
        result_type=str,
        context={
            "preferred_activities": preferences.preferred_activities,
            "budget": preferences.budget,
            "travel_duration": preferences.travel_duration,
            "preferred_region": preferences.preferred_region
        }
    )

    # Plan a daily schedule for the trip at the chosen destination
    daily_schedule = cf.run(
        "Create a daily schedule for the trip at the chosen destination",
        agents=[itinerary_planner],
        result_type=List[str],
        context={
            "destination": destination,
            "travel_duration": preferences.travel_duration,
            "preferred_activities": preferences.preferred_activities
        }
    )

    return TravelItinerary(destination=destination, daily_schedule=daily_schedule)

To create a personalized travel itinerary, you just need to provide the user's travel preferences, including preferred activities, budget, travel duration, and preferred region. Here's how to run the flow:

preferences = TravelPreferences(
    preferred_activities=["beach", "local cuisine"],
    budget="medium",
    travel_duration=7,
    preferred_region="Europe"
)

# Generate the personalized travel itinerary
itinerary = create_travel_itinerary(preferences)

With ControlFlow, building this travel agent application is simple and quick, requiring just a few lines of code.

Displaying the Destination and Daily Itinerary

Let's now print out the recommended destination and the detailed daily itinerary.

print("Recommended Destination:")
print(f"- {itinerary.destination}")
 
print("\n\nDaily Schedule:")
for schedule in itinerary.daily_schedule:
    print(f"{schedule}")

In our example, the destination is the Amalfi Coast in Italy, and here’s the full travel plan:

Recommended Destination:

• Based on your preferences for beach activities, local cuisine, a medium budget, and a 7-day travel duration in Europe, I recommend visiting the Amalfi Coast in Italy. This breathtaking destination offers stunning beaches, delicious local dishes, and charming towns like Positano and Ravello, making it the perfect vacation spot.

Daily Schedule:

Day 1: Arrival and Exploration of Positano

• Arrive in Positano and check in to your accommodation.

• Spend the afternoon relaxing on Spiaggia Grande beach.

• In the evening, enjoy dinner at a local restaurant, trying fresh seafood and pasta.

Day 2: Beach Day and Local Cuisine in Positano

• Morning: Visit Fornillo Beach, a quieter spot than Spiaggia Grande.

• Afternoon: Take a cooking class to learn traditional Italian recipes.

• Evening: Stroll through Positano and savor gelato from a local gelateria.

Day 3: Day Trip to Amalfi and Ravello

• Morning: Take a boat or bus to Amalfi and explore the Amalfi Cathedral.

• Afternoon: Head to Ravello to visit Villa Rufolo and its stunning gardens.

• Evening: Return to Positano and dine at a local trattoria.

Day 4: Capri Island Excursion

• Morning: Take a ferry to Capri and explore the Blue Grotto.

• Afternoon: Ride the chairlift to Monte Solaro and stroll through Capri town.

• Evening: Return to Positano and relax at your accommodation.

Day 5: Beach and Leisure Day in Positano

• Morning: Relax at Arienzo Beach, known for its crystal-clear waters.

• Afternoon: Enjoy lunch at a beachside café.

• Evening: Take a sunset boat tour along the coast.

Day 6: Explore Sorrento

• Morning: Travel to Sorrento by bus or ferry and visit the historic center.

• Afternoon: Have lunch at a local restaurant and sample some limoncello.

• Evening: Return to Positano.

Day 7: Departure

• Morning: Spend your final hours in Positano shopping for souvenirs.

• Afternoon: Head to the airport or continue your journey.

This project can be easily turned into a product. With a website, users could get personalized travel recommendations and itineraries.

If you're having trouble running the code, check out the ControlFlow Tutorial Colab notebook for assistance.

Final Thoughts

ControlFlow is still in its early stages, so you may run into some issues when running the examples in the documentation. The good news is there's a dedicated team actively addressing these issues in real time, ensuring a smooth onboarding experience. If you encounter any problems, simply create an issue on the GitHub repository at Issues · PrefectHQ/ControlFlow. I personally faced some challenges when using it on Colab, but they were resolved within two hours, thanks to Jeremiah Lowin.

I believe we should shift our focus toward creating AI solutions that offer real business value, rather than solely concentrating on marginal improvements in model performance. To unlock AI’s full potential, we need tools like ControlFlow, LangChain, and other frameworks that make it possible to build sophisticated AI applications with just a few lines of code.