Building 3 Exciting AI Applications with ControlFlow

by

The AI landscape is rapidly evolving, particularly in leveraging large language models (LLMs) to create innovative solutions. As companies seek tools that effortlessly integrate AI into their existing codebases without incurring substantial costs, ControlFlow emerges as a pivotal resource. With ControlFlow, you can develop intricate AI applications using minimal code.

In this tutorial, we will explore ControlFlow and demonstrate how to build three engaging AI applications. These projects will range from a straightforward text classifier to more complex AI applications involving multiple agents and tasks.

What is ControlFlow?

ControlFlow is a Python framework designed to provide a structured methodology for developing LLM workflows. It encompasses three primary components for creating AI applications:

  • Tasks: Fundamental units of AI workflows that outline discrete objectives to be achieved by one or more AI agents.
  • Agents: Intelligent entities that drive your AI workflows. You can define a model, give it specific instructions, and incorporate various tools to create these agents.
  • Flows: Manage the execution of multiple AI workflows sequentially, providing a systematic way to handle tasks, agents, tools, and shared contexts.

By utilizing ControlFlow, you can integrate AI capabilities into your Python applications seamlessly. It grants you greater control over AI processes and facilitates the generation of structured outputs beyond just plain text. Simplifying complex workflows, it is also user-friendly, allowing you to observe an AI model’s decision-making at each step.

In essence, ControlFlow serves two critical functions: orchestrating LLM workflows and enabling structured output generation, ultimately enhancing your control over AI.

Setting Up ControlFlow

Setting up ControlFlow is straightforward; just execute the following command in your terminal to install it along with all necessary dependencies:

pip install controlflow

You will also need to generate an OpenAI API key and set it as an environment variable:

export OPENAI_API_KEY="your-api-key"

To ensure proper installation, you can verify the version of ControlFlow and related Python packages with the command:

controlflow version

Expected Output:

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 executing a task in ControlFlow is quite simple. Here, we’ve established a Horror storytelling agent by providing it with custom instructions, which we will use for generating a short story:

import controlflow as cf

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

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

1. Tweet Classification

The Tweet Classifier project is an accessible introduction for students. Typically, building an effective text classifier could take months; however, with ControlFlow, you can easily create one in just a few lines of code.

Start by listing four short tweets:

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! 🌟"
]

Set up an agent with tailored instructions using the GPT-4-mini model:

# 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 that classifies tweets as either “hate” or “love,” including prompts, result types, agents, and context. Finally, run the task and view the results:

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)

ControlFlow will classify the tweets and generate a list of labels.

To present the results neatly with color coding (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
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")

This produces an accurate output, with two tweets labeled “hate” and two as “love,” and outputs it in a structured format, ready for integration into existing applications.

2. Book Recommender

My journey toward creating a book recommendation application began two years ago with a Goodreads project on Deepnote.

In this endeavor, we’ll define a data model for book recommendations using Pydantic models to ensure that each recommendation includes appropriate fields: title, author, publication year, and genre.

The recommend_books function will utilize this Pydantic model to generate a list of book recommendations based on a specified genre and 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}
    )

Let’s generate five Science Fiction book recommendations:

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

The output will first appear as a formatted text, then convert into a JSON structure that aligns with our Pydantic model:

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

The output in JSON format can be readily integrated into web applications or any codebase:

{
  "title": "Dune",
  "author": "Frank Herbert",
  "year_published": 1965,
  "genre": "Science Fiction"
}

3. Travel Agent

This project connects two tasks: generating travel destinations based on user preferences and using this information to create a detailed itinerary.

We will define two model classes for travel preferences and travel itineraries. One will handle user input, and the other will structure the output.

We will also create a flow function that houses two tasks: one for generating a suitable destination and another for planning the trip.

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 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 travel 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 daily itinerary using the 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)

Now, let’s create user preferences by specifying the activity, budget, travel duration, and region:

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

# Create a personalized travel itinerary
itinerary = create_travel_itinerary(preferences)

We will print the recommended destination along with the daily itinerary:

print("Recommended Destination:")
print(f"- {itinerary.destination}")

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

Suppose the suggested destination is the Amalfi Coast in Italy. Here’s how it may appear:

Recommended Destination:

  • Based on your preferences for beach activities and local cuisine, along with a medium budget and a travel duration of 7 days in Europe, I recommend visiting the Amalfi Coast in Italy. This stunning destination offers beautiful beaches, delicious local dishes, and picturesque towns such as Positano and Ravello, perfect for a memorable vacation.

Daily Schedule:

  • Day 1: Arrival and exploration of Positano
  • Day 2: Beach day and local cuisine in Positano
  • Day 3: Day trip to Amalfi and Ravello
  • Day 4: Capri Island excursion
  • Day 5: Beach and leisure day in Positano
  • Day 6: Explore Sorrento
  • Day 7: Departure

With minimal effort, this project can be expanded into a full-fledged product, providing users with tailored travel recommendations.

If you encounter issues while running this code, refer to the ControlFlow Tutorial on Colab for assistance.

Final Thoughts

As ControlFlow is relatively new, you may encounter certain challenges when experimenting with examples from the documentation. Fortunately, there is a dedicated team working diligently to resolve these issues, ensuring a seamless onboarding experience. Creating an issue on their GitHub repository can lead to prompt resolutions—my own experience with runtime issues in Colab was quickly addressed.

It is vital to prioritize building AI solutions that deliver real business value rather than merely emphasizing marginal model performance improvements. To unlock the full potential of AI, we need frameworks like ControlFlow, LangChain, and others that empower us to develop complex applications with minimal effort.

Feel free to reach out if you need any additional changes or content!

Leave a Comment