crewai-cache-hook

A Redis-based cache decorator for crewai tasks.

Project Status

• Version: 0.1.1
• Status: Beta
• Maintained: Yes

Features

• Add cache capability to tasks via a decorator
• Checks Redis cache before task execution, returns cached result if hit
• Automatically writes task result to Redis cache after execution
• Supports custom cache key, expiration time, and Redis connection parameters
• Robust error handling and fallback mechanisms
• Multiple serialization options (pickle, JSON)

Prerequisites

Redis Dependency

This library requires Redis as a backend for caching. Make sure you have Redis installed and running.

Installing Redis

• MacOS: brew install redis
• Ubuntu/Debian: sudo apt-get install redis-server
• Windows: Download from Redis Official Website

Starting Redis Server

• MacOS: brew services start redis
• Ubuntu/Debian: sudo systemctl start redis
• Manual: redis-server

Installation

# Install via pip
pip install crewai-cache-hook

# Ensure Redis is installed and running

Configuration

Basic Usage

from cache_hook import cache_hook

@task
@cache_hook(expire=600)  # Cache for 10 minutes
def my_task(self):
    ...

Advanced Configuration

from cache_hook import CacheHook
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)

# Custom Redis configuration
custom_cache = CacheHook(
    host='redis.example.com',     # Redis host
    port=6379,                    # Redis port
    db=1,                         # Redis database
    password='secret',            # Optional password
    connect_timeout=5,            # Connection timeout
    max_retries=2,                # Connection retry attempts
    serializer='json'             # Serialization method
)
custom_cache_hook = custom_cache.cache_hook

@task
@custom_cache_hook(expire=600)
def my_task(self):
    ...

Advanced Features

Custom Cache Key Generator

from cache_hook import cache_hook

def custom_key_generator(func, args, kwargs):
    """
    Create a custom cache key based on specific logic
    
    Args:
        func: The function being cached
        args: Positional arguments
        kwargs: Keyword arguments
    
    Returns:
        str: Custom cache key
    """
    # Example: Include specific arguments in cache key
    key_parts = [
        func.__name__,
        str(kwargs.get('destination', '')),
        str(kwargs.get('start_date', ''))
    ]
    return ':'.join(key_parts)

@task
@cache_hook(expire=600, cache_key_func=custom_key_generator)
def travel_analysis_task(self, destination, start_date):
    ...

Force Refresh

@task
@cache_hook(expire=600, force_refresh=True)
def always_fresh_task(self):
    # This task will always execute and update cache
    ...

Serialization Options

# Pickle serialization (default)
cache_hook(serializer='pickle')

# JSON serialization (for more compatibility)
cache_hook(serializer='json')

Parameters

• expire: Cache expiration time (seconds)
• cache_key_func: Custom function to generate cache key (optional)
• force_refresh: Force task execution and cache update
• host/port/db/password: Redis connection parameters
• serializer: Serialization method ('pickle' or 'json')
• connect_timeout: Connection timeout in seconds
• max_retries: Number of connection retry attempts

Notes

• Uses configurable serialization (pickle or JSON)
• Ensure task results are serializable
• Suitable for crewAI tasks
• Caches the entire Task result
• Requires an active Redis server

Troubleshooting

• Connection Errors: Check Redis server status, network, and configuration
• Serialization Errors: Ensure task results are serializable
• Performance Issues: Monitor cache size and hit/miss rates