Flash Learn - Agents made simple

FlashLearn provides a simple interface and orchestration (up to 1000 calls/min) for incorporating Agent LLMs into your typical workflows and ETL pipelines. Conduct data transformations, classifications, summarizations, rewriting, and custom multi-step tasks, just like you’d do with any standard ML library, harnessing the power of LLMs under the hood. Each step and task has a compact JSON definition which makes pipelines simple to understand and maintain. It supports LiteLLM, Ollama, OpenAI, DeepSeek, and all other OpenAI-compatible clients.

🚀 Examples

📖 Full Documentation

Installation

pip install flashlearn

Add the API keys for the provider you want to use to your .env file.

OPENAI_API_KEY=

High-Level Concept Flow

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

    H[Your Data] --> I[Load Skill / Learn Skill]
    I --> J[Create Tasks]
    J --> K[Run Tasks]
    K --> L[Structured Results]
    L --> M[Downstream Steps]

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

Learning a New “Skill”

Like a fit/predict pattern, you can quickly “learn” a custom skill. Below, we’ll create a skill that evaluates the likelihood of buying a product from user comments on social media posts, returning a score (1–100) and a short reason. We’ll instruct the LLM to transform each comment according to our custom specifications.

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

# Instantiate your pipeline “estimator” or “transformer”
learner = LearnSkill(model_name="gpt-4o-mini", client=OpenAI())
# Provide instructions and sample data for the new skill
skill = learner.learn_skill(
    df=[], #dif you want you can also pass data sample
    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'."
    ),
)

# Save skill to be used from any system
skill.save("evaluate_buy_comments_skill.json")

Input Is a List of Dictionaries

Whether you retrieved data from an API, a spreadsheet, or user-submitted forms, you can simply wrap each record into a dictionary. FlashLearn’s “skills” accept a list of such dictionaries, as shown below:

user_inputs = [
    {"comment_text": "I love this product, it's everything I wanted!"},
    {"comment_text": "Not impressed... wouldn't consider buying this."},
    # ...
]

Run in 3 Lines of Code

Once you’ve defined or learned a skill, you can load it as though it were a specialized transformer in a standard ML pipeline. Then apply it to your data in just a few lines:

from flashlearn.skills.general_skill import GeneralSkill

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

# Suppose we previously saved a learned skill to "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)

Get Structured Results

FlashLearn returns structured outputs for each of your records. The keys in the results dictionary map to the indexes of your original list. For example:

{
  "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."
  }
}

Pass on to Next Steps

Each record’s output can then be used in downstream tasks. For instance, you might:

1. Store the results in a database
2. Filter for high-likelihood leads
3. Send them to another tool for further analysis (for example, rewriting the “reason” in a formal tone)

Below is a small example showing how you might parse the dictionary and feed it into a separate function:

# Suppose 'flash_results' is the dictionary with structured LLM outputs
for idx, result in flash_results.items():
    desired_score = result["likely_to_buy"]
    reason_text = result["reason"]
    # Now do something with the score and reason, e.g., store in DB or pass to next step
    print(f"Comment #{idx} => Score: {desired_score}, Reason: {reason_text}")

Supported LLM Providers

Anywhere you might rely on an ML pipeline component, you can swap in an LLM:

