Your AI agent crashed at step 7. Now what? Meet pylinxi

Production AI agents fail in boring ways, and pylinxi is a small open-source Python library built to make those failures recoverable.

Picture a support agent handling a refund. It classifies the ticket with an LLM, looks up the charges, drafts a reply, waits for a manager to approve, then issues the refund. Somewhere around step 4 the container is recycled. Now what?

Demos rarely hit these questions. Production hits them every day: rate limits, malformed JSON from a model, a deploy in the middle of a run, a human who answers tomorrow instead of in 30 seconds. An agent that can't survive them isn't ready for real users, however good its prompts are.

This post walks through why that matters, and how pylinxi handles it with plain Python instead of a graph.

Graph frameworks such as LangGraph answer durability well: you describe your agent as nodes and edges, keep state in a typed schema, and a checkpointer saves progress between nodes. For deterministic or semi-structured workflows that's a good fit, and LangGraph's ecosystem (integrations, LangSmith tracing, a hosted platform) is a real strength.

The cost shows up as agents grow. Teams commonly run into three things:

None of this makes graphs wrong. It suggests a different trade-off is worth having: keep the durability, drop the graph.

In pylinxi a flow is an ordinary async def, and every call to a @step is saved to a journal. Branching is if, loops are for, parallelism is asyncio.gather. There is no graph to wire.

import asyncio from pylinxi import flow, step, SQLiteJournal @step async def classify ( message ): return await llm ( f " Classify this ticket: { message } " ) @step def lookup_charges ( order ): return billing. charges ( order ) @step def refund ( order ): return billing. refund ( order ) @flow ( journal = SQLiteJournal ( " runs.db " )) async def support ( ctx, message, order ): intent = await classify ( message ) if intent!= " refund ": # branching is just `if` return await answer_faq ( message ) charges = await lookup_charges ( order ) if not await ctx. wait_for ( " manager_approval " ): return " escalated " return await refund ( order ) result = await support. run ( " I was charged twice ", " A-1001 ", run_id = " ticket-1001 " ) Enter fullscreen mode Exit fullscreen mode If the process dies halfway through, you resume the run by its id from any process:

result = await support. resume ( " ticket-1001 " ) Enter fullscreen mode Exit fullscreen mode pylinxi runs the function again from the top. Every step that already finished returns its saved result instantly, without calling the LLM or the billing API again, and execution carries on at the first step that hadn't finished.

That replay model comes with one rule: code outside steps must be deterministic. Put I/O, LLM calls, randomness and clocks inside steps (or use ctx.now() and ctx.uuid(), which record their value once). Everything else stays plain Python. Steps called outside a flow are plain functions too, so you unit-test them like any other code.

Each step gets one declarative Policy, and the healing runs in this order: retry, validate, repair, fallback, circuit breaker.

The repair hook is the part most frameworks leave to you. It runs between attempts and may rewrite the step's inputs. For LLMs, that means feeding the actual error back into the prompt:

