@@ -3104,145 +3104,6 @@ def _check_free_tier_eligibility(
31043104
31053105 return True
31063106
3107- async def _get_resilient_response_stream (
3108- self ,
3109- valves : "Pipe.Valves" ,
3110- __user__ : "UserData" ,
3111- gen_content_args : dict ,
3112- is_streaming : bool ,
3113- __metadata__ : "Metadata" ,
3114- event_emitter : EventEmitter ,
3115- model_config : dict ,
3116- ) -> AsyncIterator [types .GenerateContentResponse ] | None :
3117- """
3118- Generates a stream of responses with automatic fallback logic.
3119-
3120- 1. Identifies if routing (Free -> Paid) is possible.
3121- 2. Tries Free Tier first.
3122- 3. If Free Tier fails on the FIRST chunk with a Quota/Permission/Overload error, switches to Paid.
3123- 4. Updates __metadata__['is_paid_api'] so cost calculation is accurate.
3124- """
3125-
3126- model_id = __metadata__ .get ("canonical_model_id" )
3127-
3128- # --- 1. Determine Routing Strategy ---
3129- has_free_key = bool (valves .GEMINI_FREE_API_KEY )
3130- has_paid_key = bool (valves .GEMINI_PAID_API_KEY )
3131- use_vertex = valves .USE_VERTEX_AI and valves .VERTEX_PROJECT
3132-
3133- execution_order = []
3134- if use_vertex :
3135- execution_order = ["vertex" ]
3136- elif has_free_key :
3137- is_eligible = self ._check_free_tier_eligibility (
3138- model_id , model_config , __metadata__ .get ("features" , {}) # type: ignore
3139- )
3140-
3141- if is_eligible :
3142- execution_order = ["free" ]
3143- if has_paid_key :
3144- execution_order .append ("paid" )
3145- else :
3146- execution_order = ["paid" ] if has_paid_key else ["free" ]
3147- elif has_paid_key :
3148- execution_order = ["paid" ]
3149- else :
3150- execution_order = ["standard" ]
3151-
3152- log .debug (f"Routing strategy for { model_id } : { execution_order } " )
3153-
3154- # --- 2. Execution Loop ---
3155- for attempt_idx , tier in enumerate (execution_order ):
3156- is_last_attempt = attempt_idx == len (execution_order ) - 1
3157- current_valves = copy .copy (valves )
3158-
3159- if tier == "free" :
3160- current_valves .GEMINI_PAID_API_KEY = None
3161- log .info (f"Attempting execution on FREE Tier ({ model_id } )..." )
3162- __metadata__ ["is_paid_api" ] = False
3163- elif tier == "paid" :
3164- current_valves .GEMINI_FREE_API_KEY = None
3165- log .info (f"Attempting execution on PAID Tier ({ model_id } )..." )
3166- __metadata__ ["is_paid_api" ] = True
3167-
3168- try :
3169- client = self ._get_user_client (current_valves , __user__ ["email" ])
3170- except Exception as e :
3171- if not is_last_attempt :
3172- log .warning (f"Client creation failed for { tier } : { e } . Retrying..." )
3173- continue
3174- raise e
3175-
3176- try :
3177- if is_streaming :
3178- stream = await client .aio .models .generate_content_stream (** gen_content_args ) # type: ignore
3179- else :
3180- response = await client .aio .models .generate_content (
3181- ** gen_content_args
3182- )
3183-
3184- async def one_shot_iter ():
3185- yield response
3186-
3187- stream = one_shot_iter ()
3188-
3189- # --- 3. The "Peek" Logic ---
3190- # We fetch the first chunk to catch early errors (429, 403, 503)
3191- iterator = stream .__aiter__ ()
3192- try :
3193- first_chunk = await iterator .__anext__ ()
3194- except StopAsyncIteration :
3195- return
3196-
3197- # Success! Reconstruct the stream and return
3198- async def reconstructed_stream ():
3199- yield first_chunk
3200- async for chunk in iterator :
3201- yield chunk
3202-
3203- return reconstructed_stream ()
3204-
3205- except Exception as e :
3206- error_str = str (e ).upper ()
3207-
3208- # Check for Quota (429), Permissions (403), or Overloaded/Unavailable (503)
3209- is_fallback_eligible_error = (
3210- "429" in error_str
3211- or "403" in error_str
3212- or "503" in error_str
3213- or "UNAVAILABLE" in error_str
3214- or (
3215- isinstance (e , genai_errors .ClientError ) and e .code in [429 , 403 ]
3216- )
3217- or (isinstance (e , genai_errors .ServerError ) and e .code in [503 ])
3218- )
3219-
3220- should_retry = (
3221- not is_last_attempt
3222- and tier == "free"
3223- and is_fallback_eligible_error
3224- )
3225-
3226- if should_retry :
3227- reason = (
3228- "quota exceeded" if "429" in error_str else "model overloaded"
3229- )
3230- log .warning (
3231- f"Free Tier { reason } (Error: { e } ). Switching to Paid API..."
3232- )
3233-
3234- asyncio .create_task (
3235- event_emitter .emit_status (
3236- f"Free Tier { reason } , switching to Paid API..." , done = False
3237- )
3238- )
3239- continue
3240- else :
3241- log .error (f"Error during request execution (Tier: { tier } ): { e } " )
3242- raise e
3243-
3244- raise ValueError ("Exhausted execution options without result." )
3245-
32463107 async def _unified_response_processor (
32473108 self ,
32483109 response_stream : AsyncIterator [types .GenerateContentResponse ],
0 commit comments