uPjm&ndZddlZddlmZddlmZddlmZddlm Z ddl m Z ddl m Z Gd d eZy) zb Base class across routing strategies to abstract commmon functions like batch incrementing redis N)ABC)Final)verbose_router_logger) DualCache)RedisPipelineIncrementOperation)DEFAULT_REDIS_SYNC_INTERVALceZdZdedededzfdZdedzfdZdZde e e e fd e d e efd Z d e d ed e fdZdedzfdZdZd e fdZd e dzfdZd ee fdZd ee fdZdZdZy)BaseRoutingStrategy dual_cacheshould_batch_redis_writesdefault_sync_intervalNcr||_g|_d|_|r|j|t |_yN)r redis_increment_operation_queue _sync_tasksetup_sync_tasksetin_memory_keys_to_update)selfr r r s r/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/router_strategy/base_routing_strategy.py__init__zBaseRoutingStrategy.__init__s6 %VX,59 $  !6 725%%c tj}|j |j ||_y#t$r,tj}tj|Y[wxYw)z;Setup the sync task in a way that's compatible with FastAPI)r N)asyncioget_running_loop RuntimeErrornew_event_loopset_event_loop create_task(periodic_sync_in_memory_spend_with_redisr)rr loops rrz#BaseRoutingStrategy.setup_sync_tasksh )++-D **  9 9Pe 9 f   )))+D  " "4 ( )s=2A21A2cK|j0|jj |jd{yy7#tj$rYywxYww)z.Cleanup method to be called when shutting downN)rcancelrCancelledErrorrs rcleanupzBaseRoutingStrategy.cleanup*sS ?? & OO " " $ oo%% '&))  s7'AAAAAAAAAAincrement_listttlreturncKg}|D]2\}}|j|||d{}|j|4|S7w)zB Increment a list of values in the current window keyvaluer(N)"_increment_value_in_current_windowappend)rr'r(resultsr,r-results r'_increment_value_list_in_current_windowz;BaseRoutingStrategy._increment_value_list_in_current_window3sU ( #JCBBsRW]`BaaF NN6 " #bs"A>Ar,r-cK|jjj|||d{}t|||}|jj ||j ||S7Aw)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,)r in_memory_cacheasync_incrementrrr/add_to_in_memory_keys_to_update)rr,r-r(r1 increment_ops rr.z6BaseRoutingStrategy._increment_value_in_current_window?s|#oo==MMN   >! ,,33LA ,,,5  s,A2A0AA2c*K|xst} |jd{tj|d{77"7#t$r>}t j d|tj|d{7Yd}~Gd}~wwxYww)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 NzError in periodic sync task: %s)r _sync_in_memory_spend_with_redisrsleep Exceptionrerror)rr es rr zt |j}|jjj|d{}t |jDcgc] \}}||vs |c}}|_|$t||Dcic] \}}|d| }}}|Si}|Sy7acc}}wcc}}w#t$r'} tjd| g|_Yd} ~ yd} ~ wwxYww)av How this works: - async_log_success_event collects all provider spend increments in `redis_increment_operation_queue` - This function compresses multiple increments for the same key into a single operation - Then pushes all increments to Redis in a batched pipeline to optimize performance Only runs if Redis is initialized Nrr,r4)r',Error syncing in-memory cache with Redis: %s) r redis_cachelenr enumerater/listvaluesasync_increment_pipelinezipr=rr>) rcompressed_ops ops_to_removeidxopcompressed_queueincrement_resultr, return_resultr?s r#_push_in_memory_increments_to_redisz7BaseRoutingStrategy._push_in_memory_increments_to_redisjs$ 6??..47781K |jjy|j}t|}i}|jjj |d{}t ||D]\}}t|xsd||<|jd{}|y|D]}t|j|dxsd} t|j|dxsd} t|jjj|d{xsd} | | z } | | kr| | z} n|jjj|| d{y777T7#t$r }tjd|Yd}~yd}~wwxYww)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)rYrr5)r,r-rA)r rBrWrEr6async_batch_get_cacherHfloatrPgetasync_get_cacheasync_set_cacher=r exception)r cache_keyscache_keys_listin_memory_before_dictin_memory_beforekv redis_valuesr, redis_valbeforeafterdeltamergedr?s rr;z4BaseRoutingStrategy._sync_in_memory_spend_with_rediss+ _**2113 &**%5O,. !,0OO,K,K,a,agv,a,w&w O-=> 91+0a=%a( 9)-(P(P(R"RL#' ]!,"2"23":"?a@ 488a@EAFDOO$C$C$S$SX[$S$\\a`abI%&.Foo55EE#U[E\\\! ]'x #S]] _ ! + +,Z\] ^ ^ _sFE1FAE1!E("=E1E+ E1&F'A1E1E- AE1 E/!E1'F(E1+E1-E1/E11 F:FFFF)__name__ __module__ __qualname__rboolr_rrr&rEtuplestrintr2r.r rPr8rUrrWrZr\r;rTrrr r s 8 8$( 8 %t| 8  UT\   "5c?3 :=  e CTW0TY\`T`&-6^/3/t -c#h-C .8_rr )__doc__rabcrtypingrlitellm._loggingrlitellm.caching.cachingrlitellm.caching.redis_cacherlitellm.constantsrr rTrrr~s12-G9W_#W_r