Method do_node_optim

src/agents/optimization/node_optimizer.py:240–316

Executes the node optimization based on the given method. Note: If an error occurs during optimization, an exception is raised instead of returning, because the upper layer will handle the error and return the original unoptimized node. Args: node (Node

(node: Node, optim_method_str: str, logger)

    @staticmethod
    def do_node_optim(node: Node, optim_method_str: str, logger):
        """
        Executes the node optimization based on the given method.

        Note: If an error occurs during optimization, an exception is raised instead of returning,
        because the upper layer will handle the error and return the original unoptimized node.

        Args:
            node (Node): The node to be optimized.
            optim_method_str (str): The optimization method in JSON string format.
            logger: The logger for logging information and errors.

        Returns:
            tuple: The optimized Node object, a boolean indicating if the update was successful,
                   and the optimization method (JSON if successful, otherwise a string).
        """
        optim_method = json.loads(optim_method_str)

        # Validity check, attempt update only if successful
        check_status, reasons = NodeOptimizer.validate_dict(optim_method)

        if not check_status:
            logger.error(f"Error in validating optim_method: {reasons}, optim_method: {optim_method}")
            raise ValueError(f"Error in validating optim_method: {reasons}")
        else:
            logger.debug(f"succeed in validating optim_method, load the json successfully.")
            # try to optim the node
            for rule in optim_method:
                action = rule.get("action")
                if action == "add_role":
                    role_name = rule["role_name"]
                    role_description = rule["role_description"]
                    role_prompt = rule["role_prompt"]
                    role_prompt_key = "step_" + role_name
                    assert role_name not in node.node_prompt_paddings.keys(), f"Role name '{role_name}' already exists."
                    node.node_prompt_templates[role_prompt_key] = role_prompt
                    node.node_roles_description[role_name] = role_description
                    node.node_prompt_paddings[role_name] = {
                        role_prompt_key: {"value_source": "case", "value": "input_data"}}

                elif action == "delete_role":
                    role_name = rule["role_name"]
                    node.node_roles.pop(role_name)
                    node.node_roles_description.pop(role_name)
                    node.node_primary_prompts.pop(role_name)
                    node.node_prompt_templates.pop(role_name)
                    node.node_prompt_paddings.pop(role_name)
                    if len(node.node_prompt_paddings) == 0:
                        raise ValueError("The node should have at least one role.")
                    # begin role may need to be updated
                    if node.begin_role == role_name:
                        node.begin_role = next(iter(node.node_prompt_templates))

                elif action == "update_role_description":
                    role_name = rule["role_name"]
                    role_description = rule["role_description"]
                    node.node_roles_description[role_name] = role_description