Outils idempotents et effets de bord

Pourquoi l’idempotence est importante

Qu’est-ce que l’idempotence ?

Clés d’idempotence

import uuid

def create_payment(amount, idempotency_key):
    # Stripe accepts this header — returns the existing payment if key was used
    return stripe.Charge.create(
        amount=amount,
        currency='usd',
        source=token,
        idempotency_key=idempotency_key
    )

Générer les clés à partir de l’intention

import hashlib, json

def intent_key(tool_name, args, run_id):
    payload = json.dumps([tool_name, args, run_id], sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()

print(intent_key('transfer_money', {'to': 'acc2', 'amount': 100}, 'run-1'))

Inventaire des effets de bord

Encapsuler les outils à effets de bord

def send_email(to, subject, body, run_id):
    key = intent_key('send_email', {'to': to, 'subject': subject, 'body': body}, run_id)
    if redis.set(f'sent:{key}', 1, nx=True, ex=86400):
        smtp.send(to, subject, body)
        return {'sent': True}
    return {'sent': False, 'reason': 'already-sent'}

Confirmer avant d’agir

def transfer_money(from_acc, to_acc, amount, confirmed=False):
    if not confirmed:
        return {'pending': True, 'msg': 'Call again with confirmed=True to proceed.'}
    return {'pending': False, 'msg': f'Transferred {amount} from {from_acc} to {to_acc}.'}

print(transfer_money('acc1', 'acc2', 100))
print(transfer_money('acc1', 'acc2', 100, confirmed=True))

Outils en deux phases

def draft_email(to, subject, body):
    draft_id = save_draft(to, subject, body)
    return {'draft_id': draft_id, 'preview': body[:200]}

def send_draft(draft_id):
    return smtp.send(load_draft(draft_id))

Actions compensatoires

def refund_payment(payment_id):
    return stripe.Refund.create(payment=payment_id)
# Now an erroneous create_payment can be undone.

Journaux d’audit

Budget d’effets de bord par exécution

if state.side_effect_count >= 5:
    return {'error': 'Side-effect budget exhausted for this run.'}

Mode simulation

import os
os.environ['DRY_RUN'] = '1'
DRY_RUN = os.getenv('DRY_RUN') == '1'

def send_email(to, subject, body):
    if DRY_RUN:
        print(f'would send email to {to}')
        return {'sent': False, 'dry_run': True}
    print(f'sending email to {to}')
    return {'sent': True, 'dry_run': False}

result = send_email('user@example.com', 'Hi', 'Hello there')
print(result)

Définition

Récapitulatif