Edit

Python Examples

Python Examples

All projects listed below can be found in our cookbook directory.

They have been sorted by complexity points - which are represented by the scale 🥚→🐣→🐥→🐓→🦕→🦖→☄️→🐭→🐒→🧠→⚙️→🤖→👾→🛸→🌌 - to help you easily find projects that suit your skills.

Caskada Hello World (python-hello-world)

Complexity Points: 2.5 └🥚

Your first Caskada application! This simple example demonstrates how to create a basic Caskada app from scratch.

Details

Caskada Hello World

Your first Caskada application! This simple example demonstrates how to create a basic Caskada app from scratch.

Project Structure

.
├── docs/          # Documentation files
├── utils/         # Utility functions
├── flow.py        # Caskada implementation
├── main.py        # Main application entry point
└── README.md      # Project documentation

Setup

  1. Create a virtual environment:

python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

  1. Install dependencies:

pip install -r requirements.txt

  1. Run the example:

python main.py

What This Example Demonstrates

  • How to create your first Caskada application

  • Basic Caskada concepts and usage

  • Simple example of Caskada's capabilities

Additional Resources

Text Summarization (python-node)

Complexity Points: 3 └🥚

A practical example demonstrating how to use Caskada to build a robust text summarization tool with error handling and retries. This example showcases core Caskada concepts in a real-world application.

Details

Text Summarization

A practical example demonstrating how to use Caskada to build a robust text summarization tool with error handling and retries. This example showcases core Caskada concepts in a real-world application.

Features

  • Text summarization using LLMs (Large Language Models)

  • Automatic retry mechanism (up to 3 attempts) on API failures

  • Graceful error handling with fallback responses

  • Clean separation of concerns using Caskada's Node architecture

Project Structure

.
├── docs/          # Documentation files
├── utils/         # Utility functions (LLM API wrapper)
├── flow.py        # Caskada implementation with Summarize Node
├── main.py        # Main application entry point
└── README.md      # Project documentation

Implementation Details

The example implements a simple but robust text summarization workflow:

  1. Summarize Node (flow.py):

    • prep(): Retrieves text from the shared store

    • exec(): Calls LLM to summarize text in 10 words

    • exec_fallback(): Provides graceful error handling

    • post(): Stores the summary back in shared store

  2. Flow Structure:

    • Single node flow for demonstration

    • Configured with 3 retries for reliability

    • Uses shared store for data passing

Setup

  1. Create a virtual environment:

python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

  1. Install dependencies:

pip install -r requirements.txt

  1. Configure your environment:

    • Set up your LLM API key (check utils/call_llm.py for configuration)

  2. Run the example:

python main.py

Example Usage

The example comes with a sample text about Caskada, but you can modify main.py to summarize your own text:

shared = {"data": "Your text to summarize here..."}
await flow.run(shared)
print("Summary:", shared["summary"])

What You'll Learn

This example demonstrates several key Caskada concepts:

  • Node Architecture: How to structure LLM tasks using prep/exec/post pattern

  • Error Handling: Implementing retry mechanisms and fallbacks

  • Shared Store: Using shared storage for data flow between steps

  • Flow Creation: Setting up a basic Caskada workflow

Additional Resources

Structured Output Demo (python-structured-output)

Complexity Points: 3 └🥚

A minimal demo application showing how to use Caskada to extract structured data from a resume using direct prompting and YAML formatting. Why YAML? Check out the doc or Why JSON Costs More Than TSV.

Details

Structured Output Demo

A minimal demo application showing how to use Caskada to extract structured data from a resume using direct prompting and YAML formatting. Why YAML? Check out the doc or Why JSON Costs More Than TSV.

This implementation is based on this tutorial (for Pocketflow): Structured Output for Beginners: 3 Must-Know Prompting Tips.

Features

  • Extracts structured data using prompt engineering

  • Validates output structure before processing

Run It

  1. Install the packages you need with this simple command:

  2. Make sure your OpenAI API key is set:

    Alternatively, you can edit the utils.py file to include your API key directly.

    Let's do a quick check to make sure your API key is working properly:

  3. Edit data.txt with the resume you want to parse (a sample resume is already included)

  4. Run the application:

How It Works

The Resume Parser application uses a single node that:

  1. Takes resume text from the shared state (loaded from data.txt)

  2. Sends the resume to an LLM with a prompt that requests YAML formatted output

  3. Extracts and validates the structured YAML data

  4. Outputs the structured result

Files

Example Output

OpenAI Embeddings with Caskada (python-tool-embeddings)

Complexity Points: 3 └🥚

This example demonstrates how to properly integrate OpenAI's text embeddings API with Caskada, focusing on:

Details

OpenAI Embeddings with Caskada

This example demonstrates how to properly integrate OpenAI's text embeddings API with Caskada, focusing on:

  1. Clean code organization with separation of concerns:

    • Tools layer for API interactions (tools/embeddings.py)

    • Node implementation for Caskada integration (nodes.py)

    • Flow configuration (flow.py)

    • Centralized environment configuration (utils/call_llm.py)

  2. Best practices for API key management:

    • Using environment variables

    • Supporting both .env files and system environment variables

    • Secure configuration handling

  3. Proper project structure:

    • Modular code organization

    • Clear separation between tools and Caskada components

    • Reusable OpenAI client configuration

Project Structure

Setup

  1. Create a virtual environment:

  1. Install dependencies:

  1. Set up your OpenAI API key in one of two ways:

    a. Using a .env file:

    b. Or as a system environment variable:

Usage

Run the example:

This will:

  1. Load the OpenAI API key from environment

  2. Create a Caskada node to handle embedding generation

  3. Process a sample text and generate its embedding

  4. Display the embedding dimension and first few values

Key Concepts Demonstrated

  1. Environment Configuration

    • Secure API key handling

    • Flexible configuration options

  2. Code Organization

    • Clear separation between tools and Caskada components

    • Reusable OpenAI client configuration

    • Modular project structure

  3. Caskada Integration

    • Node implementation with prep->exec->post lifecycle

    • Flow configuration

    • Shared store usage for data passing

Simple Chat (python-chat)

Complexity Points: 3.5 └🐣🐣

A basic chat application using Caskada with OpenAI's GPT-4o model.

Details

Simple Chat

A basic chat application using Caskada with OpenAI's GPT-4o model.

Features

  • Conversational chat interface in the terminal

  • Maintains full conversation history for context

  • Simple implementation demonstrating Caskada's node and flow concepts

Run It

  1. Make sure your OpenAI API key is set:

    Alternatively, you can edit the utils.py file to include your API key directly.

  2. Install requirements and run the application:

How It Works

The chat application uses:

  • A single ChatNode with a self-loop that:

    • Takes user input in the prep method

    • Sends the complete conversation history to GPT-4o

    • Adds responses to the conversation history

    • Loops back to continue the chat until the user types 'exit'

Files

  • main.py: Implementation of the ChatNode and chat flow

  • utils.py: Simple wrapper for calling the OpenAI API

Text Converter Flow (python-flow)

Complexity Points: 4 └🐣🐣

This project demonstrates an interactive text transformation tool built with Caskada.

Details

Text Converter Flow

This project demonstrates an interactive text transformation tool built with Caskada.

Features

  • Convert text to UPPERCASE

  • Convert text to lowercase

  • Reverse text

  • Remove extra spaces

  • Interactive command-line interface

  • Continuous flow with option to process multiple texts

Getting Started

  1. Install the required dependencies:

  1. Run the application:

How It Works

The workflow features an interactive loop with branching paths:

Here's what each part does:

  1. TextInput Node: Collects text input and handles menu choices

  2. TextTransform Node: Applies the selected transformation to the text

Example Output

Files

  • main.py: Main entry point for running the text converter

  • flow.py: Defines the nodes and flow for text transformation

  • requirements.txt: Lists the required dependencies

Batch Translation Process (python-batch)

Complexity Points: 4.5 └🐥🐥🐥

This project demonstrates a batch processing implementation that enables LLMs to translate documents into multiple languages simultaneously. It's designed to efficiently handle the translation of markdown files while preserving formatting.

