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

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

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 (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.

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 (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.

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 (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.

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

Backend Selection via Factory

The CacheServiceFactory in 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. You can override the backend at runtime using the environment variable LANGFLOW_CACHE_TYPE:

export LANGFLOW_CACHE_TYPE=redis

The test suite demonstrates this pattern in 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:

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.
• 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.

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 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, 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}