A postgres job queue LLM agents can depend on must survive provider latency spikes, arbitrary task lengths, and concurrent workers pulling the same work. This tutorial builds a minimal but production-shaped queue using raw Postgres and Python, then runs agent tasks that call LLM endpoints through it.
Prerequisites
- Postgres 14+ (SKIP LOCKED is mandatory for safe concurrency)
- Python 3.11+
psycopg3.x:pip install "psycopg[binary]"httpxfor async HTTP:pip install httpx- A running Postgres instance and a database URL
No ORM. We stay close to the metal so the locking semantics are explicit.
Schema design
The core table holds the job payload, a status state machine, and attempt counting. The partial index keeps the hot queue query cheap.
CREATE TABLE agent_jobs (
id BIGSERIAL PRIMARY KEY,
payload JSONB NOT NULL,
status TEXT NOT NULL DEFAULT 'queued',
attempts INT NOT NULL DEFAULT 0,
max_attempts INT NOT NULL DEFAULT 3,
run_at TIMESTAMPTZ NOT NULL DEFAULT now(),
last_error TEXT,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX ON agent_jobs (status, run_at) WHERE status = 'queued';SKIP LOCKED is the trick that lets N workers scan the table without blocking each other. A worker locks one row, others skip it. Without it, you get race conditions where two workers claim the same job.
Enqueue a task
Synchronous helper for submitting work. In practice you’d call this from your API layer.
import psycopg
from psycopg.rows import dict_row
def enqueue(conn, payload: dict, run_at=None):
with conn.cursor() as cur:
cur.execute(
"""INSERT INTO agent_jobs (payload, run_at)
VALUES (%s, COALESCE(%s, now()))
RETURNING id, status""",
(psycopg.sql.Json(payload), run_at),
)
return cur.fetchone()Checkpoint: insert a summarize job.
# conn = psycopg.connect("postgres://user:pass@localhost/db")
# row = enqueue(conn, {"task": "summarize", "text": "Long transcript..."})
# print(row)Expected output:
(1, 'queued')Worker: claim a job safely
Async claim using FOR UPDATE SKIP LOCKED. We also select attempts so the retry path knows the count.
import asyncio
import psycopg
async def fetch_job(conn):
async with conn.transaction():
async with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
await cur.execute(
"""SELECT id, payload, attempts
FROM agent_jobs
WHERE status = 'queued' AND run_at <= now()
ORDER BY run_at
FOR UPDATE SKIP LOCKED
LIMIT 1"""
)
return await cur.fetchone()If two workers run fetch_job simultaneously, each gets a distinct row. No double execution. This is the property a postgres job queue LLM agents use must guarantee.
State transitions
Mark success or schedule a retry with exponential backoff.
async def mark_done(conn, job_id):
async with conn.cursor() as cur:
await cur.execute(
"UPDATE agent_jobs SET status='done', updated_at=now() WHERE id=%s",
(job_id,),
)
async def mark_error(conn, job_id, err, run_at):
async with conn.cursor() as cur:
await cur.execute(
"""UPDATE agent_jobs
SET attempts = attempts + 1,
status = CASE WHEN attempts + 1 >= max_attempts THEN 'failed' ELSE 'queued' END,
run_at = %s,
last_error = %s,
updated_at = now()
WHERE id = %s""",
(run_at, str(err), job_id),
)Running the LLM agent task
The task itself calls an OpenAI-compatible chat endpoint. Swap the URL for any gateway.
import httpx
API_URL = "https://api.openai.com/v1/chat/completions"
HEADERS = {"Authorization": f"Bearer {open('key.txt').read().strip()}"}
async def run_agent_task(payload: dict) -> dict:
if payload.get("task") == "summarize":
async with httpx.AsyncClient(timeout=60) as client:
resp = await client.post(
API_URL,
headers=HEADERS,
json={
"model": "gpt-4o-mini",
"messages": [
{"role": "system", "content": "Summarize the text."},
{"role": "user", "content": payload["text"]},
],
},
)
resp.raise_for_status()
return resp.json()
raise ValueError(f"unknown task: {payload.get('task')}")If you route through n4n.ai, the OpenAI-compatible endpoint gives automatic fallback when a provider is rate-limited, so the worker can skip custom 429 retries and just treat failures as transient.
Worker loop
Pull, execute, transition. The backoff uses attempts to compute delay.
async def worker(conn):
while True:
job = await fetch_job(conn)
if not job:
await asyncio.sleep(1)
continue
try:
await run_agent_task(job["payload"])
await mark_done(conn, job["id"])
print(f"job {job['id']} done")
except Exception as e:
backoff = 2 ** (job["attempts"] + 1)
await mark_error(
conn,
job["id"],
e,
f"now() + interval '{backoff} seconds'",
)
print(f"job {job['id']} retry in {backoff}s: {e}")Launch concurrent workers
async def main():
conn = await psycopg.AsyncConnection.connect(
"postgres://user:pass@localhost/db",
row_factory=psycopg.rows.dict_row,
)
await asyncio.gather(*(worker(conn) for _ in range(4)))
if __name__ == "__main__":
asyncio.run(main())Checkpoint: with two jobs queued, output looks like:
job 1 done
job 2 doneA postgres job queue LLM agents run in production should log these transitions to a metrics system; the table itself is the source of truth.
Why SKIP LOCKED beats SELECT then UPDATE
A naive design does SELECT ... WHERE status='queued' LIMIT 1 then UPDATE. Under concurrency, two workers read the same row before either updates it. You either add FOR UPDATE (which blocks all workers on one row) or accept duplicates. SKIP LOCKED gives you the best of both: each worker grabs an unlocked row and moves on. At 10 workers, throughput scales nearly linearly for independent jobs.
Handling long-running LLM calls
Agent tasks can hang on slow providers. Set explicit HTTP timeouts (above) and let the exception path retry. If a job exceeds max_attempts, it lands in failed status. Inspect failures:
SELECT id, last_error FROM agent_jobs WHERE status = 'failed';Expected output for a bad API key:
id | last_error
----+-----------------------------------------
3 | 401 Unauthorized: Invalid AuthenticationThe postgres job queue LLM agents rely on should treat failed as terminal and alert, not loop forever.
Inspect queue health
SELECT status, count(*) FROM agent_jobs GROUP BY status;Expected after draining:
status | count
---------+-------
done | 2
queued | 0
failed | 0Idempotency and deduplication
LLM agent triggers often arrive twice (webhook retries, client timeouts). Add a deterministic key:
ALTER TABLE agent_jobs ADD COLUMN dedupe_key TEXT UNIQUE;Then enqueue uses INSERT ... ON CONFLICT (dedupe_key) DO NOTHING. The postgres job queue LLM agents pattern stays safe under at-least-once delivery.
Waking workers instead of polling
Polling sleep(1) is fine at low volume. For tighter latency, use LISTEN/NOTIFY:
-- in enqueue, after INSERT:
SELECT pg_notify('job_added', '');Worker opens a second connection, listens, and calls fetch_job on notify. This cuts idle latency to milliseconds without hammering Postgres.
Where to take it
The design above is the skeleton for a postgres job queue LLM agents can scale with. Add:
- Partitioning by
idrange once you pass 10M rows. - A dead-letter table for
failedjobs with fulllast_error. - Per-tenant isolation by adding
tenant_idand indexing it. pg_advisory_lockif you need a singleton scheduler.
Postgres is not Redis, but for agent workloads where the job payload is a rich JSON document and you already have ACID guarantees, it removes an extra moving part. Build the queue, point your agents at it, and let the database handle the hard concurrency problems.