quantalogic-flow 0.1.0

Standalone flow package extracted from Quantalogic

Quantalogic Flow: Your Workflow Automation Powerhouse 🚀

Welcome to Quantalogic Flow, an open-source Python library designed to make workflow automation intuitive, scalable, and powerful. Whether you're orchestrating AI-driven tasks with Large Language Models (LLMs), processing data pipelines, or formatting outputs with templates, Quantalogic Flow has you covered with two flexible approaches: a declarative YAML interface for simplicity and a fluent Python API for dynamic control.

This README is your guide to mastering Quantalogic Flow. Packed with examples, visualizations, and insider tips, it’ll take you from beginner to pro in no time. Let’s dive in and start building workflows that work smarter, not harder!

---

Table of Contents

1. Why Quantalogic Flow?
2. Installation and Setup
3. Core Concepts
   • Nodes: The Building Blocks
   • Workflows: The Roadmap
   • Context: The Glue
4. Approaches: YAML vs. Fluent API
   • YAML Approach
   • Fluent API Approach
5. Advanced Features
   • Input Mapping
   • Dynamic Model Selection
   • Sub-Workflows
   • Observers
6. Validation and Debugging
7. Conversion Tools
8. Case Study: AI-Powered Story Generator
9. Best Practices and Insider Tips
10. Resources and Community

---

Why Quantalogic Flow?

Why: Workflows—like generating reports, processing data, or creating AI-driven content—often involve repetitive steps, conditional logic, and data handoffs. Writing this logic from scratch is time-consuming and error-prone. Quantalogic Flow simplifies this by letting you define workflows declaratively or programmatically, saving hours and reducing bugs.

What: Quantalogic Flow is a Python library that enables:

• Declarative YAML workflows: Human-readable, shareable, and perfect for static processes.
• Fluent Python API: Dynamic, code-driven workflows for developers.
• LLM integration: Seamlessly use AI models for text generation or structured data extraction.
• Template rendering: Format outputs with Jinja2 for polished results.
• Advanced logic: Support for branching, looping, and parallel execution.

How: You define nodes (tasks) and workflows (sequences), then execute them with a shared context to pass data. Whether you're a non-coder editing YAML or a developer chaining Python methods, Quantalogic Flow adapts to your style.

---

Installation and Setup

Prerequisites

• Python 3.10 or higher.
• Basic knowledge of Python and YAML.
• Optional: API keys for LLM providers (e.g., Gemini, OpenAI).

Installation

Install Quantalogic Flow via pip:

pip install quantalogic-flow

For template nodes, install Jinja2:

pip install jinja2

For structured LLM nodes, install Instructor:

pip install instructor[litellm]

Setup

Set up your environment by configuring LLM API keys (if using LLM nodes):

export GEMINI_API_KEY="your-api-key"

---

Core Concepts

Nodes: The Building Blocks

Nodes are the individual tasks in a workflow, like workers in a factory. Quantalogic Flow supports four types:

1. Function Nodes: Run custom Python code (e.g., data processing).
2. LLM Nodes: Generate text using AI models (e.g., story writing).
3. Structured LLM Nodes: Extract structured data (e.g., JSON or Pydantic models).
4. Template Nodes: Render formatted text with Jinja2 (e.g., reports).

Workflows: The Roadmap

Workflows define how nodes connect, like a recipe directing kitchen staff. They specify:

• A start node to begin execution.
• Transitions for sequential, parallel, or conditional flow.
• Convergence nodes where parallel paths merge.

Context: The Glue

The context (ctx) is a dictionary that carries data between nodes, like a shared clipboard. Nodes read inputs from the context and write outputs to it.

Mermaid Diagram: Core Workflow Structure

graph TD
    A[Start Node] --> B[Node 2]
    B -->|Condition A| C[Node 3]
    B -->|Condition B| D[Node 4]
    C --> E[Convergence Node]
    D --> E
    E --> F[End Node]
    style A fill:#90CAF9
    style B fill:#90CAF9
    style C fill:#90CAF9
    style D fill:#90CAF9
    style E fill:#90CAF9,stroke-dasharray:5 5
    style F fill:#90CAF9

