用ControlFlow构建3个AI应用：情感分类、书籍推荐与旅行规划

1. 用ControlFlow构建3个有趣的AI应用

作为一名长期在数据科学领域实践的开发者，我一直在寻找能够简化AI应用开发的工具。最近发现ControlFlow这个Python框架，它让我能用几行代码就构建出功能完整的AI应用。今天我就带大家用ControlFlow实现三个实用又有趣的项目。

ControlFlow的核心价值在于它提供了一种结构化定义LLM工作流的方式。不同于直接调用API生成文本，它允许你通过任务(Tasks)、代理(Agents)和流(Flows)三个核心组件，构建复杂的AI工作流程。最棒的是，整个过程完全透明——你可以观察到AI模型在每个任务中的决策过程。

2. 环境准备与基础配置

2.1 安装ControlFlow

安装ControlFlow非常简单，只需运行以下pip命令：

pip install controlflow

安装完成后，建议检查版本信息以确保所有依赖项正确安装：

controlflow version

典型输出如下：

ControlFlow version: 0.9.4 Prefect version: 3.0.2 LangChain Core version: 0.2.40 Python version: 3.10.12

注意：如果在Colab中运行出现RuntimeError，尝试重新运行单元格通常可以解决问题。

2.2 设置API密钥

使用ControlFlow需要OpenAI的API密钥。设置环境变量的方法：

export OPENAI_API_KEY="your-api-key"

或者在Python脚本中直接设置：

import os os.environ["OPENAI_API_KEY"] = "your-api-key"

3. 项目一：推文情感分类器

3.1 项目概述

情感分析是自然语言处理的经典应用。传统方法需要收集数据、训练模型、评估调优等复杂步骤。而使用ControlFlow，我们可以用不到20行代码构建一个可用的分类器。

3.2 实现步骤

首先创建分类器代理：

import controlflow as cf classifier = cf.Agent( name="Tweet Classifier", model="openai/gpt-4o-mini", instructions="你是一位擅长快速分类推文情感的专家。" )

定义测试推文列表：

tweets = [ "这里的负面情绪太容易传播了。#叹气", "有时候发泄是必要的。#讨厌疗法", "今天空气中充满了爱！💖 #感恩", "感谢我所有的微博朋友！🌟" ]

运行分类任务：

from typing import Literal classifications = cf.run( '对这些推文进行情感分类', result_type=list[Literal['负面', '正面']], agents=[classifier], context=dict(tweets=tweets) )

3.3 结果可视化

为提升可读性，添加颜色标记：

GREEN = '\033[92m' RED = '\033[91m' RESET = '\033[0m' for twt, cls in zip(tweets, classifications): color = RED if cls == '负面' else GREEN print(f"推文: {twt} | 分类: {color}{cls}{RESET}")

输出示例：

推文: 这里的负面情绪太容易传播了。#叹气 | 分类: 负面 推文: 有时候发泄是必要的。#讨厌疗法 | 分类: 负面 推文: 今天空气中充满了爱！💖 #感恩 | 分类: 正面 推文: 感谢我所有的微博朋友！🌟 | 分类: 正面

3.4 经验分享

在实际应用中，我发现几个优化点：

1. 对于中文推文，明确指示代理使用简体中文输出更可靠
2. 添加示例推文到context能显著提升分类准确性
3. 结果类型使用Literal比普通字符串更结构化

4. 项目二：书籍推荐系统

4.1 数据模型设计

使用Pydantic定义推荐书籍的数据结构：

from pydantic import BaseModel, Field class BookRecommendation(BaseModel): title: str = Field(description='书籍标题') author: str = Field(description='作者姓名') year_published: int = Field(description='出版年份') genre: str = Field(description='书籍类型')

4.2 推荐函数实现

创建推荐函数，指定返回类型为定义好的Pydantic模型列表：

def recommend_books(genre: str, count: int) -> list[BookRecommendation]: return cf.run( f"推荐{count}本{genre}类型的书籍及其详细信息", result_type=list[BookRecommendation], context={"genre": genre, "count": count} )

4.3 获取推荐结果

