fix circular dependency for role/env

metagpt/base/base_env.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   : base environment
+
+from abc import abstractmethod
+from typing import Any, Optional
+
+from metagpt.base.base_env_space import BaseEnvAction, BaseEnvObsParams
+from metagpt.schema import Message
+
+
+class BaseEnvironment:
+    """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/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/base/base_role.py

@@ -0,0 +1,33 @@
+from abc import abstractmethod
+from typing import Optional, Union
+
+from metagpt.schema import Message
+
+
+class BaseRole:
+    """Abstract base class for all roles."""
+
+    name: str
+    is_idle: bool
+
+    @abstractmethod
+    def think(self):
+        """Consider what to do and decide on the next course of action."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def act(self):
+        """Perform the current action."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def react(self) -> Message:
+        """Entry to one of three strategies by which Role reacts to the observed Message."""
+
+    @abstractmethod
+    async def run(self, with_message: Optional[Union[str, Message, list[str]]] = None) -> Optional[Message]:
+        """Observe, and think and act based on the results of the observation."""
+
+    @abstractmethod
+    def get_memories(self, k: int = 0) -> list[Message]:
+        """Return the most recent k memories of this role."""