Step-Result Cache

Cache a workflow step’s output by its inputs — pay the LLM once for a deterministic step

A workflow step-result cache stores a step’s output keyed by its inputs, so a deterministic step reuses its answer instead of paying the full LLM cost on every run.

from praisonaiagents import Agent, AgentFlow

formatter = Agent(name="Formatter", instructions="Format the text as clean JSON.")
summariser = Agent(name="Summariser", instructions="Summarise in one sentence.")

workflow = AgentFlow(steps=[formatter, summariser], cache=True)
workflow.start("Same input each time")
workflow.start("Same input each time")   # second call served from cache

from praisonaiagents import Agent, AgentFlow
from praisonaiagents.workflows.step_cache import FileStepCache

formatter = Agent(name="Formatter", instructions="Format the text as clean JSON.")
summariser = Agent(name="Summariser", instructions="Summarise in one sentence.")

workflow = AgentFlow(steps=[formatter, summariser], cache=FileStepCache("run.json"))
workflow.start("Same input each time")   # interrupted mid-flow
workflow.start("Same input each time")   # new process: continues from where it stopped

This caches entire step outputs inside the workflow, keyed by each step’s actual inputs. It is different from Prompt Cache Optimization, which caches prompt prefixes at the LLM provider.

​

Quick Start

1. Enable with True

   Pass cache=True for a process-scoped LRU cache (default max_entries=128).

   from praisonaiagents import Agent, AgentFlow
   
   formatter = Agent(name="Formatter", instructions="Format the text as clean JSON.")
   
   workflow = AgentFlow(steps=[formatter], cache=True)
   workflow.start("Same input each time")
   workflow.start("Same input each time")   # served from cache
   
2. Tune the cache size

   Pass an InMemoryStepCache instance to set capacity or share one cache across flows.

   from praisonaiagents import Agent, AgentFlow
   from praisonaiagents.workflows.step_cache import InMemoryStepCache
   
   formatter = Agent(name="Formatter", instructions="Format the text as clean JSON.")
   
   cache = InMemoryStepCache(max_entries=256)
   workflow = AgentFlow(steps=[formatter], cache=cache)
   workflow.start("Same input each time")
   
3. Bring your own cache

   Any object with get() and set() works — back the cache with Redis, disk, or anything else.

   from praisonaiagents import Agent, AgentFlow
   
   class RedisStepCache:
       def __init__(self, client):
           self.client = client
   
       def get(self, key):
           raw = self.client.get(key)
           return None if raw is None else __import__("json").loads(raw)
   
       def set(self, key, value):
           self.client.set(key, __import__("json").dumps(value))
   
   formatter = Agent(name="Formatter", instructions="Format the text as clean JSON.")
   
   workflow = AgentFlow(steps=[formatter], cache=RedisStepCache(client))
   workflow.start("Same input each time")
   

InMemoryStepCache is not re-exported at the top level. Import it from praisonaiagents.workflows.step_cache.

---

​

Durable Resume (FileStepCache)

FileStepCache is a drop-in durable replacement for InMemoryStepCache — it keeps cached step results in a JSON file so an interrupted flow resumes from where it stopped in a new process.

from praisonaiagents import Agent, AgentFlow
from praisonaiagents.workflows.step_cache import FileStepCache

researcher = Agent(name="Researcher", instructions="Research the topic in depth.")
writer     = Agent(name="Writer",     instructions="Write a one-page briefing from the research.")

flow = AgentFlow(steps=[researcher, writer], cache=FileStepCache("briefing.json"))
flow.start("quantum error correction")
# Ctrl-C after Researcher finishes, before Writer finishes
# ...in a new process, same path...
flow.start("quantum error correction")   # Researcher served from briefing.json; Writer runs

FileStepCache is not re-exported at the top level. Import it from praisonaiagents.workflows.step_cache.

1. Start a long flow

   Pass cache=FileStepCache("run.json"). Each successful step’s result is written atomically to run.json.

2. Get interrupted

   A deploy, crash, Ctrl-C, OOM or a closed laptop stops the run part-way.

3. Re-run the same command

   Completed steps are served from the file; the flow continues at the first unfinished step.

4. Edit a step and re-run

   A step whose code or instructions changed — and every step after it — runs again automatically. No stale results are served.

