fix(agent): format problem

dbgpt/agent/resource/app.py

@@ -1,20 +1,23 @@
+"""Application Resources for the agent."""
+
 import dataclasses
 import uuid
-from typing import Optional, Tuple, Dict, Type, Any, List, cast
+from typing import Any, Dict, List, Optional, Tuple, Type, cast
 
-from dbgpt.agent import ConversableAgent, AgentMessage, AgentContext
+from dbgpt.agent import AgentMessage, ConversableAgent
 from dbgpt.serve.agent.agents.app_agent_manage import get_app_manager
 from dbgpt.util import ParameterDescription
+
 from .base import Resource, ResourceParameters, ResourceType
 
 
-def get_app_list():
+def _get_app_list():
     apps = get_app_manager().get_dbgpts()
     results = [
         {
             "label": f"{app.app_name}({app.app_code})",
             "key": app.app_code,
-            "description": app.app_describe
+            "description": app.app_describe,
         }
         for app in apps
     ]
@@ -23,20 +26,21 @@ def get_app_list():
 
 @dataclasses.dataclass
 class AppResourceParameters(ResourceParameters):
+    """Application resource class."""
+
     app_code: str = dataclasses.field(
-        default=None,
         metadata={
             "help": "app code",
-            "valid_values": get_app_list(),
+            "valid_values": _get_app_list(),
         },
     )
 
     @classmethod
     def to_configurations(
-            cls,
-            parameters: Type["AppResourceParameters"],
-            version: Optional[str] = None,
-            **kwargs,
+        cls,
+        parameters: Type["AppResourceParameters"],
+        version: Optional[str] = None,
+        **kwargs,
     ) -> Any:
         """Convert the parameters to configurations."""
         conf: List[ParameterDescription] = cast(
@@ -53,7 +57,7 @@ def to_configurations(
 
     @classmethod
     def from_dict(
-            cls, data: dict, ignore_extra_fields: bool = True
+        cls, data: dict, ignore_extra_fields: bool = True
     ) -> ResourceParameters:
         """Create a new instance from a dictionary."""
         copied_data = data.copy()
@@ -66,33 +70,51 @@ class AppResource(Resource[AppResourceParameters]):
     """AppResource resource class."""
 
     def __init__(self, name: str, app_code: str, **kwargs):
+        """Initialize AppResource resource."""
         self._resource_name = name
         self._app_code = app_code
 
         app = get_app_manager().get_app(self._app_code)
         self._app_name = app.app_name
         self._app_desc = app.app_describe
 
+    @property
+    def app_desc(self):
+        """Return the app description."""
+        return self._app_desc
+
+    @property
+    def app_name(self):
+        """Return the app name."""
+        return self._app_name
+
     @classmethod
     def type(cls) -> ResourceType:
+        """Return the resource type."""
         return ResourceType.App
 
     @property
     def name(self) -> str:
+        """Return the resource name."""
         return self._resource_name
 
     @classmethod
     def resource_parameters_class(cls, **kwargs) -> Type[ResourceParameters]:
         """Return the resource parameters class."""
         return AppResourceParameters
 
-    async def get_prompt(self, *, lang: str = "en", prompt_type: str = "default", question: Optional[str] = None,
-                         resource_name: Optional[str] = None, **kwargs) -> Tuple[str, Optional[Dict]]:
+    async def get_prompt(
+        self,
+        *,
+        lang: str = "en",
+        prompt_type: str = "default",
+        question: Optional[str] = None,
+        resource_name: Optional[str] = None,
+        **kwargs,
+    ) -> Tuple[str, Optional[Dict]]:
         """Get the prompt."""
-
         prompt_template_zh = (
-            "{name}：调用此资源与应用 {app_name} 进行交互。"
-            "应用 {app_name} 有什么用？{description}"
+            "{name}：调用此资源与应用 {app_name} 进行交互。" "应用 {app_name} 有什么用？{description}"
         )
         prompt_template_en = (
             "{name}：Call this resource to interact with the application {app_name} ."
@@ -102,9 +124,7 @@ async def get_prompt(self, *, lang: str = "en", prompt_type: str = "default", qu
 
         return (
             template.format(
-                name=self.name,
-                app_name=self._app_name,
-                description=self._app_desc
+                name=self.name, app_name=self._app_name, description=self._app_desc
             ),
             None,
         )
@@ -114,15 +134,16 @@ def is_async(self) -> bool:
         """Return whether the tool is asynchronous."""
         return True
 
-    async def execute(self, *args, resource_name: Optional[str] = None, **kwargs) -> Any:
+    def execute(self, *args, resource_name: Optional[str] = None, **kwargs) -> Any:
+        """Execute the resource."""
         if self.is_async:
-            raise RuntimeError("Async execution is not supported")
+            raise RuntimeError("Sync execution is not supported")
 
     async def async_execute(
-            self,
-            *args,
-            resource_name: Optional[str] = None,
-            **kwargs,
+        self,
+        *args,
+        resource_name: Optional[str] = None,
+        **kwargs,
     ) -> Any:
         """Execute the tool asynchronously.
 
@@ -132,35 +153,41 @@ async def async_execute(
             specific tool).
             **kwargs: The keyword arguments.
         """
+        user_input: Optional[str] = kwargs.get("user_input")
+        parent_agent: Optional[ConversableAgent] = kwargs.get("parent_agent")
 
-        user_input = kwargs.get("user_input")
-        parent_agent = kwargs.get("parent_agent")
+        if user_input is None:
+            raise RuntimeError("AppResource async execution user_input is None")
+        if parent_agent is None:
+            raise RuntimeError("AppResource async execution parent_agent is None")
 
-        reply_message = await self.chat_2_app_once(self._app_code, user_input=user_input, sender=parent_agent)
+        reply_message = await _start_app(self._app_code, user_input, parent_agent)
         return reply_message.content
 
-    async def chat_2_app_once(self,
-                              app_code: str,
-                              user_input: str,
-                              conv_uid: str = None,
-                              sender: ConversableAgent = None) -> AgentMessage:
-        # create a new conv_uid
-        conv_uid = str(uuid.uuid4()) if conv_uid is None else conv_uid
-
-        gpts_app = get_app_manager().get_app(app_code)
-
-        app_agent = await get_app_manager().create_agent_by_app_code(gpts_app, conv_uid=conv_uid)
-
-        agent_message = AgentMessage(
-            content=user_input,
-            current_goal=user_input,
-            context={
-                "conv_uid": conv_uid,
-            },
-            rounds=0,
-        )
 
-        reply_message: AgentMessage = await app_agent.generate_reply(received_message=agent_message,
-                                                                     sender=sender)
+async def _start_app(
+    app_code: str,
+    user_input: str,
+    sender: ConversableAgent,
+    conv_uid: Optional[str] = None,
+) -> AgentMessage:
+    """Start App By AppResource."""
+    conv_uid = str(uuid.uuid4()) if conv_uid is None else conv_uid
+    gpts_app = get_app_manager().get_app(app_code)
+    app_agent = await get_app_manager().create_agent_by_app_code(
+        gpts_app, conv_uid=conv_uid
+    )
+
+    agent_message = AgentMessage(
+        content=user_input,
+        current_goal=user_input,
+        context={
+            "conv_uid": conv_uid,
+        },
+        rounds=0,
+    )
+    reply_message: AgentMessage = await app_agent.generate_reply(
+        received_message=agent_message, sender=sender
+    )
 
-        return reply_message
+    return reply_message

dbgpt/app/component_configs.py

@@ -106,12 +106,12 @@ def _initialize_resource_manager(system_app: SystemApp):
         get_current_host_system_load,
     )
     from dbgpt.agent.expand.resources.search_tool import baidu_search
+    from dbgpt.agent.resource.app import AppResource
     from dbgpt.agent.resource.base import ResourceType
     from dbgpt.agent.resource.manage import get_resource_manager, initialize_resource
     from dbgpt.serve.agent.resource.datasource import DatasourceResource
     from dbgpt.serve.agent.resource.knowledge import KnowledgeSpaceRetrieverResource
     from dbgpt.serve.agent.resource.plugin import PluginToolPack
-    from dbgpt.agent.resource.app import AppResource
 
     initialize_resource(system_app)
     rm = get_resource_manager(system_app)