Skip to content

honeyhive.tracer.instrumentation.decorators

Decorators for HoneyHive tracing.

This module provides decorators for adding tracing capabilities to functions and classes. Uses dynamic logic and reflection to minimize complexity for graceful degradation.

The main :func:trace decorator automatically detects function type or asynchronous and applies the appropriate wrapper. This eliminates the need for separate sync/async decorators in most cases.

Key Features
  • Unified :func:trace decorator that auto-detects sync/async functions
  • Dynamic attribute management using reflection and mapping
  • Graceful degradation when no tracer is available
  • Class-level tracing with :func:trace_class
  • Comprehensive span enrichment with OpenTelemetry integration
Example

Basic usage with auto-detection::

from honeyhive.tracer.decorators import trace

@trace(event_type="model", event_name="gpt_call")
def sync_function(prompt: str) -> str:
    return "response"

@trace(event_type="model", event_name="async_gpt_call")
async def async_function(prompt: str) -> str:
    return "async response"

Class-level tracing::

@trace_class
class MyService:
    def process_data(self, data):
        return data.upper()
Note

This module follows graceful degradation standards. If no tracer is available, functions execute normally without tracing rather than raising exceptions.

See Also

:mod:honeyhive.tracer.core: Core tracer implementation :mod:honeyhive.tracer.enrichment_core: Span enrichment functionality

T module-attribute

T = TypeVar('T')

P module-attribute

P = TypeVar('P')

BASIC_ATTRIBUTES module-attribute

BASIC_ATTRIBUTES = {
    "event_type": "honeyhive_event_type",
    "event_name": "honeyhive_event_name",
    "event_id": "honeyhive_event_id",
    "source": "honeyhive_source",
    "project": "honeyhive_project",
    "session_id": "honeyhive_session_id",
    "user_id": "honeyhive_user_id",
    "session_name": "honeyhive_session_name",
}

COMPLEX_ATTRIBUTES module-attribute

COMPLEX_ATTRIBUTES = {
    "inputs": "honeyhive_inputs",
    "config": "honeyhive_config",
    "metadata": "honeyhive_metadata",
    "metrics": "honeyhive_metrics",
    "feedback": "honeyhive_feedback",
    "outputs": "honeyhive_outputs",
}

trace

trace(
    event_type: Optional[str] = None,
    event_name: Optional[str] = None,
    **kwargs: Any
) -> Union[
    Callable[[Callable[..., T]], Callable[..., T]],
    Callable[..., T],
]

Unified trace decorator that auto-detects sync/async functions.

Automatically detects whether the decorated function is synchronous or asynchronous and applies the appropriate wrapper. This decorator can be used on both sync and async functions without needing separate decorators.

Parameters:

Name Type Description Default
event_type Optional[str]

Type of event being traced (e.g., "model", "tool", "chain")

None
event_name Optional[str]

Name of the event (defaults to function name)

None
**kwargs Any

Additional tracing parameters (source, project, session_id, etc.)

{}

Returns:

Type Description
Union[Callable[[Callable[..., T]], Callable[..., T]], Callable[..., T]]

Decorated function with tracing capabilities

Example

@trace(event_type="model", event_name="gpt_call") ... def sync_function(): ... return "result"

@trace(event_type="model", event_name="async_gpt_call") ... async def async_function(): ... return "async result"

Source code in src/honeyhive/tracer/instrumentation/decorators.py
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
def trace(
    event_type: Optional[str] = None,
    event_name: Optional[str] = None,
    **kwargs: Any,
) -> Union[Callable[[Callable[..., T]], Callable[..., T]], Callable[..., T]]:
    """Unified trace decorator that auto-detects sync/async functions.

    Automatically detects whether the decorated function is synchronous or
    asynchronous and applies the appropriate wrapper. This decorator can be
    used on both sync and async functions without needing separate decorators.

    Args:
        event_type: Type of event being traced (e.g., "model", "tool", "chain")
        event_name: Name of the event (defaults to function name)
        **kwargs: Additional tracing parameters (source, project, session_id, etc.)

    Returns:
        Decorated function with tracing capabilities

    Example:
        >>> @trace(event_type="model", event_name="gpt_call")
        ... def sync_function():
        ...     return "result"

        >>> @trace(event_type="model", event_name="async_gpt_call")
        ... async def async_function():
        ...     return "async result"
    """

    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        # Auto-detect if function is async
        is_async = inspect.iscoroutinefunction(func)

        # Filter out tracer argument for _create_tracing_params
        tracing_kwargs = {k: v for k, v in kwargs.items() if k != "tracer"}
        params = _create_tracing_params(
            event_type=event_type, event_name=event_name, **tracing_kwargs
        )
        return _create_wrapper(func, params, is_async=is_async, **kwargs)

    # Handle both @trace and @trace() usage patterns
    if event_type is not None and callable(event_type):
        # Used as @trace (without parentheses)
        func = event_type
        is_async = inspect.iscoroutinefunction(func)
        params = _create_tracing_params(event_type="tool")
        return _create_wrapper(func, params, is_async=is_async)

    # Used as @trace(...) (with parentheses)
    return decorator

