add action_outcls decorator to support init same class with same class name and fields

metagpt/actions/action_node.py

@@ -15,6 +15,7 @@ from typing import Any, Dict, List, Optional, Tuple, Type, Union
 from pydantic import BaseModel, create_model, model_validator
 from tenacity import retry, stop_after_attempt, wait_random_exponential
 
+from metagpt.actions.action_outcls_registry import register_action_outcls
 from metagpt.llm import BaseLLM
 from metagpt.logs import logger
 from metagpt.provider.postprocess.llm_output_postprocess import llm_output_postprocess
@@ -201,6 +202,7 @@ class ActionNode:
         return {} if exclude and self.key in exclude else self.get_self_mapping()
 
     @classmethod
+    @register_action_outcls
     def create_model_class(cls, class_name: str, mapping: Dict[str, Tuple[Type, Any]]):
         """基于pydantic v1的模型动态生成，用来检验结果类型正确性"""
 

metagpt/actions/action_outcls_registry.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   : registry to store Dynamic Model from ActionNode.create_model_class to keep it as same Class
+#           with same class name and mapping
+
+from functools import wraps
+
+
+action_outcls_registry = dict()
+
+
+def register_action_outcls(func):
+    """
+    Due to `create_model` return different Class even they have same class name and mapping.
+    In order to do a comparison, use outcls_id to identify same Class with same class name and field definition
+    """
+    @wraps(func)
+    def decorater(*args, **kwargs):
+        """
+        arr example
+            [<class 'metagpt.actions.action_node.ActionNode'>, 'test', {'field': (str, Ellipsis)}]
+        """
+        arr = list(args) + list(kwargs.values())
+        """
+        outcls_id example
+            "<class 'metagpt.actions.action_node.ActionNode'>_test_{'field': (str, Ellipsis)}"
+        """
+        for idx, item in enumerate(arr):
+            if isinstance(item, dict):
+                arr[idx] = dict(sorted(item.items()))
+        outcls_id = "_".join([str(i) for i in arr])
+        # eliminate typing influence
+        outcls_id = outcls_id.replace("typing.List", "list").replace("typing.Dict", "dict")
+
+        if outcls_id in action_outcls_registry:
+            return action_outcls_registry[outcls_id]
+
+        out_cls = func(*args, **kwargs)
+        action_outcls_registry[outcls_id] = out_cls
+        return out_cls
+
+    return decorater