5. Changed a tool? Clear first

   Tool implementations are not fingerprinted, so call FileStepCache("run.json").clear() before re-running.

​

When to pick which

Pick the lightest cache that covers where your results need to live.

​

What invalidates a cached step

Resume is content-addressed, not index-based, so a step re-runs whenever anything it depends on changes.

Change	Step re-runs?
The input changes	✅ Yes
An upstream step’s output changes	✅ Yes (downstream keys see a new previous output)
A handler’s code, or an immutable default / closure / __kwdefaults__ value, changes — even under the same name	✅ Yes
A module-level helper function the handler calls is edited	✅ Yes
An agent’s role, goal, backstory, instructions, llm, action, description, expected_output or tool names change	✅ Yes
A workflow variable (or its type) changes	✅ Yes

Not detected — call clear() after changing	Why
Mutable captured state (a list the handler appends to, a client in a closure)	Left out so the key does not drift during a run
Methods the handler calls, or imported libraries’ behaviour	Only module-level Python functions are fingerprinted
A tool’s underlying implementation	Only the tool’s name is in the key

• An anonymous step with no stable name never hits a durable cache in a new process — it re-runs.
• Mutable captured state, imported libraries and tool implementations are not fingerprinted — after changing one, call clear() (or delete the file).
• Values that don’t JSON round-trip unchanged (tuples, int-keyed dicts, arbitrary objects) are rejected at set() time, not stored lossily. The workflow logs a warning and continues.
• Pointing FileStepCache at an existing file that isn’t a step-cache JSON object raises ValueError at construction — use a dedicated path.

​

Crash-safety and concurrent writers

Every set() is a locked, atomic read-modify-write of the file, so a crash mid-write keeps the last good file. Two caches on the same path — in one process or two — both add to the file; neither erases the other’s entries. Only successful steps are cached, as with the in-memory cache.

​

FileStepCache Options

Option / method	Type	Default	Description
path	str	required	File path for the JSON store. ~ is expanded; an absolute path is used internally. The parent directory is created on first write.
get(key)	method	—	Deep copy of the stored entry, or None on miss.
set(key, value)	method	—	Atomic locked write. Raises TypeError if the value doesn’t JSON round-trip unchanged.
clear()	method	—	Removes all entries and deletes the file.
__len__()	method	—	Number of cached entries.

---

​

When to Cache

Cache deterministic steps; skip steps that call tools with side effects, read a clock, or depend on randomness.

---

​

How It Works

On a miss the step runs and its output is stored; on a hit the cached output is returned without an LLM call.

The key is a stable SHA-256 (first 32 hex chars) built from five fields:

Field	What it captures
Step identity	String steps use their text; other steps use .name / .__name__, else type(step).__name__:id(step)
Step definition	A handler’s code and immutable configuration, or an agent’s role, instructions, model and tool names — so an edited step misses the cache
Previous output	The output of the prior step
Input text	The current step’s rendered input
Variables	The workflow’s variables at that point, tagged with each value’s type

Variable types are part of the key, so True and "True" (or 1 and "1") never collapse to the same entry. Two runs with the same prompt but different variables are different calls.

---

​

Correctness Guarantees

Opt-in only

Nothing caches unless cache= is set. Agent steps are not always deterministic, so silent caching would be wrong.

Failures are never cached

Only successful steps are stored. Caching a failure would serve it again on every re-run, turning a transient error into a permanent one.

Both executors covered

The linear executor and the pattern executor (used by Parallel, If, etc.) are both wrapped, so caching works regardless of where a step sits.

Bounded LRU

InMemoryStepCache caps at max_entries (default 128). An unbounded cache would be a memory leak on long-running flows.

Thread-safe

InMemoryStepCache uses an internal lock, so it is safe to share across Parallel branches.

Deep-copied snapshots

Values are deep-copied on set and on get, so a caller mutating a returned value cannot corrupt future hits.

A hit replays the full step delta

On a cache hit the full step delta is replayed — output, the step’s output_variable (or {step_name}_output), any handler-supplied variables, the step’s status and retries, and the StepResult.stop_workflow signal. Downstream {{step_name_output}} substitutions, ctx.variables[...] reads, and result["steps"][...] metadata all resolve the same way on cold and cached runs.

