Method optimize

src/agents/optimization/node_optimizer.py:66–119

Optimizes nodes in the given list of cases. Args: case_list (list[Case]): The list of cases to be optimized. solution (Solution): The solution to be optimized. save_dir (Path): The directory to save the results. parallel_max_num (int)

(
            self,
            case_list: list[Case],
            solution: Solution,
            save_dir: Path,
            parallel_max_num,
    )

        self.logger = logging.getLogger(logger_name) if logger_name else logging.getLogger(__name__)

    def optimize(
            self,
            case_list: list[Case],
            solution: Solution,
            save_dir: Path,
            parallel_max_num,
    ):
        """
        Optimizes nodes in the given list of cases.

        Args:
            case_list (list[Case]): The list of cases to be optimized.
            solution (Solution): The solution to be optimized.
            save_dir (Path): The directory to save the results.
            parallel_max_num (int): The maximum number of parallel processes.

        Returns:
            tuple: The updated solution and optimization status.
        """
        self.logger.info("Start Node Optimization")
        saved_ori_solution = copy.deepcopy(solution)

        # backward
        backward_save_dir = save_dir / "case_after_backward"
        partial_funcs = [partial(self.backward, case, solution, backward_save_dir) for case in case_list]
        OptimUtils.parallel_execution(partial_funcs, max_workers=parallel_max_num)

        # optimization (based on the suggestions)
        op_info = self.optimize_node(case_list, solution)

        # Determine the optimization status based on the success of any node optimization
        op_status = any(info["optim_status"] for info in op_info.values())

        # if optimized successfully, update the AgentTeam
        if op_status:
            all_node_roles_description = {}
            for node_name, node in solution.sop.nodes.items():
                all_node_roles_description[node_name] = node.node_roles_description
            agent_team_config = AgentTeamConfig.generate_config(
                solution.task.task_description, all_node_roles_description)
            solution.agent_team = AgentTeam(agent_team_config)

        # save new solution and op_info
        try:
            with open(save_dir / "node_optim_info.json", "w", encoding="utf-8") as f:
                json.dump(op_info, f, ensure_ascii=False, indent=4)
            solution.dump(save_dir)
            solution = Solution(config=SolutionConfig(str(save_dir / "solution.json")))
        except Exception as e:
            self.logger.error(f"Error in saving solution: {e}")
            solution = saved_ori_solution
            solution.dump(save_dir / "accepted_solution")

        return solution, op_status

    def backward(self, case: Case, solution: Solution, save_dir: str):
        """