refact: extract the core logic into core subfolder

Signed-off-by: Lin Yi <[email protected]>

Author: WannaTen

metagpt/core/__init__.py

@@ -0,0 +1,5 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Time    : 2023/4/24 22:26
+# @Author  : alexanderwu
+# @File    : __init__.py

metagpt/core/actions/__init__.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+@Time    : 2023/5/11 17:44
+@Author  : alexanderwu
+@File    : __init__.py
+"""
+
+from metagpt.core.actions.base import Action
+from metagpt.core.actions.action_output import ActionOutput
+
+__all__ = [
+    "Action",
+    "ActionOutput",
+]

metagpt/core/actions/action_output.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python
+# coding: utf-8
+"""
+@Time    : 2023/7/11 10:03
+@Author  : chengmaoyu
+@File    : action_output
+"""
+
+from pydantic import BaseModel
+
+
+class ActionOutput:
+    content: str
+    instruct_content: BaseModel
+
+    def __init__(self, content: str, instruct_content: BaseModel):
+        self.content = content
+        self.instruct_content = instruct_content

metagpt/core/actions/base.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+@Time    : 2023/5/11 14:43
+@Author  : alexanderwu
+@File    : action.py
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class BaseAction(BaseModel):
+    def __str__(self):
+        return self.__class__.__name__
+
+    def __repr__(self):
+        return self.__str__()
+
+    async def _aask(self, prompt: str, system_msgs: Optional[list[str]] = None) -> str:
+        """Append default prefix"""
+        return await self.llm.aask(prompt, system_msgs)
+
+    async def run(self, *args, **kwargs):
+        """Run action"""
+        raise NotImplementedError("The run method should be implemented in a subclass.")

metagpt/core/base/__init__.py

@@ -0,0 +1,8 @@
+from metagpt.core.base.base_env import BaseEnvironment
+from metagpt.core.base.base_role import BaseRole
+
+
+__all__ = [
+    "BaseEnvironment",
+    "BaseRole",
+]

metagpt/core/base/base_env.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   : base environment
+
+import typing
+from abc import abstractmethod
+from typing import Any, Optional
+
+from metagpt.base.base_env_space import BaseEnvAction, BaseEnvObsParams
+from metagpt.base.base_serialization import BaseSerialization
+
+if typing.TYPE_CHECKING:
+    from metagpt.schema import Message
+
+
+class BaseEnvironment(BaseSerialization):
+    """Base environment"""
+
+    @abstractmethod
+    def reset(
+        self,
+        *,
+        seed: Optional[int] = None,
+        options: Optional[dict[str, Any]] = None,
+    ) -> tuple[dict[str, Any], dict[str, Any]]:
+        """Implement this to get init observation"""
+
+    @abstractmethod
+    def observe(self, obs_params: Optional[BaseEnvObsParams] = None) -> Any:
+        """Implement this if you want to get partial observation from the env"""
+
+    @abstractmethod
+    def step(self, action: BaseEnvAction) -> tuple[dict[str, Any], float, bool, bool, dict[str, Any]]:
+        """Implement this to feed a action and then get new observation from the env"""
+
+    @abstractmethod
+    def publish_message(self, message: "Message", peekable: bool = True) -> bool:
+        """Distribute the message to the recipients."""
+
+    @abstractmethod
+    async def run(self, k=1):
+        """Process all task at once"""

metagpt/core/base/base_env_space.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   :
+
+from enum import IntEnum
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class BaseEnvActionType(IntEnum):
+    # # NONE = 0  # no action to run, just get observation
+    pass
+
+
+class BaseEnvAction(BaseModel):
+    """env action type and its related params of action functions/apis"""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    action_type: int = Field(default=0, description="action type")
+
+
+class BaseEnvObsType(IntEnum):
+    # # NONE = 0                     # get whole observation from env
+    pass
+
+
+class BaseEnvObsParams(BaseModel):
+    """observation params for different EnvObsType to get its observe result"""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    obs_type: int = Field(default=0, description="observation type")

metagpt/core/base/base_serialization.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, model_serializer, model_validator
+
+
+class BaseSerialization(BaseModel, extra="forbid"):
+    """
+    PolyMorphic subclasses Serialization / Deserialization Mixin
+    - First of all, we need to know that pydantic is not designed for polymorphism.
+    - If Engineer is subclass of Role, it would be serialized as Role. If we want to serialize it as Engineer, we need
+        to add `class name` to Engineer. So we need Engineer inherit SerializationMixin.
+
+    More details:
+    - https://docs.pydantic.dev/latest/concepts/serialization/
+    - https://github.com/pydantic/pydantic/discussions/7008 discuss about avoid `__get_pydantic_core_schema__`
+    """
+
+    __is_polymorphic_base = False
+    __subclasses_map__ = {}
+
+    @model_serializer(mode="wrap")
+    def __serialize_with_class_type__(self, default_serializer) -> Any:
+        # default serializer, then append the `__module_class_name` field and return
+        ret = default_serializer(self)
+        ret["__module_class_name"] = f"{self.__class__.__module__}.{self.__class__.__qualname__}"
+        return ret
+
+    @model_validator(mode="wrap")
+    @classmethod
+    def __convert_to_real_type__(cls, value: Any, handler):
+        if isinstance(value, dict) is False:
+            return handler(value)
+
+        # it is a dict so make sure to remove the __module_class_name
+        # because we don't allow extra keywords but want to ensure
+        # e.g Cat.model_validate(cat.model_dump()) works
+        class_full_name = value.pop("__module_class_name", None)
+
+        # if it's not the polymorphic base we construct via default handler
+        if not cls.__is_polymorphic_base:
+            if class_full_name is None:
+                return handler(value)
+            elif str(cls) == f"<class '{class_full_name}'>":
+                return handler(value)
+            else:
+                # f"Trying to instantiate {class_full_name} but this is not the polymorphic base class")
+                pass
+
+        # otherwise we lookup the correct polymorphic type and construct that
+        # instead
+        if class_full_name is None:
+            raise ValueError("Missing __module_class_name field")
+
+        class_type = cls.__subclasses_map__.get(class_full_name, None)
+
+        if class_type is None:
+            # TODO could try dynamic import
+            raise TypeError(f"Trying to instantiate {class_full_name}, which has not yet been defined!")
+
+        return class_type(**value)
+
+    def __init_subclass__(cls, is_polymorphic_base: bool = False, **kwargs):
+        cls.__is_polymorphic_base = is_polymorphic_base
+        cls.__subclasses_map__[f"{cls.__module__}.{cls.__qualname__}"] = cls
+        super().__init_subclass__(**kwargs)

metagpt/core/config.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+@Time    : 2024/1/4 01:25
+@Author  : alexanderwu
+@File    : config2.py
+"""
+import os
+from pathlib import Path
+from typing import Dict, Iterable, List, Literal, Optional
+
+from pydantic import BaseModel, Field, model_validator
+
+from metagpt.configs.embedding_config import EmbeddingConfig
+from metagpt.core.configs.llm_config import LLMConfig, LLMType
+from metagpt.configs.omniparse_config import OmniParseConfig
+from metagpt.configs.role_custom_config import RoleCustomConfig
+from metagpt.configs.workspace_config import WorkspaceConfig
+from metagpt.const import CONFIG_ROOT, METAGPT_ROOT
+from metagpt.core.utils.yaml_model import YamlModel
+
+
+class CLIParams(BaseModel):
+    """CLI parameters"""
+
+    project_path: str = ""
+    project_name: str = ""
+    inc: bool = False
+    reqa_file: str = ""
+    max_auto_summarize_code: int = 0
+    git_reinit: bool = False
+
+    @model_validator(mode="after")
+    def check_project_path(self):
+        """Check project_path and project_name"""
+        if self.project_path:
+            self.inc = True
+            self.project_name = self.project_name or Path(self.project_path).name
+        return self
+
+
+class CoreConfig(CLIParams, YamlModel):
+    """Configurations for MetaGPT"""
+
+    # Key Parameters
+    llm: LLMConfig
+
+    # Global Proxy. Will be used if llm.proxy is not set
+    proxy: str = ""
+
+    # Misc Parameters
+    repair_llm_output: bool = False
+    prompt_schema: Literal["json", "markdown", "raw"] = "json"
+    workspace: WorkspaceConfig = Field(default_factory=WorkspaceConfig)
+
+    @classmethod
+    def from_home(cls, path):
+        """Load config from ~/.metagpt/config2.yaml"""
+        pathname = CONFIG_ROOT / path
+        if not pathname.exists():
+            return None
+        return CoreConfig.from_yaml_file(pathname)
+
+    @classmethod
+    def default(cls, reload: bool = False, **kwargs) -> "CoreConfig":
+        """Load default config
+        - Priority: env < default_config_paths
+        - Inside default_config_paths, the latter one overwrites the former one
+        """
+        default_config_paths = (
+            METAGPT_ROOT / "config/config2.yaml",
+            CONFIG_ROOT / "config2.yaml",
+        )
+        if reload or default_config_paths not in _CONFIG_CACHE:
+            dicts = [dict(os.environ), *(CoreConfig.read_yaml(path) for path in default_config_paths), kwargs]
+            final = merge_dict(dicts)
+            _CONFIG_CACHE[default_config_paths] = CoreConfig(**final)
+        return _CONFIG_CACHE[default_config_paths]
+
+    @classmethod
+    def from_llm_config(cls, llm_config: dict):
+        """user config llm
+        example:
+        llm_config = {"api_type": "xxx", "api_key": "xxx", "model": "xxx"}
+        gpt4 = CoreConfig.from_llm_config(llm_config)
+        A = Role(name="A", profile="Democratic candidate", goal="Win the election", actions=[a1], watch=[a2], config=gpt4)
+        """
+        llm_config = LLMConfig.model_validate(llm_config)
+        dicts = [dict(os.environ)]
+        dicts += [{"llm": llm_config}]
+        final = merge_dict(dicts)
+        return CoreConfig(**final)
+
+    def update_via_cli(self, project_path, project_name, inc, reqa_file, max_auto_summarize_code):
+        """update config via cli"""
+
+        # Use in the PrepareDocuments action according to Section 2.2.3.5.1 of RFC 135.
+        if project_path:
+            inc = True
+            project_name = project_name or Path(project_path).name
+        self.project_path = project_path
+        self.project_name = project_name
+        self.inc = inc
+        self.reqa_file = reqa_file
+        self.max_auto_summarize_code = max_auto_summarize_code
+
+
+def merge_dict(dicts: Iterable[Dict]) -> Dict:
+    """Merge multiple dicts into one, with the latter dict overwriting the former"""
+    result = {}
+    for dictionary in dicts:
+        result.update(dictionary)
+    return result
+
+
+_CONFIG_CACHE = {}

metagpt/core/configs/__init__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+@Time    : 2024/1/4 16:33
+@Author  : alexanderwu
+@File    : __init__.py
+"""