atrace

atrace(
    event_type: Optional[str] = None,
    event_name: Optional[str] = None,
    **kwargs: Any
) -> Union[
    Callable[[Callable[..., Any]], Callable[..., Any]],
    Callable[..., Any],
]

Legacy async-specific trace decorator (deprecated).

Note

This decorator is maintained for backwards compatibility. Use the unified :func:trace decorator instead, which auto-detects sync/async functions.

Parameters:

Name Type Description Default
event_type Optional[str]

Type of event being traced (e.g., "model", "tool", "chain")

None
event_name Optional[str]

Name of the event (defaults to function name)

None
**kwargs Any

Additional tracing parameters (source, project, session_id, etc.)

{}

Returns:

Type Description
Union[Callable[[Callable[..., Any]], Callable[..., Any]], Callable[..., Any]]

Decorated async function with tracing capabilities

See Also

:func:trace: Unified decorator that handles both sync and async functions

Source code in src/honeyhive/tracer/instrumentation/decorators.py
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
def atrace(
    event_type: Optional[str] = None,
    event_name: Optional[str] = None,
    **kwargs: Any,
) -> Union[Callable[[Callable[..., Any]], Callable[..., Any]], Callable[..., Any]]:
    """Legacy async-specific trace decorator (deprecated).

    Note:
        This decorator is maintained for backwards compatibility.
        Use the unified :func:`trace` decorator instead, which auto-detects
        sync/async functions.

    Args:
        event_type: Type of event being traced (e.g., "model", "tool", "chain")
        event_name: Name of the event (defaults to function name)
        **kwargs: Additional tracing parameters (source, project, session_id, etc.)

    Returns:
        Decorated async function with tracing capabilities

    See Also:
        :func:`trace`: Unified decorator that handles both sync and async functions
    """

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        params = _create_tracing_params(
            event_type=event_type, event_name=event_name, **kwargs
        )
        return _create_wrapper(func, params, is_async=True, **kwargs)

    # Handle both @atrace and @atrace() usage patterns
    if event_type is not None and callable(event_type):
        # Used as @atrace (without parentheses)
        func = event_type
        params = _create_tracing_params(event_type="tool")
        return _create_wrapper(func, params, is_async=True)

    # Used as @atrace(...) (with parentheses)
    return decorator

trace_class

trace_class(cls: type) -> type

Class decorator to automatically trace all public methods.

Uses dynamic reflection to discover and wrap all public methods of a class. Automatically detects sync/async methods and applies appropriate tracing.

Parameters:

Name Type Description Default
cls type

The class to be decorated

required

Returns:

Type Description
type

The decorated class with all public methods traced

Example

@trace_class ... class MyService: ... def process_data(self, data): ... return data.upper() ... ... async def async_process(self, data): ... return await some_async_operation(data)

Source code in src/honeyhive/tracer/instrumentation/decorators.py
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
def trace_class(cls: type) -> type:
    """Class decorator to automatically trace all public methods.

    Uses dynamic reflection to discover and wrap all public methods of a class.
    Automatically detects sync/async methods and applies appropriate tracing.

    Args:
        cls: The class to be decorated

    Returns:
        The decorated class with all public methods traced

    Example:
        >>> @trace_class
        ... class MyService:
        ...     def process_data(self, data):
        ...         return data.upper()
        ...
        ...     async def async_process(self, data):
        ...         return await some_async_operation(data)
    """
    # Dynamically discover and wrap methods
    for attr_name in dir(cls):
        attr_value = getattr(cls, attr_name)

        # Dynamic method detection
        if (
            callable(attr_value)
            and not attr_name.startswith("_")
            and not isinstance(attr_value, (classmethod, staticmethod))
        ):
            # Determine if method is async
            is_async_method = inspect.iscoroutinefunction(attr_value)

            # Create tracing params for method
            params = _create_tracing_params(
                event_type="tool",
                event_name=f"{cls.__name__}.{attr_name}",
            )

            # Wrap method with appropriate wrapper
            wrapped_method = _create_wrapper(
                attr_value, params, is_async=is_async_method
            )
            setattr(cls, attr_name, wrapped_method)

    return cls