import json from pylinxi import step, Policy, Retry, CircuitBreaker def is_json ( text ): try: json. loads ( text ) return True except ValueError: return False def feed_error_back ( error, args, kwargs, attempt ): prompt = args [ 0 ] + f " \n\n Your last answer failed: { error }. Return strict JSON. " return ( prompt,), kwargs @step ( policy = Policy ( retry = Retry ( attempts = 4, backoff = 0.5, on = ( ConnectionError, TimeoutError )), timeout = 20, # seconds, per attempt validate = is_json, # False -> retried repair = feed_error_back, # rewrite inputs before the next attempt )) async def classify ( prompt ): return await llm ( prompt ) Enter fullscreen mode Exit fullscreen mode For dependencies that go down, add a fallback and share one breaker per service:

billing = CircuitBreaker ( threshold = 5, reset_after = 30 ) @step ( policy = Policy ( breaker = billing, fallback = lambda order: { " order ": order, " charges ": " unavailable " })) def lookup_charges ( order ): return billing_api. charges ( order ) Enter fullscreen mode Exit fullscreen mode After 5 consecutive failures the breaker opens and calls go straight to the fallback for 30 seconds, across every run, instead of hammering a dead service. Then it lets exactly one trial call through. The journal records what happened on each step ( attempts, and healed as "retry" or "fallback" ), so you can see where your agent is quietly struggling.

Per-step policies handle a flaky call. A Worker handles the fleet: it resumes runs whose timers are due and, optionally, runs that ended in an unhandled failure, with exponential backoff and a give-up limit.

from pylinxi import Worker worker = Worker ( journal, [ support, onboarding ], retry_failed = True, max_failures = 5, backoff = 30 ) await worker. run_forever ( interval = 5 ) Enter fullscreen mode Exit fullscreen mode Three details make replay safe rather than merely possible:

from pylinxi import idempotency_key @step def refund ( order ): return stripe. Refund. create ( charge = order. charge_id, idempotency_key = idempotency_key ()) Enter fullscreen mode Exit fullscreen mode Runs also hold a lock while they execute, so two workers can't resume the same run at once: the second gets RunLocked.

Real agents wait on people and on time. In pylinxi both are durable: a waiting run frees its worker and survives restarts.

Signals. ctx.wait_for("approval") suspends the run until you deliver the signal, minutes or days later, from any process. Signals are saved when delivered, so one that arrives before the flow reaches its wait_for isn't lost.

result = await support. run ( msg, order, run_id = " t-1001 " ) # status: "suspended" #...a day later, from your web handler: result = await support. resume ( " t-1001 ", signals = { " manager_approval ": True }) Enter fullscreen mode Exit fullscreen mode Timers. A sales agent that follows up if a lead goes quiet is three lines:

@flow ( journal = journal ) async def nurture ( ctx, lead ): await send_email ( lead, await draft_reply ( lead )) answer = await ctx. wait_for ( " customer_reply ", timeout = 3 * 86400, default = None ) if answer is None: await send_email ( lead, f " Just following up, { lead }. Any questions? " ) Enter fullscreen mode Exit fullscreen mode ctx.sleep(seconds) works the same way. The Worker resumes the run when the deadline passes, even if every process restarted in between.

Streaming. Write a step as a generator and pylinxi streams each chunk live while saving only the final text. If a retry restarts the stream, you get a stream_reset event so your UI can clear the partial answer.

@step async def answer ( prompt ): async for token in llm. stream ( prompt ): yield token async for event in chat. stream ( " hello ", run_id = " chat-1 " ): if event [ " event " ] == " chunk ": await websocket. send ( event [ " chunk " ]) Enter fullscreen mode Exit fullscreen mode 4. A journal of deltas, not snapshots pylinxi never writes a full copy of your state per step. Each journal entry holds one step's result plus only the keys of ctx.state that changed since the last entry. A long agent run with a large conversation history doesn't rewrite that history on every step.

You can still rebuild state at any point by folding the deltas, which makes debugging a bad run a matter of time travel:

journal. state_at ( " ticket-1001 " ) # state now journal. state_at ( " ticket-1001 ", seq = 3 ) # state as of entry 3 Enter fullscreen mode Exit fullscreen mode Long-lived loops such as chat sessions or monitors use ctx.continue_as_new to roll over into a fresh journal ( chat.g1, chat.g2,...), so replay stays short however long the session runs. Resuming the original id follows the chain.

The codec is lossless: tuples, sets, bytes and non-string dict keys round-trip exactly, and your own dataclasses or pydantic models do too once registered. An unregistered class raises an error rather than silently coming back as a dict on replay.

An agent's journal is a record of what it decided and did, so pylinxi treats it as evidence.

The pylinxi command-line tool makes this usable in CI and audits:

export PYLINXI_JOURNAL = postgresql://app@db/agents PYLINXI_KEYS = "2026-q4=..." pylinxi runs --status failed pylinxi show ticket-1001 # every entry: kind, step, tries, healed, changed state pylinxi state ticket-1001 --seq 3 # time travel pylinxi verify --all --require-signed # exits 1 on any failure pylinxi delete ticket-1001 --yes Enter fullscreen mode Exit fullscreen mode A note on certifications: SOC 2 or ISO 27001 apply to an organization or a hosted service, never to a library. What a library can do is make the audit easy, and add nothing to vet: pylinxi makes no network calls and its core has zero dependencies.

Many workers, one database. PostgresJournal serializes appends under a per-run advisory lock and reads the chain from the database, so any number of workers can share it. A worker that dies releases its run lock automatically.

from pylinxi.postgres import PostgresJournal from pylinxi.otel import OTelExporter journal = PostgresJournal ( " postgresql://app@db:5432/agents ", secret = KEY_FROM_KMS ) @flow ( journal = journal, on_event = OTelExporter ()) async def support ( ctx, message, order ):... Enter fullscreen mode Exit fullscreen mode Tracing. OTelExporter uses only the OpenTelemetry API, so it works with any backend you already run (OTLP, Jaeger, Datadog, Honeycomb). Each run is a span with a child span per executed step, and retries, repairs, fallbacks and breaker trips appear as events on the step's span. Replayed steps show as events, not new spans, so a trace shows what actually ran. Set record_errors=False if error text may contain personal data.

Deploying while runs are in flight. Changing flow code under a suspended run is the classic way to break replay. ctx.patched() makes it safe:

if ctx. patched ( " fraud-check-v2 " ): await fraud_check ( order ) # runs that reach this point on the new code # runs already past this point keep their old behaviour on replay Enter fullscreen mode Exit fullscreen mode If flow code changes in a way that doesn't match the journal, pylinxi emits a divergence event instead of failing silently.

They solve overlapping problems with different trade-offs, and LangGraph is far more mature. Here is how they differ.

Choose pylinxi if you want your agent to read like ordinary Python, need audit-grade records, or want a small dependency you can read in an afternoon. It works with any LLM client: steps are just functions.

Choose LangGraph if you rely on its integrations, want LangSmith's tracing UI, or prefer a managed runtime.

pip install pylinxi # core, zero dependencies, Python 3.9+ pip install "pylinxi[postgres,otel]" # production extras Enter fullscreen mode Exit fullscreen mode import asyncio from pylinxi import flow, step @step def greet ( name ): return f " hello { name } " @flow async def hello ( ctx, name ): return await greet ( name ) print ( asyncio. run ( hello. run ( " world " )). output ) # hello world Enter fullscreen mode Exit fullscreen mode The repository includes two runnable examples. support_agent.py repairs broken LLM JSON, survives a simulated crash, resumes and waits for a manager's approval. followup_agent.py streams a reply, waits for a customer with a deadline, and lets a worker send the follow-up after a restart.

pylinxi is MIT-licensed, fully typed, and tested on Python 3.9 to 3.13 against a real Postgres database.

If your agents have ever lost an approval, double-charged a customer or restarted a 20-minute run from scratch, I'd love to hear how you're handling it today. Issues, ideas and stars are all welcome.