Method _process_single_request

datastudio/models/openai_api.py:254–377

Process a single request with concurrency control and retry. Args: session: Shared aiohttp session. semaphore: Concurrency limiter. processed_msg: Preprocessed message (OpenAI format). all_kwargs: Additional generation arguments. p

(
        self,
        session: aiohttp.ClientSession,
        semaphore: asyncio.Semaphore,
        processed_msg: Any,
        all_kwargs: dict,
        pbar: Optional[async_tqdm] = None,
        error_counter: Optional[dict] = None,
    )

        return super()._calculate_retry_delay(attempt, max_delay=2.0)

    async def _process_single_request(
        self,
        session: aiohttp.ClientSession,
        semaphore: asyncio.Semaphore,
        processed_msg: Any,
        all_kwargs: dict,
        pbar: Optional[async_tqdm] = None,
        error_counter: Optional[dict] = None,
    ) -> str:
        """Process a single request with concurrency control and retry.

        Args:
            session: Shared aiohttp session.
            semaphore: Concurrency limiter.
            processed_msg: Preprocessed message (OpenAI format).
            all_kwargs: Additional generation arguments.
            pbar: Optional progress bar.
            error_counter: Optional error tracking dict.

        Returns:
            Response content string or fail_msg.
        """
        ret_code = None
        response_struct = None

        # Per-attempt timeout
        _, read_timeout = self.timeout
        per_attempt_timeout = read_timeout + 60

        connect_timeout, _ = self.timeout

        async with semaphore:
            for i in range(self.retry):
                try:
                    if error_counter is not None:
                        error_counter["active"] += 1
                        pbar.set_postfix(**error_counter)
                    t0 = asyncio.get_event_loop().time()
                    ret_code, response_struct = await asyncio.wait_for(
                        self._make_request(session, processed_msg, **all_kwargs),
                        timeout=per_attempt_timeout,
                    )
                    if error_counter is not None:
                        error_counter["active"] -= 1

                    if ret_code == 0 and response_struct and response_struct != "":
                        if pbar:
                            if error_counter is not None:
                                error_counter["ok"] += 1
                                pbar.set_postfix(**error_counter)
                            pbar.update(1)
                        return response_struct
                    else:
                        raise Exception(f"Invalid response: {response_struct}")

                except aiohttp.ServerTimeoutError:
                    elapsed = asyncio.get_event_loop().time() - t0
                    ret_code = 1