Skip to main content

Multi-Step Workflows

This tutorial covers building complex workflows with branching, composition, and error handling.

Branching Workflows​

Route execution based on decisions:

@agent(name="support-router")
class SupportRouter:

    @workflow
    def handle_ticket(self, ctx):
        # Step 1: Classify
        category = ctx.call(self.classify)

        # Step 2: Route based on classification
        if category == "billing":
            return ctx.call(self.handle_billing)
        elif category == "technical":
            return ctx.call(self.handle_technical)
        else:
            return ctx.call(self.handle_general)

    @decision
    def classify(self, ctx):
        return ctx.llm.classify(ctx.input.message,
            categories=["billing", "technical", "general"])

    @workflow
    def handle_billing(self, ctx):
        # Billing-specific workflow
        pass

    @workflow
    def handle_technical(self, ctx):
        # Technical-specific workflow
        pass

    @workflow
    def handle_general(self, ctx):
        # General handling
        pass

Composing Workflows​

Workflows can call other workflows:

@agent(name="order-processor")
class OrderProcessor:

    @workflow
    def process_order(self, ctx):
        # Validate
        validated = ctx.call(self.validate_order)

        # Process payment
        payment = ctx.call(self.process_payment, order=validated)

        # Fulfill
        fulfillment = ctx.call(self.fulfill_order,
            order=validated,
            payment=payment
        )

        return fulfillment

    @workflow
    def validate_order(self, ctx):
        # Validation logic
        return ctx.input.order

    @workflow
    def process_payment(self, ctx, order):
        # Payment processing
        return {"status": "success", "order_id": order["id"]}

    @workflow
    def fulfill_order(self, ctx, order, payment):
        # Fulfillment logic
        return {"shipped": True}

Error Handling​

Handle errors gracefully:

@workflow
def resilient_workflow(self, ctx):
    try:
        result = ctx.call(self.risky_operation)
        return result
    except Exception as e:
        # Log and handle error
        return ctx.call(self.handle_error, error=str(e))

Next Steps​