A cached stop still stops

A handler that returned StepResult(stop_workflow=True) on the cold run also stops the workflow on the cached run — downstream steps never execute on the second run. With verbose=True the cache hit prints 🛑 Workflow stopped at: {step.name}, matching the cold-run message.

Handler variables survive a hit

StepResult.variables returned by the handler are cached and re-applied on a hit, so a downstream consumer reading ctx.variables["answer"] sees the same value on cold and cached runs. If guardrails rejected one attempt and accepted the next, the union of their variables is cached — matching what the live run accumulated, not just the last attempt’s keys.

No spurious output variable on early stop

If the cold run stopped before writing <step>_output (because stop_workflow=True), the cached hit does not insert it either — variables keeps the exact shape the cold run produced.

A hit replays the whole payload the cold run produced, not just the output.

Replayed on a hit	Where it lands
output	result["output"], next step’s context
<step>_output (or output_variable)	all_variables — only when the cold run did not stop first
Handler-supplied variables	all_variables — union across guardrail-retried attempts
StepResult.stop_workflow	Stops the cached run at this step
Step status, retries	result["steps"][...] metadata

from praisonaiagents import AgentFlow
from praisonaiagents.workflows.workflows import StepResult

def finish(ctx):
    return StepResult(output="done", stop_workflow=True)

def downstream(ctx):
    return "unexpected"   # never reached, cold or cached

flow = AgentFlow(steps=[finish, downstream], cache=True)
flow.run("same")   # cold: stops at finish
flow.run("same")   # cached: also stops at finish

from praisonaiagents import AgentFlow
from praisonaiagents.workflows.workflows import StepResult

def producer(ctx):
    return StepResult(output="p", variables={"answer": 42})

def consumer(ctx):
    return f"answer={ctx.variables.get('answer')}"

flow = AgentFlow(steps=[producer, consumer], cache=True)
first  = flow.run("same")   # "answer=42"
second = flow.run("same")   # "answer=42" (cached)
assert second["output"] == first["output"]

These examples do not call an LLM — they exercise the handler-only path to make the guarantee concrete.

---

​

InMemoryStepCache Options

Option	Type	Default	Description
max_entries	int	128	LRU capacity. Must be > 0 (raises ValueError otherwise)

Public attributes: hits, misses. Public methods: get, set, clear, __len__.

Accepted cache= values:

Value	Meaning
None / False	No caching
True	Process-scoped InMemoryStepCache() (default max_entries=128)
An InMemoryStepCache instance	Use the given cache — share across flows, tune size
A FileStepCache instance	Durable JSON-backed cache — resume across processes
Any object with get() and set()	Custom cache (Redis, disk, etc.)
Anything else	TypeError at construction

---

​

Seeing Cache Hits

Turn on verbose to print a line for each hit.

from praisonaiagents import Agent, AgentFlow

formatter = Agent(name="Formatter", instructions="Format the text as clean JSON.")

workflow = AgentFlow(steps=[formatter], cache=True, verbose=True)
workflow.start("Same input each time")
workflow.start("Same input each time")   # prints: ↩︎  cache hit: Formatter

---

​

Best Practices

Cache deterministic steps only

Formatters, lookups, and summarisers over unchanged text are safe. Their output depends only on their input.

Skip steps with side effects

Do not cache steps that call tools with side effects or read a clock — a cached answer would skip the effect the caller expected.

Share one cache across flows

Create a single InMemoryStepCache and pass it to multiple AgentFlows to reuse results across pipelines.

Use variables to force a re-run

Change a workflow variable when you want a fresh call — different variables produce a different key.

Bound the cache to your working set

Set max_entries to the number of results you actually reuse; a larger cache only wastes memory.

Use FileStepCache for long runs and expensive steps

Any flow that costs real money or minutes to re-run deserves durable resume. The overhead is one atomic JSON write per successful step.

Call clear() after editing a tool or a mutable dependency

The fingerprint doesn’t cross into tool implementations or mutable captured state.

Keep each run on its own path for strict isolation

Or deliberately share a path to pool results across runs and processes — both writers add; neither erases.

---

​

Related

Workflows

Build multi-step agent pipelines with AgentFlow.

Prompt Cache Optimization

Different layer — provider prompt-prefix caching.