Build an AgentVerse from a task name. The task name should correspond to a directory in `tasks` directory. Then this method will load the configuration from the yaml file in that directory.
(cls, task: str, tasks_dir: str, postfix='')
| 25 | |
| 26 | @classmethod |
| 27 | def from_task(cls, task: str, tasks_dir: str, postfix=''): |
| 28 | """Build an AgentVerse from a task name. |
| 29 | The task name should correspond to a directory in `tasks` directory. |
| 30 | Then this method will load the configuration from the yaml file in that directory. |
| 31 | """ |
| 32 | # Prepare the config of the task |
| 33 | task_config = prepare_task_config(task, tasks_dir, postfix) |
| 34 | |
| 35 | # Build the environment |
| 36 | env_config = task_config["environment"] |
| 37 | |
| 38 | # Build agents for all pipeline (task) |
| 39 | agents = {} |
| 40 | for i, agent_config in enumerate(task_config["agents"]): |
| 41 | if agent_config.get("agent_type", "") == "critic" and 0: |
| 42 | agent = load_agent(agent_config) |
| 43 | agents[AGENT_TYPES.CRITIC] = [ |
| 44 | copy.deepcopy(agent) |
| 45 | for _ in range(task_config.get("cnt_agents", 1) - 1) |
| 46 | ] |
| 47 | # if agent_config.get("llm_type").startswith('[') and agent_config.get("llm_type").endswith(']'): |
| 48 | # for j, agent in enumerate(agents[AGENT_TYPES.CRITIC]): |
| 49 | # agent. |
| 50 | for j, agent in enumerate(agents[AGENT_TYPES.CRITIC]): |
| 51 | agent.set_name(f"Agent {j+1}") |
| 52 | else: |
| 53 | agent_type = AGENT_TYPES.from_string(agent_config.get("agent_type", "")) |
| 54 | if agent_type not in agents.keys(): |
| 55 | agents[agent_type] = load_agent(agent_config) |
| 56 | else: |
| 57 | if not isinstance(agents[agent_type], list): |
| 58 | agents[agent_type] = [agents[agent_type]] |
| 59 | agents[agent_type].append(load_agent(agent_config)) |
| 60 | # print('!!!!!!!!!!!!!!!The agents:!!!!!!!!!!!!!!!!!') |
| 61 | # print(agents) |
| 62 | # print('!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!') |
| 63 | env_config["agents"] = agents |
| 64 | |
| 65 | env_config["task_description"] = task_config.get("task_description", "") |
| 66 | env_config["max_rounds"] = task_config.get("max_rounds", 3) |
| 67 | |
| 68 | environment: BasicEnvironment = load_environment(env_config) |
| 69 | |
| 70 | return cls(environment=environment, task=task) |
| 71 | |
| 72 | def run(self): |
| 73 | """Run the environment from scratch until it is done.""" |
nothing calls this directly
no test coverage detected