Flash Learn - 让代理变得简单

FlashLearn 提供了一个简单的接口和编排功能 (最高可达每分钟1000次调用)，用于将 代理型大语言模型 集成到您的典型工作流和 ETL 流水线中。您可以进行数据转换、分类、摘要生成、改写以及自定义的多步骤任务，就像使用任何标准的机器学习库一样，在底层利用大语言模型的强大能力。每个 步骤和任务都有紧凑的 JSON 定义，这使得流水线易于理解和维护。它支持 LiteLLM、Ollama、OpenAI、DeepSeek 以及其他所有与 OpenAI 兼容的客户端。

🚀 示例

📖 完整文档

安装

pip install flashlearn

将您想要使用的提供商的 API 密钥添加到您的 .env 文件中。

OPENAI_API_KEY=

高层次概念流程

flowchart TB
    classDef smallBox font-size:12px, padding:0px;

    H[您的数据] --> I[加载技能 / 学习技能]
    I --> J[创建任务]
    J --> K[运行任务]
    K --> L[结构化结果]
    L --> M[下游步骤]

    class H,I,J,K,L,M smallBox;

---

学习一项新“技能”

类似于 fit/predict 模式，您可以快速“学习”一项自定义技能。下面我们将创建一个技能，用于根据社交媒体帖子中的用户评论评估购买产品的可能性，并返回一个分数（1–100）和简短的理由。我们将指示大语言模型按照我们的自定义规范对每条评论进行转换。

from flashlearn.skills.learn_skill import LearnSkill
from openai import OpenAI

# 实例化您的流水线“估计器”或“转换器”
learner = LearnSkill(model_name="gpt-4o-mini", client=OpenAI())
# 提供新技能的指令和样本数据
skill = learner.learn_skill(
    df=[], #如果您需要，也可以传递数据样本
    task=(
        "根据用户评论中的情感，评估其购买我产品可能性的高低，"
        "在 'likely_to_buy' 键下返回一个1到100之间的整数，"
        "并在 'reason' 键下返回一段简短的解释。"
    ),
)

# 保存技能以便在任何系统中使用
skill.save("evaluate_buy_comments_skill.json")

---

输入是一个字典列表

无论您是从 API、电子表格还是用户提交的表单中获取的数据，都可以简单地将每条记录封装成一个字典。FlashLearn 的“技能”接受这样的字典列表，如下所示：

user_inputs = [
    {"comment_text": "我爱这款产品，正是我想要的！"},
    {"comment_text": "不太满意……不会考虑购买这个。"},
    # ...
]

---

三行代码即可运行

一旦您定义或学习了一项技能，就可以像使用标准机器学习流水线中的专用转换器一样加载它。然后只需几行代码即可将其应用于您的数据：

from flashlearn.skills.general_skill import GeneralSkill

with open("evaluate_buy_comments_skill.json", "r", encoding="utf-8") as file:
    definition= json.load(file)

# 假设我们之前已将学习到的技能保存为 "evaluate_buy_comments_skill.json"。
skill = GeneralSkill.load_skill(definition)

tasks = skill.create_tasks(user_inputs)
results = skill.run_tasks_in_parallel(tasks)
print(results)

---

获取结构化结果

FlashLearn 会为您的每一条记录返回结构化的输出。结果字典中的键对应于您原始列表的索引。例如：

{
  "0": {
    "likely_to_buy": 90,
    "reason": "评论表现出强烈的热情和积极的情感。"
  },
  "1": {
    "likely_to_buy": 25,
    "reason": "表达了失望和不愿购买的态度。"
  }
}

---

传递给下一步

每条记录的输出随后可以用于下游任务。例如，您可以：

1. 将结果存储到数据库中
2. 过滤出高可能性的潜在客户
3. 将它们发送到另一个工具进行进一步分析（例如，以正式语气改写“理由”）

下面是一个小示例，展示如何解析该字典并将其传递给另一个函数：

# 假设 'flash_results' 是包含结构化 LLM 输出的字典
for idx, result in flash_results.items():
    desired_score = result["likely_to_buy"]
    reason_text = result["reason"]
    # 现在可以对分数和理由做些事情，比如存储到数据库或传递给下一步
    print(f"评论 #{idx} => 分数: {desired_score}, 理由: {reason_text}")

支持的大语言模型提供商

无论您在何处依赖机器学习流水线组件，都可以用大语言模型来替代：

