BatchFactory

Composable, cache‑aware pipelines for parallel LLM workflows, API calls, and dataset generation.

Status — v0.6 beta. More robust and battle-tested on small projects. Optimized for pressure under large dataset scale.

📦 GitHub Repository →

Install

pip install batchfactory            # latest tag
pip install --upgrade batchfactory  # grab the newest patch

Quick‑start

import batchfactory as bf
from batchfactory.op import *

PROMPT = """
Write a poem about {keyword}.
"""

with bf.ProjectFolder("quickstart", 1, 0, 7) as project:
    g = bf.Graph()
    g |= ReadMarkdownLines("./demo_data/greek_mythology_stories.md")
        # load keywords
    g |= Shuffle(seed=42) | TakeFirstN(5)
        # random sample
    g |= AskLLM(PROMPT, model="gpt-4o-mini@openai")
        # generate poems
    g |= MapField(lambda headings, keyword: headings + [keyword], ["headings", "keyword"], "headings")
        # tag heading
    g |= WriteMarkdownEntries(project["out/poems.md"])
        # save results

g.execute(dispatch_brokers=True)

Run it twice – everything after the first run is served from the on‑disk ledger.

🚀 Why BatchFactory?

BatchFactory lets you build cache‑aware, composable pipelines for LLM calls, embeddings, and data transforms—so you can go from idea to production with zero boilerplate.

• Composable Ops – chain 30‑plus ready‑made Ops (and your own) using simple pipe syntax.
• Transparent Caching & Cost Tracking – every expensive call is hashed, cached, resumable, and audited.
• Pluggable Brokers – swap in LLM, embedding, search, or human‑in‑the‑loop brokers at will.
• Self‑contained datasets – pack arrays, images, audio—any data—into each entry so your entire workflow travels as a single, copy‑anywhere .jsonl file.
• Ready‑to‑Copy Demos – learn the idioms fast with five concise example pipelines.

🧩 Three killer moves

Core concepts (one‑liner view)

Spawn snippet (Text Segmentation)

g |= MapField(create_input_chunks, "text", "text_segments")
    spawn_chain = AskLLM(LABEL_SEG_PROMPT, model=model, output_key="labels")
    spawn_chain |= MapField(F.text_to_integer_list, "labels")
    g | ListParallel(spawn_chain, "text_segments", "text", "labels", "labels")
    g |= MapField(post_processing, ["text", "labels"], ["text_segments", "labels"])
    g |= ExplodeList(["text_segments"],["text"])

Loop snippet (Role‑Playing)

with bf.ProjectFolder("roleplay", 1, 0, 5) as project:
    ###### Setup topics ######
    g = ReadMarkdownLines("./demo_data/greek_mythology_stories.md") | TakeFirstN(1)

    ###### Create the characters and their settings ######
    Teacher = AICharacter("Teacher", "You are a teacher. "+FORMAT_REQ, model=model)
    Student = AICharacter("Student", "You are a student. "+FORMAT_REQ, model=model)

    ###### Introduction of the Role Playing Session ######
    g |= Teacher("Please introduce the text from {headings} titled {keyword}.")
    ###### Main Role Playing Loop ######
    loop_body = Student("Please ask questions or respond.")
    loop_body |= Teacher("Please respond to the student or continue explaining.")
    g |= Repeat(loop_body, 3)
    ###### Wrap Up ######
    g |= Teacher("Please summarize.")
    g |= ChatHistoryToText(template="**{role}**: {content}\n\n")

    ###### Export the Role Playing Session ######
    g |= MapField(lambda headings,keyword: headings+[keyword], ["headings", "keyword"], "headings")
    g |= WriteMarkdownEntries(project["out/roleplay.md"])

Text Embedding snippet

g |= EmbedText("keyword", model="text-embedding-3-small@openai", output_format="list")

📚 Example Gallery

Available Ops

© 2025 · MIT License