定义了 ComposedPhase 类和相关函数。

check_bool 函数用于判断字符串是否为 True。它接收一个参数 s，并将其转换为小写后与字符串 “true” 进行比较。

ComposedPhase 是一个抽象基类，定义了复合阶段的基本结构和函数，不能被实例化。具体的复合阶段需要实现 update_phase_env 和 update_chat_env 这两个抽象函数。

在 __init__ 函数中，定义了复合阶段的具体属性和实例化过程：

1. phase_name 表示复合阶段的名称，cycle_num 表示循环次数，composition 表示由哪些 SimplePhases 构成复合阶段，config_phase 和 config_role 表示 SimplePhases 和 Role 的配置信息。
2. model_type 表示使用哪种模型类型，log_filepath 表示日志文件的路径。
3. self.phase_env 是一个字典，用于存储阶段环境信息。
4. 初始化 chat_turn_limit_default 表示每轮对话的最大回合数。
5. self.role_prompts 是一个字典，表示各个角色的提示信息。
6. 实例化所有 SimplePhases 并保存到 self.phases 字典中。每个 SimplePhase 都会接收一些参数，包括助手和用户的角色名称、阶段提示信息等。

update_phase_env 和 update_chat_env 是抽象方法，需要在具体阶段中实现。update_phase_env 接收一个 chat_env 字典作为参数，用于更新当前阶段的环境变量；update_chat_env 会根据当前阶段的结果更新全局 chat_env 字典，并返回更新后的 chat_env。这两个函数属于具体阶段的实现范畴，必须在子类中实现具体逻辑。

1. @abstractmethod:
   • 这是一个装饰器，用于标识一个方法为抽象方法。在该代码段中，break_cycle方法被标记为抽象方法，表示它在抽象类ComposedPhase中没有具体的实现，而是需要在子类中进行实现。
2. break_cycle(self, phase_env) -> bool:
   • 这是一个抽象方法，用于在ComposedPhase类中定义特殊条件来提前终止循环。
   • 参数phase_env是阶段环境。
   • 返回值是一个布尔类型，表示是否要提前结束循环。
3. execute(self, chat_env) -> ChatEnv:
   • 这个方法执行ComposedPhase的逻辑。
   • 参数chat_env是全局聊天环境。
   • 返回值是ChatEnv类型的对象，表示更新后的全局聊天环境。
4. self.update_phase_env(chat_env):
   • 这个方法用于从全局环境中更新阶段环境。
5. for cycle_index in range(self.cycle_num):
   • 这是一个循环，用于控制执行的循环次数，循环次数由self.cycle_num确定。
6. for phase_item in self.composition:
   • 这是一个循环，用于遍历ComposedPhase中组成阶段的每个SimplePhase。
7. assert phase_item[“phaseType”] == “SimplePhase”：
   • 这是一个断言语句，用于确保phase_item的类型是”SimplePhase”，即简单阶段。
8. phase = phase_item[‘phase’]
   • 获取phase_item字典中的”phase”键对应的值，表示SimplePhase的名称。
9. max_turn_step = phase_item[‘max_turn_step’]
   • 获取phase_item字典中的”max_turn_step”键对应的值，表示最大的对话轮数。
10. need_reflect = check_bool(phase_item[‘need_reflect’])
    • 获取phase_item字典中的”need_reflect”键对应的值，并通过check_bool函数将其转换为布尔类型，表示是否需要反射。
11. self.phases[phase].phase_env = self.phase_env
    • 将当前ComposedPhase的阶段环境赋值给对应的SimplePhase的阶段环境。
12. self.phases[phase].update_phase_env(chat_env)
    • 调用对应的SimplePhase的update_phase_env方法，更新SimplePhase的阶段环境。
13. if self.break_cycle(self.phases[phase].phase_env): return chat_env
    • 调用break_cycle方法判断是否需要提前终止循环，如果需要则直接返回chat_env。
14. chat_env = self.phases[phase].execute(chat_env, self.chat_turn_limit_default if max_turn_step <= 0 else max_turn_step, need_reflect)
    • 调用对应的SimplePhase的execute方法执行聊天，并更新chat_env。
