Retrying LLM requests without duplicating side effects

Retrying LLM requests without duplicate side effects is harder than wrapping a call in a for-loop. The model inference is stateless, but the code around it—database writes, webhooks, queue publishes—is not, and a timeout after a successful generation can double-execute those mutations on retry.

Step 1: Isolate the inference call from mutations

Treat the LLM call as a pure function: same input, same output (or acceptable variance), no external writes. If your function both calls the model and inserts a row, a retry after a network drop can insert twice.

Define a narrow interface:

from dataclasses import dataclass

@dataclass
class InferenceResult:
    text: str
    model: str
    finish_reason: str

def generate(prompt: str, model: str) -> InferenceResult:
    # only network I/O to the model gateway, no side effects
    ...

Keep any business logic that consumes InferenceResult in a separate function that receives the result as an argument.

Step 2: Generate a deterministic operation key

The retry must be keyed to the logical user action, not the transport attempt. Generate one idempotency key per high-level operation (e.g., “summarize ticket 123 for user 9”) and thread it through the call.

import uuid

def operation_key(user_id: str, action: str, ref: str) -> str:
    return f"{user_id}:{action}:{ref}"

# or a random but persisted key from the client
key = uuid.uuid4().hex

Persist this key in your request context, job queue message, or GraphQL resolver so every retry of the same operation reuses it.

Step 3: Cache LLM responses by operation key

Before paying for inference, check a key-value store. On hit, return the stored completion. On miss, call the model and write back with a TTL. This makes the inference itself replayable.

import redis, json, openai

r = redis.Redis(host="localhost", port=6379, db=0)

def generate_idempotent(prompt: str, op_key: str, model: str = "gpt-4o-mini") -> InferenceResult:
    cache_hit = r.get(f"llm:{op_key}")
    if cache_hit:
        data = json.loads(cache_hit)
        return InferenceResult(**data)

    resp = openai.ChatCompletion.create(
        api_base="https://your-gateway.example/v1",  # OpenAI-compatible
        model=model,
        messages=[{"role": "user", "content": prompt}],
        seed=42,  # improves determinism when supported
    )
    result = InferenceResult(
        text=resp.choices[0].message.content,
        model=resp.model,
        finish_reason=resp.choices[0].finish_reason,
    )
    r.setex(f"llm:{op_key}", 3600, json.dumps(result.__dict__))
    return result

If you route through n4n.ai, its automatic fallback when a provider is rate-limited or degraded reduces manual provider-retry logic, but the side-effect duplication problem remains yours.

Step 4: Make side effects idempotent with a unique constraint

The safest commit is a database row with a primary key on the operation key. Any retry that reaches the commit step collides and is a no-op.

CREATE TABLE ticket_summaries (
    op_key TEXT PRIMARY KEY,
    ticket_id INTEGER NOT NULL,
    summary TEXT NOT NULL,
    created_at TIMESTAMPTZ DEFAULT now()
);

def commit_summary(op_key: str, ticket_id: int, summary: str) -> bool:
    try:
        cursor.execute(
            "INSERT INTO ticket_summaries (op_key, ticket_id, summary) VALUES (%s, %s, %s)",
            (op_key, ticket_id, summary),
        )
        conn.commit()
        return True
    except psycopg2.errors.UniqueViolation:
        conn.rollback()
        return False  # already done by a prior attempt

For non-DB side effects (email, webhook), push the payload to an outbox table with the same unique key, then a separate worker drains it exactly once.

Step 5: Build the retry loop around the isolated pieces

Retry only the inference fetch, not the commit. Because the inference is cached by op_key, repeated calls are cheap and safe. The commit is guarded by the unique constraint.

import time, random

class TransientLLMError(Exception):
    pass

def call_with_backoff(fn, max_attempts=4):
    for attempt in range(max_attempts):
        try:
            return fn()
        except TransientLLMError as e:
            if attempt == max_attempts - 1:
                raise
            sleep = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(sleep)

def run_operation(user_id: str, ticket_id: int, prompt: str):
    op_key = operation_key(user_id, "summarize", str(ticket_id))
    result = call_with_backoff(
        lambda: generate_idempotent(prompt, op_key)
    )
    committed = commit_summary(op_key, ticket_id, result.text)
    return {"committed": committed, "text": result.text}

This structure means retrying LLM requests without duplicate side effects is mostly a data-modeling problem, not a networking one.

Step 6: Handle concurrency and partial writes

Two workers may process the same op_key simultaneously. The DB unique constraint handles the final state, but you should also use INSERT ... ON CONFLICT DO NOTHING and check cursor.rowcount to know if you were the writer.

def commit_summary_conflict(op_key: str, ticket_id: int, summary: str) -> bool:
    cursor.execute(
        "INSERT INTO ticket_summaries (op_key, ticket_id, summary) "
        "VALUES (%s, %s, %s) ON CONFLICT (op_key) DO NOTHING",
        (op_key, ticket_id, summary),
    )
    return cursor.rowcount == 1

If you publish to a message queue, set the message key to op_key so the broker deduplicates or partitions correctly.

Step 7: Verify the setup with a fault injection test

Write a test that forces a timeout after the LLM returns but before the commit, then retries.

def test_retry_does_not_duplicate(monkeypatch, pg_conn):
    calls = {"n": 0}
    def fake_generate(prompt, op_key, model=None):
        calls["n"] += 1
        if calls["n"] == 1:
            # simulate success then transport failure before return
            raise TransientLLMError("timeout")
        return InferenceResult(text="summary", model="x", finish_reason="stop")

    monkeypatch.setattr("__main__.generate_idempotent", fake_generate)
    run_operation("u1", 42, "summarize")
    run_operation("u1", 42, "summarize")  # retry same op_key

    cur = pg_conn.cursor()
    cur.execute("SELECT count(*) FROM ticket_summaries WHERE ticket_id=42")
    assert cur.fetchone()[0] == 1
    assert calls["n"] == 2  # inference retried, row inserted once

Run with pytest. If the count is 1 and the inference was attempted twice, your idempotency layer works.

Step 8: Apply the same pattern to streaming

Streaming complicates caching because tokens arrive incrementally. Buffer the full response in memory, write it to the cache only after the stream closes successfully, and key the stream by op_key. If the stream breaks mid-way, the retry starts fresh and the prior partial buffer is discarded. Never write partial side effects (e.g., incremental DB updates) per token.

Pitfalls to avoid

• Keying on request timestamp: every retry gets a new key, defeating the cache.
• Retrying outside the gateway but inside a transaction: the DB transaction may roll back, but the LLM call already happened and is not rolled back.
• Assuming the model is deterministic: even with seed, providers may differ; cache the actual returned text, don’t regenerate from the prompt.
• Ignoring background workers: if a webhook sends the result, the outbox must be drained by a single consumer per op_key.

When retrying LLM requests without duplicate side effects, the rule is simple: make the inference cheap to replay and make the mutation impossible to repeat. The code above is the minimum viable scaffold; adapt the store and constraint to your stack.