Details

Batch Translation Process

This project demonstrates a batch processing implementation that enables LLMs to translate documents into multiple languages simultaneously. It's designed to efficiently handle the translation of markdown files while preserving formatting.

Features

  • Translates markdown content into multiple languages in parallel

  • Saves translated files to specified output directory

Getting Started

  1. Install the required packages:

  1. Set up your API key:

  1. Run the translation process:

How It Works

The implementation uses a fan-out pattern with nodes orchestrated by a Flow for sequential processing.

Example Output

When you run the translation process, you'll see output similar to this:

Files

The translations are saved to the translations directory, with each file named according to the target language.

Batch Node Example (python-batch-node)

Complexity Points: 4.5 └🐥🐥🐥

This example demonstrates the batch Node concept in Caskada by implementing a CSV processor that handles large files by processing them in chunks.

Details

Batch Node Example

This example demonstrates the batch Node concept in Caskada by implementing a CSV processor that handles large files by processing them in chunks.

What this Example Demonstrates

  • How to use batch Node to process large inputs in chunks

  • The three key methods of batch Node:

    1. prep: Splits input into chunks

    2. exec: Processes each chunk independently

    3. post: Combines results from all chunks

Project Structure

How it Works

The example processes a large CSV file containing sales data:

  1. Chunking (prep): The CSV file is read and split into chunks of N rows

  2. Processing (exec): Each chunk is processed to calculate:

    • Total sales

    • Average sale value

    • Number of transactions

  3. Combining (post): Results from all chunks are aggregated into final statistics

Installation

Usage

Sample Output

Key Concepts Illustrated

  1. Chunk-based Processing: Shows how batch Node handles large inputs by breaking them into manageable pieces

  2. Independent Processing: Demonstrates how each chunk is processed separately

  3. Result Aggregation: Shows how individual results are combined into a final output

LLM Streaming and Interruption (python-llm-streaming)

Complexity Points: 4.5 └🐥🐥🐥

Demonstrates real-time LLM response streaming with user interrupt capability.

Details

LLM Streaming and Interruption

Demonstrates real-time LLM response streaming with user interrupt capability.

Features

  • Real-time display of LLM responses as they're generated

  • User interrupt with ENTER key at any time

Run It

How It Works

StreamNode:

  1. Creates interrupt listener thread

  2. Fetches content chunks from LLM

  3. Displays chunks in real-time

  4. Handles user interruption

API Key

By default, demo uses fake streaming responses. To use real OpenAI streaming:

  1. Edit main.py to replace the fake_stream_llm with stream_llm:

  1. Make sure your OpenAI API key is set:

Files

  • main.py: StreamNode implementation

  • utils.py: Real and fake LLM streaming functions

Shared Store Communication (python-communication)

Complexity Points: 5 └🐥🐥🐥

This example demonstrates the Memory concept in Caskada, specifically focusing on the Shared Store pattern.

Details

Shared Store Communication

This example demonstrates the Memory concept in Caskada, specifically focusing on the Shared Store pattern.

Overview

The example implements a simple word counter that shows how nodes can communicate using a shared store. It demonstrates:

  • How to initialize and structure a shared store

  • How nodes can read from and write to the shared store

  • How to maintain state across multiple node executions

  • Best practices for shared store usage

Project Structure

Installation

Usage

Enter text when prompted. The program will:

  1. Count words in the text

  2. Store statistics in the shared store

  3. Display running statistics (total texts, total words, average)

Enter 'q' to quit.

How it Works

The example uses three nodes:

  1. TextInput: Reads user input and initializes the shared store

  2. WordCounter: Counts words and updates statistics in the shared store

  3. ShowStats: Displays statistics from the shared store

This demonstrates how nodes can share and maintain state using the shared store pattern.

Web Search with Analysis (python-tool-search)

Complexity Points: 6 └🐓🐓🐓🐓

A web search tool built with Caskada that performs searches using SerpAPI and analyzes results using LLM.

Details

Web Search with Analysis

A web search tool built with Caskada that performs searches using SerpAPI and analyzes results using LLM.

Features

  • Web search using Google via SerpAPI

  • Extracts titles, snippets, and links

  • Analyzes search results using GPT-4 to provide:

    • Result summaries

    • Key points/facts

    • Suggested follow-up queries

  • Clean command-line interface

Installation

  1. Clone the repository

  2. Install dependencies:

  3. Set required API keys:

Usage

Run the search tool:

You will be prompted to:

  1. Enter your search query

  2. Specify number of results to fetch (default: 5)

The tool will then:

  1. Perform the search using SerpAPI

  2. Analyze results using GPT-4

  3. Present a summary with key points and follow-up queries

Project Structure

Limitations

  • Requires SerpAPI subscription

  • Rate limited by both APIs

  • Basic error handling

  • Text results only

Dependencies

  • caskada: Flow-based processing

  • google-search-results: SerpAPI client

  • openai: GPT-4 API access

  • pyyaml: YAML processing

Majority Vote Reasoning (python-majority-vote)

Complexity Points: 6.5 └🐓🐓🐓🐓

This project demonstrates a majority vote implementation that enables LLMs to solve complex reasoning problems by aggregating multiple independent attempts. It's designed to improve problem-solving accuracy through consensus-based reasoning.

Details

Majority Vote Reasoning

This project demonstrates a majority vote implementation that enables LLMs to solve complex reasoning problems by aggregating multiple independent attempts. It's designed to improve problem-solving accuracy through consensus-based reasoning.

Features

  • Improves model reliability on complex problems through multiple attempts

  • Works with models like Claude 3.7 Sonnet

  • Solves problems that single attempts often fail on

  • Provides detailed reasoning traces for verification

  • Uses a consensus approach to reduce the impact of occasional reasoning errors

Getting Started

  1. Install the required packages:

  1. Set up your API key:

  1. Run a test problem to see majority voting in action:

  1. Try your own reasoning problem:

How It Works

The implementation uses a MajorityVoteNode that processes multiple attempts and finds consensus:

The MajorityVoteNode:

  1. Makes multiple independent attempts to solve the same problem

  2. Collects structured answers from each attempt

  3. Determines the most frequent answer as the final solution

  4. Returns the consensus answer

This approach helps overcome occasional reasoning errors that might occur in individual attempts.

Example Problem

Example Problem from Quant Interview:

Below is an example of how the majority vote approach uses Claude 3.7 Sonnet to solve this complex problem:

This shows that 4 out of 5 attempts yielded the same answer (0.333), which is chosen as the final solution.

Files

  • main.py: Implementation of the majority vote node and flow

  • utils.py: Simple wrapper for calling the Anthropic model

Streamlit Human-in-the-Loop (HITL) Application (python-streamlit-hitl)

Complexity Points: 6.5 └🐓🐓🐓🐓

Minimal Human-in-the-Loop (HITL) web application using Caskada and Streamlit. Submit text, review processed output, and approve/reject.

Details

Streamlit Human-in-the-Loop (HITL) Application

Minimal Human-in-the-Loop (HITL) web application using Caskada and Streamlit. Submit text, review processed output, and approve/reject.

Features

  • Streamlit UI: Simple, interactive interface for submitting tasks and providing feedback, built entirely in Python.

  • Caskada Workflow: Manages distinct processing stages (initial processing, finalization) using synchronous Caskada Flows.

  • Session State Management: Utilizes Streamlit's st.session_state to manage the current stage of the workflow and to act as the shared data store for Caskada.

  • Iterative Feedback Loop: Allows users to reject processed output and resubmit, facilitating refinement.

