o ưi(@s|dZddlZddlmZddlmZmZmZmZm Z m Z ddl m Z ddl mZddlmZddlmZGd d d eZdS) zb Base class across routing strategies to abstract commmon functions like batch incrementing redis N)ABC)DictListOptionalSetTupleUnion)verbose_router_logger) DualCache)RedisPipelineIncrementOperation)DEFAULT_REDIS_SYNC_INTERVALc @s eZdZdededeeeeffddZ deeeeffddZ dd Z d e e eefd ed e efd dZdedeeefd efddZdeeeeffddZddZdefddZd eefddZd eefddZd eefddZdd Zd!d"Zd#S)$BaseRoutingStrategy dual_cacheshould_batch_redis_writesdefault_sync_intervalcCs,||_g|_d|_|r||t|_dSN)rredis_increment_operation_queue _sync_tasksetup_sync_tasksetin_memory_keys_to_update)selfrrrrd/home/app/Keep/.python/lib/python3.10/site-packages/litellm/router_strategy/base_routing_strategy.py__init__s zBaseRoutingStrategy.__init__cCsJzt}Wntyt}t|Ynw||j|d|_dS)z;Setup the sync task in a way that's compatible with FastAPI)rN)asyncioget_running_loop RuntimeErrornew_event_loopset_event_loop create_task(periodic_sync_in_memory_spend_with_redisr)rrlooprrrr s   z#BaseRoutingStrategy.setup_sync_taskcsD|jdur |jz |jIdHWdStjyYdSwdS)z.Cleanup method to be called when shutting downN)rcancelrCancelledErrorrrrrcleanup.s  zBaseRoutingStrategy.cleanupincrement_listttlreturncs8g}|D]\}}|j|||dIdH}||q|S)zB Increment a list of values in the current window keyvaluer(N)"_increment_value_in_current_windowappend)rr'r(resultsr+r,resultrrr'_increment_value_list_in_current_window7s  z;BaseRoutingStrategy._increment_value_list_in_current_windowr+r,csF|jjj|||dIdH}t|||d}|j||j|d|S)a Increment spend within existing budget window Runs once the budget start time exists in Redis Cache (on the 2nd and subsequent requests to the same provider) - Increments the spend in memory cache (so spend instantly updated in memory) - Queues the increment operation to Redis Pipeline (using batched pipeline to optimize performance. Using Redis for multi instance environment of LiteLLM) r*N)r+increment_valuer(r+)rin_memory_cacheZasync_incrementr rr.add_to_in_memory_keys_to_update)rr+r,r(r0Z increment_oprrrr-Es    z6BaseRoutingStrategy._increment_value_in_current_windowc sz|pt} z|IdHt|IdHWn$ty;}ztdt|t|IdHWYd}~nd}~wwq)z Handler that triggers sync_in_memory_spend_with_redis every DEFAULT_REDIS_SYNC_INTERVAL seconds Required for multi-instance environment usage of provider budgets TNzError in periodic sync task: )r _sync_in_memory_spend_with_redisrsleep Exceptionr errorstr)rrerrrr!_s zs zKBaseRoutingStrategy._push_in_memory_increments_to_redis..cSsi|] \}}|d|qSr3r)r<r+r>rrr szKBaseRoutingStrategy._push_in_memory_increments_to_redis..*Error syncing in-memory cache with Redis: )r redis_cachelenr enumerater.listvaluesZasync_increment_pipelinezipr8r r9r:)rZcompressed_opsr=r>Zcompressed_queueZincrement_resultZ return_resultr;rr?r#_push_in_memory_increments_to_redistsJ       ' z7BaseRoutingStrategy._push_in_memory_increments_to_rediscCs|j|dSr)radd)rr+rrrr5sz3BaseRoutingStrategy.add_to_in_memory_keys_to_updatecCsdS)z- Get the key pattern to sync Nrr%rrrget_key_pattern_to_syncsz+BaseRoutingStrategy.get_key_pattern_to_synccCs|jSr)rr%rrrget_in_memory_keys_to_updatesz0BaseRoutingStrategy.get_in_memory_keys_to_updatecCs|j}t|_|S)z-Atomic get and reset in-memory keys to update)rr)rkeysrrr&get_and_reset_in_memory_keys_to_updatesz:BaseRoutingStrategy.get_and_reset_in_memory_keys_to_updatecCs t|_dSr)rrr%rrrreset_in_memory_keys_to_updates z2BaseRoutingStrategy.reset_in_memory_keys_to_updatec sDz|jjdur WdS|}t|}i}|jjj|dIdH}t||D] \}}t|p-d||<q%|IdH}|dur@WdS|D]?}t| |dpLd} t| |dpVd} t|jjj |dIdHped} | | } | | krt| | } nqB|jjj || dIdHqBWdSt y}zt dt|WYd}~dSd}~ww)a  Ensures in-memory cache is updated with latest Redis values for all provider spends. Why Do we need this? - Optimization to hit sub 100ms latency. Performance was impacted when redis was used for read/write per request - Use provider budgets in multi-instance environment, we use Redis to sync spend across all instances What this does: 1. Push all provider spend increments to Redis 2. Fetch all current provider spend from Redis to update in-memory cache N)rMrr3)r+r,rB)rrCrLrFr4Zasync_batch_get_cacherHfloatrIgetZasync_get_cacheZasync_set_cacher8r exceptionr:)rZ cache_keysZcache_keys_listZin_memory_before_dictZin_memory_beforekvZ redis_valuesr+Z redis_valbeforeafterdeltamergedr;rrrr6sJ    z4BaseRoutingStrategy._sync_in_memory_spend_with_redisN)__name__ __module__ __qualname__r boolrrintrPrrr&rrr:r1r-r!rIr5rKrrLrNrOr6rrrrr sD      : r )__doc__rabcrtypingrrrrrrZlitellm._loggingr Zlitellm.caching.cachingr Zlitellm.caching.redis_cacher Zlitellm.constantsr r rrrrs