ChatDev/camel/generators.py
2023-09-04 19:24:28 +08:00

268 lines
11 KiB
Python

# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
# Licensed under the Apache License, Version 2.0 (the “License”);
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an “AS IS” BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
from typing import Dict, Generator, List, Optional, Set, Tuple
from camel.messages import SystemMessage, SystemMessageType
from camel.prompts import PromptTemplateGenerator, TextPrompt
from camel.typing import RoleType, TaskType
class SystemMessageGenerator:
r"""System message generator for agents.
Args:
task_type (TaskType, optional): The task type.
(default: :obj:`TaskType.AI_SOCIETY`)
sys_prompts (Optional[Dict[RoleType, str]], optional): The prompts of
the system messages for each role type. (default: :obj:`None`)
sys_msg_meta_dict_keys (Optional[Set[str]], optional): The set of keys
of the meta dictionary used to fill the prompts.
(default: :obj:`None`)
"""
def __init__(
self,
task_type: TaskType = TaskType.AI_SOCIETY,
sys_prompts: Optional[Dict[RoleType, str]] = None,
sys_msg_meta_dict_keys: Optional[Set[str]] = None,
) -> None:
self.sys_prompts: Dict[RoleType, str]
if sys_prompts is not None:
self.sys_prompts = sys_prompts
self.sys_msg_meta_dict_keys = sys_msg_meta_dict_keys or set()
else:
templates = PromptTemplateGenerator()
agenttech_prompt_template = templates.get_system_prompt(task_type, RoleType.CHATDEV)
counselor_prompt_template = templates.get_system_prompt(task_type, RoleType.CHATDEV_COUNSELOR)
ceo_prompt_template = templates.get_system_prompt(task_type, RoleType.CHATDEV_CEO)
chro_prompt_template = templates.get_system_prompt(task_type, RoleType.CHATDEV_CHRO)
cpo_prompt_template = templates.get_system_prompt(task_type, RoleType.CHATDEV_CPO)
cto_prompt_template = templates.get_system_prompt(task_type, RoleType.CHATDEV_CTO)
programmer_prompt_template = templates.get_system_prompt(task_type, RoleType.CHATDEV_PROGRAMMER)
reviewer_prompt_template = templates.get_system_prompt(task_type, RoleType.CHATDEV_REVIEWER)
tester_prompt_template = templates.get_system_prompt(task_type, RoleType.CHATDEV_TESTER)
cco_prompt_template = templates.get_system_prompt(task_type, RoleType.CHATDEV_CCO)
self.sys_prompts = dict()
self.sys_prompts[RoleType.CHATDEV] = agenttech_prompt_template
self.sys_prompts[RoleType.CHATDEV_COUNSELOR] = counselor_prompt_template
self.sys_prompts[RoleType.CHATDEV_CEO] = ceo_prompt_template
self.sys_prompts[RoleType.CHATDEV_CHRO] = chro_prompt_template
self.sys_prompts[RoleType.CHATDEV_CPO] = cpo_prompt_template
self.sys_prompts[RoleType.CHATDEV_CTO] = cto_prompt_template
self.sys_prompts[RoleType.CHATDEV_PROGRAMMER] = programmer_prompt_template
self.sys_prompts[RoleType.CHATDEV_REVIEWER] = reviewer_prompt_template
self.sys_prompts[RoleType.CHATDEV_TESTER] = tester_prompt_template
self.sys_prompts[RoleType.CHATDEV_CCO] = cco_prompt_template
self.sys_msg_meta_dict_keys = (agenttech_prompt_template.key_words |
counselor_prompt_template.key_words |
ceo_prompt_template.key_words |
chro_prompt_template.key_words |
cpo_prompt_template.key_words |
cto_prompt_template.key_words |
programmer_prompt_template.key_words |
reviewer_prompt_template.key_words |
tester_prompt_template.key_words |
cco_prompt_template.key_words)
if RoleType.DEFAULT not in self.sys_prompts:
self.sys_prompts[RoleType.DEFAULT] = "You are a helpful assistant."
def validate_meta_dict_keys(self, meta_dict: Dict[str, str]) -> None:
r"""Validates the keys of the meta_dict.
Args:
meta_dict (Dict[str, str]): The dictionary to validate.
"""
if not set(meta_dict.keys()).issubset(self.sys_msg_meta_dict_keys):
raise ValueError("The keys of the meta_dict should be in "
f"{self.sys_msg_meta_dict_keys}. "
f"Got {set(meta_dict.keys())} instead.")
def from_dict(
self,
meta_dict: Dict[str, str],
role_tuple: Tuple[str, RoleType] = ("", RoleType.DEFAULT),
) -> SystemMessageType:
r"""Generates a system message from a dictionary.
Args:
meta_dict (Dict[str, str]): The dictionary containing the
information to generate the system message.
role_tuple (Tuple[str, RoleType], optional): The tuple containing
the role name and role type. (default: ("", RoleType.DEFAULT))
Returns:
SystemMessageType: The generated system message.
"""
self.validate_meta_dict_keys(meta_dict)
role_name, role_type = role_tuple
sys_prompt = self.sys_prompts[role_type]
sys_prompt = sys_prompt.format(**meta_dict)
return SystemMessage(role_name=role_name, role_type=RoleType.DEFAULT,
meta_dict=meta_dict, content=sys_prompt)
def from_dicts(
self,
meta_dicts: List[Dict[str, str]],
role_tuples: Tuple[str, str],
) -> List[SystemMessageType]:
r"""Generates a list of system messages from a list of dictionaries.
Args:
meta_dicts (List[Dict[str, str]]): A list of dictionaries
containing the information to generate the system messages.
role_tuples (List[Tuple[str, RoleType]]): A list of tuples
containing the role name and role type for each system message.
Returns:
List[SystemMessageType]: A list of generated system messages.
Raises:
ValueError: If the number of meta_dicts and role_tuples are
different.
"""
if len(meta_dicts) != len(role_tuples):
raise ValueError(
"The number of meta_dicts and role_types should be the same.")
return [
self.from_dict(meta_dict, role_tuple)
for meta_dict, role_tuple in zip(meta_dicts, role_tuples)
]
class RoleNameGenerator:
def __init__(self, assistant_role_names_path:
str = "data/ai_society/assistant_roles.txt",
user_role_names_path: str = "data/ai_society/user_roles.txt",
assistant_role_names: Optional[List[str]] = None,
user_role_names: Optional[List[str]] = None) -> None:
if assistant_role_names is None:
with open(assistant_role_names_path, "r") as f:
assistant_role_names_: List[str] = f.read().splitlines()
self.assistant_role_names = [
" ".join(name.split(" ")[1:])
for name in assistant_role_names_
]
else:
self.assistant_role_names = assistant_role_names
if user_role_names is None:
with open(user_role_names_path, "r") as f:
user_role_names_: List[str] = f.read().splitlines()
self.user_role_names = [
" ".join(name.split(" ")[1:]) for name in user_role_names_
]
else:
self.user_role_names = user_role_names
def from_role_files(self) -> Generator[Tuple, None, None]:
for assistant_role_name in self.assistant_role_names:
for user_role_name in self.user_role_names:
yield (assistant_role_name, user_role_name)
class AISocietyTaskPromptGenerator:
def __init__(
self,
num_tasks: int = 10,
) -> None:
self.generate_tasks_prompt = PromptTemplateGenerator(
).get_generate_tasks_prompt(TaskType.AI_SOCIETY)
self.num_tasks = num_tasks
# TODO: Return role names for user and assistant with the generator.
def from_role_files(
self,
assistant_role_names_path: str = "data/ai_society/assistant_roles.txt",
user_role_names_path: str = "data/ai_society/user_roles.txt"
) -> Generator[Tuple[str, Tuple[str, str]], None, None]:
roles_generator = RoleNameGenerator(
assistant_role_names_path, user_role_names_path).from_role_files()
for role_1, role_2 in roles_generator:
generate_tasks_prompt = self.generate_tasks_prompt.format(
assistant_role=role_1, user_role=role_2,
num_tasks=self.num_tasks)
yield (generate_tasks_prompt, (role_1, role_2))
def from_role_generator(
self, role_generator: Generator[Tuple, None, None]
) -> Generator[Tuple[str, Tuple[str, str]], None, None]:
for role_1, role_2 in role_generator:
generate_tasks_prompt = self.generate_tasks_prompt.format(
assistant_role=role_1, user_role=role_2,
num_tasks=self.num_tasks)
yield (generate_tasks_prompt, (role_1, role_2))
class SingleTxtGenerator:
def __init__(
self,
text_file_path: str,
) -> None:
with open(text_file_path, "r") as f:
data_list: List[str] = f.read().splitlines()
self.data_list = [
" ".join(name.split(" ")[1:]) for name in data_list
]
def from_role_files(self) -> Generator[str, None, None]:
for data in self.data_list:
yield data
class CodeTaskPromptGenerator:
def __init__(
self,
num_tasks: int = 50,
) -> None:
self.generate_tasks_prompt = PromptTemplateGenerator(
).get_generate_tasks_prompt(TaskType.CODE)
self.num_tasks = num_tasks
def from_role_files(
self, languages_path: str = "data/code/languages.txt",
domains_path: str = "data/code/domains.txt"
) -> Generator[Tuple[TextPrompt, str, str], None, None]:
language_generator = SingleTxtGenerator(
languages_path).from_role_files()
for language in language_generator:
domains_generator = SingleTxtGenerator(
domains_path).from_role_files()
for domain in domains_generator:
generated_tasks_prompt = self.generate_tasks_prompt.format(
language=language, domain=domain, num_tasks=self.num_tasks)
yield generated_tasks_prompt, language, domain
def from_role_generator(
self, role_generator: Generator[Tuple, None, None]
) -> Generator[str, None, None]:
raise NotImplementedError