---

Approaches: YAML vs. Fluent API

Quantalogic Flow offers two ways to define workflows: YAML for simplicity and Fluent API for flexibility. Let’s explore both with a simple example: a workflow that reads a string, converts it to uppercase, and prints it.

YAML Approach

Why: YAML is declarative, readable, and ideal for static workflows or non-coders. What: Define functions, nodes, and workflow structure in a YAML file. How:

functions:
  read_data:
    type: embedded
    code: |
      def read_data():
          return "hello world"
  process_data:
    type: embedded
    code: |
      def process_data(data):
          return data.upper()
  write_data:
    type: embedded
    code: |
      def write_data(processed_data):
          print(processed_data)
nodes:
  start:
    function: read_data
    output: data
  process:
    function: process_data
    inputs_mapping: {"data": "data"}
    output: processed_data
  end:
    function: write_data
    inputs_mapping: {"processed_data": "processed_data"}
workflow:
  start: start
  transitions:
    - from_node: start
      to_node: process
    - from_node: process
      to_node: end

Execution:

from quantalogic_flow.flow.flow_manager import WorkflowManager
import asyncio

manager = WorkflowManager()
manager.load_from_yaml("simple_workflow.yaml")
workflow = manager.instantiate_workflow()
result = asyncio.run(workflow.build().run({}))
print(result)  # Outputs: HELLO WORLD

Fluent API Approach

Why: The Fluent API is programmatic, dynamic, and perfect for developers integrating workflows with Python logic. What: Use method chaining to define nodes and transitions. How:

from quantalogic_flow.flow import Nodes, Workflow

@Nodes.define(output="data")
def read_data():
    return "hello world"

@Nodes.define(output="processed_data")
def process_data(data):
    return data.upper()

@Nodes.define()
def write_data(processed_data):
    print(processed_data)

workflow = (
    Workflow("read_data")
    .node("read_data")
    .then("process_data")
    .then("write_data")
)

async def main():
    result = await workflow.build().run({})
    print(result)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())  # Outputs: HELLO WORLD

Comparison Table:

Feature	YAML	Fluent API
Style	Declarative, static	Programmatic, dynamic
Best For	Non-coders, static flows	Developers, dynamic logic
Readability	High, non-technical	Moderate, Python-based
Flexibility	Limited by YAML structure	High, full Python power

Mermaid Diagram: Workflow Flow

graph TD
    A[read_data] --> B[process_data]
    B --> C[write_data]
    style A fill:#90CAF9
    style B fill:#90CAF9
    style C fill:#90CAF9

Insider Tip: Use YAML for team collaboration or quick prototyping, and switch to Fluent API when you need runtime decisions or integration with existing Python code.

---

Advanced Features

Input Mapping

Why: Hardcoding inputs limits flexibility. Input mapping lets nodes dynamically pull data from the context or compute values. What: Map node parameters to context keys or lambda expressions. How:

nodes:
  process:
    function: process_data
    inputs_mapping:
      data: "data"
      prefix: "lambda ctx: 'Processed: ' + ctx['data']"
    output: processed_data

Fluent API:

.node("process_data", inputs_mapping={"data": "data", "prefix": lambda ctx: "Processed: " + ctx["data"]})

Dynamic Model Selection

Why: Different tasks may need different LLMs (e.g., GPT for creativity, Gemini for speed). What: Specify the LLM model dynamically using a lambda. How:

nodes:
  generate:
    llm_config:
      model: "lambda ctx: ctx['model_name']"
      prompt_template: "Write about {topic}."
    inputs_mapping:
      topic: "user_topic"
    output: text

Context Example:

{"model_name": "gemini/gemini-2.0-flash", "user_topic": "space travel"}

Sub-Workflows

Why: Break complex workflows into reusable modules. What: Define nested workflows within a node. How:

nodes:
  parent_node:
    sub_workflow:
      start: sub_start
      transitions:
        - from_node: sub_start
          to_node: sub_end

Fluent API:

sub_workflow = Workflow("sub_start").then("sub_end")
workflow.add_sub_workflow("parent_node", sub_workflow, inputs={"key": "value"}, output="result")