How to Run

  1. Install Dependencies:

  2. Run the Streamlit Application:

  3. Access the Web UI: Open the URL provided by Streamlit (usually http://localhost:8501).

Files

SQLite Database with Caskada (python-tool-database)

Complexity Points: 6.5 └🐓🐓🐓🐓

This example demonstrates how to properly integrate SQLite database operations with Caskada, focusing on:

Details

SQLite Database with Caskada

This example demonstrates how to properly integrate SQLite database operations with Caskada, focusing on:

  1. Clean code organization with separation of concerns:

    • Tools layer for database operations (tools/database.py)

    • Node implementation for Caskada integration (nodes.py)

    • Flow configuration (flow.py)

    • Safe SQL query execution with parameter binding

  2. Best practices for database operations:

    • Connection management with proper closing

    • SQL injection prevention using parameterized queries

    • Error handling and resource cleanup

    • Simple schema management

  3. Example task management system:

    • Database initialization

    • Task creation

    • Task listing

    • Status tracking

Project Structure

Setup

  1. Create a virtual environment:

  1. Install dependencies:

Usage

Run the example:

This will:

  1. Initialize a SQLite database with a tasks table

  2. Create an example task

  3. List all tasks in the database

  4. Display the results

Key Concepts Demonstrated

  1. Database Operations

    • Safe connection handling

    • Query parameterization

    • Schema management

  2. Code Organization

    • Clear separation between database operations and Caskada components

    • Modular project structure

    • Type hints and documentation

  3. Caskada Integration

    • Node implementation with prep->exec->post lifecycle

    • Flow configuration

    • Shared store usage for data passing

Example Output

Article Writing Workflow (python-workflow)

Complexity Points: 6.5 └🐓🐓🐓🐓

A Caskada example that demonstrates an article writing workflow using a sequence of LLM calls.

Details

Article Writing Workflow

A Caskada example that demonstrates an article writing workflow using a sequence of LLM calls.

Features

  • Generate a simple outline with up to 3 main sections using YAML structured output

  • Write concise (100 words max) content for each section in simple terms

  • Apply a conversational, engaging style to the final article

Getting Started

  1. Install the required dependencies:

  1. Set your OpenAI API key as an environment variable:

  1. Run the application with a default topic ("AI Safety"):

  1. Or specify your own topic:

How It Works

The workflow consists of three sequential nodes:

Here's what each node does:

  1. Generate Outline: Creates a simple outline with up to 3 main sections using YAML structured output

  2. Write Simple Content: Writes a concise 100-word explanation for each section

  3. Apply Style: Rewrites the combined content in a conversational, engaging style

Files

  • main.py: Main entry point for running the article workflow

  • flow.py: Defines the flow that connects the nodes

  • nodes.py: Contains the node classes for each step in the workflow

  • utils.py: Utility functions including the LLM wrapper

  • requirements.txt: Lists the required dependencies

Example Output

Batch Flow Example (python-batch-flow)

Complexity Points: 7 └🦕🦕🦕🦕🦕

This example demonstrates the batch Flow concept in Caskada by implementing an image processor that applies different filters to multiple images.

Details

Batch Flow Example

This example demonstrates the batch Flow concept in Caskada by implementing an image processor that applies different filters to multiple images.

What this Example Demonstrates

  • How to use batch Flow to run a Flow multiple times with different parameters

  • Key concepts of batch Flow:

    1. Creating a base Flow for single-item processing

    2. Using batch Flow to process multiple items with different parameters

    3. Managing parameters across multiple Flow executions

Project Structure

How it Works

The example processes multiple images with different filters:

  1. Base Flow: Processes a single image

    • Load image

    • Apply filter (grayscale, blur, or sepia)

    • Save processed image

  2. batch Flow: Processes multiple image-filter combinations

    • Takes a list of parameters (image + filter combinations)

    • Runs the base Flow for each parameter set

    • Organizes output in a structured way

Installation

Usage

Sample Output

Key Concepts Illustrated

  1. Parameter Management: Shows how batch Flow manages different parameter sets

  2. Flow Reuse: Demonstrates running the same Flow multiple times

  3. Batch Processing: Shows how to process multiple items efficiently

  4. Real-world Application: Provides a practical example of batch processing

Travel Advisor Chat with Guardrails (python-chat-guardrail)

Complexity Points: 7 └🦕🦕🦕🦕🦕

A travel-focused chat application using Caskada with OpenAI's GPT-4o model, enhanced with input validation to ensure only travel-related queries are processed.

Details

Travel Advisor Chat with Guardrails

A travel-focused chat application using Caskada with OpenAI's GPT-4o model, enhanced with input validation to ensure only travel-related queries are processed.

Features

  • Travel advisor chatbot that answers questions about destinations, planning, accommodations, etc.

  • Topic-specific guardrails to ensure only travel-related queries are accepted

Run It

  1. Make sure your OpenAI API key is set:

    Alternatively, you can edit the utils.py file to include your API key directly.

  2. Install requirements and run the application:

How It Works

The chat application uses:

  • A UserInputNode that collects user input in its exec method

  • A GuardrailNode that validates if the query is travel-related using:

    • Basic validation checks (empty input, too short)

    • LLM-based validation to determine if the query relates to travel

  • An LLMNode that processes valid travel queries using GPT-4o with a travel advisor system prompt

  • Flow connections that route inputs through validation before processing and handle retries for non-travel related queries

Files

  • main.py: Implementation of the nodes and chat flow

  • utils.py: Utilities for calling the OpenAI API

Example Outputs

Chain-of-Thought (python-thinking)

Complexity Points: 7 └🦕🦕🦕🦕🦕

This project demonstrates an implementation that orchestrates a Chain-of-Thought process, enabling LLMs to solve complex reasoning problems by thinking step-by-step. It's designed to improve problem-solving accuracy through deliberate, structured reasoning managed externally.

Details

Chain-of-Thought

This project demonstrates an implementation that orchestrates a Chain-of-Thought process, enabling LLMs to solve complex reasoning problems by thinking step-by-step. It's designed to improve problem-solving accuracy through deliberate, structured reasoning managed externally.

This implementation is based on this tutorial (for Pocketflow): Build Chain-of-Thought From Scratch - Tutorial for Dummies.

Features

  • Improves model reasoning on complex problems.

  • Leverages capable instruction-following models (e.g., Claude 3.7 Sonnet, GPT-4 series) to perform structured Chain-of-Thought reasoning.

  • Solves problems that direct prompting often fails on by breaking them down systematically.

  • Provides detailed reasoning traces, including step-by-step evaluation and planning, for verification.

Getting Started

  1. Install Packages:

  2. Set API Key:

  3. Verify API Key (Optional): Run a quick check to ensure your key and environment are set up correctly.

  4. Run Default Example: Execute the main script to see the process in action with the default Jane Street problem.

    The default question is:

    You keep rolling a fair die until you roll three, four, five in that order consecutively on three rolls. What is the probability that you roll the die an odd number of times?

  5. Run Custom Problem: Provide your own reasoning problem using the -- argument.

How It Works

The implementation uses a self-looping Caskada node (ChainOfThoughtNode) that guides an LLM through a structured problem-solving process:

In each loop (thought step), the node directs the LLM to:

  1. Evaluate the previous thought's reasoning and results.

  2. Execute the next pending step according to a maintained plan.

  3. Update the plan, marking the step done (with results) or noting issues.

  4. Refine the plan if steps need breaking down or errors require correction.

  5. Decide if further thinking (next_thought_needed) is required based on the plan state.

This external orchestration enforces a systematic approach, helping models tackle problems that are difficult with a single prompt.

Comparison with Different Approaches

  • Standard Prompting: Techniques like asking the model to "think step by step" within a single prompt can help, but the reasoning might lack depth or structure, and the model can easily lose track or make unrecoverable errors.

  • Native Extended Thinking Modes: Some models (like Claude 3.7, GPT-o1, etc.) offer dedicated modes or features explicitly for extended reasoning, often yielding strong results directly via API calls.

  • This Implementation: Demonstrates how to orchestrate a structured Chain-of-Thought process using standard LLMs (even those without a specific native 'extended thinking' mode), managing the steps, planning, and evaluation externally via prompt engineering and flow control.

Example Thinking Process

Let's try out this challenging Jane Street Quant Trading Interview Question:

Problem: You keep rolling a fair die until you roll three, four, five in that order consecutively on three rolls. What is the probability that you roll the die an odd number of times?

This problem demonstrates why structured Chain-of-Thought is valuable:

  • Standard models (single prompt): Often get the wrong answer or provide flawed reasoning.

  • Models using native thinking modes: Can find the correct answer (216/431 ≈ 0.5012), though performance and reasoning clarity may vary.

  • This implementation (orchestrating a capable LLM): Can guide the model towards the correct answer by enforcing a step-by-step plan, evaluation, and refinement loop.

For comparison:

Below is an example output trace showing how this implementation guides Claude 3.7 Sonnet through the problem-solving process:

Note: Even with structured thinking orchestration, models don't always get the right answer, especially on very complex or novel problems. However, this approach significantly improves the robustness of the reasoning process and provides a traceable path for verification and debugging.

Caskada Visualization (python-visualization)

Complexity Points: 7 └🦕🦕🦕🦕🦕

This directory contains tools for visualizing Caskada workflow graphs using interactive D3.js visualizations.

Details

Caskada Visualization

This directory contains tools for visualizing Caskada workflow graphs using interactive D3.js visualizations.

Overview

The visualization tools allow you to:

  1. View Caskada nodes and flows as an interactive graph

  2. See how different flows connect to each other

  3. Understand the relationships between nodes within flows

Features

  • Interactive Graph: Nodes can be dragged to reorganize the layout

  • Group Visualization: Flows are displayed as groups with dashed borders

  • Inter-Group Links: Connections between flows are shown as dashed lines connecting group boundaries

  • Action Labels: Edge labels show the actions that trigger transitions between nodes

Requirements

  • Python 3.6 or higher

  • Modern web browser (Chrome, Firefox, Edge) for viewing the visualizations

Usage

1. Basic Visualization

To visualize a Caskada graph, you can use the visualize_flow function in visualize.py:

This will:

  1. Print a Mermaid diagram to the console

  2. Generate a D3.js visualization in the ./viz directory

2. Running the Example

The included example shows an order processing pipeline with payment, inventory, and shipping flows:

This will generate visualization files in the ./viz directory.

3. Viewing the Visualization

After running the script:

  1. Host with

  2. Interact with the visualization:

    • Drag nodes to reorganize

    • Hover over nodes to see node names

    • Observe connections between nodes and flows

Customizing the Visualization

Adjusting Layout Parameters

You can adjust the force simulation parameters in visualize.py to change how nodes and groups are positioned:

Styling

Adjust the CSS styles in the HTML template inside create_d3_visualization function to change colors, shapes, and other visual properties.

How It Works

The visualization process consists of three main steps:

  1. Flow to JSON Conversion: The flow_to_json function traverses the Caskada graph and converts it to a structure with nodes, links, and group information.

  2. D3.js Visualization: The JSON data is used to create an interactive D3.js visualization with:

    • Nodes represented as circles

    • Flows represented as dashed rectangles containing nodes

    • Links showing connections within and between flows

  3. Group Boundary Connections: The visualization calculates intersection points with group boundaries to ensure inter-group links connect at the borders rather than centers.

Extending the Visualization

You can extend the visualization tools by:

  1. Adding new node shapes

  2. Implementing additional layout algorithms

  3. Adding tooltips with more detailed information

  4. Creating animation for flow execution

Troubleshooting

If you encounter any issues:

  • Make sure your flow objects are properly constructed with nodes connected correctly

  • Check the browser console for any JavaScript errors

  • Verify that the generated JSON data structure matches what you expect

Example Output

The visualization displays:

  • Payment processing flow nodes

  • Inventory management flow nodes

  • Shipping flow nodes

  • Group boundaries around each flow

  • Connections between flows (Payment → Inventory → Shipping)

Multi-Agent Taboo Game (python-multi-agent)

Complexity Points: 8 └🦕🦕🦕🦕🦕

A Caskada example that demonstrates how to implement asynchronous multi-agent communication using the Taboo word guessing game.

Details

Multi-Agent Taboo Game

A Caskada example that demonstrates how to implement asynchronous multi-agent communication using the Taboo word guessing game.

Features

  • Implement asynchronous communication between two AI agents (Hinter and Guesser)

  • Create dynamic conversation flow through asyncio message queues

  • Demonstrate complex turn-based game mechanics with LLMs

  • Automatically terminate the game when the correct word is guessed

Getting Started

  1. Install the required dependencies:

  1. Set your OpenAI API key as an environment variable:

  1. Run the application:

How It Works

The workflow follows an asynchronous multi-agent communication pattern:

Here's what each component does:

  1. AsyncHinter Node: Generates hints about the target word while avoiding forbidden words

  2. AsyncGuesser Node: Makes guesses based on the hints received from the Hinter

  3. Message Queue: Facilitates asynchronous communication between the agents

Files

  • main.py: Main entry point implementing the AsyncHinter and AsyncGuesser nodes and game flow

  • utils.py: Utility functions including LLM wrappers for generating hints and guesses

  • requirements.txt: Lists the required dependencies

Example Output

Parallel Batch Translation Process (python-parallel-batch)

Complexity Points: 8 └🦕🦕🦕🦕🦕

This project demonstrates using Caskada's async and parallel features (ParallelFlow, Node) to translate a document into multiple languages concurrently.

Details

Parallel Batch Translation Process

This project demonstrates using Caskada's async and parallel features (ParallelFlow, Node) to translate a document into multiple languages concurrently.

Goal

Translate ../../README.md into multiple languages (Chinese, Spanish, etc.) in parallel, saving each to a file in the translations/ directory. The main goal is to compare execution time against a sequential process.

Getting Started

  1. Install requirements:

  1. Set API Key: Set the environment variable for your Anthropic API key.

    (Replace "your-api-key-here" with your actual key) (Alternatively, place ANTHROPIC_API_KEY=your-api-key-here in a .env file)

  2. Verify API Key (Optional): Run a quick check using the utility script.

    (Note: This requires a valid API key to be set.)

  3. Run the translation process:

How It Works

The implementation uses a Node that processes translation requests concurrently. The TranslateTextNodeParallel:

  1. Prepares batches, pairing the source text with each target language.

  2. Executes translation calls to the LLM for all languages concurrently using async operations.

  3. Saves the translated content to individual files (translations/README_LANGUAGE.md).

This approach leverages asyncio and parallel execution to speed up I/O-bound tasks like multiple API calls.

Example Output & Comparison

Running this parallel version significantly reduces the total time compared to a sequential approach:

(Actual times will vary based on API response speed and system.)

Files

  • main.py: Implements the parallel batch translation node and flow.

  • utils.py: Async wrapper for calling the Anthropic model.

  • requirements.txt: Project dependencies (includes aiofiles).

  • translations/: Output directory (created automatically).

Web Human-in-the-Loop (HITL) Feedback Service (python-fastapi-hitl)

Complexity Points: 9 └🦖🦖🦖🦖🦖🦖

This project demonstrates a minimal web application for human-in-the-loop workflows using Caskada, FastAPI, and Server-Sent Events (SSE). Users can submit text, have it processed (simulated), review the output, and approve or reject it, potentially triggering reprocessing until approved.

Details

Web Human-in-the-Loop (HITL) Feedback Service

This project demonstrates a minimal web application for human-in-the-loop workflows using Caskada, FastAPI, and Server-Sent Events (SSE). Users can submit text, have it processed (simulated), review the output, and approve or reject it, potentially triggering reprocessing until approved.

Features

  • Web UI: Simple interface for submitting tasks and providing feedback.

  • Caskada Workflow: Manages the process -> review -> result/reprocess logic.

  • FastAPI Backend: Serves the UI and handles API requests asynchronously.

  • Server-Sent Events (SSE): Provides real-time status updates to the client without polling.

How to Run

  1. Install Dependencies:

  2. Run the FastAPI Server: Use Uvicorn (or another ASGI server):

    (The --reload flag is useful for development.)

  3. Access the Web UI: Open your web browser and navigate to http://127.0.0.1:8000.

  4. Use the Application:

    • Enter text into the textarea and click "Submit".

    • Observe the status updates pushed via SSE.

    • When prompted ("waiting_for_review"), use the "Approve" or "Reject" buttons.

    • If rejected, the process loops back. If approved, the final result is displayed.

How It Works

The application uses Caskada to define and execute the feedback loop workflow. FastAPI handles web requests and manages the real-time SSE communication.

Caskada Workflow:

The core logic is orchestrated by a Flow defined in flow.py:

  1. ProcessNode: Receives input text, calls the minimal process_task utility, and stores the output.

  2. ReviewNode (Async):

    • Pushes a "waiting_for_review" status with the processed output to the SSE queue.

    • Waits asynchronously for an external signal (triggered by the /feedback API endpoint).

    • Based on the received feedback ("approved" or "rejected"), determines the next step in the flow. Stores the result if approved.

  3. ResultNode: Logs the final approved result.

FastAPI & SSE Integration:

  • The /submit endpoint creates a unique task, initializes the Caskada shared state (including an asyncio.Event for review and an asyncio.Queue for SSE), and schedules the flow execution using BackgroundTasks.

  • Nodes within the flow (specifically ReviewNode's prep logic) put status updates onto the task-specific sse_queue.

  • The /stream/{task_id} endpoint uses StreamingResponse to read from the task's sse_queue and push formatted status updates to the connected client via Server-Sent Events.

  • The /feedback/{task_id} endpoint receives the human's decision, updates the shared state, and sets the asyncio.Event to unblock the waiting ReviewNode.

This setup allows for a decoupled workflow logic (Caskada) and web interaction layer (FastAPI), with efficient real-time updates pushed to the user.

Files

  • server.py: The main FastAPI application handling HTTP requests, SSE, state management, and background task scheduling.

  • nodes.py: Defines the Caskada Node classes (ProcessNode, ReviewNode, ResultNode) for the workflow steps.

  • flow.py: Defines the Caskada Flow that connects the nodes into the feedback loop.

  • utils/process_task.py: Contains the minimal simulation function for task processing.

  • templates/index.html: The HTML structure for the frontend user interface.

  • static/style.css: Basic CSS for styling the frontend.

  • requirements.txt: Project dependencies (FastAPI, Uvicorn, Jinja2, Caskada).

Tool Calling: PDF Vision (python-tool-pdf-vision)

Complexity Points: 9 └🦖🦖🦖🦖🦖🦖

A Caskada example project demonstrating PDF processing with OpenAI's Vision API for OCR and text extraction.

Details

Tool Calling: PDF Vision

A Caskada example project demonstrating PDF processing with OpenAI's Vision API for OCR and text extraction.

Features

  • Convert PDF pages to images while maintaining quality and size limits

  • Extract text from scanned documents using GPT-4 Vision API

  • Support for custom extraction prompts

  • Maintain page order and formatting in extracted text

  • Batch processing of multiple PDFs from a directory

Installation

  1. Clone the repository

  2. Install dependencies:

  3. Set your OpenAI API key as an environment variable:

Usage

  1. Place your PDF files in the pdfs directory

  2. Run the example:

    The script will process all PDF files in the pdfs directory and output the extracted text for each one.

Project Structure

Flow Description

  1. LoadPDFNode: Loads PDF and converts pages to images

  2. ExtractTextNode: Processes images with Vision API

  3. CombineResultsNode: Combines extracted text from all pages

Customization

You can customize the extraction by modifying the prompt in shared:

Limitations

  • Maximum PDF page size: 2000px (configurable in tools/pdf.py)

  • Vision API token limit: 1000 tokens per response

  • Image size limit: 20MB per image for Vision API

License

MIT

Nested Batch Flow Example (python-nested-batch)

Complexity Points: 9.5 └🦖🦖🦖🦖🦖🦖

This example demonstrates Nested batch Flow using a simple school grades calculator.

Details

Nested Batch Flow Example

This example demonstrates Nested batch Flow using a simple school grades calculator.

What this Example Does

Calculates average grades for:

  1. Each student in a class

  2. Each class in the school

Structure

Running the Example

Expected Output

Web Crawler with Content Analysis (python-tool-crawler)

Complexity Points: 9.5 └🦖🦖🦖🦖🦖🦖

A web crawler tool built with Caskada that crawls websites and analyzes content using LLM.

Details

Web Crawler with Content Analysis

A web crawler tool built with Caskada that crawls websites and analyzes content using LLM.

Features

  • Crawls websites while respecting domain boundaries

  • Extracts text content and links from pages

  • Analyzes content using GPT-4 to generate:

    • Page summaries

    • Main topics/keywords

    • Content type classification

  • Processes pages in batches for efficiency

  • Generates a comprehensive analysis report

Installation

  1. Clone the repository

  2. Install dependencies:

  3. Set your OpenAI API key:

Usage

Run the crawler:

You will be prompted to:

  1. Enter the website URL to crawl

  2. Specify maximum number of pages to crawl (default: 10)

The tool will then:

  1. Crawl the specified website

  2. Extract and analyze content using GPT-4

  3. Generate a report with findings

Project Structure

Limitations

  • Only crawls within the same domain

  • Text content only (no images/media)

  • Rate limited by OpenAI API

  • Basic error handling

Dependencies

  • caskada: Flow-based processing

  • requests: HTTP requests

  • beautifulsoup4: HTML parsing

  • openai: GPT-4 API access

Resume Qualification - Map Reduce Example (python-map-reduce)

Complexity Points: 10 └☄️☄️☄️☄️☄️☄️☄️

A Caskada example that demonstrates how to implement a Map-Reduce pattern for processing and evaluating resumes.

Details

Resume Qualification - Map Reduce Example

A Caskada example that demonstrates how to implement a Map-Reduce pattern for processing and evaluating resumes.

Features

  • Read and process multiple resume files using a Map-Reduce pattern

  • Evaluate each resume individually using an LLM with structured YAML output

  • Determine if candidates qualify for technical roles based on specific criteria

  • Aggregate results to generate qualification statistics and summaries

Getting Started

  1. Install the required dependencies:

  1. Set your OpenAI API key as an environment variable:

  1. Run the application:

How It Works

The workflow follows a classic Map-Reduce pattern with sequential nodes.

Files

  • main.py: Main entry point for running the resume qualification workflow

  • flow.py: Defines the flow that connects the nodes

  • nodes.py: Contains the node classes for each step in the workflow

  • utils.py: Utility functions including the LLM wrapper

  • requirements.txt: Lists the required dependencies

  • data/: Directory containing sample resume files for evaluation

Example Output

Async Recipe Finder (python-async-basic)

Complexity Points: 10.5 └☄️☄️☄️☄️☄️☄️☄️

This example demonstrates async operations using a simple Recipe Finder that:

Details

Async Recipe Finder

This example demonstrates async operations using a simple Recipe Finder that:

  1. Fetches recipes from an API (async HTTP)

  2. Processes them with an LLM (async LLM)

  3. Waits for user confirmation (async input)

What this Example Does

When you run the example:

  1. You enter an ingredient (e.g., "chicken")

  2. It searches for recipes (async API call)

  3. It suggests a recipe (async LLM call)

  4. You approve or reject the suggestion

  5. If rejected, it tries again with a different recipe

How it Works

  1. FetchRecipes (Node)

  2. SuggestRecipe (Node)

  3. GetApproval (Node)

Running the Example

Sample Interaction

Key Concepts

  1. Async Operations: Using async/await for:

    • API calls (non-blocking I/O)

    • LLM calls (potentially slow)

    • User input (waiting for response)

  2. Node Methods:

    • prep: Setup and data gathering

    • exec: Main async processing

    • post: Post-processing and decisions

  3. Flow Control:

    • Actions ("accept"/"retry") control flow

    • Retry loop for rejected suggestions

Chat with Memory Retrieval (python-chat-memory)

Complexity Points: 10.5 └☄️☄️☄️☄️☄️☄️☄️

A chat application with memory retrieval using Caskada. This example maintains a sliding window of recent conversations while retrieving relevant past conversations based on context.

Details

Chat with Memory Retrieval

A chat application with memory retrieval using Caskada. This example maintains a sliding window of recent conversations while retrieving relevant past conversations based on context.

Features

  • Maintains a window of 3 most recent conversation pairs

  • Archives older conversations with embeddings

  • Uses vector similarity to retrieve the most relevant past conversation

  • Combines recent context (3 pairs) with retrieved context (1 pair) for better responses

Run It

  1. Make sure your OpenAI API key is set:

  2. Install requirements and run the application:

How It Works

The chat application uses:

  • Four specialized nodes:

    • GetUserQuestionNode: Handles interactive user input

    • RetrieveNode: Finds relevant past conversations using vector similarity

    • AnswerNode: Generates responses using both recent and retrieved context

    • EmbedNode: Archives older conversations with embeddings

  • A sliding window approach that maintains only the 3 most recent conversation pairs in active context

Files

  • nodes.py: Four node implementations with clear separation of concerns

  • flow.py: Chat flow structure definition

  • main.py: Entry point for running the demo

  • utils/: Utility functions for embeddings, LLM calls, and vector operations

Example Output

Parallel Image Processor (python-parallel-batch-flow)

Complexity Points: 10.5 └☄️☄️☄️☄️☄️☄️☄️

Demonstrates how a batch ParallelFlow processes multiple images with multiple filters >8x faster than sequential processing.

Details

Parallel Image Processor

Demonstrates how a batch ParallelFlow processes multiple images with multiple filters >8x faster than sequential processing.

Features

  • Processes images with multiple filters in parallel

  • Applies three different filters (grayscale, blur, sepia)

  • Shows significant speed improvement over sequential processing

  • Manages system resources with semaphores

Run It

Output

Key Points

  • Sequential: Total time = sum of all item times

    • Good for: Rate-limited APIs, maintaining order

  • Parallel: Total time ≈ longest single item time

    • Good for: I/O-bound tasks, independent operations

Research Agent (python-agent)

Complexity Points: 11.5 └🐭🐭🐭🐭🐭🐭🐭🐭

This project demonstrates a simple yet powerful LLM-powered research agent.

Details

Research Agent

This project demonstrates a simple yet powerful LLM-powered research agent.

👉 Run the tutorial in your browser: Try Google Colab Notebook

Features

  • Performs web searches to gather information

  • Makes decisions about when to search vs. when to answer

  • Generates comprehensive answers based on research findings

Getting Started

  1. Install the packages you need with this simple command:

  1. Let's get your OpenAI API key ready:

  1. Let's do a quick check to make sure your API key is working properly:

This will test both the LLM call and web search features. If you see responses, you're good to go!

  1. Try out the agent with the default question (about Nobel Prize winners):

  1. Got a burning question? Ask anything you want by using the -- prefix:

How It Works?

The magic happens through a simple but powerful graph structure with three main parts:

Here's what each part does:

  1. DecideAction: The brain that figures out whether to search or answer

  2. SearchWeb: The researcher that goes out and finds information

  3. AnswerQuestion: The writer that crafts the final answer

Here's what's in each file:

  • main.py: The starting point - runs the whole show!

  • flow.py: Connects everything together into a smart agent

  • nodes.py: The building blocks that make decisions and take actions

  • utils.py: Helper functions for talking to the LLM and searching the web

MCP Demo (python-mcp)

Complexity Points: 11.5 └🐭🐭🐭🐭🐭🐭🐭🐭

This project shows how to build an agent that performs addition using Caskada and Model Context Protocol (MCP). It presents a comparison between using MCP and basic function calling approaches.

Details

MCP Demo

This project shows how to build an agent that performs addition using Caskada and Model Context Protocol (MCP). It presents a comparison between using MCP and basic function calling approaches.

This implementation is based on this tutorial (for Pocketflow): MCP Simply Explained: Function Calling Rebranded or Genuine Breakthrough?

Features

  • Mathematical operation tools through a simple terminal interface

  • Integration with Model Context Protocol (MCP)

  • Comparison between MCP and direct function calling

  • Simple toggle between MCP and local function calling

How to Run

  1. Set your API key:

    Or update it directly in utils.py

  2. Install and run:

MCP vs Function Calling

To compare both approaches, this demo provides local function alternatives that don't require MCP:

  • Toggle with a simple flag: Set MCP = True or MCP = False at the top of utils.py to switch between MCP and local implementations.

  • No code changes needed! The application automatically uses either:

    • MCP server tools when MCP = True

    • Local function implementations when MCP = False

This allows you to see the difference between the two approaches while keeping the same workflow.

Function Calling

  • Functions are directly embedded in application code

  • Each new tool requires modifying the application

  • Tools are defined within the application itself

MCP Approach

  • Tools live in separate MCP servers

  • Standard protocol for all tool interactions

  • New tools can be added without changing the agent

  • AI can interact with tools through a consistent interface

How It Works

The agent uses Caskada to create a workflow where:

  1. It takes user input about numbers

  2. Connects to the MCP server for mathematical operations (or uses local functions based on the MCP flag)

  3. Returns the result

Files

  • main.py: Implementation of the addition agent using Caskada

  • utils.py: Helper functions for API calls and MCP integration

  • simple_server.py: MCP server that provides the addition tool

Voice Chat (python-voice-chat)

Complexity Points: 11.5 └🐭🐭🐭🐭🐭🐭🐭🐭

This project demonstrates a voice-based interactive chat application built with Caskada. Users can speak their queries, and the system will respond with spoken answers from an LLM, maintaining conversation history.

Details

Voice Chat

This project demonstrates a voice-based interactive chat application built with Caskada. Users can speak their queries, and the system will respond with spoken answers from an LLM, maintaining conversation history.

Features

  • Voice Activity Detection (VAD): Automatically detects when the user starts and stops speaking.

  • Speech-to-Text (STT): Converts spoken audio into text using OpenAI.

  • LLM Interaction: Processes the transcribed text with an LLM (e.g., GPT-4o), maintaining conversation history.

  • Text-to-Speech (TTS): Converts the LLM's text response back into audible speech using OpenAI.

  • Continuous Conversation: Loops back to listen for the next user query after responding, allowing for an ongoing dialogue.

How to Run

  1. Set your OpenAI API key:

    Ensure this environment variable is set, as the utility scripts for STT, LLM, and TTS rely on it. You can test individual utility functions (e.g., python utils/call_llm.py, python utils/text_to_speech.py) to help verify your API key and setup.

  2. Install dependencies: Make sure you have Python installed. Then, install the required libraries using pip:

    This will install libraries such as openai, caskada, sounddevice, numpy, scipy, and soundfile.

    Note for Linux users: sounddevice may require PortAudio. If you encounter issues, you might need to install it first:

  3. Run the application:

    Follow the console prompts. The application will start listening when you see "Listening for your query...".

How It Works

The application uses a Caskada workflow to manage the conversation steps:

Here's what each node in the flow does:

  1. CaptureAudioNode: Records audio from the user's microphone. It uses Voice Activity Detection (VAD) to start recording when speech is detected and stop when silence is detected.

  2. SpeechToTextNode: Takes the recorded audio data, converts it to a suitable format, and sends it to OpenAI's STT API (gpt-4o-transcribe) to get the transcribed text.

  3. QueryLLMNode: Takes the transcribed text from the user, along with the existing conversation history, and sends it to an LLM (OpenAI's GPT-4o model) to generate an intelligent response.

  4. TextToSpeechNode: Receives the text response from the LLM, converts it into audio using OpenAI's TTS API (gpt-4o-mini-tts), and plays the audio back to the user. If the conversation is set to continue, it transitions back to the CaptureAudioNode.

Example Interaction

When you run main.py:

  1. The console will display:

  2. When you see Listening for your query..., speak clearly into your microphone.

  3. After you stop speaking, the console will show updates:

  4. You will hear the LLM's response spoken aloud.

  5. The application will then loop back, and you'll see Listening for your query... again, ready for your next input.

The conversation continues in this manner. To stop the application, you typically need to interrupt it (e.g., Ctrl+C in the terminal), as it's designed to loop continuously.

Text-to-SQL Workflow (python-text2sql)

Complexity Points: 12 └🐭🐭🐭🐭🐭🐭🐭🐭

A Caskada example demonstrating a text-to-SQL workflow that converts natural language questions into executable SQL queries for an SQLite database, including an LLM-powered debugging loop for failed queries.

Details

Text-to-SQL Workflow

A Caskada example demonstrating a text-to-SQL workflow that converts natural language questions into executable SQL queries for an SQLite database, including an LLM-powered debugging loop for failed queries.

Features

  • Schema Awareness: Automatically retrieves the database schema to provide context to the LLM.

  • LLM-Powered SQL Generation: Uses an LLM (GPT-4o) to translate natural language questions into SQLite queries (using YAML structured output).

  • Automated Debugging Loop: If SQL execution fails, an LLM attempts to correct the query based on the error message. This process repeats up to a configurable number of times.

Getting Started

  1. Install Packages:

  2. Set API Key: Set the environment variable for your OpenAI API key.

    (Replace "your-api-key-here" with your actual key)

  3. Verify API Key (Optional): Run a quick check using the utility script. If successful, it will print a short joke.

    (Note: This requires a valid API key to be set.)

  4. Run Default Example: Execute the main script. This will create the sample ecommerce.db if it doesn't exist and run the workflow with a default query.

    The default query is:

    Show me the names and email addresses of customers from New York

  5. Run Custom Query: Provide your own natural language query as command-line arguments after the script name.

    Or, for queries with spaces, ensure they are treated as a single argument by the shell if necessary (quotes might help depending on your shell):

How It Works

The workflow uses several nodes connected in a sequence, with a loop for debugging failed SQL queries.

Node Descriptions:

  1. GetSchema: Connects to the SQLite database (ecommerce.db by default) and extracts the schema (table names and columns).

  2. GenerateSQL: Takes the natural language query and the database schema, prompts the LLM to generate an SQLite query (expecting YAML output with the SQL), and parses the result.

  3. ExecuteSQL: Attempts to run the generated SQL against the database.

    • If successful, the results are stored, and the flow ends successfully.

    • If an sqlite3.Error occurs (e.g., syntax error), it captures the error message and triggers the debug loop.

  4. DebugSQL: If ExecuteSQL failed, this node takes the original query, schema, failed SQL, and error message, prompts the LLM to generate a corrected SQL query (again, expecting YAML).

  5. (Loop): The corrected SQL from DebugSQL is passed back to ExecuteSQL for another attempt.

  6. (End Conditions): The loop continues until ExecuteSQL succeeds or the maximum number of debug attempts (default: 3) is reached.

Files

  • main.py: Main entry point to run the workflow. Handles command-line arguments for the query.

  • flow.py: Defines the Caskada Flow connecting the different nodes, including the debug loop logic.

  • nodes.py: Contains the Node classes for each step (GetSchema, GenerateSQL, ExecuteSQL, DebugSQL).

  • utils.py: Contains the minimal call_llm utility function.

  • populate_db.py: Script to create and populate the sample ecommerce.db SQLite database.

  • requirements.txt: Lists Python package dependencies.

  • README.md: This file.

Example Output (Successful Run)

Research Supervisor (python-supervisor)

Complexity Points: 13 └🐒🐒🐒🐒🐒🐒🐒🐒🐒

This project demonstrates a supervisor that oversees an unreliable research agent to ensure high-quality answers.

Details

Research Supervisor

This project demonstrates a supervisor that oversees an unreliable research agent to ensure high-quality answers.

Features

  • Evaluates responses for quality and relevance

  • Rejects nonsensical or unreliable answers

  • Requests new answers until a quality response is produced

Getting Started

  1. Install the packages you need with this simple command:

  1. Let's get your OpenAI API key ready:

  1. Let's do a quick check to make sure your API key is working properly:

This will test both the LLM call and web search features. If you see responses, you're good to go!

  1. Try out the agent with the default question (about Nobel Prize winners):

  1. Got a burning question? Ask anything you want by using the -- prefix:

How It Works?

The magic happens through a simple but powerful graph structure with these main components:

Here's what each part does:

  1. DecideAction: The brain that figures out whether to search or answer based on current context

  2. SearchWeb: The researcher that goes out and finds information using web search

  3. UnreliableAnswerNode: Generates answers (with a 50% chance of being unreliable)

  4. SupervisorNode: Quality control that validates answers and rejects nonsensical ones

Example Output

Files

  • main.py: The starting point - runs the whole show!

  • flow.py: Connects everything together into a smart agent with supervision

  • nodes.py: The building blocks that make decisions, take actions, and validate answers

  • utils.py: Helper functions for talking to the LLM and searching the web

Retrieval Augmented Generation (RAG) (python-rag)

Complexity Points: 16 └🤖🤖🤖🤖🤖🤖🤖🤖🤖🤖🤖🤖

This project demonstrates a simplified RAG system that retrieves relevant documents based on user queries and generates answers using an LLM. This implementation is based directly on this tutorial (for Pocketflow): Retrieval Augmented Generation (RAG) from Scratch — Tutorial For Dummies.

Details

Retrieval Augmented Generation (RAG)

This project demonstrates a simplified RAG system that retrieves relevant documents based on user queries and generates answers using an LLM. This implementation is based directly on this tutorial (for Pocketflow): Retrieval Augmented Generation (RAG) from Scratch — Tutorial For Dummies.

Features

  • Document chunking for processing long texts

  • FAISS-powered vector-based document retrieval

  • LLM-powered answer generation

How to Run

  1. Set your API key:

    Or update it directly in utils.py

    Let's do a quick check to make sure your API key is working properly:

  2. Install and run with the default query:

  3. Run the application with a sample query:

How It Works

The magic happens through a two-phase pipeline implemented with Caskada:

Here's what each part does:

  1. ChunkDocumentsNode: Breaks documents into smaller chunks for better retrieval

  2. EmbedDocumentsNode: Converts document chunks into vector representations

  3. CreateIndexNode: Creates a searchable FAISS index from embeddings

  4. EmbedQueryNode: Converts user query into the same vector space

  5. RetrieveDocumentNode: Finds the most similar document using vector search

  6. GenerateAnswerNode: Uses an LLM to generate an answer based on the retrieved content

Example Output

Agent-to-Agent with A2A Protocol (python-a2a)

Complexity Points: 20.5 └🌌🌌🌌🌌🌌🌌🌌🌌🌌🌌🌌🌌🌌🌌🌌

This project demonstrates how to take an existing agent built with the Caskada library and make it accessible to other agents using the Agent-to-Agent (A2A) communication protocol.

Details

Agent-to-Agent with A2A Protocol

This project demonstrates how to take an existing agent built with the Caskada library and make it accessible to other agents using the Agent-to-Agent (A2A) communication protocol.

This implementation is based on this tutorial for Pocketflow: A2A Protocol Simply Explained: Here are 3 key differences to MCP!

How it Works: A2A Integration

This project combines two main parts:

  1. Caskada Agent Logic: The original agent code (nodes.py, utils.py, flow.py) defines the internal workflow (Decide -> Search -> Answer). This code is taken directly from the Caskada Agent Tutorial.

  2. A2A Server Wrapper: Code from the google/A2A samples repository (common/ directory) provides the necessary infrastructure to host the agent as an A2A-compliant server. Note: Minor modifications were made to the common server/client code to add detailed logging for educational purposes.

  3. The Bridge (task_manager.py): A custom CaskadaTaskManager class acts as the bridge. It receives A2A requests (like tasks/send), extracts the user query, runs the Caskada agent_flow, takes the final result from the flow's shared state, and packages it back into an A2A Task object with the answer as an Artifact.

This demonstrates how a non-A2A agent framework can be exposed over the A2A protocol by implementing a specific TaskManager.

Simplified Interaction Sequence

Getting Started

Prerequisites

  • Python 3.10+ (due to type hinting used in the A2A common code)

  • An OpenAI API Key

Installation

  1. Install dependencies:

  2. Set your OpenAI API key as an environment variable:

    Let's do a quick check to make sure your API key is working properly:

  3. Run the server from this directory:

    You should see logs indicating the server has started on http://localhost:10003.

  4. Run the Client in a separate terminal

  5. Follow the instructions in the client terminal to ask questions. Type :q or quit to exit the client.

Example Interaction Logs

(Server Log - showing internal Caskada steps)

(Client Log - showing request/response)

Key A2A Integration Points

To make the Caskada agent A2A-compatible, the following were essential:

  1. A2A Server (common/server/server.py): An ASGI application (using Starlette/Uvicorn) that listens for HTTP POST requests, parses JSON-RPC, and routes requests based on the method field.

  2. A2A Data Types (common/types.py): Pydantic models defining the structure of A2A messages, tasks, artifacts, errors, and the agent card, ensuring compliance with the a2a.json specification.

  3. Task Manager (task_manager.py): A custom class (CaskadaTaskManager) inheriting from the common InMemoryTaskManager. Its primary role is implementing the on_send_task method (and potentially others like on_send_task_subscribe if streaming were supported). This method:

    • Receives the validated A2A SendTaskRequest.

    • Extracts the user's query (TextPart) from the request's message.

    • Initializes the Caskada shared_data dictionary.

    • Creates and runs the Caskada agent_flow.

    • Retrieves the final answer from the shared_data dictionary after the flow completes.

    • Updates the task's state (e.g., to COMPLETED or FAILED) in the InMemoryTaskManager's store.

    • Packages the final answer into an A2A Artifact containing a TextPart.

    • Constructs the final A2A Task object for the response.

  4. Agent Card (a2a_server.py): A Pydantic model (AgentCard) defining the agent's metadata (name, description, URL, capabilities, skills) served at /.well-known/agent.json.

  5. Server Entry Point (a2a_server.py): A script that initializes the AgentCard, the CaskadaTaskManager, and the A2AServer, then starts the Uvicorn server process.

The Complexity Points System

All listed projects had their calculation automatically calculated. Click Details below for more information in the points system.

Details

Caskada Cookbook: Project Complexity Point System

This document outlines a revised point system to categorize projects within the Caskada cookbook by complexity. This system aims for simplicity, generalizability, and clear guidance for learners.

I. Core Caskada Structure & Flow

Evaluates the fundamental Caskada constructs.

  • A. Node Usage:

    • 0.5 points per distinct Node class implemented and used in the primary flow logic. (e.g., 4 unique Node classes = 2 points).

  • B. Flow Complexity:

    • Simple linear flow (e.g., A >> B >> C): 0 points

    • Flow with branching OR looping (e.g., A - "action" >> B, B >> A): 1 point

    • Flow with both branching AND looping: 2 points

    • (Note: These are mutually exclusive for a single primary flow; a flow is either linear, has branching/looping, or both.)

  • C. Advanced Flow Constructs:

    • Use of ParallelFlow (or significant parallel execution of nodes/flows): 3 points (one-time for the project if the pattern is present).

    • Use of Nested Flows (one Flow instance used as a node within another Flow): 2 points (one-time for the project if the pattern is present and significant).

II. Code & Logic Complexity

Assesses the custom code beyond basic Caskada definitions.

  • A. Logic within Node Methods (prep, exec, post):

    • 0.5 points per node that contains minor custom logic (e.g., simple data transformation, formatting beyond direct pass-through).

    • An additional 1 point for that same node (total 1.5 for that node in this sub-category) if its custom processing logic is significant (e.g., complex algorithms, detailed state management, intricate parsing).

  • B. Helper Modules/Utilities (e.g., code in utils.py or other non-node, non-flow Python/TypeScript modules, excluding simple API client wrappers which are covered by III.A/B):

    • Project contains minor utilities or simple helper functions: 1 point (total for this category if only minor utilities are present).

    • Project contains complex helper functions/classes (e.g., custom VAD, advanced data structures, significant business logic): 2 points (total for this category if complex utilities are present, supersedes the 1 point for minor).

  • C. Project File Structure & Modularity:

    • 0.25 points per relevant Python/TypeScript file (e.g., main.py, flow.py, nodes.py, utils.py, and other custom modules containing substantial logic).

    • Exclude: README.md, requirements.txt, .ipynb, __init__.py, package.json, configuration files, empty or near-empty stub files.

    • Maximum of 2 points for this category. (e.g., 4 files = 1 point; 8 files = 2 points; 10 files = 2 points).

  • D. Code Quality & Advanced Language Features:

    • Demonstrable use of strong typing (e.g., comprehensive type hints in Python, TypeScript's type system) that significantly contributes to the project's robustness or complexity of implementation: 1 point (one-time for the project).

    • Advanced error handling and resilience patterns (e.g., custom retry logic beyond basic node retries, sophisticated exception management, graceful degradation): 1 point (one-time for the project).

III. External Integrations & Tools

Points for incorporating external services, APIs, or complex data handling.

  • A. LLM API Usage (Points are cumulative if multiple distinct API capabilities are used):

    • Basic call (chat completion, text generation): 1 point

    • Embeddings API: +1 point

    • Vision API: +1 point

    • TTS (Text-to-Speech) API: +1 point

    • STT (Speech-to-Text) API: +1 point

  • B. Other External APIs/Services:

    • Web Search API (e.g., DuckDuckGo, SerpAPI): 1 point

    • Other distinct external APIs (e.g., weather API, financial data API): 1 point per distinct API type used.

  • C. Data Handling & Persistence:

    • Database Integration (e.g., SQLite, or other DBs): 2 points

    • Complex File I/O (e.g., processing PDFs, audio files, multiple structured files, image manipulation beyond simple read/write): 1 point

  • D. Implemented "Tool" Modules for LLM/Agent:

    • 1.5 points per distinct, non-trivial Python/TypeScript module designed as a reusable "tool" for an agent (e.g., crawler.py in python-tool-crawler, a custom PDF processor tool). This is distinct from general helper utilities (II.B) and focuses on agent-callable tools. (Max 4.5 points for this category).

IV. Advanced Caskada Patterns & User Interaction

Markers for sophisticated architectural patterns or user interfaces.

  • A. Key Design Patterns (Apply points for each distinct, significantly implemented pattern):

    • Basic Agent (LLM-based decision-making for flow control or simple tool selection): 2 points

    • RAG (Retrieval Augmented Generation - explicit retrieve, augment, generate steps): 3 points

    • Supervisor Pattern (a flow/node validating or overseeing another): 2 points

    • Chain-of-Thought / ReAct (multi-step, structured reasoning orchestrated by the flow): 3 points

    • Majority Vote (or similar consensus mechanism over multiple LLM calls): 2 points

    • MapReduce (explicit map and reduce nodes/flows for batch data processing): 2 points

  • B. Communication & Protocols:

    • MCP (Model Context Protocol integration): 3 points

    • A2A (Agent-to-Agent communication setup): 4 points

  • C. User Interaction Methods:

    • Enhanced CLI (beyond simple input(), e.g., argument parsing, interactive menus): 0.5 points

    • Web-based UI (e.g., Streamlit, FastAPI with HTML, React): 2 points for a basic web UI.

    • +1 additional point if the web UI involves significant backend API development (e.g., FastAPI managing state, SSE, background tasks for the UI, making it 3 total for complex web UIs).

  • D. Multi-Agent System Complexity:

    • 1 point per distinct agent role (e.g., a system with a Hinter agent and a Guesser agent gets 2 points).

    • +1 point if the agents' interaction involves complex communication patterns (e.g., asynchronous message queues, shared state beyond simple turn-passing, negotiation protocols).

    • +1 point if the system involves more than two agents that actively and distinctly contribute to the core task.

    • (Max 5 points for this category)

Complexity Tiers

Based on the total points, projects can be categorized into tiers:

  • Beginner (1-7 points): Introduces fundamental Caskada concepts.

  • Intermediate (8-16 points): Covers more complex flow structures, basic patterns, and integrations.

  • Advanced (17-25 points): Deals with sophisticated patterns, UI integrations, or complex tool building.

  • Expert (26+ points): Combines multiple advanced features, involves intricate system design.

Last updated