def mutate()

in 1_synthetic-qa-generation/reasoningplaning/evolve.py [0:0]
59 lines of code
12 McCabe index (conditional complexity)

    def mutate(self, iteration):
        assert len(self.prompts) == self.num_rows or len(self.prompts) == len(self.seed_text_dict)
        list_prompts = []
        mutations = []
        original_prompts = []
        # for i in range(self.num_rows):
        for k in self.prompts:
            mutation = np.random.choice(Mutation)
            mutations.append(mutation)
            # if mutation == Mutation.FRESH_START:
            #     mutation = Mutation.COMPLICATE
            before = k #self.prompts[i]
            prompt = self.prompt_templates[mutation].replace("<PROMPT>", before)
            
            if mutation == Mutation.SWITCH_TOPIC:
                prompt += "#Created Prompt#:\r\n"
            else:
                prompt += "#Rewritten Prompt:\r\n"
            
            logger.info(f"Full prompt={prompt}")
            list_prompts.append(prompt)
            original_prompts.append(k)

        ds = self.convertListToDataset(list_prompts)
        assert ds['train'].num_rows == len(list_prompts) == self.num_rows == len(self.prompts)
        
        # Processing transformed prompts using the LLM pipeline
        t0 = time.time()
        after = self.llm_pipeline(ds['train'])
        assert len(after) == self.num_rows
        t1 = time.time()
        
        llm_pipeline_name = self.llm_pipeline.__class__.__name__
        logger.info(f"{llm_pipeline_name} took {t1 - t0:.4f} seconds")

        for i in range(len(after)):
            after[i] = after[i].split("Prompt#:")[-1].strip()
            for pp in ['New Prompt:\n', 'New Prompt: ']:
                if after[i][:len(pp)] == pp:
                    after[i] = after[i][len(pp):]
            after[i] = after[i].strip()
            
            #use_new_prompt, why = self.changeApproved(self.prompts[i], after[i])
            use_new_prompt = True

            original_p = original_prompts[i]
            if self.verbose:
                logger.info("===========================")
                logger.info("Old Prompt: %s" % original_p)
                logger.info("Mutation: %s" % mutations[i].name)
                logger.info("New Prompt: %s" % after[i])
                logger.info("===========================")

            if use_new_prompt:
                original_itm = self.prompts[original_p]
                self.maxIdx = self.maxIdx + 1
                self.final_prompts[after[i]] = {
                    "idx": self.maxIdx,
                    "preidx": original_itm["idx"],
                    "preproblem": original_p,
                    GRND_TRUTH_COL: original_itm[GRND_TRUTH_COL]
                }
                del self.prompts[original_p]
                chosen_prmp = np.random.choice(list(self.seed_text_dict.keys()))
                self.prompts[chosen_prmp] = {
                        "idx": self.seed_text_dict[chosen_prmp]["idx"],
                        GRND_TRUTH_COL: self.seed_text_dict[chosen_prmp][GRND_TRUTH_COL]
                    }
                # if self.max_len_bytes >= len(after[i]) >= self.min_len_bytes:
                #     self.final_prompts.append(after[i])
                #     logger.info("Prompt was accepted, now have %d good prompts." % len(self.final_prompts))
                #     self.prompts[i] = np.random.choice(self.seed_text_dict)
                #     logger.info("Creating new prompt.")
                # else:
                #     self.prompts[i] = after[i]
                #     logger.info("Prompt was successfully modified.")
            else:
                logger.info("Mutation rejected, will try again. Reason: %s" % why)
            # logger.info("", flush=True)
        logger.info("final_prompt=")
        logger.info(self.final_prompts)
        return len(self.final_prompts) <= self.num_rows