client = OpenAI()  # This is equivalent to instantiating a pipeline component 
deep_seek = OpenAI(api_key='YOUR DEEPSEEK API KEY', base_url="https://api.deepseek.com")
lite_llm = FlashLiteLLMClient()  # LiteLLM integration Manages keys as environment variables, akin to a top-level pipeline manager
ollama =  OpenAI(base_url = 'http://localhost:11434/v1', api_key='ollama', # required, but unused) # Just use ollama's openai client

KEY IDEA: JSON in, JSON out

Examples by use case

• Customer service

  • Classifying customer tickets

• Finance

  • Parse financial report data

• Marketing

  • Customer segmentation

• Personal assistant

  • Research assistant

• Product intelligence

  • Discover trends in product_reviews
  • User behaviour analysis

• Sales

  • Personalized cold emails
  • Sentiment classification

• Software development

  • Automated PR reviews

--> Full Documentation

Customization

“All JSON, All the Time”: Example Classification Workflow

The following example classifies IMDB movie reviews into “positive” or “negative” sentiment. Notice how at each step you can view, store, or chain the partial results—always in JSON format.

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"

  # Step 1: Load or generate your data
  data = imdb_reviews_50k(sample=100)  # 100 sample reviews

  # Step 2: Load JSON definition of skill in dict format
  skill = GeneralSkill.load_skill(ClassifyReviewSentiment)

  # Step 3: Save skill definition in JSON for later loading
  # skill.save("BinaryClassificationSkill.json")

  # Step 5: Convert data rows into JSON tasks
  tasks = skill.create_tasks(data)

  # Step 6: Save results to a JSONL file and run now or later
  with open('tasks.jsonl', 'w') as jsonl_file:
    for entry in tasks:
      jsonl_file.write(json.dumps(entry) + '\n')

  # Step 7: Run tasks (in parallel by default)
  results = skill.run_tasks_in_parallel(tasks)

  # Step 8: Every output is strict JSON
  # You can easily map results back to inputs
  # e.g., store results as 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')

  # Step 9: Inspect or chain the JSON results
  print("Sample result:", results.get("0"))


if __name__ == "__main__":
  main()

The output is consistently keyed by task ID ("0", "1", etc.), with the JSON content that your pipeline can parse or store with no guesswork.

“Skill” Is Just a Simple Dictionary

Internally, a skill is just a compact JSON-like object containing instructions and, optionally, function definitions to strictly validate LLM outputs. You can generate this skill from sample data (as shown above) or create it directly:

EvaluateToBuySkill = {
  "skill_class": "GeneralSkill",
  "system_prompt": "Evaluate how likely the user is to buy our product, returning an integer 1-100 and a short reason.",
  "function_definition": {
    "type": "function",
    "function": {
      "name": "EvaluateToBuySkill",
      "description": "Assess user text with respect to willingness to buy a certain product.",
      "strict": True,
      "parameters": {
        "type": "object",
        "properties": {
          "likely_to_buy": {
            "type": "integer",
            "description": "A number from 1 to 100 indicating how likely the user is to buy."
          },
          "reason": {
            "type": "string",
            "description": "A brief explanation of why this is the likely score."
          }
        },
        "required": ["likely_to_buy", "reason"],
        "additionalProperties": False
      }
    }
  }
}

You can load or save this skill to JSON as needed, version it, share it, or plug it into your pipelines. FlashLearn makes the entire process—training, storing, loading, and using such custom LLM transformations—simple and uniform.

Single-Step Classification Using Prebuilt Skills

Classic classification tasks are as straightforward as calling “fit_predict” on a ML estimator:

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

os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
data = [{"message": "Where is my refund?"}, {"message": "My product was damaged!"}]

skill = ClassificationSkill(
    model_name="gpt-4o-mini",
    client=OpenAI(),
    categories=["billing","product issue"],
    system_prompt="Classify the request."
)

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

High-throughput

Process up to 999 tasks in 60 seconds on your local machine. For higher loads and enterprise needs contact us for enterprise solution.

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.

Parallel Execution:

run_tasks_in_parallel organizes concurrent requests to the LLM.

Cost Estimation

Quickly preview your token usage:

cost_estimate = skill.estimate_tasks_cost(tasks)
print("Estimated cost:", cost_estimate)

Loading a Skill

Here’s how the library handles comedic rewrites:

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


def main():
  data = [{"original_text": "We are behind schedule."}]
  skill = GeneralSkill.load_skill(HumorizeText)
  tasks = skill.create_tasks(data)
  results = skill.run_tasks_in_parallel(tasks)
  print(results)

You’ll see output like:

{
  "0": {
    "comedic_version": "Hilarious take on your sentence..."
  }
}

Everything is well-structured JSON, suitable for further analysis.

Contributing & Community

• Licensed under MIT.
• Fork us to add new skills, fix bugs, or create new examples.
• We aim to make robust LLM workflows accessible to all startups.
• All code needs at least 95% test coverage
• Explore the examples folder for more advanced usage patterns.

License

MIT License.
Use it in commercial products, and personal projects.

“Hello World” Demos

Image Classification

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="Classify what's in the picture."
    )

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

    # Save skill definition for reuse
    skill.save("MyCustomSkillIMG.json")

if __name__ == "__main__":
    main()

Text Classification

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"] = 'YOUR API KEY'
    reviews = imdb_reviews_50k(sample=100)

    skill = ClassificationSkill(
        model_name="gpt-4o-mini",
        client=OpenAI(),
        categories=["positive", "negative"],
        max_categories=1,
        system_prompt="Classify short movie reviews by sentiment."
    )

    # Convert each row into a JSON-based task
    tasks = skill.create_tasks([{'review': x['review']} for x in reviews])
    results = skill.run_tasks_in_parallel(tasks)

    # Compare predicted sentiment with ground truth for accuracy
    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'Accuracy: {round(correct / len(reviews), 2)}')

    # Store final results as JSON Lines
    with open('results.jsonl', 'w') as jsonl_file:
        for entry in reviews:
            jsonl_file.write(json.dumps(entry) + '\n')

    # Save the skill definition
    skill.save("BinaryClassificationSkill.json")

if __name__ == "__main__":
    main()

Final Words

FlashLearn brings clarity to LLM workflows by enforcing consistent JSON output at every step. Whether you run a single classification or a multi-step pipeline, you can store partial results, debug easily, and maintain confidence in your data.