Observers

Why: Monitor workflow execution for debugging or logging. What: Functions that receive events (e.g., node started, completed). How:

functions:
  monitor:
    type: embedded
    code: |
      def monitor(event):
          print(f"Event: {event.event_type.value} @ {event.node_name}")
observers:
  - monitor

Fluent API:

workflow.add_observer(monitor)

Mermaid Diagram: Observer Integration

graph TD
    A[Workflow Execution] --> B[Node 1]
    B --> C[Node 2]
    A --> D[Observer]
    B --> D
    C --> D
    style A fill:#90CAF9
    style B fill:#90CAF9
    style C fill:#90CAF9
    style D fill:#FFCCBC

Insider Tip: Use observers to track LLM token usage or log errors for quick debugging.

---

Validation and Debugging

• Validation: Ensure your workflow is sound with validate_workflow_definition().

  from quantalogic_flow.flow.flow_validator import validate_workflow_definition
  issues = validate_workflow_definition(manager.workflow)
  for issue in issues:
      print(f"Node '{issue.node_name}': {issue.description}")
  

• Debugging: Attach observers to log context changes or use print statements in function nodes.

Insider Tip: Validate early and often to catch circular transitions or missing inputs before execution.

---

Conversion Tools

Switch between YAML and Python effortlessly:

• YAML to Python: Generate executable scripts with flow_generator.py.

  from quantalogic_flow.flow.flow_generator import generate_executable_script
  manager = WorkflowManager()
  manager.load_from_yaml("workflow.yaml")
  generate_executable_script(manager.workflow, {}, "script.py")
  

• Python to YAML: Extract Fluent API workflows with flow_extractor.py.

  from quantalogic_flow.flow.flow_extractor import extract_workflow_from_file
  workflow_def, globals = extract_workflow_from_file("script.py")
  WorkflowManager(workflow_def).save_to_yaml("workflow.yaml")
  

Insider Tip: Use conversion tools to prototype in YAML, then fine-tune in Python for dynamic tweaks.

---

Case Study: AI-Powered Story Generator

Let’s build a workflow that generates a multi-chapter story, analyzes its tone, and formats it with a template.

Objective

1. Generate a story outline with an LLM.
2. Analyze its tone (light or dark) with a structured LLM.
3. Generate chapters based on tone.
4. Summarize each chapter with a template.
5. Loop until all chapters are done, then finalize the story.

YAML Definition

functions:
  update_progress:
    type: embedded
    code: |
      async def update_progress(**context):
          chapters = context.get('chapters', [])
          completed_chapters = context.get('completed_chapters', 0)
          chapter_summary = context.get('chapter_summary', '')
          updated_chapters = chapters + [chapter_summary]
          return {**context, "chapters": updated_chapters, "completed_chapters": completed_chapters + 1}
  check_if_complete:
    type: embedded
    code: |
      async def check_if_complete(completed_chapters=0, num_chapters=0):
          return completed_chapters < num_chapters
  finalize_story:
    type: embedded
    code: |
      async def finalize_story(chapters):
          return "\n".join(chapters)
  monitor:
    type: embedded
    code: |
      def monitor(event):
          print(f"Event: {event.event_type.value} @ {event.node_name}")
nodes:
  generate_outline:
    llm_config:
      model: "lambda ctx: ctx['model_name']"
      system_prompt: "You are a creative writer."
      prompt_template: "Create a story outline for a {genre} story with {num_chapters} chapters."
    inputs_mapping:
      genre: "story_genre"
      num_chapters: "chapter_count"
    output: outline
  analyze_tone:
    llm_config:
      model: "lambda ctx: ctx['model_name']"
      system_prompt: "Analyze the tone."
      prompt_template: "Determine if this outline is light or dark: {outline}."
      response_model: "path.to.ToneModel"
    inputs_mapping:
      outline: "outline"
    output: tone
  generate_chapter:
    llm_config:
      model: "lambda ctx: ctx['model_name']"
      system_prompt: "You are a writer."
      prompt_template: "Write chapter {chapter_num} for this outline: {outline}. Style: {style}."
    inputs_mapping:
      chapter_num: "completed_chapters"
      style: "style"
    output: chapter
  summarize_chapter:
    template_config:
      template: "Chapter {{ chapter_num }}: {{ chapter }}"
    inputs_mapping:
      chapter_num: "completed_chapters"
      chapter: "chapter"
    output: chapter_summary
  update_progress:
    function: update_progress
    output: updated_context
  check_if_complete:
    function: check_if_complete
    output: continue_generating
  finalize_story:
    function: finalize_story
    output: final_story
