add error category (#1455)

# Description


![image](https://github.com/microsoft/promptflow/assets/23182548/a66fd866-0cad-4663-8fc4-1354e9cd86c1)


![image](https://github.com/microsoft/promptflow/assets/23182548/95b00f9f-7426-4085-85b6-426cf22e9ec8)


![image](https://github.com/microsoft/promptflow/assets/23182548/37f1dfae-5aef-4005-b2ed-f6843491fabb)



# All Promptflow Contribution checklist:
- [x] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [ ] **I have read the [contribution guidelines](../CONTRIBUTING.md).**
- [ ] **Create an issue and link to the pull request to get dedicated
review from promptflow team. Learn more: [suggested
workflow](../CONTRIBUTING.md#suggested-workflow).**

## General Guidelines and Best Practices
- [x] Title of the pull request is clear and informative.
- [ ] There are a small number of commits, each of which have an
informative message. This means that previously merged commits do not
appear in the history of the PR. For more information on cleaning up the
commits in your PR, [see this
page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).

### Testing Guidelines
- [x] Pull request includes test coverage for the included changes.

src/promptflow/promptflow/_sdk/_errors.py

@@ -2,70 +2,84 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
 from promptflow._sdk._constants import BULK_RUN_ERRORS
-from promptflow.exceptions import ErrorTarget, PromptflowException
+from promptflow.exceptions import ErrorTarget, UserErrorException
 
 
-class RunExistsError(PromptflowException):
+class SDKError(UserErrorException):
+    """SDK base class, target default is CONTROL_PLANE_SDK."""
+
+    def __init__(
+        self,
+        message="",
+        message_format="",
+        target: ErrorTarget = ErrorTarget.CONTROL_PLANE_SDK,
+        module=None,
+        **kwargs,
+    ):
+        super().__init__(message=message, message_format=message_format, target=target, module=module, **kwargs)
+
+
+class RunExistsError(SDKError):
     """Exception raised when run already exists."""
 
     pass
 
 
-class RunNotFoundError(PromptflowException):
+class RunNotFoundError(SDKError):
     """Exception raised if run cannot be found."""
 
     pass
 
 
-class InvalidRunStatusError(PromptflowException):
+class InvalidRunStatusError(SDKError):
     """Exception raised if run status is invalid."""
 
     pass
 
 
-class UnsecureConnectionError(PromptflowException):
+class UnsecureConnectionError(SDKError):
     """Exception raised if connection is not secure."""
 
     pass
 
 
-class DecryptConnectionError(PromptflowException):
+class DecryptConnectionError(SDKError):
     """Exception raised if connection decryption failed."""
 
     pass
 
 
-class StoreConnectionEncryptionKeyError(PromptflowException):
+class StoreConnectionEncryptionKeyError(SDKError):
     """Exception raised if no keyring backend."""
 
     pass
 
 
-class InvalidFlowError(PromptflowException):
+class InvalidFlowError(SDKError):
     """Exception raised if flow definition is not legal."""
 
     pass
 
 
-class ConnectionNotFoundError(PromptflowException):
+class ConnectionNotFoundError(SDKError):
     """Exception raised if connection is not found."""
 
     pass
 
 
-class InvalidRunError(PromptflowException):
+class InvalidRunError(SDKError):
     """Exception raised if run name is not legal."""
 
     pass
 
 
-class GenerateFlowToolsJsonError(PromptflowException):
+class GenerateFlowToolsJsonError(SDKError):
     """Exception raised if flow tools json generation failed."""
 
     pass
 
 
-class BulkRunException(PromptflowException):
+class BulkRunException(SDKError):
     """Exception raised when bulk run failed."""
 
     def __init__(self, *, message="", failed_lines, total_lines, errors, module: str = None, **kwargs):
@@ -87,19 +101,19 @@ def additional_info(self):
         return self._additional_info
 
 
-class RunOperationParameterError(PromptflowException):
+class RunOperationParameterError(SDKError):
     """Exception raised when list run failed."""
 
     pass
 
 
-class RunOperationError(PromptflowException):
+class RunOperationError(SDKError):
     """Exception raised when run operation failed."""
 
     pass
 
 
-class FlowOperationError(PromptflowException):
+class FlowOperationError(SDKError):
     """Exception raised when flow operation failed."""
 
     pass

src/promptflow/promptflow/_sdk/_telemetry/activity.py

@@ -9,6 +9,7 @@
 from concurrent.futures import ThreadPoolExecutor
 
 from promptflow._sdk._telemetry.telemetry import TelemetryMixin
+from promptflow.exceptions import _ErrorInfo
 from promptflow._sdk._utils import ClientUserAgentUtil
 from promptflow._utils.version_hint_utils import hint_for_update, check_latest_version, HINT_ACTIVITY_NAME
 
@@ -92,6 +93,12 @@ def log_activity(
     except BaseException as e:  # pylint: disable=broad-except
         exception = e
         completion_status = ActivityCompletionStatus.FAILURE
+        error_category, error_type, error_target, error_message, error_detail = _ErrorInfo.get_error_info(exception)
+        activity_info["error_category"] = error_category
+        activity_info["error_type"] = error_type
+        activity_info["error_target"] = error_target
+        activity_info["error_message"] = error_message
+        activity_info["error_detail"] = error_detail
     finally:
         try:
             if first_call:
@@ -177,6 +184,7 @@ def wrapper(self, *args, **kwargs):
                     with ThreadPoolExecutor() as pool:
                         pool.submit(check_latest_version)
                 return f(self, *args, **kwargs)
+
         return wrapper
 
     return monitor

src/promptflow/promptflow/exceptions.py

@@ -1,12 +1,19 @@
 # ---------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
-
+import inspect
 import string
+import traceback
 from enum import Enum
 from functools import cached_property
 
 
+class ErrorCategory(str, Enum):
+    USER_ERROR = "UserError"
+    SYSTEM_ERROR = "SystemError"
+    UNKNOWN = "Unknown"
+
+
 class ErrorTarget(str, Enum):
     """The target of the error, indicates which part of the system the error occurs."""
 
@@ -161,6 +168,9 @@ def error_codes(self):
         i.e. For ToolExcutionError which inherits from UserErrorException,
         The result would be ["UserErrorException", "ToolExecutionError"].
         """
+        if getattr(self, "_error_codes", None):
+            return self._error_codes
+
         from promptflow._utils.exception_utils import infer_error_code_from_class
 
         def reversed_error_codes():
@@ -169,9 +179,9 @@ def reversed_error_codes():
                     break
                 yield infer_error_code_from_class(clz)
 
-        result = list(reversed_error_codes())
-        result.reverse()
-        return result
+        self._error_codes = list(reversed_error_codes())
+        self._error_codes.reverse()
+        return self._error_codes
 
     def get_arguments_from_message_format(self, message_format):
         """Get the arguments from the message format."""
@@ -209,3 +219,113 @@ class ValidationException(UserErrorException):
     """Exception raised when validation fails."""
 
     pass
+
+
+class _ErrorInfo:
+    @classmethod
+    def get_error_info(cls, e: Exception):
+        if not isinstance(e, Exception):
+            return None, None, None, None, None
+
+        e = cls.select_exception(e)
+        if cls._is_system_error(e):
+            return (
+                ErrorCategory.SYSTEM_ERROR,
+                cls._error_type(e),
+                cls._error_target(e),
+                cls._error_message(e),
+                cls._error_detail(e),
+            )
+        if cls._is_user_error(e):
+            return (
+                ErrorCategory.USER_ERROR,
+                cls._error_type(e),
+                cls._error_target(e),
+                cls._error_message(e),
+                cls._error_detail(e),
+            )
+
+        return ErrorCategory.UNKNOWN, cls._error_type(e), ErrorTarget.UNKNOWN, "", cls._error_detail(e)
+
+    @classmethod
+    def select_exception(cls, e: Exception):
+        """Select the exception  in e and e.__cause__, and prioritize the Exception defined in the promptflow."""
+
+        if isinstance(e, PromptflowException):
+            return e
+
+        # raise Exception("message") from PromptflowException("message")
+        if e.__cause__ and isinstance(e.__cause__, PromptflowException):
+            return e.__cause__
+
+        return e
+
+    @classmethod
+    def _is_system_error(cls, e: Exception):
+        if isinstance(e, SystemErrorException):
+            return True
+
+        return False
+
+    @classmethod
+    def _is_user_error(cls, e: Exception):
+        if isinstance(e, UserErrorException):
+            return True
+
+        return False
+
+    @classmethod
+    def _error_type(cls, e: Exception):
+        return type(e).__name__
+
+    @classmethod
+    def _error_target(cls, e: Exception):
+        return getattr(e, "target", ErrorTarget.UNKNOWN)
+
+    @classmethod
+    def _error_message(cls, e: Exception):
+        return getattr(e, "message_format", "")
+
+    @classmethod
+    def _error_detail(cls, e: Exception):
+        exception_codes = cls._get_exception_codes(e)
+        exception_code = None
+        for item in exception_codes[::-1]:
+            if "promptflow" in item["module"]:  # Only record information within the promptflow package
+                exception_code = item
+                break
+        if not exception_code:
+            return ""
+        return (
+            f"module={exception_code['module']}, "
+            f"code={exception_code['exception_code']}, "
+            f"lineno={exception_code['lineno']}."
+        )
+
+    @classmethod
+    def _get_exception_codes(cls, e: Exception) -> list:
+        """
+        Obtain information on each line of the traceback, including the module name,
+        exception code and lineno where the error occurred.
+
+        :param e: Exception object
+        :return: A list, each item contains information for each row of the traceback, which format is like this:
+                {
+                'module': 'promptflow.executor.errors',
+                'exception_code': 'return self.inner_exception.additional_info',
+                'lineno': 223
+                }
+        """
+        exception_codes = []
+        traceback_info = traceback.extract_tb(e.__traceback__)
+        for item in traceback_info:
+            lineno = item.lineno
+            filename = item.filename
+            line_code = item.line
+            module = inspect.getmodule(None, _filename=filename)
+            exception_code = {"module": "", "exception_code": line_code, "lineno": lineno}
+            if module is not None:
+                exception_code["module"] = module.__name__
+            exception_codes.append(exception_code)
+
+        return exception_codes

src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py

@@ -1098,9 +1098,7 @@ def validate_log(log_msg, prefix, expect_dict):
 
         with caplog.at_level(level=logging.INFO, logger=LOGGER_NAME):
             run_pf_command("flow", "test", "--flow", f"{FLOWS_DIR}/web_classification", *extra_args)
-            for (log, expected_input, expected_log_prefix) in zip(
-                caplog.records, expected_inputs, expected_log_prefixes
-            ):
+            for log, expected_input, expected_log_prefix in zip(caplog.records, expected_inputs, expected_log_prefixes):
                 validate_log(
                     prefix=expected_log_prefix,
                     log_msg=log.message,
@@ -1276,7 +1274,6 @@ def test_pf_run_with_stream_log(self, capfd):
             assert keyword not in out
 
     def test_pf_run_no_stream_log(self, capfd):
-
         # without --stream, logs will be in the run's log file
 
         run_pf_command(
@@ -1312,7 +1309,8 @@ def test_format_cli_exception(self, capsys):
             outerr = capsys.readouterr()
             assert outerr.err
             error_msg = json.loads(outerr.err)
-            assert error_msg["code"] == "ConnectionNotFoundError"
+            assert error_msg["code"] == "UserError"
+            assert error_msg["innerError"]["innerError"]["code"] == "ConnectionNotFoundError"
 
             def mocked_connection_get(*args, **kwargs):
                 raise Exception("mock exception")

src/promptflow/tests/sdk_cli_test/e2etests/test_flow_run.py

@@ -86,7 +86,8 @@ def assert_run_with_invalid_column_mapping(client: PFClient, run: Run) -> None:
 
     exception = local_storage.load_exception()
     assert "The input for batch run is incorrect. Couldn't find these mapping relations" in exception["message"]
-    assert exception["code"] == "BulkRunException"
+    assert exception["code"] == "UserError"
+    assert exception["innerError"]["innerError"]["code"] == "BulkRunException"
 
 
 @pytest.mark.usefixtures(
@@ -449,7 +450,6 @@ def test_eval_run_data_deleted(self, pf):
             assert "Please make sure it exists and not deleted." in str(e.value)
 
     def test_eval_run_data_not_exist(self, pf):
-
         base_run = pf.run(
             flow=f"{FLOWS_DIR}/print_env_var",
             data=f"{DATAS_DIR}/env_var_names.jsonl",

src/promptflow/tests/sdk_cli_test/recording_utilities/record_storage.py

@@ -10,8 +10,7 @@
 
 from filelock import FileLock
 
-from promptflow._sdk._errors import PromptflowException
-
+from promptflow.exceptions import PromptflowException
 from .constants import ENVIRON_TEST_MODE, RecordMode