> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# 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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
subgraph "Step-Result Cache"
Req[đ Step input] --> Key[đ Build key]
Key --> Hit{đ Cached?}
Hit -->|Hit| Return[â
Cached output]
Hit -->|Miss| LLM[đ¤ Run step]
LLM --> Store[đž Store output]
Store --> Return
end
classDef input fill:#8B0000,stroke:#7C90A0,color:#fff
classDef cache fill:#189AB4,stroke:#7C90A0,color:#fff
classDef decision fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef result fill:#10B981,stroke:#7C90A0,color:#fff
class Req input
class Key,Store cache
class Hit decision
class LLM,Return result
```
This caches **entire step outputs inside the workflow**, keyed by each step's actual inputs. It is different from [Prompt Cache Optimization](/docs/features/prompt-cache-optimization), which caches **prompt prefixes at the LLM provider**.
## Quick Start
Pass `cache=True` for a process-scoped LRU cache (default `max_entries=128`).
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```
Pass an `InMemoryStepCache` instance to set capacity or share one cache across flows.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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")
```
Any object with `get()` and `set()` works â back the cache with Redis, disk, or anything else.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
Start[𧊠A step] --> Q{đ Same input, same output?}
Q -->|Yes| Cache[â
Cache it]
Q -->|No| Skip[đĢ Do not cache]
Skip --> Why[Side effects, tools, clock, randomness]
classDef start fill:#8B0000,stroke:#7C90A0,color:#fff
classDef decision fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef good fill:#10B981,stroke:#7C90A0,color:#fff
classDef bad fill:#6366F1,stroke:#7C90A0,color:#fff
class Start start
class Q decision
class Cache good
class Skip,Why bad
```
***
## 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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant Flow as AgentFlow
participant Cache as Step Cache
participant Step as Step
Flow->>Cache: get(key)
alt Miss
Cache-->>Flow: None
Flow->>Step: run
Step-->>Flow: output
Flow->>Cache: set(key, output)
else Hit
Cache-->>Flow: cached output
end
```
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
Nothing caches unless `cache=` is set. Agent steps are not always deterministic, so silent caching would be wrong.
Only successful steps are stored. Caching a failure would serve it again on every re-run, turning a transient error into a permanent one.
The linear executor and the pattern executor (used by `Parallel`, `If`, etc.) are both wrapped, so caching works regardless of where a step sits.
`InMemoryStepCache` caps at `max_entries` (default 128). An unbounded cache would be a memory leak on long-running flows.
`InMemoryStepCache` uses an internal lock, so it is safe to share across `Parallel` branches.
Values are deep-copied on `set` and on `get`, so a caller mutating a returned value cannot corrupt future hits.
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 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.
`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.
If the cold run stopped before writing `_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 |
| `_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 |
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
Formatters, lookups, and summarisers over unchanged text are safe. Their output depends only on their input.
Do not cache steps that call tools with side effects or read a clock â a cached answer would skip the effect the caller expected.
Create a single `InMemoryStepCache` and pass it to multiple `AgentFlow`s to reuse results across pipelines.
Change a workflow variable when you *want* a fresh call â different variables produce a different key.
Set `max_entries` to the number of results you actually reuse; a larger cache only wastes memory.
***
## Related
Build multi-step agent pipelines with AgentFlow.
Different layer â provider prompt-prefix caching.
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.