Spaces:
Runtime error
Runtime error
| 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 | |
| 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 | |
| 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 | |
| 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 | |