pylinxi 開源 Python 函式庫讓 AI 智能體流程可從中斷處恢復

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?

• Do you start again from step 1, paying for the LLM calls twice?
• Did the refund API already run? Will a retry charge the customer back twice?
• The manager clicked "approve" an hour ago. Is that approval still anywhere?

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.

Why not just draw 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:

• Wiring overhead. Every branch is a conditional edge and every loop is a cycle in the graph. Logic you'd write as an if in five seconds becomes a node, a router function and two edges.
• State plumbing. State lives in a schema with reducers, and checkpoints snapshot it as the graph advances, which gets heavy as conversations and tool outputs grow.
• Healing is mostly retry. Retry policies exist, but repairing a bad LLM output, falling back to a cached answer, or tripping a circuit breaker on a dead dependency is still your code.

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

Meet pylinxi: your agent is just a Python function

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

If the process dies halfway through, you resume the run by its id from any process:

result = await support.resume("ticket-1001")

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.

1. Self-healing that does more than retry

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

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)

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.

2. Crashes heal themselves, without doing anything twice

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)

Three details make replay safe rather than merely possible:

• Parallel branches replay correctly. Tasks your flow creates with asyncio.gather or create_task each get their own key scope, so a resume matches every step to its branch, whichever branch finished first last time. ctx.map(fn, items) fans out with bounded concurrency, and a resume re-runs only the items that failed.
• Handled failures stay handled. If your flow catches a failed step and takes a fallback path (say, emailing "payment failed"), a resume replays that same failure instead of retrying the payment and also shipping the order.
• Idempotency keys for the outside world. Inside a step, idempotency_key() returns "<run_id>:<step_key>", identical across retries and resumes. Pass it to your payment or email provider and the side effect happens once even if the step runs twice.

from pylinxi import idempotency_key

@step
def refund(order):
    return stripe.Refund.create(charge=order.charge_id,
                                idempotency_key=idempotency_key())

Runs also hold a lock while they execute, so two workers can't resume the same run at once: the second gets RunLocked.

3. Humans, deadlines and streaming tokens

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

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

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

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

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.

5. Built for audits from day one

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

• Tamper-evident. Entries form a hash chain signed with HMAC-SHA256. Editing, deleting or reordering an entry breaks verification.
• Key rotation. Pass secret={"2026-q3": old_key, "2026-q4": new_key}. New entries use the newest key and old ones keep verifying. verify(run_id, require_signed=True) rejects anyone rewriting a run as an unsigned chain.
• Truncation detection. Store journal.head(run_id) somewhere else and check with verify(run_id, anchor=...) to catch the newest entries being dropped.
• Redaction. State keys that look like secrets (password, api_key, access_token, apiKey and so on) are stored as [REDACTED], including inside tuples and dataclasses. max_tokens is left alone.
• Secrets never enter the journal by accident. Flow inputs are saved so runs can resume, so pylinxi refuses inputs, signals or state that contain secret-looking keys. Load secrets inside a @step(sensitive=True) instead: its result is never saved and is recomputed on resume.
• Retention. journal.delete(run_id) permanently removes a run, for retention policies or erasure requests.

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

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.

Going to production

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

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

If flow code changes in a way that doesn't match the journal, pylinxi emits a divergence event instead of failing silently.

pylinxi or LangGraph? An honest comparison

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.

Try it in two minutes

pip install pylinxi                       # core, zero dependencies, Python 3.9+
pip install "pylinxi[postgres,otel]"      # production extras

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

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.

• GitHub: karawash-projects/pylinxi
• PyPI: pylinxi

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.