workflow:
  start: generate_outline
  transitions:
    - from_node: generate_outline
      to_node: analyze_tone
    - from_node: analyze_tone
      to_node:
        - to_node: generate_chapter
          condition: "ctx['tone'] == 'light'"
        - to_node: generate_chapter
          condition: "ctx['tone'] == 'dark'"
    - from_node: generate_chapter
      to_node: summarize_chapter
    - from_node: summarize_chapter
      to_node: update_progress
    - from_node: update_progress
      to_node: check_if_complete
    - from_node: check_if_complete
      to_node: generate_chapter
      condition: "ctx['continue_generating']"
  convergence_nodes:
    - finalize_story
observers:
  - monitor

Pydantic Model

from pydantic import BaseModel

class ToneModel(BaseModel):
    tone: str  # e.g., "light" or "dark"

Execution

from quantalogic_flow.flow.flow_manager import WorkflowManager
import asyncio

manager = WorkflowManager()
manager.load_from_yaml("story_generator.yaml")
workflow = manager.instantiate_workflow()

async def main():
    context = {
        "story_genre": "fantasy",
        "chapter_count": 2,
        "chapters": [],
        "completed_chapters": 0,
        "style": "epic",
        "model_name": "gemini/gemini-2.0-flash"
    }
    result = await workflow.build().run(context)
    print(f"Story:\n{result['final_story']}")

asyncio.run(main())

Mermaid Diagram: Story Generator Workflow

graph TD
    A["generate_outline<br>(LLM)"] --> B["analyze_tone<br>(Structured LLM)"]
    B -->|light| C["generate_chapter (LLM)"]
    B -->|dark| C
    C --> D["summarize_chapter<br>(Template)"]
    D --> E["update_progress<br>(Function)"]
    E --> F["check_if_complete<br>(Function)"]
    F -->|yes| C
    F --> G["finalize_story<br>(Function)"]
    E --> G
    style A fill:#CE93D8
    style B fill:#A5D6A7
    style C fill:#CE93D8
    style D fill:#FCE4EC
    style E fill:#90CAF9
    style F fill:#90CAF9
    style G fill:#90CAF9,stroke-dasharray:5 5

Sample Output:

Event: workflow_started @ workflow
Event: node_started @ generate_outline
Event: node_completed @ generate_outline
...
Story:
Chapter 1: A mage discovers a prophecy...
Chapter 2: The mage defeats the dragon...

Insider Tip: Use prompt_file for reusable LLM prompts stored in Jinja2 templates to keep your YAML clean and modular.

---

Best Practices and Insider Tips

1. Start Small: Begin with a simple workflow (e.g., two nodes) to grasp the context flow.
2. Validate Early: Run validate_workflow_definition() before execution to catch errors.
3. Optimize LLMs: Set temperature=0.3 for consistent outputs, 0.7 for creative tasks.
4. Reuse Sub-Workflows: Encapsulate common patterns (e.g., data validation) for modularity.
5. Log Everything: Attach observers to track context changes and debug issues.
6. Test Incrementally: Add nodes one at a time and test to isolate problems.
7. Document YAML: Use comments to explain node purposes for team collaboration.

---

Resources and Community

• Documentation: Quantalogic Flow Docs
• GitHub: Repository
• Issues: Report bugs or request features on GitHub.

---

Conclusion

Quantalogic Flow empowers you to automate complex tasks with ease, whether you prefer the simplicity of YAML or the power of Python. From AI-driven content creation to data pipelines, this library is your toolkit for smarter workflows. Try the story generator, experiment with your own projects, and join the community to share your creations. Happy flowing! 🚀