[Feature] Enable Tracing Mechanism (Phase 1)

lazyllm/common/globals.py

@@ -127,8 +127,16 @@ def __reduce__(self):
 
 
 class Globals(metaclass=SingletonABCMeta):
-    __global_attrs__ = ThreadSafeDict(user_id=None, chat_history={}, global_parameters={},
-                                      lazyllm_files={}, usage={}, config={}, call_stack=[])
+    __global_attrs__ = ThreadSafeDict(
+        user_id=None,
+        chat_history={},
+        global_parameters={},
+        lazyllm_files={},
+        usage={},
+        trace={},
+        config={},
+        call_stack=[]
+    )
 
     def __new__(cls, *args, **kw):
         if cls is not Globals: return super().__new__(cls)

lazyllm/configs.py

@@ -244,8 +244,8 @@ def prefix(self):
         return self._config._prefix
 
 
-config = _NamespaceConfig().add('mode', Mode, Mode.Normal, dict(DISPLAY=Mode.Display, DEBUG=Mode.Debug),
-                                description='The default mode for LazyLLM.'
+config = (_NamespaceConfig().add('mode', Mode, Mode.Normal, dict(DISPLAY=Mode.Display, DEBUG=Mode.Debug),
+                                 description='The default mode for LazyLLM.'
         ).add('repr_ml', bool, False, 'REPR_USE_ML', description='Whether to use Markup Language for repr.'
         ).add('repr_show_child', bool, False, 'REPR_SHOW_CHILD',
               description='Whether to show child modules in repr.'
@@ -263,7 +263,8 @@ def prefix(self):
               description='Whether to skip check keywords for deployment.'
         ).add('allow_internal_network', bool, False, 'ALLOW_INTERNAL_NETWORK',
               description='Whether to allow loading images from internal network addresses. '
-                          'Set to False for security in production environments.')
+                          'Set to False for security in production environments.'
+        ))
 
 def refresh_config(key):
     if key in Config:

lazyllm/docs/__init__.py

@@ -3,8 +3,8 @@
 
 config.add('init_doc', bool, False, 'INIT_DOC', description='whether to init docs')
 if config['init_doc'] and (add_doc.__doc__ is None or 'Add document' not in add_doc.__doc__):