15. if self.break_cycle(self.phases[phase].phase_env): return chat_env
    • 再次调用break_cycle方法判断是否需要提前终止循环，如果需要则直接返回chat_env。
16. else: print(f”Phase ‘{phase}’ is not yet implemented.
    Please write its config in phaseConfig.json
    and implement it in chatdev.phase”)

• 如果SimplePhase未实现，则打印一条错误信息。

1. chat_env = self.update_chat_env(chat_env)
   • 调用update_chat_env方法，更新全局聊天环境。
2. return chat_env
   • 返回更新后的全局聊天环境。

下面是几个具体子类（Art、CodeCompleteAll、CodeReview和Test）的定义，它们都继承自ComposedPhase，并对其中的一些方法进行了具体实现。

1. Art类：
   • update_phase_env方法为空实现。
   • update_chat_env方法返回chat_env。
   • break_cycle方法始终返回False。
2. CodeCompleteAll类：
   • update_phase_env方法获取目录中所有以”.py”结尾的文件名，并初始化num_tried字典。
   • update_chat_env方法返回chat_env。
   • break_cycle方法判断是否存在未实现的文件，如果不存在则返回True，否则返回False。
3. CodeReview类：
   • update_phase_env方法初始化modification_conclusion为空字符串。
   • update_chat_env方法返回chat_env。
   • break_cycle方法判断modification_conclusion是否包含”<INFO> Finished”，如果是则返回True，否则返回False。
4. Test类：

• update_phase_env方法初始化一个空字典。
• update_chat_env方法返回chat_env。
• break_cycle方法判断exist_bugs_flag是否为False，如果是则打印一条测试通过的信息并返回True，否则返回False。

import importlib
import os
from abc import ABC, abstractmethod
from collections import defaultdict

from camel.typing import ModelType
from chatdev.chat_env import ChatEnv
from chatdev.utils import log_and_print_online


def check_bool(s):
    return s.lower() == "true"


