Class PageController

browser_utils/page_controller.py:46–375

Encapsulates all operations for interacting with the AI Studio page.

class PageController(
    ParameterController,
    InputController,
    ChatController,
    ResponseController,
    ThinkingController,
    FunctionCallingController,
    BaseController,
):
    """Encapsulates all operations for interacting with the AI Studio page."""

    def __init__(self, page: AsyncPage, logger, req_id: str):
        self.page = page
        self.logger = logger
        self.req_id = req_id

    async def _check_disconnect(self, check_client_disconnected: Callable, stage: str):
        if check_client_disconnected(stage):
            raise ClientDisconnectedError(
                f"[{self.req_id}] Client disconnected at stage: {stage}"
            )

    async def adjust_parameters(
        self,
        request_params: Dict[str, Any],
        page_params_cache: Dict[str, Any],
        params_cache_lock: asyncio.Lock,
        model_id_to_use: Optional[str],
        parsed_model_list: List[Dict[str, Any]],
        check_client_disconnected: Callable,
        is_streaming: bool = True,
    ):
        self.logger.info(f"[{self.req_id}] Adjusting parameters...")
        await self._check_disconnect(
            check_client_disconnected, "Start Parameter Adjustment"
        )
        temp = request_params.get("temperature", DEFAULT_TEMPERATURE)
        await self._adjust_temperature(
            temp, page_params_cache, params_cache_lock, check_client_disconnected
        )
        max_tokens = request_params.get("max_output_tokens", DEFAULT_MAX_OUTPUT_TOKENS)
        await self._adjust_max_tokens(
            max_tokens,
            page_params_cache,
            params_cache_lock,
            model_id_to_use,
            parsed_model_list,
            check_client_disconnected,
        )
        stop = request_params.get("stop", DEFAULT_STOP_SEQUENCES)
        await self._adjust_stop_sequences(
            stop, page_params_cache, params_cache_lock, check_client_disconnected
        )
        top_p = request_params.get("top_p", DEFAULT_TOP_P)
        await self._adjust_top_p(top_p, check_client_disconnected)
        await self._ensure_tools_panel_expanded(check_client_disconnected)

        # Force disable URL context if function calling is active