-    from . import common, components, configs, flow, hook, launcher, module, patch, prompt_template, tools, utils  # noqa F401
-    del common, components, configs, flow, hook, launcher, module, patch, prompt_template, tools, utils
+    from . import common, components, configs, flow, hook, launcher, module, patch, prompt_template, tools, tracing, utils  # noqa F401
+    del common, components, configs, flow, hook, launcher, module, patch, prompt_template, tools, tracing, utils
 
 __all__ = [
     'add_doc'

lazyllm/docs/hook.py

@@ -85,6 +85,25 @@
     output: The return value of the monitored function.
 ''')
 
+add_chinese_doc('LazyLLMHook.on_error', '''\
+异常处理钩子，在被监控函数执行抛出异常时调用。
+
+默认实现为空操作，子类可以按需覆盖，用于记录错误状态、补充诊断信息或执行清理逻辑。
+
+Args:
+    exc: 被监控函数抛出的异常对象。
+''')
+
+add_english_doc('LazyLLMHook.on_error', '''\
+Error-handling hook, called when the monitored function raises an exception.
+
+The default implementation is a no-op. Subclasses can override it to record error status,
+attach diagnostic information, or perform cleanup.
+
+Args:
+    exc: The exception raised by the monitored function.
+''')
+
 add_chinese_doc('LazyLLMHook.report', '''\
 生成钩子的执行报告。
 
@@ -96,3 +115,57 @@
 
 This is an abstract method and must be implemented in subclasses.
 ''')
+
+add_chinese_doc('HookPhaseError', '''\
+Hook 阶段错误，当一个 hook 阶段中有一个或多个 strict 模式的 hook 执行失败时抛出。
+
+Args:
+    phase (str): 发生错误的 hook 阶段名称（如 ``'post_hook'``、``'on_error'``、``'report'``）。
+    errors: 包含 ``(hook_obj, exception)`` 元组的序列，记录所有失败的 hook 及其异常。
+''')
+
+add_english_doc('HookPhaseError', '''\
+Raised when one or more strict-mode hooks fail during a hook phase.
+
+Args:
+    phase (str): The name of the hook phase where the error(s) occurred (e.g. ``'post_hook'``, ``'on_error'``, ``'report'``).
+    errors: A sequence of ``(hook_obj, exception)`` tuples recording each failed hook and its exception.
+''')
+
+add_chinese_doc('LazyTracingHook', '''\
+为 flow 或 module 创建 tracing hook。
+
+该 hook 会在执行生命周期中创建、更新并结束对应的 tracing span。
+
+Args:
+    obj: 要进行 tracing 的 flow 或 module 对象。
+''')
+
+add_english_doc('LazyTracingHook', '''\
+Create a tracing hook for a flow or module object.
+
+This hook is responsible for creating, updating, and finishing the corresponding tracing span
+during the execution lifecycle.
+
+Args:
+    obj: The flow or module object to be traced.
+''')
+
+add_chinese_doc('LazyTracingHook.on_error', '''\
+在 tracing span 上记录异常状态。
+
+当被包裹的 flow 或 module 执行失败时，该方法会把异常信息写入当前 span。
+
+Args:
+    exc: 执行过程中抛出的异常对象。
+''')
+
+add_english_doc('LazyTracingHook.on_error', '''\
+Record the error state on the active tracing span.
+
+When the wrapped flow or module execution fails, this method writes the exception information
+to the current span.
+
+Args:
+    exc: The exception raised during execution.
+''')

lazyllm/docs/tracing.py

@@ -0,0 +1,180 @@
+# flake8: noqa E501
+from . import utils
+import functools
+import lazyllm
+
+# ============= TracingBackend
+
+add_chinese_doc = functools.partial(utils.add_chinese_doc, module=lazyllm.tracing.backends.base)
+add_english_doc = functools.partial(utils.add_english_doc, module=lazyllm.tracing.backends.base)
+
+add_chinese_doc('TracingBackend', '''\
+追踪后端的抽象基类，定义了将 LazyLLM 追踪数据导出至外部可观测平台所需的接口。
+
+子类需要实现所有抽象方法，以适配具体的可观测后端（如 Langfuse、Jaeger 等）。
+
+**注意**: 此类是抽象基类，不能直接实例化。
+''')
+
+add_english_doc('TracingBackend', '''\
+Abstract base class for tracing backends, defining the interface required to export
+LazyLLM tracing data to external observability platforms.
+
+Subclasses must implement all abstract methods to adapt to a specific observability
+backend (e.g. Langfuse, Jaeger, etc.).
+
+**Note**: This class is an abstract base class and cannot be instantiated directly.
+''')
+
+add_chinese_doc('TracingBackend.build_exporter', '''\
+构建并返回一个 OpenTelemetry SpanExporter 实例，用于将 Span 数据发送至目标后端。
+
+Returns:
+    opentelemetry.sdk.trace.export.SpanExporter: 配置好的 Span 导出器。
+
+Raises:
+    RuntimeError: 当必要的后端配置缺失时抛出。
+''')
+
+add_english_doc('TracingBackend.build_exporter', '''\
+Build and return an OpenTelemetry SpanExporter instance for sending span data to
+the target backend.
+
+Returns:
+    opentelemetry.sdk.trace.export.SpanExporter: A configured span exporter.
+
+Raises:
+    RuntimeError: If required backend configuration is missing.
+''')
+
+add_chinese_doc('TracingBackend.context_attributes', '''\
+将追踪上下文转换为后端特定的 Span 属性。
+
+Args:
+    trace_ctx (Dict[str, Any]): 当前请求的追踪上下文，包含 ``session_id``、``user_id``、``request_tags`` 等字段。
+    is_root_span (bool): 当前 Span 是否为调用链的根 Span。
+
+Returns:
+    Dict[str, Any]: 后端特定的 Span 属性字典。
+''')
+
+add_english_doc('TracingBackend.context_attributes', '''\
+Convert the trace context into backend-specific span attributes.
+
+Args:
+    trace_ctx (Dict[str, Any]): The current request trace context containing fields such as ``session_id``, ``user_id``, and ``request_tags``.
+    is_root_span (bool): Whether the current span is the root span of the trace.
+
+Returns:
+    Dict[str, Any]: A dictionary of backend-specific span attributes.
+''')
+
+add_chinese_doc('TracingBackend.input_attributes', '''\
+将调用输入转换为后端特定的 Span 属性。
+
+Args:
+    args (tuple): 传递给目标对象的位置参数。
+    kwargs (Dict[str, Any]): 传递给目标对象的关键字参数。
+    capture_payload (bool): 是否记录输入 payload 内容。
+    is_root_span (bool): 当前 Span 是否为调用链的根 Span。
+
+Returns:
+    Dict[str, Any]: 后端特定的输入属性字典。若 ``capture_payload`` 为 False，返回空字典。
+''')
+
+add_english_doc('TracingBackend.input_attributes', '''\
+Convert call inputs into backend-specific span attributes.
+
+Args:
+    args (tuple): Positional arguments passed to the target object.
+    kwargs (Dict[str, Any]): Keyword arguments passed to the target object.
+    capture_payload (bool): Whether to record the input payload content.
+    is_root_span (bool): Whether the current span is the root span of the trace.
+
+Returns:
+    Dict[str, Any]: A dictionary of backend-specific input attributes. Returns an empty dict when ``capture_payload`` is False.
+''')
+
+add_chinese_doc('TracingBackend.set_root_span_name', '''\
+为根 Span 设置后端特定的显示名称。
+
+Args:
+    span: OpenTelemetry Span 对象。
+    span_name (str): 要设置的 Span 名称。
+''')
+
+add_english_doc('TracingBackend.set_root_span_name', '''\
+Set a backend-specific display name on the root span.
+
+Args:
+    span: The OpenTelemetry span object.
+    span_name (str): The name to assign to the span.
+''')
+
+add_chinese_doc('TracingBackend.output_attributes', '''\
+将调用输出转换为后端特定的 Span 属性。
+
+Args:
+    text (str): 序列化后的输出文本。
+    is_root_span (bool): 当前 Span 是否为调用链的根 Span。
+
+Returns:
+    Dict[str, Any]: 后端特定的输出属性字典。
+''')
+
+add_english_doc('TracingBackend.output_attributes', '''\
+Convert call output into backend-specific span attributes.
+
+Args:
+    text (str): The serialized output text.
+    is_root_span (bool): Whether the current span is the root span of the trace.
+
+Returns:
+    Dict[str, Any]: A dictionary of backend-specific output attributes.
+''')
+
+add_chinese_doc('TracingBackend.error_attributes', '''\
+将异常信息转换为后端特定的 Span 属性。
+
+Args:
+    exc (Exception): 执行过程中抛出的异常对象。
+
+Returns:
+    Dict[str, Any]: 后端特定的错误属性字典。
+''')
+
+add_english_doc('TracingBackend.error_attributes', '''\
+Convert exception information into backend-specific span attributes.
+
+Args:
+    exc (Exception): The exception raised during execution.
+
+Returns:
+    Dict[str, Any]: A dictionary of backend-specific error attributes.
+''')
+
+# ============= LangfuseBackend
+
+add_chinese_doc_lf = functools.partial(utils.add_chinese_doc, module=lazyllm.tracing.backends.langfuse)
+add_english_doc_lf = functools.partial(utils.add_english_doc, module=lazyllm.tracing.backends.langfuse)
+
+add_chinese_doc_lf('LangfuseBackend', '''\
+Langfuse 追踪后端实现，通过 OTLP/HTTP 协议将追踪数据导出至 Langfuse 平台。
+
+使用 HTTP Basic Auth 认证，需要配置以下环境变量：
+
+- ``LANGFUSE_HOST`` 或 ``LANGFUSE_BASE_URL``: Langfuse 服务地址
+- ``LANGFUSE_PUBLIC_KEY``: Langfuse 公钥
+- ``LANGFUSE_SECRET_KEY``: Langfuse 密钥
+''')
+
+add_english_doc_lf('LangfuseBackend', '''\
+Langfuse tracing backend implementation that exports trace data to the Langfuse platform
+via OTLP/HTTP protocol.
+
+Uses HTTP Basic Auth authentication. The following environment variables must be configured:
+
+- ``LANGFUSE_HOST`` or ``LANGFUSE_BASE_URL``: Langfuse service URL
+- ``LANGFUSE_PUBLIC_KEY``: Langfuse public key
+- ``LANGFUSE_SECRET_KEY``: Langfuse secret key
+''')

lazyllm/flow/flow.py

@@ -21,7 +21,7 @@
 import concurrent.futures
 from collections import deque
 import uuid
-from ..hook import LazyLLMHook
+from ..hook import LazyLLMHook, prepare_hooks, register_hooks, resolve_default_hooks, run_hooks
 from itertools import repeat
 
 
@@ -157,6 +157,8 @@ def _defined_at_the_same_scope(self, other: 'FlowBase'):
 
     def __setattr__(self, name: str, value):
         if '_capture' in self.__dict__ and self._capture and not name.startswith('_'):
+            if hasattr(value, '_module_id') and hasattr(value, 'name') and not value.name:
+                value.name = name
             if len(_get_flow_stack()) > 1 and not self._auto_registered:
                 super(__class__, self).__setattr__('_auto_registered', True)
                 locals = self._curr_frame.f_locals.copy()
@@ -224,36 +226,36 @@ def __init__(self, *args, post_action=None, auto_capture=False, id: Optional[str
                                         id=id, name=name, group_id=group_id)
         self.post_action = post_action() if isinstance(post_action, type) else post_action
         self._sync = False
-        self._hooks = set()
+        self._hooks = []
+        register_hooks(self, resolve_default_hooks(self))
 
     def __call__(self, *args, **kw):
-        hook_objs = []
-        for hook_type in self._hooks:
-            if isinstance(hook_type, LazyLLMHook):
-                hook_objs.append(hook_type)
-            else:
-                hook_objs.append(hook_type(self))
-            hook_objs[-1].pre_hook(*args, **kw)
-        with globals.stack_enter(self.identities):
-            output = self._run(args[0] if len(args) == 1 else package(args), **kw)
-        if self.post_action is not None: self.invoke(self.post_action, output)
-        if self._sync: self.wait()
-        r = self._post_process(output)
-        for hook_obj in hook_objs[::-1]:
-            hook_obj.post_hook(r)
-        for hook_obj in hook_objs:
-            hook_obj.report()
-        return r
+        hook_objs = prepare_hooks(self, self._hooks, *args, **kw)
+
+        try:
+            with globals.stack_enter(self.identities):
+                output = self._run(args[0] if len(args) == 1 else package(args), **kw)
+            if self.post_action is not None: self.invoke(self.post_action, output)
+            if self._sync: self.wait()
+            r = self._post_process(output)
+            run_hooks(hook_objs, 'post_hook', r)
+            return r
+        except Exception as e:
+            LOG.error(f'Flow `{self.__class__.__name__}` raised {type(e).__name__}: {e}')
+            run_hooks(hook_objs, 'on_error', e)
+            raise
+        finally:
+            run_hooks(hook_objs, 'report')
 
     def register_hook(self, hook_type: LazyLLMHook):
-        self._hooks.add(hook_type)
+        self._hooks.append(hook_type)
 
     def unregister_hook(self, hook_type: LazyLLMHook):
         if hook_type in self._hooks:
             self._hooks.remove(hook_type)
 
     def clear_hooks(self):
-        self._hooks = set()
+        self._hooks = []
 
     def _post_process(self, output):
         return output