Durable workflows
with AI cost controls

activities:
  - key: analyze_content
    worker: std
    activity_name: llm_prompt
    parameters:
      model:
        - anthropic/claude-haiku-4-5-20251001  # Try first
        - openai/gpt-4o-mini                   # Fallback
        - google/gemini-2.0-flash-lite         # Last resort
      prompt: |
        You are a content moderation assistant.
        Analyze the following text and determine
        if it violates community guidelines.

        Text to moderate:
        {{INPUT.user_content}}

        Respond with a JSON object containing:
        - "violates": true/false
        - "reason": explanation of your decision
        - "severity": "none", "low", "medium", or "high"
        - "confidence": probability score (0.00 to 1.00)
      max_tokens: 500
    outputs:
      - result
    settings:
      timeout_seconds: 30
      retry:
        max_attempts: 3
        strategy: exponential
        base_seconds: 2
        factor: 2
        max_seconds: 60
      budget:
        limit: 0.10
        action: abort

# Python SDK with ScriptActivity
from kruxiaflow import ScriptActivity, Workflow

# SQL transformation using DuckDB
@ScriptActivity.from_function(
    inputs={"clean_data": cleanse_data["clean_data"]},
    depends_on=["cleanse_data"],
)
async def sql_transform(clean_data):
    import duckdb, pandas as pd

    df = pd.DataFrame(clean_data)

    # SQL query on DataFrame
    results = duckdb.sql("""
        SELECT category, region,
               SUM(amount) as total_sales
        FROM df
        GROUP BY category, region
    """).df()

    return {"results": results.to_dict()}

# Python SDK with ML training
from kruxiaflow import ScriptActivity, Workflow

# Train Random Forest model
@ScriptActivity.from_function(
    inputs={
        "X_train": split_data["X_train"],
        "y_train": split_data["y_train"],
    },
    depends_on=["split_data"],
)
async def train_model(X_train, y_train):
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.model_selection import GridSearchCV

    # Hyperparameter tuning with cross-validation
    param_grid = {
        "n_estimators": [50, 100],
        "max_depth": [5, 10],
    }
    grid_search = GridSearchCV(
        RandomForestClassifier(), param_grid
    )
    grid_search.fit(X_train, y_train)

    return {"best_params": grid_search.best_params_}