The Saga Pattern for AI Python Agents

The External API Problem

Intercepting an agent hallucination before it writes to your database is relatively easy. But what happens if the agent already interacted with the real world? What if it sent an email, charged a Stripe credit card, provisioned an AWS server, or deleted a file, and then the workflow crashed in a subsequent step?

You cannot simply "rollback" a Stripe charge by resetting a Python variable. You cannot un-send an email. To fix these side-effects, you must issue a refund, or send a follow-up apology email. This architectural challenge is where the Saga Pattern comes in.

What is a Saga?

The Saga pattern is a microservices concept popularized by distributed systems engineers. Instead of a single massive database transaction, a Saga is a sequence of local transactions. Every step (local transaction) has a corresponding "compensation action"—an undo function.

If a multi-step transaction fails at step 3, the system catches the failure and automatically executes the compensations for steps 2 and 1 in reverse order to clean up the mess.

AI Agents are Distributed Systems

When you build an agent that talks to OpenAI, reads from Pinecone, writes to Postgres, and charges via Stripe, you are no longer building a simple script. You are orchestrating a highly unpredictable distributed system. You must use distributed systems patterns.

Building Sagas with StateGuard

StateGuard brings first-class Saga support natively to Python and LangGraph. You can register compensation functions that are automatically triggered if a memory invariant is violated or an unhandled exception occurs.

1. Defining the Compensation

A compensation function takes the current state and the exception context, and performs whatever network calls are necessary to undo the action.

from stateguard import StateGuardCheckpointer
import stripe

# The Compensation (Undo) Function
def refund_stripe_charge(state, exception_context):
    charge_id = state.get('last_charge_id')
    if charge_id:
        print(f"Rolling back! Refunding charge: {charge_id} due to {exception_context}")
        stripe.Refund.create(charge=charge_id)
        return True
    return False

2. Registering the Saga

When you initialize your StateGuardCheckpointer or GuardedState, you map compensation functions to specific tags or nodes.

# Drop StateGuard into your existing code
checkpointer = StateGuardCheckpointer(
    invariants=[no_negative_balance],
    compensations={
        # If the 'charge_user' node was executed in this transaction,
        # and the transaction fails later, run refund_stripe_charge.
        'charge_user': refund_stripe_charge
    }
)

If the workflow proceeds from charge_user to deliver_product, and deliver_product hallucinates and causes a rollback, StateGuard looks at the transaction history. It sees that charge_user succeeded, looks up its compensation, and automatically triggers the refund.

Advanced: Idempotent Compensations

It is critical that your compensation functions are idempotent. If a network error occurs while attempting to refund the Stripe charge, StateGuard may retry the compensation. The compensation must be safe to call multiple times.

def refund_stripe_charge(state, exception_context):
    charge_id = state.get('last_charge_id')
    
    # 1. Check if already refunded (idempotency check)
    charge = stripe.Charge.retrieve(charge_id)
    if charge.refunded:
        return True
        
    # 2. Issue refund
    stripe.Refund.create(charge=charge_id)

Conclusion

Agents will fail. They will hallucinate. They will hit rate limits. By implementing the Saga pattern with StateGuard, you ensure that when the inevitable failure happens, your system gracefully cleans up its own mess without leaving users charged for products they never received.