A long running research agent timeout is the silent killer of multi-hour LLM workflows. If your agent loops over hundreds of sources and blocks on a synchronous API call, a single 30-second HTTP timeout or a provider outage wipes out hours of progress. This guide shows how to architect research agents that survive network blips, provider limits, and process restarts.
Step 1: Move execution off the request path
Never run a multi-hour agent inside a web request handler. The HTTP server has its own idle and connection timeouts, and most load balancers kill requests at 30 or 60 seconds. Use a job queue and a separate worker process. The API endpoint only enqueues a job and returns a job ID; the worker runs without any inbound timeout.
# Using RQ as an example; Celery, NATS, or SQS work similarly
from redis import Redis
from rq import Queue
redis = Redis()
q = Queue("research", connection=redis)
job = q.enqueue("agents.research.run", topic="quantum computing", depth=5)
print(job.id) # return this to clientSpin up one or more workers with rq worker research. The queue connection uses a short timeout, but that is independent of job runtime. A long running research agent timeout now only affects the initial enqueue, not the research itself.
Make the worker resilient: run it under systemd or Kubernetes with restart policies. If the host reboots, the queue redelivers the job and your checkpointing (next step) resumes it.
Step 2: Persist state after every step
Checkpoint the agent’s progress to a durable store. A simple JSON blob in Postgres, MySQL, or even SQLite is enough for most cases. The key is writing after each completed sub-task, not at the end of the run.
import json, sqlite3
def save_state(conn, job_id, state):
conn.execute(
"INSERT OR REPLACE INTO agent_state (job_id, data) VALUES (?, ?)",
(job_id, json.dumps(state)),
)
conn.commit()
def load_state(conn, job_id):
row = conn.execute("SELECT data FROM agent_state WHERE job_id=?", (job_id,)).fetchone()
return json.loads(row[0]) if row else NoneState should include: current sub-question, visited source URLs, intermediate findings, and the next action. On restart, load state and resume from the last completed boundary. Use a single row per job and rely on the DB transaction to avoid partial writes.
If you run multiple workers consuming the same queue, serialize state writes with a job-level lock (e.g., SELECT ... FOR UPDATE in Postgres or a Redis lock) to prevent lost updates. Lost updates are worse than a timeout because they silently corrupt findings.
If you expect schema changes, store a version field and migrate lazily. Don’t over-engineer; a flat dict is fine for v1.
Step 3: Decompose the research into bounded tasks
A monolithic “research everything” loop is undebuggable and unrestartable. Split work into discrete tasks: generate sub-questions, fetch sources per sub-question, summarize each source, then synthesize. Each task is a unit of checkpointing.
{
"job_id": "abc123",
"subquestions": ["what is X", "how does Y affect Z"],
"completed": ["what is X"],
"findings": {"what is X": "..."},
"status": "processing",
"version": 1
}This structure makes a long running research agent timeout recoverable: you lose at most one sub-task, not the whole run. The worker logic becomes a state machine:
- If
completedlength <subquestionslength, pick next unanswered question. - Fetch and summarize.
- Append to
findings, mark completed, save state. - When all done, call synthesis model and write final report.
This map-reduce style keeps each LLM call small and bounded. Fetching can be parallelized with asyncio, but keep state saves serial to avoid race conditions. The synthesis step can itself be chunked if the final report is large.
Step 4: Call LLMs asynchronously with retries and fallback
Model calls are the most likely point of failure. Use an OpenAI-compatible client with a gateway that provides automatic fallback when a provider is rate-limited or degraded. For example, n4n.ai exposes one endpoint covering 240+ models and fails over automatically, so a single base_url change removes a class of timeout errors.
from openai import OpenAI
client = OpenAI(
base_url="https://api.n4n.ai/v1", # OpenAI-compatible, auto fallback
api_key="YOUR_KEY",
)
def summarize(text):
resp = client.chat.completions.create(
model="anthropic/claude-3.5-sonnet", # or any of 240+ models
messages=[{"role": "user", "content": f"Summarize: {text}"}],
timeout=60, # per-call, not whole job
)
return resp.choices[0].message.contentWrap calls in a retry loop with exponential backoff. Set per-call timeouts, never a global multi-hour timeout. The gateway’s fallback handles provider-side 429s; your retry handles transient network errors.
import time
def call_with_retry(fn, attempts=5):
for i in range(attempts):
try:
return fn()
except Exception as e:
if i == attempts - 1:
raise
time.sleep(2 ** i)Honor client routing directives when you must pin a provider for compliance; a good gateway forwards your model choice and only falls back when that provider is unavailable. Pick models strategically: cheap models for extraction, stronger ones for synthesis.
Step 5: Stream and checkpoint incrementally
For long summaries or synthesis, stream tokens and append to the state store in chunks. This avoids holding large strings in memory and gives you progress markers if the process is killed.
def stream_summary(text, state, conn, job_id):
stream = client.chat.completions.create(
model="openai/gpt-4o-mini",
messages=[{"role": "user", "content": f"Analyze: {text}"}],
stream=True,
)
buf = ""
for chunk in stream:
delta = chunk.choices[0].delta.content or ""
buf += delta
if len(buf) > 500: # checkpoint every ~500 chars
state["partial"] = buf
save_state(conn, job_id, state)
buf = ""
state["partial"] = buf
save_state(conn, job_id, state)Streaming also surfaces errors early; a broken connection mid-stream raises before you’ve wasted a full call. Always persist the partial so a long running research agent timeout leaves a recoverable fragment.
Step 6: Make tasks idempotent with unique keys
Workers die. Queues redeliver. Use idempotency keys per sub-task so a duplicate execution doesn’t double-count sources or burn tokens.
def process_subquestion(conn, job_id, sq):
key = f"{job_id}:{sq}"
if conn.execute("SELECT 1 FROM done WHERE k=?", (key,)).fetchone():
return # already processed
# ... do work ...
conn.execute("INSERT INTO done (k) VALUES (?)", (key,))
conn.commit()This guarantees that a long running research agent timeout during a restart doesn’t corrupt the final report. At-least-once delivery is fine when your processing is idempotent.
Step 7: Monitor and verify success
Success means the job reaches status: "completed" in the state store and the synthesized report exists. Add a simple poll endpoint or CLI.
# Check job state via sqlite
sqlite3 agent.db "SELECT data FROM agent_state WHERE job_id='abc123';" | jq .statusExpect to see "completed". If it’s stuck in "processing" after hours, inspect the worker log for the last checkpoint time. Set an external watchdog (e.g., a cron) that alerts if no state update occurred in 30 minutes.
To verify end-to-end, run a short test job with depth=1 and confirm the state transitions to completed and the report file appears. Then scale to multi-hour runs with confidence.
Practical caveats
Provider rate limits are real. Even with fallback, you must throttle concurrent calls. Use a semaphore in the worker:
import asyncio
sem = asyncio.Semaphore(5) # max 5 concurrent LLM calls
async def bounded_call(coro):
async with sem:
return await coroAlso, cache-control hints matter. Forward provider cache directives if your gateway supports them; n4n.ai forwards cache-control hints so repeated context (like system prompts) hits provider caches and cuts latency and cost.
A long running research agent timeout is not a single bug; it’s an architecture problem. Decouple, checkpoint, decompose, and call models with per-call limits. Do that and multi-hour agents become boringly reliable.