使用 ControlFlow 构建 3 个有趣的 AI 应用
图片由作者 | Canva Pro
人工智能行业正迅速朝着利用大型语言模型 (LLM) 创建解决方案和最大化 AI 模型潜力迈进。公司正在寻求能够将 AI 无缝集成到现有代码库中的工具，而无需承担聘请专业人士和获取资源的巨额成本。这就是 Controlflow 发挥作用的地方。借助 ControlFlow，您只需几行代码即可开发复杂的 AI 应用程序。
在本教程中，我们将探讨 ControlFlow，并使用它来构建三个令人兴奋的 AI 应用程序。项目范围从简单的文本分类器到具有多个代理、任务和流程的复杂 AI。
什么是 ControlFlow？
ControlFlow 是一个 Python 框架，它为定义 LLM 工作流提供了一种结构化方法。它包含创建 AI 应用程序的三个主要组件：
- 任务 (Tasks)：这些是 AI 工作流的基本构建块。它们定义了一个或多个 AI 代理需要完成的离散、明确的目标。
- 代理 (Agents)：这些是为您的 AI 工作流提供动力的智能、自主实体。您可以定义一个模型，为其提供自定义说明，并添加各种工具来创建代理。
- 流程 (Flows)：Flows 负责按指定的顺序运行多个 AI 工作流。它们提供了一种管理任务、代理、工具和共享上下文的结构化方式。
通过使用 ControlFlow，您可以将 AI 功能无缝集成到您的 Python 应用程序中，更多地控制 AI 工作流，并生成结构化输出，而不仅仅是文本。它允许您轻松构建复杂的工作流，并且非常用户友好。使用 ControlFlow 的最大好处是，它使您能够在每个任务中观察 AI 模型的决策过程。
简单来说，ControlFlow 有两个主要用途：它协调您的 LLM 工作流并帮助您生成结构化输出，从而让您对 AI 有更多的控制权。
设置 ControlFlow
我们可以通过在终端中键入以下命令来简单安装 ControlFlow。它将自动安装所有依赖项。
|
1 |
$ pip install controlflow |
生成 OpenAI API 密钥并将其设置为环境变量。
|
1 |
$ export OPENAI_API_KEY="your-api-key" |
在使用 ControlFlow 之前，请确保已正确安装。在终端中键入以下命令以查看与 ControlFlow 相关的所有 Python 包版本。
|
1 |
$ controlflow version |
输出
|
1 2 3 4 5 6 |
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-<b>with</b>-glibc2.35 Path: /usr/local/lib/python3.10 |
在 ControlFlow 中创建代理并运行任务非常简单。在此示例中，我们通过提供自定义说明来创建了恐怖故事讲述代理。然后，我们将使用它通过提供提示来运行一个简单的任务。最终，我们将生成一个短篇故事。
|
1 2 3 4 5 6 7 |
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]) |
这是结果。
如果在 Colab 中运行以下代码时遇到 `RuntimeError` 或运行时问题，请再次运行该单元格。
1. 推文分类
推文分类器是学生们一个流行的项目，通常需要几个月的时间才能构建一个合适的文本分类器。通过使用 ControlFlow，我们可以用几行代码创建一个合适的推文分类器。
- 创建 4 条简短推文的列表。
- 使用 GPT-4-mini 模型设置一个具有自定义说明的代理。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
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! 🌟" ] # 创建一个专用代理 classifier = cf.Agent( name="Tweet Classifier", model="openai/gpt-4o-mini", instructions="You are an expert at quickly classifying tweets.", ) |
- 创建一个分类推文为“仇恨”或“爱”的任务，包含提示、结果类型、代理和上下文。我们将提供推文列表作为上下文。
- 运行任务并显示结果。
|
1 2 3 4 5 6 7 8 9 10 11 |
from typing import Literal # 设置 ControlFlow 任务来分类推文 classifications = cf.run( 'Classify the tweets', result_type=list[Literal['hate', 'love']], agents=[classifier], context=dict(tweets=tweets), ) print(classifications) |
ControlFlow 任务已对推文进行分类，并生成了一个标签列表。
让我们使用颜色代码（红色表示仇恨，绿色表示爱）以正确的方式显示它们。
|
1 2 3 4 5 6 7 8 9 10 11 |
# 用于绿色文本的 ANSI 转义代码 GREEN = '\033[92m' RED = '\033[91m' RESET = '\033[0m' # 打印推文及其分类，分类显示为绿色 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") |
两篇推文被标记为“仇恨”，两篇被标记为“爱”。这相当准确，并且以结构化格式提供输出，我们可以将其集成到现有应用程序中。
2. 书籍推荐器
两年前，我在 Deepnote 上构建了一个 Goodreads 应用程序，这激发了我创建书籍推荐应用程序的兴趣。
在这个项目中，我们将使用 Pydantic 模型定义书籍推荐的数据模型。此模型将确保每条推荐都包含标题、作者、出版年份和正确字段类型的流派。
`recommend_books` 函数将使用此 Pydantic 模型来定义结果类型，并根据指定的流派和书籍数量生成书籍推荐列表。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
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} ) |
我们将推荐 5 本科幻书籍。
|
1 2 |
recommended_books = recommend_books(genre="Science Fiction", count=5) print(recommended_books) |
代理首先生成了文本形式的推荐书籍，将其转换为 JSON 格式，最后将其转换为包含所有正确数据类型的 Pydentic 模型类型。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
[ 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' ) ] |
要将输出转换为 JSON 格式，我们只需使用 `.model_dump_json(indent=2)` 函数即可。
|
1 2 |
for book in recommended_books: print(book.model_dump_json(indent=2)) |
我们得到了 JSON 格式的输出，可以轻松地集成到 Web 应用程序或任何代码库中。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 |
{ "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. 旅行代理
在这个项目中，我们将连接两个任务。一个任务将根据用户偏好生成目的地，然后使用该目的地创建详细的行程。为了创建这个项目，我们将使用多个代理、多个任务，并通过流程将它们组合起来。
我们将定义两个用于旅行偏好和旅行行程的模型类。一个将用于用户输入，另一个将用作输出的结构。
此外，我们将创建一个流程函数，其中包含两个任务：一个用于生成目的地，另一个用于根据用户偏好规划行程。
该过程将涉及生成目的地，然后使用该目的地为“n”天规划行程。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 |
import controlflow as cf from pydantic import BaseModel from typing import List # 创建专用代理 destination_recommender = cf.Agent(name="DestinationRecommender", model="openai/gpt-4o-mini") itinerary_planner = cf.Agent(name="ItineraryPlanner", model="openai/gpt-4o") # 定义我们的数据模型 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 within the preferred region 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 schedule 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) |
当您习惯了编写 ControlFlow 工作流时，您会发现用几行代码就可以轻松创建新的应用程序和项目。
我们将通过提供偏好的活动、预算、旅行时间和偏好的地区来创建用户偏好。然后，我们将使用这些用户偏好来运行流程。
|
1 2 3 4 5 6 7 8 9 |
preferences = TravelPreferences( preferred_activities=["beach", "local cuisine"], budget="medium", travel_duration=7, preferred_region="Europe" ) # 创建个性化的旅行行程 itinerary = create_travel_itinerary(preferences) |
现在我们将打印出目的地和每日行程。
|
1 2 3 4 5 6 |
print("推荐目的地:") print(f"- {itinerary.destination}") print("\n\n每日行程:") for schedule in itinerary.daily_schedule: print(f"{schedule}") |
我们的目的地是意大利的阿马尔菲海岸，并且我们有一个非常好的旅行计划。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 |
推荐 目的地: - 根据您偏爱 海滩 活动和 当地 美食, 以及 适中 的预算 和 7 天的 欧洲 旅行 时长, 我 推荐您游览 意大利 的 阿马尔菲 海岸。 这个 令人惊叹 的目的地 提供 美丽的 海滩, 美味的 当地 佳肴, 以及 如 波西塔诺 和 拉韦洛 这样 如画 的城镇, 非常适合 一次 难忘 的假期。 每日 行程: 第一天 1: 抵达 并 探索 波西塔诺 - 抵达 波西塔诺 并 办理 入住 手续 - 下午 在 大沙滩 （Spiaggia Grande）放松身心 - 晚上: 在 当地 的 一家 餐厅 享用 晚餐, 品尝 新鲜 的海鲜 和 意面 第二天 2: 波西塔诺 海滩 日 和 当地 美食 - 上午: 参观 Forlillo 海滩, 这里 比 大沙滩 更 宁静 - 下午: 参加 一次 烹饪 课程,学习 制作 传统 意大利 菜肴 - 晚上: 漫步 小镇, 在 当地 的 冰淇淋店 品尝 美味 的 冰淇淋 第三天 3: 阿马尔菲 和 拉韦洛 一日游 - 上午: 乘坐 船 或 巴士 前往 阿马尔菲 - 参观 阿马尔菲 大教堂, 并 游览 小镇 - 下午: 继续 前往 拉韦洛, 参观 鲁菲洛 别墅 及其 花园 - 晚上: 返回 波西塔诺,并在 当地 的 小餐馆 享用 晚餐 第四天 4: 卡普里 岛 一日游 - 上午: 乘坐 渡轮 前往 卡普里 - 探索 蓝洞, 并 乘坐 缆车 前往 索拉罗山 - 下午: 漫步 卡普里 镇, 并 享用 午餐 - 晚上: 返回 波西塔诺,并在 您的 住所 放松 第五天 5: 波西塔诺 海滩 和 休闲 日 - 上午: 在 阿瑞佐 海滩（Arienzo Beach） 放松, 这里 以 清澈 的 海水 闻名 - 下午: 在 海滨 咖啡馆 享用 悠闲 的 午餐 - 晚上: 乘船 进行 日落 海岸 游览 第六天 6: 探索 索伦托 - 上午: 乘坐 巴士 或 渡轮前往 索伦托 - 参观 历史 中心, 包括 塔索广场 和 大港口 - 下午: 在 当地 餐厅 用餐, 并 品尝 柠檬酒 - 晚上: 返回 波西塔诺 第七天 7: 离开 - 上午: 在 波西塔诺 度过 最后的 时光, 购买 纪念品 - 下午: 前往 机场 或 下一个 目的地 |
我们可以轻松地将此项目转化为产品，并构建一个提供旅行推荐的网站。
如果您在运行代码时遇到问题，请查看 ControlFlow 教程 Colab 笔记本。
总结
ControlFlow 尚属新事物，您可能会在运行文档中的某些示例时遇到问题。好消息是，有一个专门的团队正在实时解决这些问题，以确保顺畅的上手体验。您只需在 GitHub 仓库 Issues · PrefectHQ/ControlFlow 上创建一个 issue。我在 Colab 上运行它时遇到了一些问题，但得益于 Jeremiah Lowin，这些问题在两个小时内就得到了解决。
我认为我们应该致力于构建能提供商业价值的 AI 解决方案，而不是仅仅专注于将模型性能提高几个百分点。要充分发挥 AI 的潜力，我们需要 ControlFlow、LangChain 和其他 AI 框架等工具，这些工具可以帮助我们用几行代码构建复杂的 AI 应用程序。
暂无评论。