Skip to content

Getting Started

This guide provides a practical walkthrough for setting up VibeBlocks and executing your first resilient workflow. Following the AA Digital engineering standards, we prioritize type safety and explicit state management from the very first line of code.

Installation

VibeBlocks is a "Zero-Gravity" framework. It requires only a modern Python environment (3.8+).

pip install vibeblocks

Your First Workflow: Data Enrichment

In this example, we will build a simple enrichment pipeline that validates an input and processes it.

1. Define the Execution Context

VibeBlocks flows revolve around a central ExecutionContext. We recommend using Python dataclasses or Pydantic models to define your state for full type-hinting support.

from dataclasses import dataclass
from typing import Optional

@dataclass
class ProcessingContext:
raw_input: str
processed_result: Optional[str] \= None
is_valid: bool \= False

2. Create Atomic Blocks

Use the @block decorator to transform standard Python functions into managed units of work.

from vibeblocks import block, ExecutionContext

@block(description="Validates that the input string is not empty.")
def validate_input(ctx: ExecutionContext[ProcessingContext]):
if len(ctx.data.raw_input) > 0:
ctx.data.is_valid \= True
else:
raise ValueError("Input cannot be empty")

@block(description="Transforms raw input into uppercase.")
def transform_data(ctx: ExecutionContext[ProcessingContext]):
ctx.data.processed_result \= ctx.data.raw_input.upper()

3. Assemble and Execute

For simple sequences, the execute_flow utility provides a streamlined entry point that handles the initialization of the ExecutionContext and the SyncRunner.

from vibeblocks import Flow, execute_flow

# Define the orchestration structure
pipeline \= Flow(
name="EnrichmentFlow",
blocks=[validate_input, transform_data]
)

# Execute the flow with initial data
initial_state \= ProcessingContext(raw_input="vibe_blocks_initial_test")
outcome \= execute_flow(pipeline, initial_state)

# Inspect the result
if outcome.status \== "SUCCESS":
print(f"Result: {outcome.context.data.processed_result}")
print(f"Execution took: {outcome.duration_ms}ms")
else:
print(f"Flow failed with errors: {outcome.errors}")

Understanding the Execution Lifecycle

When you run a flow, VibeBlocks performs several automated blocks to ensure reliability:

sequenceDiagram
participant U as User Code
participant R as Runner
participant C as ExecutionContext
participant B as Block

U-\>\>R: execute\_flow(Flow, Data)  
R-\>\>C: Initialize(Data)  
Note over R,C: Create Trace & Metrics  
loop For each Step  
    R-\>\>B: execute(Context)  
    B-\>\>C: Log Event (Start)  
    B-\>\>B: Run User Logic  
    B-\>\>C: Update Data State  
    B-\>\>C: Log Event (Success)  
end  
R-\>\>U: Return Outcome Object

Next blocks

Now that you have executed your first flow, you are ready to explore the core building blocks that allow for high-end AI orchestration:

Engineered with precision by AA Digital Business. High-end AI Architecture.