Идемпотентные инструменты и побочные эффекты

Почему важна идемпотентность

Что такое идемпотентность

Ключи идемпотентности

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
    )

Создавайте ключи на основе намерения

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'))

Инвентаризация побочных эффектов

Оборачивайте инструменты с побочными эффектами

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'}

Подтверждайте действие перед выполнением

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))

Двухфазные инструменты

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))

Компенсирующие действия

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

Журналы аудита

Бюджет побочных эффектов на один запуск

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

Режим пробного запуска

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)

Определение

Итоги