# How Langflow's Caching Layer Works: Architecture and Backend Configuration

> Explore Langflow's caching layer architecture. Discover how it abstracts data storage and configure backends like Redis, memory, or disk for efficient temporary data management.

- Repository: [Langflow/langflow](https://github.com/langflow-ai/langflow)
- Tags: internals
- Published: 2026-02-24

---

**Langflow implements a pluggable caching layer that abstracts temporary data storage through uniform synchronous and asynchronous interfaces, supporting four configurable backends including in-memory LRU, Redis, async memory, and disk-based caches.**

Langflow is an open-source visual framework for building LangChain workflows. Its **caching layer** provides a flexible infrastructure for storing component results, session data, and intermediate artifacts. The architecture separates interface definitions from concrete implementations, allowing developers to swap storage backends via environment variables without modifying application logic.

## Core Abstraction Layer

The caching system defines two abstract base classes in [`src/backend/base/langflow/services/cache/base.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/services/cache/base.py) that establish the contract for all implementations.

**`CacheService`** provides the synchronous interface. It defines primitive operations including `get`, `set`, `upsert`, `delete`, `clear`, and `contains`, plus support for the mapping protocol. This class inherits from Langflow's generic `Service` infrastructure, enabling management by the service container.

**`AsyncBaseCacheService`** mirrors the same contract for asynchronous operations. All methods are awaitable, ensuring non-blocking cache access in async contexts. Both base classes return a sentinel object named `CACHE_MISS` when a key is absent or expired, making cache-miss handling explicit and consistent across backends.

## Configurable Cache Backends

Langflow provides four concrete implementations that satisfy the abstract interfaces. Each backend is optimized for different deployment scenarios, from single-process development to distributed production environments.

### ThreadingInMemoryCache (Synchronous)

**`ThreadingInMemoryCache`** offers a thread-safe, in-memory LRU cache for synchronous code paths. It uses an `OrderedDict` for LRU eviction and a `threading.RLock` for thread safety. You can configure `max_size` to limit entries and `expiration_time` (in seconds) for automatic TTL.

```python
from langflow.services.cache.service import ThreadingInMemoryCache

cache = ThreadingInMemoryCache(max_size=100, expiration_time=300)
cache.set("component:result", {"data": "processed"})
result = cache.get("component:result")

```

*Implementation reference:* [`src/backend/base/langflow/services/cache/service.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/services/cache/service.py) (lines 22-73)

### RedisCache (Asynchronous)

**`RedisCache`** provides distributed caching via `redis.asyncio.StrictRedis`. Values are serialized using *dill* and stored with Redis `setex` for automatic expiry. The class exposes `is_connected` and `contains` methods alongside standard CRUD operations.

```python
import asyncio
from langflow.services.cache.service import RedisCache

async def main():
    cache = RedisCache(host="localhost", port=6379, expiration_time=600)
    await cache.set("session:abc", {"user_id": 42})
    data = await cache.get("session:abc")

asyncio.run(main())

```

*Implementation reference:* [`src/backend/base/langflow/services/cache/service.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/services/cache/service.py) (lines 78-92)

### AsyncInMemoryCache (Asynchronous)

**`AsyncInMemoryCache`** implements the same LRU semantics as the synchronous version but uses `asyncio.Lock` instead of threading primitives. This backend operates fully within the async event loop, eliminating thread-safety overhead in async contexts.

```python
import asyncio
from langflow.services.cache.service import AsyncInMemoryCache

async def main():
    cache = AsyncInMemoryCache(max_size=50, expiration_time=60)
    await cache.set("temp:data", [1, 2, 3])
    return await cache.get("temp:data")

asyncio.run(main())

```

*Implementation reference:* [`src/backend/base/langflow/services/cache/service.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/services/cache/service.py) (lines 94-118)

### AsyncDiskCache (Asynchronous)

**`AsyncDiskCache`** persists entries to the filesystem using the **diskcache** library. It supports max size limits, expiration policies, and async locking mechanisms. This backend is ideal for single-node deployments requiring cache persistence across process restarts.

```python
import asyncio
from pathlib import Path
from langflow.services.cache.disk import AsyncDiskCache

async def main():
    cache = AsyncDiskCache(cache_dir=Path("/tmp/langflow_cache"), expiration_time=3600)
    await cache.set("model:metadata", {"version": "1.0"})
    return await cache.get("model:metadata")

asyncio.run(main())

```

*Implementation reference:* [`src/backend/base/langflow/services/cache/disk.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/services/cache/disk.py)

## Backend Selection via Factory

The **`CacheServiceFactory`** in [`src/backend/base/langflow/services/cache/factory.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/services/cache/factory.py) instantiates the appropriate backend based on the `cache_type` setting. The factory reads `settings_service.settings.cache_type`, which accepts a `Literal["async", "redis", "memory", "disk"]` with a default value of `"async"`.

The selection logic follows this priority:

1. **"redis"** → Returns `RedisCache`
2. **"memory"** → Returns `ThreadingInMemoryCache`
3. **"async"** → Returns `AsyncInMemoryCache` (default)
4. **"disk"** → Returns `AsyncDiskCache`

Configuration occurs through the `Settings` class in [`src/lfx/src/lfx/services/settings/base.py`](https://github.com/langflow-ai/langflow/blob/main/src/lfx/src/lfx/services/settings/base.py). You can override the backend at runtime using the environment variable `LANGFLOW_CACHE_TYPE`:

```bash
export LANGFLOW_CACHE_TYPE=redis

```

The test suite demonstrates this pattern in [`src/backend/tests/conftest.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/tests/conftest.py), where `LANGFLOW_CACHE_TYPE="redis"` configures the test environment for distributed cache validation.

## Consuming the Cache in Application Code

Services retrieve the configured cache through the dependency-injection helper `get_cache_service` defined in [`src/backend/base/langflow/services/deps.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/services/deps.py):

```python
from langflow.services.deps import get_cache_service

def process_component():
    cache = get_cache_service()
    cached_result = cache.get("computation:heavy")
    
    if cached_result is cache.CACHE_MISS:
        result = expensive_operation()
        cache.set("computation:heavy", result)
        return result
    return cached_result

```

Because the API remains identical across sync and async implementations, calling code remains agnostic to the underlying storage mechanism. The service container handles lifecycle management and backend instantiation automatically.

## Summary

- **Langflow's caching layer** abstracts storage through `CacheService` (sync) and `AsyncBaseCacheService` (async) interfaces defined in [`base.py`](https://github.com/langflow-ai/langflow/blob/main/base.py).
- **Four configurable backends** are available: `ThreadingInMemoryCache` for thread-safe sync caching, `RedisCache` for distributed async caching, `AsyncInMemoryCache` for lightweight async caching, and `AsyncDiskCache` for persistent filesystem storage.
- **Backend selection** occurs via `CacheServiceFactory`, which reads the `cache_type` setting (configurable through the `LANGFLOW_CACHE_TYPE` environment variable).
- **Uniform API** ensures that consumers use identical `get`, `set`, and `delete` methods regardless of the concrete implementation, with `CACHE_MISS` handling explicit absence detection.

## Frequently Asked Questions

### What cache backends does Langflow support?

Langflow supports four distinct backends: **ThreadingInMemoryCache** (synchronous, thread-safe LRU), **RedisCache** (asynchronous, distributed), **AsyncInMemoryCache** (asynchronous, single-process LRU), and **AsyncDiskCache** (asynchronous, filesystem-persistent). Each implements the abstract `CacheService` or `AsyncBaseCacheService` interface defined in [`src/backend/base/langflow/services/cache/base.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/services/cache/base.py).

### How do I configure Langflow to use Redis for caching?

Set the environment variable `LANGFLOW_CACHE_TYPE=redis` before starting the application. The `CacheServiceFactory` in [`src/backend/base/langflow/services/cache/factory.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/services/cache/factory.py) detects this value and instantiates `RedisCache` with connection parameters from the `Settings` class. Ensure your Redis server is accessible via the host and port configured in your environment.

### What is the difference between AsyncInMemoryCache and ThreadingInMemoryCache?

**`ThreadingInMemoryCache`** uses `threading.RLock` for synchronization and is designed for synchronous code paths, while **`AsyncInMemoryCache`** uses `asyncio.Lock` and operates exclusively in async contexts. Both implement LRU eviction with configurable `max_size` and `expiration_time`, but the async version eliminates thread-blocking overhead in coroutine-based workloads.

### How does Langflow handle cache misses?

All cache implementations return a sentinel object named `CACHE_MISS` when a key is absent or expired. This explicit approach, implemented in [`src/backend/base/langflow/services/cache/base.py`](https://github.com/langflow-ai/langflow/blob/main/src/backend/base/langflow/services/cache/base.py), allows calling code to distinguish between `None` values and missing keys without raising exceptions. Consumers check `if result is CACHE_MISS` before proceeding with cache population logic.