6363
6464_REFUSAL_PREFIX = '[[REFUSAL]]: '
6565
66+ # Timeouts, in seconds, for the completions HTTP client. httpx applies no
67+ # timeout at all unless one is given, so a stalled proxy would otherwise hold
68+ # the connection and the streaming loop open indefinitely.
69+ _CONNECT_TIMEOUT_SECONDS = 30.0
70+ _REQUEST_TIMEOUT_SECONDS = 600.0
71+
72+
73+ def _httpx_timeout (timeout_seconds : Optional [float ] = None ) -> httpx .Timeout :
74+ """Returns the httpx timeout budget for a completions request.
75+
76+ A bare float would spend the caller's whole budget on the connect phase too,
77+ so the connect budget is always kept short enough to fail fast on an
78+ unreachable proxy.
79+
80+ Args:
81+ timeout_seconds: The total budget for the request, or None for the default.
82+ """
83+ return httpx .Timeout (
84+ _REQUEST_TIMEOUT_SECONDS if timeout_seconds is None else timeout_seconds ,
85+ connect = _CONNECT_TIMEOUT_SECONDS ,
86+ )
87+
6688
6789class ApigeeLlm (Gemini ):
6890 """A BaseLlm implementation for calling Apigee proxy.
@@ -427,8 +449,8 @@ def _client(self) -> httpx.AsyncClient:
427449 client = httpx .AsyncClient (
428450 base_url = self ._base_url ,
429451 headers = self ._headers ,
430- timeout = None ,
431- follow_redirects = True ,
452+ timeout = _httpx_timeout () ,
453+ follow_redirects = False ,
432454 )
433455 atexit .register (self ._cleanup_client , client )
434456 return client
@@ -519,6 +541,7 @@ async def generate_content_async(
519541 ) -> AsyncGenerator [LlmResponse , None ]:
520542 """Generates content using the OpenAI-compatible HTTP API."""
521543 payload = self ._construct_payload (llm_request , stream )
544+ timeout = self ._get_request_timeout_seconds (llm_request )
522545 headers = self ._headers .copy ()
523546 headers ['Content-Type' ] = 'application/json'
524547
@@ -530,26 +553,50 @@ async def generate_content_async(
530553 url = f"{ url .rstrip ('/' )} /chat/completions"
531554
532555 if stream :
533- async for stream_res in self ._handle_streaming (url , payload , headers ):
556+ async for stream_res in self ._handle_streaming (
557+ url , payload , headers , timeout = timeout
558+ ):
534559 yield stream_res
535560 else :
536- response = await self ._httpx_post_with_retry (url , payload , headers )
561+ response = await self ._httpx_post_with_retry (
562+ url , payload , headers , timeout = timeout
563+ )
537564 data = response .json ()
538565 yield self ._parse_response (data )
539566
567+ @staticmethod
568+ def _get_request_timeout_seconds (llm_request : LlmRequest ) -> float | None :
569+ """Returns the request timeout converted from milliseconds to seconds."""
570+ if not llm_request .config or not llm_request .config .http_options :
571+ return None
572+ timeout_ms = llm_request .config .http_options .timeout
573+ return timeout_ms / 1000 if timeout_ms is not None else None
574+
540575 async def _httpx_post_with_retry (
541- self , url : str , payload : dict [str , Any ], headers : dict [str , str ]
576+ self ,
577+ url : str ,
578+ payload : dict [str , Any ],
579+ headers : dict [str , str ],
580+ * ,
581+ timeout : float | None ,
542582 ) -> httpx .Response :
543583 """Sends a POST request and handles retries."""
544584 retry_kwargs = self ._get_retry_kwargs ()
545585 async for attempt in tenacity .AsyncRetrying (** retry_kwargs ):
546586 with attempt :
547- response = await self ._client .post (url , json = payload , headers = headers )
587+ response = await self ._client .post (
588+ url , json = payload , headers = headers , timeout = _httpx_timeout (timeout )
589+ )
548590 response .raise_for_status ()
549591 return response
550592
551593 async def _handle_streaming (
552- self , url : str , payload : dict [str , Any ], headers : dict [str , str ]
594+ self ,
595+ url : str ,
596+ payload : dict [str , Any ],
597+ headers : dict [str , str ],
598+ * ,
599+ timeout : float | None ,
553600 ) -> AsyncGenerator [LlmResponse , None ]:
554601 """Handles streaming response from OpenAI-compatible API."""
555602 accumulator = ChatCompletionsResponseHandler ()
@@ -558,6 +605,7 @@ async def _handle_streaming(
558605 url ,
559606 json = payload ,
560607 headers = headers ,
608+ timeout = _httpx_timeout (timeout ),
561609 ) as resp :
562610 resp .raise_for_status ()
563611 async for line in resp .aiter_lines ():
0 commit comments