获取5本科幻小说推荐：

recommended_books = recommend_books(genre="科幻", count=5)

输出为结构化的Python对象，可直接转换为JSON：

import json print(json.dumps([book.dict() for book in recommended_books], indent=2, ensure_ascii=False))

示例输出：

[ { "title": "三体", "author": "刘慈欣", "year_published": 2008, "genre": "科幻" }, { "title": "神经漫游者", "author": "威廉·吉布森", "year_published": 1984, "genre": "科幻" } ]

4.4 实践技巧

1. 字段描述越详细，模型输出越准确
2. 对于中文内容，确保JSON输出使用ensure_ascii=False
3. 可以添加评分字段并设置取值范围，让推荐更精准

5. 项目三：智能旅行规划师

5.1 系统设计

这个项目更复杂，涉及多个代理和任务串联：

1. 目的地推荐代理：根据用户偏好推荐地点
2. 行程规划代理：为选定目的地制定详细日程

5.2 数据模型定义

class TravelPreferences(BaseModel): preferred_activities: List[str] budget: str # "经济"、"中等"、"豪华" travel_duration: int # 天数 preferred_region: str # 如"亚洲"、"欧洲" class TravelItinerary(BaseModel): destination: str daily_schedule: List[str]

5.3 创建代理

destination_recommender = cf.Agent( name="目的地推荐", model="openai/gpt-4o-mini", instructions="你是一位专业的旅行目的地推荐专家。" ) itinerary_planner = cf.Agent( name="行程规划", model="openai/gpt-4o", instructions="你擅长为各种目的地制定详细的旅行计划。" )

5.4 实现流程函数

使用@cf.flow装饰器定义工作流：

@cf.flow def create_travel_itinerary(preferences: TravelPreferences) -> TravelItinerary: # 任务1：推荐目的地 destination = cf.run( "根据用户偏好推荐旅行目的地", agents=[destination_recommender], result_type=str, context=preferences.dict() ) # 任务2：规划行程 daily_schedule = cf.run( f"为{destination}创建{duration}天的旅行计划", agents=[itinerary_planner], result_type=List[str], context={ "destination": destination, **preferences.dict() } ) return TravelItinerary( destination=destination, daily_schedule=daily_schedule )

5.5 使用示例

preferences = TravelPreferences( preferred_activities=["海滩", "当地美食"], budget="中等", travel_duration=5, preferred_region="东南亚" ) itinerary = create_travel_itinerary(preferences)

打印结果：

print(f"推荐目的地: {itinerary.destination}") print("\n每日行程:") for i, schedule in enumerate(itinerary.daily_schedule, 1): print(f"第{i}天: {schedule}")

5.6 优化建议

1. 添加预算检查功能，确保推荐符合用户预算
2. 对于多目的地行程，可以扩展模型支持
3. 集成地图API，提供地理位置信息

6. 常见问题与解决方案

6.1 运行时报错处理

如果在Colab中遇到RuntimeError，尝试以下步骤：

1. 重启运行时并重新安装ControlFlow
2. 检查API密钥是否正确设置
3. 确保使用的Python版本≥3.8

6.2 性能优化技巧

1. 对于简单任务，使用gpt-4o-mini模型更经济
2. 合理设置max_tokens避免不必要开销
3. 对频繁使用的代理添加缓存机制

6.3 结构化输出技巧

1. 使用Pydantic模型定义输出结构
2. 为关键字段添加详细描述
3. 对于枚举值，使用Literal类型限制选项

7. 扩展应用思路

ControlFlow的应用远不止于此，以下是一些扩展方向：

1. 客户服务自动化：构建多轮对话系统
2. 内容生成流水线：自动化文章创作与优化
3. 数据分析助手：自然语言查询生成可视化
4. 教育应用：个性化学习路径规划

我在实际项目中发现，ControlFlow特别适合需要多个AI能力协作的场景。比如一个电商客服系统可能同时需要：

• 分类用户问题
• 查询产品信息
• 生成友好回复
• 收集用户反馈

这些都可以通过定义不同的代理和任务流来实现。