client = OpenAI()  # 这相当于实例化一个流水线组件 
deep_seek = OpenAI(api_key='YOUR DEEPSEEK API KEY', base_url="https://api.deepseek.com")
lite_llm = FlashLiteLLMClient()  # LiteLLM 集成通过环境变量管理密钥，类似于顶级流水线管理者
ollama =  OpenAI(base_url = 'http://localhost:11434/v1', api_key='ollama', # 必需，但未使用) # 只需使用 ollama 的 openai 客户端

核心理念：输入 JSON，输出 JSON

按用例划分的示例

• 客户服务

  • 分类客户工单

• 金融

  • 解析财务报告数据

• 营销

  • 客户细分

• 个人助理

  • 研究助理

• 产品情报

  • 发现产品评论中的趋势
  • 用户行为分析

• 销售

  • 个性化冷邮件
  • 情感分类

• 软件开发

  • 自动化 PR 审核

--> 完整文档

自定义

“全程 JSON”：分类工作流示例

以下示例将 IMDB 电影评论分类为“正面”或“负面”情感。请注意，在每一步中，您都可以查看、存储或串联部分结果——始终以 JSON 格式呈现。

from flashlearn.utils import imdb_reviews_50k
from flashlearn.skills import GeneralSkill
from flashlearn.skills.toolkit import ClassifyReviewSentiment
import json
import os


def main():
  os.environ["OPENAI_API_KEY"] = "API-KEY"

  # 步骤 1：加载或生成数据
  data = imdb_reviews_50k(sample=100)  # 100 条样本评论

  # 步骤 2：以字典格式加载技能的 JSON 定义
  skill = GeneralSkill.load_skill(ClassifyReviewSentiment)

  # 步骤 3：将技能定义保存为 JSON，以便后续加载
  # skill.save("BinaryClassificationSkill.json")

  # 步骤 5：将数据行转换为 JSON 任务
  tasks = skill.create_tasks(data)

  # 步骤 6：将结果保存到 JSONL 文件中，可立即运行或稍后执行
  with open('tasks.jsonl', 'w') as jsonl_file:
    for entry in tasks:
      jsonl_file.write(json.dumps(entry) + '\n')

  # 步骤 7：运行任务（默认并行）
  results = skill.run_tasks_in_parallel(tasks)

  # 步骤 8：每个输出都是严格的 JSON
  # 您可以轻松地将结果映射回输入
  # 例如，将结果存储为 JSON Lines
  with open('sentiment_results.jsonl', 'w') as f:
    for task_id, output in results.items():
      input_json = data[int(task_id)]
      input_json['result'] = output
      f.write(json.dumps(input_json) + '\n')

  # 步骤 9：检查或串联 JSON 结果
  print("示例结果:", results.get("0"))


if __name__ == "__main__":
  main()

输出始终以任务 ID（“0”、“1”等）为键，其 JSON 内容可供您的流水线解析或存储，无需猜测。

---

“技能”只是一个简单的字典

在内部，技能只是一个紧凑的类似 JSON 的对象，包含指令以及可选的函数定义，用于严格验证 LLM 的输出。您可以根据示例数据生成此技能（如上所示），也可以直接创建：

EvaluateToBuySkill = {
  "skill_class": "GeneralSkill",
  "system_prompt": "评估用户购买我们产品的可能性，返回一个 1 到 100 之间的整数及简短理由。",
  "function_definition": {
    "type": "function",
    "function": {
      "name": "EvaluateToBuySkill",
      "description": "根据用户文本评估其购买特定产品的意愿。",
      "strict": True,
      "parameters": {
        "type": "object",
        "properties": {
          "likely_to_buy": {
            "type": "integer",
            "description": "一个 1 到 100 之间的数字，表示用户购买的可能性。"
          },
          "reason": {
            "type": "string",
            "description": "解释为何得出该评分的简短原因。"
          }
        },
        "required": ["likely_to_buy", "reason"],
        "additionalProperties": False
      }
    }
  }
}

您可以根据需要将此技能加载或保存为 JSON，对其进行版本控制、共享，或将其集成到您的流水线中。FlashLearn 使整个流程——训练、存储、加载和使用此类自定义 LLM 转换——变得简单且统一。

---

使用预构建技能的单步分类

经典的分类任务就像调用机器学习估计器上的“fit_predict”一样简单：

import os
from openai import OpenAI
from flashlearn.skills.classification import ClassificationSkill

os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
data = [{"message": "我的退款在哪里？"}, {"message": "我的产品被损坏了！"}]

skill = ClassificationSkill(
    model_name="gpt-4o-mini",
    client=OpenAI(),
    categories=["billing","product issue"],
    system_prompt="对请求进行分类。"
)

tasks = skill.create_tasks(data)
print(skill.run_tasks_in_parallel(tasks))

---

高吞吐量

在本地机器上，您可以在 60 秒内处理多达 999 个任务。如需更高负载或企业级需求，请联系我们获取企业解决方案。

Processing tasks in parallel: 100%|██████████| 999/999 [01:00<00:00, 16.38 it/s, In total: 368969, Out total: 17288]
INFO:ParallelProcessor:All tasks complete. 999 succeeded, 0 failed.

---

并行执行：

run_tasks_in_parallel 会组织对 LLM 的并发请求。

成本估算

快速预览您的令牌使用情况：

cost_estimate = skill.estimate_tasks_cost(tasks)
print("预计成本:", cost_estimate)

---

加载技能

以下是该库如何处理喜剧性改写的方式：

from flashlearn.skills import GeneralSkill
from flashlearn.skills.toolkit import HumorizeText


def main():
  data = [{"original_text": "我们进度落后了。"}]
  skill = GeneralSkill.load_skill(HumorizeText)
  tasks = skill.create_tasks(data)
  results = skill.run_tasks_in_parallel(tasks)
  print(results)

您将看到如下输出：

{
  "0": {
    "comedic_version": "你这句话的爆笑解读……"
  }
}

所有内容均为结构良好的 JSON，适合进一步分析。

---

贡献与社区

• 采用 MIT 许可证。
• Fork 我们以添加新技能、修复错误或创建新示例。
• 我们致力于让所有初创公司都能轻松使用强大的 LLM 工作流。
• 所有代码的测试覆盖率至少需达到 95%
• 探索 examples 文件夹以了解更高级的使用模式。

---

许可证

MIT 许可证。
可用于商业产品和个人项目。

---

“Hello World” 示例

图像分类

import os
from openai import OpenAI
from flashlearn.skills.classification import ClassificationSkill
from flashlearn.utils import cats_and_dogs

def main():
    # os.environ["OPENAI_API_KEY"] = 'YOUR API KEY'
    data = cats_and_dogs(sample=6)

    skill = ClassificationSkill(
        model_name="gpt-4o-mini",
        client=OpenAI(),
        categories=["cat", "dog"],
        max_categories=1,
        system_prompt="请分类图片中的内容。"
    )

    column_modalities = {"image_base64": "image_base64"}
    tasks = skill.create_tasks(data, column_modalities=column_modalities)
    results = skill.run_tasks_in_parallel(tasks)
    print(results)

    # 保存技能定义以便重复使用
    skill.save("MyCustomSkillIMG.json")

if __name__ == "__main__":
    main()

文本分类

import json
import os
from openai import OpenAI
from flashlearn.skills.classification import ClassificationSkill
from flashlearn.utils import imdb_reviews_50k

def main():
    # os.environ["OPENAI_API_KEY"] = '你的 API 密钥'
    reviews = imdb_reviews_50k(sample=100)

    skill = ClassificationSkill(
        model_name="gpt-4o-mini",
        client=OpenAI(),
        categories=["positive", "negative"],
        max_categories=1,
        system_prompt="根据情感对短篇电影评论进行分类。"
    )

    # 将每一行转换为基于 JSON 的任务
    tasks = skill.create_tasks([{'review': x['review']} for x in reviews])
    results = skill.run_tasks_in_parallel(tasks)

    # 将预测的情感与真实标签进行比较，以计算准确率
    correct = 0
    for i, review in enumerate(reviews):
        predicted = results[str(i)]['categories']
        reviews[i]['predicted_sentiment'] = predicted
        if review['sentiment'] == predicted:
            correct += 1

    print(f'准确率: {round(correct / len(reviews), 2)}')

    # 将最终结果存储为 JSON Lines 格式
    with open('results.jsonl', 'w') as jsonl_file:
        for entry in reviews:
            jsonl_file.write(json.dumps(entry) + '\n')

    # 保存技能定义
    skill.save("BinaryClassificationSkill.json")

if __name__ == "__main__":
    main()

---

结语

FlashLearn 通过在每个步骤强制使用一致的 JSON 输出，使 LLM 工作流更加清晰明了。无论您是运行单个分类任务，还是复杂的多步流程，都可以存储中间结果、轻松调试，并确保数据的可靠性。

FlashLearn 快速上手指南

FlashLearn 是一个纯 Python 编写的开源库，旨在简化 Agent LLM（大语言模型智能体）在工作流和 ETL 管道中的集成。它支持高达 1000 次/分钟的调用速率，通过紧凑的 JSON 定义来管理任务，适用于数据转换、分类、摘要及多步骤自定义任务。

环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统：Linux, macOS 或 Windows
• Python 版本：推荐 Python 3.8 及以上版本
• 前置依赖：
  • pip 包管理工具
  • 已注册的 LLM 服务商 API Key（支持 OpenAI, DeepSeek, Ollama, LiteLLM 等兼容接口）
• 环境变量：建议准备一个 .env 文件或直接在代码中配置 API Key。

安装步骤

使用 pip 直接安装 FlashLearn：

pip install flashlearn

安装完成后，请配置您的 API Key。您可以将其添加到 .env 文件中：

OPENAI_API_KEY=your_api_key_here

或者在代码运行前导出到环境变量中。

基本使用

FlashLearn 的核心概念是"Skill（技能）”。您可以像训练机器学习模型一样“学习”一个新技能，然后将其应用于数据列表。以下是评估用户评论购买意向的完整示例：

1. 定义并保存技能 (Learn Skill)

首先，创建一个技能定义。这一步类似于 ML 中的 fit 过程，LLM 将根据您的指令学习如何处理数据。

from flashlearn.skills.learn_skill import LearnSkill
from openai import OpenAI
import json

# 初始化客户端 (此处以 OpenAI 为例，也可替换为 DeepSeek 或 Ollama)
client = OpenAI() 

# 实例化学习器
learner = LearnSkill(model_name="gpt-4o-mini", client=client)

# 定义任务指令并学习技能
skill = learner.learn_skill(
    df=[], # 可选：在此处传入少量样本数据以辅助学习
    task=(
        "Evaluate how likely the user is to buy my product based on the sentiment in their comment, "
        "return an integer 1-100 on key 'likely_to_buy', "
        "and a short explanation on key 'reason'."
    ),
)

# 将技能保存为 JSON 文件，以便后续重复使用
skill.save("evaluate_buy_comments_skill.json")

2. 加载技能并执行任务 (Run Skill)

技能保存后，您可以随时加载它并批量处理数据。输入数据应为字典列表。

from flashlearn.skills.general_skill import GeneralSkill
import json

# 准备输入数据 (列表包含多个字典)
user_inputs = [
    {"comment_text": "I love this product, it's everything I wanted!"},
    {"comment_text": "Not impressed... wouldn't consider buying this."},
]

# 加载之前保存的技能定义
with open("evaluate_buy_comments_skill.json", "r", encoding="utf-8") as file:
    definition = json.load(file)

skill = GeneralSkill.load_skill(definition)

# 创建任务并并行执行
tasks = skill.create_tasks(user_inputs)
results = skill.run_tasks_in_parallel(tasks)

# 输出结构化结果
print(json.dumps(results, indent=2))

3. 获取结果

执行完成后，您将得到标准的 JSON 格式结果，键名对应输入列表的索引：

{
  "0": {
    "likely_to_buy": 90,
    "reason": "Comment shows strong enthusiasm and positive sentiment."
  },
  "1": {
    "likely_to_buy": 25,
    "reason": "Expressed disappointment and reluctance to purchase."
  }
}

进阶提示：切换模型提供商

FlashLearn 原生支持多种兼容 OpenAI 接口的模型。只需更改 client 初始化参数即可：

• DeepSeek:

  deep_seek = OpenAI(api_key='YOUR_DEEPSEEK_KEY', base_url="https://api.deepseek.com")
  

• Ollama (本地部署):

  ollama = OpenAI(base_url='http://localhost:11434/v1', api_key='ollama')
  

• LiteLLM:

  from flashlearn.utils import FlashLiteLLMClient
  lite_llm = FlashLiteLLMClient()
  

通过以上步骤，您即可在几分钟内将 LLM 能力集成到您的数据处理管道中。