Step-Result Cache

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

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

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

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

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.

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 four fields:

Field	What it captures
Step identity	String steps use their text; other steps use `.name` / `.__name__`, else `type(step).__name__:id(step)`
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
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.

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.

Workflows

Build multi-step agent pipelines with AgentFlow.

Prompt Cache Optimization

Different layer — provider prompt-prefix caching.

Getting Started

Install

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Telemetry & Analytics

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

Quick Start

When to Cache

How It Works

Correctness Guarantees

`InMemoryStepCache` Options

Seeing Cache Hits

Best Practices

Workflows

Prompt Cache Optimization

​Quick Start

​When to Cache

​How It Works

​Correctness Guarantees

​InMemoryStepCache Options

​Seeing Cache Hits

​Best Practices

​Related

Workflows

Prompt Cache Optimization

Quick Start

When to Cache

How It Works

Correctness Guarantees

`InMemoryStepCache` Options

Seeing Cache Hits

Best Practices

Related