Merge pull request #29 from hieheihei/feat/add_dify_support

Feat: Add Dify Instrumentor

Author: ralf0131

instrumentation-genai/opentelemetry-instrumentation-dify/README.md

@@ -0,0 +1,10 @@
+# OpenTelemerty Dify Instrumentation
+
+Dify Python Agent provides observability for Dify applications. This document provides examples of usage and results in the Dify instrumentation. For details on usage and installation of LoongSuite and Jaeger, please refer to [LoongSuite Documentation](https://github.com/alibaba/loongsuite-python-agent/blob/main/README.md).
+
+## Installation
+
+```shell
+git clone https://github.com/alibaba/loongsuite-python-agent.git
+pip install ./instrumentation-genai/opentelemetry-instrumentation-dify
+```

instrumentation-genai/opentelemetry-instrumentation-dify/_assets/image/demo.png

@@ -0,0 +1,54 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "opentelemetry-instrumentation-dify"
+dynamic = ["version"]
+description = "OpenTelemetry Dify Instrumentation"
+readme = "README.md"
+license = "Apache-2.0"
+requires-python = ">=3.8, <3.13"
+authors = [
+    { name = "LoongSuite Python Agent Authors", email = "" },
+]
+classifiers = [
+    "Development Status :: 5 - Production/Stable",
+    "Intended Audience :: Developers",
+    "License :: OSI Approved :: Apache Software License",
+    "Programming Language :: Python",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+]
+dependencies = [
+    "wrapt",
+]
+
+[project.optional-dependencies]
+instruments = [
+]
+test = [
+    "pytest",
+    "opentelemetry-sdk",
+]
+type-check = []
+
+
+[tool.hatch.version]
+path = "src/opentelemetry/instrumentation/dify/version.py"
+
+[tool.hatch.build.targets.sdist]
+include = [
+    "src",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/opentelemetry"]
+
+[project.entry-points.opentelemetry_instrumentor]
+dify = "opentelemetry.instrumentation.dify:DifyInstrumentor"

instrumentation-genai/opentelemetry-instrumentation-dify/pyproject.toml

@@ -0,0 +1,36 @@
+import logging
+from typing import Any, Collection
+
+from opentelemetry.instrumentation.dify.package import _instruments
+from opentelemetry.instrumentation.dify.wrapper import set_wrappers
+from opentelemetry import trace as trace_api
+from opentelemetry.instrumentation.instrumentor import BaseInstrumentor  # type: ignore
+
+from opentelemetry.instrumentation.dify.config import is_version_supported, MIN_SUPPORTED_VERSION, MAX_SUPPORTED_VERSION
+
+logger = logging.getLogger(__name__)
+logger.addHandler(logging.NullHandler())
+
+
+class DifyInstrumentor(BaseInstrumentor):  # type: ignore
+    """
+    An instrumentor for Dify
+    """
+
+    def instrumentation_dependencies(self) -> Collection[str]:
+        return _instruments
+
+    def _instrument(self, **kwargs: Any) -> None:
+        if not is_version_supported():
+            logger.warning(
+                f"Dify version is not supported. Current version must be between {MIN_SUPPORTED_VERSION} and {MAX_SUPPORTED_VERSION}."
+            )
+            return
+        if not (tracer_provider := kwargs.get("tracer_provider")):
+            tracer_provider = trace_api.get_tracer_provider()
+        tracer = trace_api.get_tracer(__name__, None, tracer_provider=tracer_provider)
+
+        set_wrappers(tracer)
+
+    def _uninstrument(self, **kwargs: Any) -> None:
+        pass

instrumentation-genai/opentelemetry-instrumentation-dify/src/opentelemetry/instrumentation/dify/__init__.py

@@ -0,0 +1,211 @@
+from typing import Any, Dict
+from abc import ABC
+from opentelemetry.metrics import get_meter
+
+from opentelemetry import trace
+from opentelemetry.trace import Tracer
+
+from opentelemetry.instrumentation.dify.semconv import GEN_AI_USER_ID, GEN_AI_SESSION_ID
+from opentelemetry.instrumentation.dify.utils import get_llm_common_attributes
+from opentelemetry.instrumentation.dify.version import __version__
+from opentelemetry.context import get_value
+from opentelemetry.instrumentation.dify.contants import _get_dify_app_name_key, DIFY_APP_ID_KEY
+
+
+_DIFY_APP_NAME_KEY = _get_dify_app_name_key()
+
+
+class BaseWrapper(ABC):
+    def __init__(self, tracer: Tracer):
+        self.tracer = tracer
+        self._span_kind = "TASK"
+        self._meter = get_meter(
+            __name__,
+            __version__,
+            None,
+            schema_url="https://opentelemetry.io/schemas/1.11.0",
+        )
+        self._app_list: Dict[str, str] = {}
+        self._init_metrics()
+
+    def _init_metrics(self):
+        meter = self._meter
+
+    def set_span_kind(self, span_kind: str):
+        self._span_kind = span_kind
+
+    def span_kind(self):
+        return self._span_kind
+
+    def get_common_attributes(self):
+        attributes = get_llm_common_attributes()
+        attributes["spanKind"] = self.span_kind()
+        return attributes
+
+    def extract_attributes_from_context(self) -> Dict:
+        attributes = {}
+        app_name = get_value(_DIFY_APP_NAME_KEY)
+        app_id = get_value(DIFY_APP_ID_KEY)
+        user_id = get_value(GEN_AI_USER_ID)
+        session_id = get_value(GEN_AI_SESSION_ID)
+        if app_name:
+            attributes[_DIFY_APP_NAME_KEY] = app_name
+        if app_id:
+            attributes[DIFY_APP_ID_KEY] = app_id
+        if user_id:
+            attributes[GEN_AI_USER_ID] = user_id
+        if session_id:
+            attributes[GEN_AI_SESSION_ID] = session_id
+        return attributes
+
+    def before_process(self):
+        pass
+
+    def after_process(self):
+        pass
+
+    def record_call_count(self, attributes: Dict[str, Any] = None, span_kind: str = None):
+        """记录调用次数"""
+        common_attrs = self.get_common_attributes()
+        if span_kind:
+            common_attrs["spanKind"] = span_kind
+        if attributes:
+            common_attrs.update(attributes)
+
+    def record_duration(self, duration: float, attributes: Dict[str, Any] = None, span_kind: str = None):
+        """记录调用持续时间"""
+        common_attrs = self.get_common_attributes()
+        if span_kind:
+            common_attrs["spanKind"] = span_kind
+        if attributes:
+            common_attrs.update(attributes)
+
+    def record_call_error_count(self, attributes: Dict[str, Any] = None, span_kind: str = None):
+        """记录调用错误次数"""
+        common_attrs = self.get_common_attributes()
+        if span_kind:
+            common_attrs["spanKind"] = span_kind
+        if attributes:
+            common_attrs.update(attributes)
+
+class LLMBaseWrapper(BaseWrapper):
+    def __init__(self, tracer: Tracer):
+        super().__init__(tracer)
+
+    def get_trace_headers(self, current_span=None):
+        # Get current context
+        if current_span is None:
+            current_span = trace.get_current_span()
+        if not current_span:
+            # logger.debug("No current span found")
+            return {}
+        current_context = current_span.get_span_context()
+        # Only inject if we have a valid context
+        if current_context and hasattr(current_context, "trace_id") and hasattr(current_context, "span_id"):
+            # Create trace headers
+            trace_headers = {}
+            # Create traceparent header
+            trace_id_hex = format(current_context.trace_id, "032x")
+            span_id_hex = format(current_context.span_id, "016x")
+            flags = format(int(current_context.trace_flags) if hasattr(current_context, "trace_flags") else 1,
+                           "02x")
+            traceparent = f"00-{trace_id_hex}-{span_id_hex}-{flags}"
+            trace_headers["traceparent"] = traceparent
+            # Add tracestate if available
+            if hasattr(current_context, "trace_state") and current_context.trace_state:
+                trace_headers["tracestate"] = str(current_context.trace_state)
+
+            return trace_headers
+
+
+    def record_call_count(self, model_name: str, attributes: Dict[str, Any] = None, span_kind: str = "LLM"):
+        """记录调用次数"""
+        if attributes is None:
+            attributes = {}
+        attributes["modelName"] = model_name
+        super().record_call_count(attributes, span_kind)
+
+    def record_duration(self, duration: float, model_name: str, attributes: Dict[str, Any] = None,
+                        span_kind: str = "LLM"):
+        """记录调用持续时间"""
+        if attributes is None:
+            attributes = {}
+        attributes["modelName"] = model_name
+        super().record_duration(duration, attributes, span_kind)
+
+    def record_call_error_count(self, model_name: str, attributes: Dict[str, Any] = None, span_kind: str = "LLM"):
+        """记录调用错误次数"""
+        if attributes is None:
+            attributes = {}
+        attributes["modelName"] = model_name
+        super().record_call_error_count(attributes, span_kind)
+
+    def record_llm_output_token_seconds(self, duration: float, attributes: Dict[str, Any] = None,
+                                        span_kind: str = "LLM"):
+        """记录LLM输出token的持续时间"""
+        common_attrs = self.get_common_attributes()
+        if span_kind:
+            common_attrs["spanKind"] = span_kind
+        if attributes:
+            common_attrs.update(attributes)
+
+    def record_first_token_seconds(self, duration: float, model_name: str, attributes: Dict[str, Any] = None,
+                                   span_kind: str = "LLM"):
+        """记录首包耗时"""
+        common_attrs = self.get_common_attributes()
+        if span_kind:
+            common_attrs["spanKind"] = span_kind
+        if attributes:
+            common_attrs.update(attributes)
+        common_attrs["modelName"] = model_name
+
+    def _record_llm_tokens(self, tokens: int, usage_type: str, model_name: str, attributes: Dict[str, Any] = None,
+                           span_kind: str = "LLM"):
+        """记录LLM token数量的通用方法"""
+        common_attrs = self.get_common_attributes()
+        if span_kind:
+            common_attrs["spanKind"] = span_kind
+        if attributes:
+            common_attrs.update(attributes)
+        common_attrs["usageType"] = usage_type
+        common_attrs["modelName"] = model_name
+
+    def record_llm_input_tokens(self, tokens: int, model_name: str, attributes: Dict[str, Any] = None,
+                                span_kind: str = "LLM"):
+        """记录LLM输入token的数量"""
+        self._record_llm_tokens(tokens, "input", model_name, attributes, span_kind)
+
+    def record_llm_output_tokens(self, tokens: int, model_name: str, attributes: Dict[str, Any] = None,
+                                 span_kind: str = "LLM"):
+        """记录LLM输出token的数量"""
+        self._record_llm_tokens(tokens, "output", model_name, attributes, span_kind)
+
+
+class TOOLBaseWrapper(BaseWrapper):
+    def __init__(self, tracer: Tracer):
+        super().__init__(tracer)
+
+    def _init_metrics(self):
+        super()._init_metrics()
+
+    def record_call_count(self, tool_name: str, attributes: Dict[str, Any] = None, span_kind: str = "TOOL"):
+        """记录调用次数"""
+        if attributes is None:
+            attributes = {}
+        attributes["rpc"] = tool_name
+        super().record_call_count(attributes, span_kind)
+
+    def record_duration(self, duration: float, tool_name: str, attributes: Dict[str, Any] = None,
+                        span_kind: str = "TOOL"):
+        """记录调用持续时间"""
+        if attributes is None:
+            attributes = {}
+        attributes["rpc"] = tool_name
+        super().record_duration(duration, attributes, span_kind)
+
+    def record_call_error_count(self, tool_name: str, attributes: Dict[str, Any] = None, span_kind: str = "TOOL"):
+        """记录调用错误次数"""
+        if attributes is None:
+            attributes = {}
+        attributes["rpc"] = tool_name
+        super().record_call_error_count(attributes, span_kind)

instrumentation-genai/opentelemetry-instrumentation-dify/src/opentelemetry/instrumentation/dify/_base_wrapper.py

@@ -0,0 +1,61 @@
+from opentelemetry import trace as trace_api
+from opentelemetry.instrumentation.dify.env_utils import is_capture_content_enabled
+from opentelemetry.instrumentation.dify.semconv import INPUT_VALUE, OUTPUT_VALUE
+
+content_key = [
+    INPUT_VALUE,
+    OUTPUT_VALUE,
+    'gen_ai.request.tool_calls',
+    'gen_ai.request.stop_sequences',
+    'tool.parameters',
+    'vector_search.query',
+    'full_text_search.query'
+    ]
+
+content_prefixes_key = [
+    "gen_ai.prompts",
+    "gen_ai.completions",
+    "retrieval.documents",
+    "vector_search.document",
+    "embedding.embeddings",
+    "reranker.input_documents",
+    "reranker.output_documents",
+    "reranker.query",
+    ]
+
+max_content_length = 4 * 1024
+
+def set_dict_value(attr:dict, key:str, value:str) -> None:
+    if is_capture_content_enabled():
+        attr[key] = value
+    elif not is_content_key(key):
+        attr[key] = value
+    else:
+        attr[key] = to_size(value)
+
+def set_span_value(span: trace_api.Span, key:str, value:str) -> None:
+    if is_capture_content_enabled():
+        span.set_attribute(key, value)
+    elif not is_content_key(key):
+        span.set_attribute(key, value)
+    else:
+        span.set_attribute(key, to_size(value))
+
+def is_content_key(key:str) -> bool:
+    return (key in content_key) or any(key.startswith(prefix) for prefix in content_prefixes_key)
+
+def process_content(content: str | None) -> str:
+    if is_capture_content_enabled():
+        if content is not None and len(content) > max_content_length:
+            content = content[:max_content_length] + "..."
+        return content
+    elif content is None:
+        return "<0size>"
+    else:
+        return to_size(content)
+
+def to_size(content:str) -> str:
+    if content is None:
+        return "<0size>"
+    size = len(content)
+    return f"<{size}size>"