Method generate

datastudio/models/base.py:537–590

Main method to generate responses. Args: messages: Input messages (list of message dicts). **kwargs: Additional arguments. Returns: list: Generated responses for each message.

(self, messages, **kwargs)

            self.logger.info("Image encoding complete!")

    def generate(self, messages, **kwargs):
        """Main method to generate responses.

        Args:
            messages: Input messages (list of message dicts).
            **kwargs: Additional arguments.

        Returns:
            list: Generated responses for each message.
        """
        all_kwargs = cp.deepcopy(self.default_kwargs)
        all_kwargs.update(kwargs)

        # Handle empty messages list
        if not messages:
            return []

        # Pre-encode images (thread-safe)
        self._pre_encode_images(messages)

        # Use dedicated thread pool to avoid resource leaks
        max_workers = min(self.thread_num, len(messages))
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            # Submit all tasks with retry on thread creation failure
            futures = {}
            for i, msg in enumerate(messages):
                while True:
                    try:
                        future = executor.submit(
                            self.process_single_message, msg, all_kwargs
                        )
                        futures[future] = i
                        break
                    except RuntimeError as e:
                        if "can't start new thread" in str(e):
                            # Wait for some tasks to complete before submitting more
                            time.sleep(0.1)
                        else:
                            raise

            # Use tqdm to display progress
            results = [None] * len(futures)

            with tqdm(total=len(futures), desc="Processing API Requests") as pbar:
                for future in concurrent.futures.as_completed(futures):
                    index = futures[future]
                    results[index] = future.result()
                    pbar.update(1)

        # Release base64 data before next batch
        self._clear_message_buffers(messages)
        self._collect_memory(trim=True)

        return results

    def shutdown(self):
        """Shutdown and release resources."""