class ComposedPhase(ABC):
    def __init__(self,
                 phase_name: str = None,
                 cycle_num: int = None,
                 composition: list = None,
                 config_phase: dict = None,
                 config_role: dict = None,
                 model_type: ModelType = ModelType.GPT_3_5_TURBO,
                 log_filepath: str = ""
                 ):
        """

        Args:
            phase_name: name of this phase
            cycle_num: loop times of this phase
            composition: list of SimplePhases in this ComposePhase
            config_phase: configuration of all SimplePhases
            config_role: configuration of all Roles
        """

        self.phase_name = phase_name
        self.cycle_num = cycle_num
        self.composition = composition
        self.model_type = model_type
        self.log_filepath = log_filepath

        self.config_phase = config_phase
        self.config_role = config_role

        self.phase_env = dict()

        # init chat turn
        self.chat_turn_limit_default = 10

        # init role
        self.role_prompts = dict()
        for role in self.config_role:
            self.role_prompts[role] = "\n".join(self.config_role[role])

        # init all SimplePhases instances in this ComposedPhase
        self.phases = dict()
        for phase in self.config_phase:
            assistant_role_name = self.config_phase[phase]['assistant_role_name']
            user_role_name = self.config_phase[phase]['user_role_name']
            phase_prompt = "\n".join(self.config_phase[phase]['phase_prompt'])
            phase_module = importlib.import_module("chatdev.phase")
            phase_class = getattr(phase_module, phase)
            phase_instance = phase_class(assistant_role_name=assistant_role_name,
                                         user_role_name=user_role_name,
                                         phase_prompt=phase_prompt,
                                         role_prompts=self.role_prompts,
                                         phase_name=phase,
                                         model_type=self.model_type,
                                         log_filepath=self.log_filepath)
            self.phases[phase] = phase_instance

    @abstractmethod
    def update_phase_env(self, chat_env):
        """
        update self.phase_env (if needed) using chat_env, then the chatting will use self.phase_env to follow the context and fill placeholders in phase prompt
        must be implemented in customized phase
        the usual format is just like:
        ```
            self.phase_env.update({key:chat_env[key]})
        ```
        Args:
            chat_env: global chat chain environment

        Returns: None

        """
        pass

    @abstractmethod
    def update_chat_env(self, chat_env) -> ChatEnv:
        """
        update chan_env based on the results of self.execute, which is self.seminar_conclusion
        must be implemented in customized phase
        the usual format is just like:
        ```
            chat_env.xxx = some_func_for_postprocess(self.seminar_conclusion)
        ```
        Args:
            chat_env:global chat chain environment

        Returns:
            chat_env: updated global chat chain environment

        """
        pass

    @abstractmethod
    def break_cycle(self, phase_env) -> bool:
        """
        special conditions for early break the loop in ComposedPhase
        Args:
            phase_env: phase environment

        Returns: None

        """
        pass

    def execute(self, chat_env) -> ChatEnv:
        """
        similar to Phase.execute, but add control for breaking the loop
        1. receive information from environment(ComposedPhase): update the phase environment from global environment
        2. for each SimplePhase in ComposedPhase
            a) receive information from environment(SimplePhase)
            b) check loop break
            c) execute the chatting
            d) change the environment(SimplePhase)
            e) check loop break
        3. change the environment(ComposedPhase): update the global environment using the conclusion

        Args:
            chat_env: global chat chain environment

        Returns:

        """
        self.update_phase_env(chat_env)
        for cycle_index in range(self.cycle_num):
            for phase_item in self.composition:
                assert phase_item["phaseType"] == "SimplePhase"  # right now we do not support nested composition
                phase = phase_item['phase']
                max_turn_step = phase_item['max_turn_step']
                need_reflect = check_bool(phase_item['need_reflect'])
                log_and_print_online(
                    f"**[Execute Detail]**\n\nexecute SimplePhase:[{phase}] in ComposedPhase:[{self.phase_name}], cycle {cycle_index}")
                if phase in self.phases:
                    self.phases[phase].phase_env = self.phase_env
                    self.phases[phase].update_phase_env(chat_env)
                    if self.break_cycle(self.phases[phase].phase_env):
                        return chat_env
                    chat_env = self.phases[phase].execute(chat_env,
                                                          self.chat_turn_limit_default if max_turn_step <= 0 else max_turn_step,
                                                          need_reflect)
                    if self.break_cycle(self.phases[phase].phase_env):
                        return chat_env
                else:
                    print(f"Phase '{phase}' is not yet implemented. \
                            Please write its config in phaseConfig.json \
                            and implement it in chatdev.phase")
        chat_env = self.update_chat_env(chat_env)
        return chat_env


class Art(ComposedPhase):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def update_phase_env(self, chat_env):
        pass

    def update_chat_env(self, chat_env):
        return chat_env

    def break_cycle(self, chat_env) -> bool:
        return False


class CodeCompleteAll(ComposedPhase):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def update_phase_env(self, chat_env):
        pyfiles = [filename for filename in os.listdir(chat_env.env_dict['directory']) if filename.endswith(".py")]
        num_tried = defaultdict(int)
        num_tried.update({filename: 0 for filename in pyfiles})
        self.phase_env = {
            "max_num_implement": 5,
            "pyfiles": pyfiles,
            "num_tried": num_tried
        }

    def update_chat_env(self, chat_env):
        return chat_env

    def break_cycle(self, phase_env) -> bool:
        if phase_env['unimplemented_file'] == "":
            return True
        else:
            return False


class CodeReview(ComposedPhase):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def update_phase_env(self, chat_env):
        self.phase_env = {"modification_conclusion": ""}

    def update_chat_env(self, chat_env):
        return chat_env

    def break_cycle(self, phase_env) -> bool:
        if "<INFO> Finished".lower() in phase_env['modification_conclusion'].lower():
            return True
        else:
            return False


class Test(ComposedPhase):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def update_phase_env(self, chat_env):
        self.phase_env = dict()

    def update_chat_env(self, chat_env):
        return chat_env

    def break_cycle(self, phase_env) -> bool:
        if not phase_env['exist_bugs_flag']:
            log_and_print_online(f"**[Test Info]**\n\nAI User (Software Test Engineer):\nTest Pass!\n")
            return True
        else:
            return False